From 307d788da1954ddaeac8a55d5040a7ce7d1a213f Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 15 Jun 2026 16:53:14 +0800 Subject: [PATCH] feat(table-catalog): add production iceberg surfaces (#3473) --- rustfs/src/admin/handlers/table_catalog.rs | 396 ++++++++++++++++++++ rustfs/src/admin/route_policy.rs | 117 +++++- rustfs/src/admin/route_registration_test.rs | 124 ++++++ rustfs/src/table_catalog.rs | 344 ++++++++++++++++- scripts/table-catalog/README.md | 7 +- scripts/table-catalog/pyiceberg_smoke.py | 16 +- 6 files changed, 992 insertions(+), 12 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index dff5cdb56..02e928ac7 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -96,11 +96,17 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /{warehouse}/namespaces/{namespace}/tables", "POST /{warehouse}/namespaces/{namespace}/tables", "POST /{warehouse}/namespaces/{namespace}/register", + "GET /{warehouse}/namespaces/{namespace}/views", + "POST /{warehouse}/namespaces/{namespace}/views", "GET /{warehouse}/namespaces/{namespace}/tables/{table}", "HEAD /{warehouse}/namespaces/{namespace}/tables/{table}", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/credentials", "POST /{warehouse}/namespaces/{namespace}/tables/{table}", "DELETE /{warehouse}/namespaces/{namespace}/tables/{table}", + "GET /{warehouse}/namespaces/{namespace}/views/{view}", + "POST /{warehouse}/namespaces/{namespace}/views/{view}", + "DELETE /{warehouse}/namespaces/{namespace}/views/{view}", + "GET /{warehouse}/namespaces/{namespace}/tables/{table}/refs", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", "PUT /{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", @@ -109,6 +115,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "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/external", "GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", @@ -125,11 +132,17 @@ static DROP_NAMESPACE_HANDLER: RestDropNamespaceHandler = RestDropNamespaceHandl static LIST_TABLES_HANDLER: RestListTablesHandler = RestListTablesHandler {}; static CREATE_TABLE_HANDLER: RestCreateTableHandler = RestCreateTableHandler {}; static REGISTER_TABLE_HANDLER: RestRegisterTableHandler = RestRegisterTableHandler {}; +static LIST_VIEWS_HANDLER: RestListViewsHandler = RestListViewsHandler {}; +static CREATE_VIEW_HANDLER: RestCreateViewHandler = RestCreateViewHandler {}; static LOAD_TABLE_HANDLER: RestLoadTableHandler = RestLoadTableHandler {}; static TABLE_EXISTS_HANDLER: RestTableExistsHandler = RestTableExistsHandler {}; static LOAD_CREDENTIALS_HANDLER: RestLoadCredentialsHandler = RestLoadCredentialsHandler {}; static COMMIT_TABLE_HANDLER: RestCommitTableHandler = RestCommitTableHandler {}; static DROP_TABLE_HANDLER: RestDropTableHandler = RestDropTableHandler {}; +static LOAD_VIEW_HANDLER: RestLoadViewHandler = RestLoadViewHandler {}; +static REPLACE_VIEW_HANDLER: RestReplaceViewHandler = RestReplaceViewHandler {}; +static DROP_VIEW_HANDLER: RestDropViewHandler = RestDropViewHandler {}; +static LIST_TABLE_REFS_HANDLER: ListTableRefsHandler = ListTableRefsHandler {}; static GET_TABLE_METADATA_LOCATION_HANDLER: GetTableMetadataLocationHandler = GetTableMetadataLocationHandler {}; static UPDATE_TABLE_METADATA_LOCATION_HANDLER: UpdateTableMetadataLocationHandler = UpdateTableMetadataLocationHandler {}; static TABLE_METADATA_MAINTENANCE_HANDLER: RestTableMetadataMaintenanceHandler = RestTableMetadataMaintenanceHandler {}; @@ -138,6 +151,7 @@ static PUT_TABLE_MAINTENANCE_CONFIG_HANDLER: PutTableMaintenanceConfigHandler = static GET_TABLE_MAINTENANCE_JOB_HANDLER: GetTableMaintenanceJobHandler = GetTableMaintenanceJobHandler {}; static EXPORT_TABLE_CATALOG_HANDLER: ExportTableCatalogHandler = ExportTableCatalogHandler {}; static IMPORT_TABLE_CATALOG_HANDLER: ImportTableCatalogHandler = ImportTableCatalogHandler {}; +static EXTERNAL_CATALOG_BRIDGE_HANDLER: ExternalCatalogBridgeHandler = ExternalCatalogBridgeHandler {}; static GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER: GetTableCatalogDiagnosticsHandler = GetTableCatalogDiagnosticsHandler {}; static RECOVER_TABLE_CATALOG_HANDLER: RecoverTableCatalogHandler = RecoverTableCatalogHandler {}; static ROLLBACK_TABLE_CATALOG_HANDLER: RollbackTableCatalogHandler = RollbackTableCatalogHandler {}; @@ -259,6 +273,47 @@ struct RollbackTableRequest { idempotency_key: Option, } +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +struct UnsupportedCatalogFeatureResponse { + feature: &'static str, + status: &'static str, + reason: &'static str, + replacement: &'static str, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +struct TableRefsResponse { + table_bucket: String, + namespace: String, + table: String, + current_metadata_location: String, + current_snapshot_id: Option, + protected_ref_count: usize, + user_defined_ref_count: usize, + refs: BTreeMap, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +struct ExternalCatalogBridgeResponse { + table_bucket: String, + namespace: String, + table: String, + status: &'static str, + supported_import: &'static str, + unsupported_bridges: Vec, +} + +#[derive(Debug, Serialize)] +#[serde(rename_all = "kebab-case")] +struct ExternalCatalogBridgeCapability { + catalog: &'static str, + status: &'static str, + reason: &'static str, +} + #[derive(Debug, Serialize)] struct TableBucketResponse { #[serde(rename = "table-bucket")] @@ -547,6 +602,16 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/register").as_str(), AdminOperation(®ISTER_TABLE_HANDLER), )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/views").as_str(), + AdminOperation(&LIST_VIEWS_HANDLER), + )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/views").as_str(), + AdminOperation(&CREATE_VIEW_HANDLER), + )?; r.insert( Method::GET, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(), @@ -572,6 +637,26 @@ 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}}/views/{{view}}").as_str(), + AdminOperation(&LOAD_VIEW_HANDLER), + )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/views/{{view}}").as_str(), + AdminOperation(&REPLACE_VIEW_HANDLER), + )?; + r.insert( + Method::DELETE, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/views/{{view}}").as_str(), + AdminOperation(&DROP_VIEW_HANDLER), + )?; + r.insert( + Method::GET, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/refs").as_str(), + AdminOperation(&LIST_TABLE_REFS_HANDLER), + )?; r.insert( Method::GET, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/metadata-location").as_str(), @@ -612,6 +697,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix 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/external").as_str(), + AdminOperation(&EXTERNAL_CATALOG_BRIDGE_HANDLER), + )?; r.insert( Method::GET, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/catalog/diagnostics").as_str(), @@ -654,6 +744,21 @@ fn empty_response(status: StatusCode) -> S3Response<(StatusCode, Body)> { S3Response::new((status, Body::default())) } +fn unsupported_catalog_feature_response( + feature: &'static str, + replacement: &'static str, +) -> S3Result> { + build_json_response( + StatusCode::NOT_IMPLEMENTED, + &UnsupportedCatalogFeatureResponse { + feature, + status: "unsupported", + reason: "the catalog advertises this advanced Iceberg surface explicitly but does not implement it yet", + replacement, + }, + ) +} + fn duration_millis_u64(duration: StdDuration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) } @@ -842,6 +947,13 @@ fn table_name_from_params(params: &Params<'_, '_>) -> S3Result { Ok(table.to_string()) } +fn view_name_from_params(params: &Params<'_, '_>) -> S3Result { + let view = params.get("view").unwrap_or(""); + crate::table_catalog::IdentifierSegment::parse(view.to_string()) + .map_err(|err| s3_error!(InvalidRequest, "invalid view name: {}", err))?; + Ok(view.to_string()) +} + fn job_id_from_params(params: &Params<'_, '_>) -> S3Result { let job = params.get("job").unwrap_or(""); if job.is_empty() { @@ -2623,6 +2735,91 @@ where Ok(report) } +async fn table_refs_response( + store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, + 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")); + }; + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; + let current_snapshot_id = metadata.get("current-snapshot-id").and_then(serde_json::Value::as_i64); + let refs = metadata + .get("refs") + .and_then(serde_json::Value::as_object) + .cloned() + .unwrap_or_default() + .into_iter() + .collect::>(); + let protected_ref_count = refs + .values() + .filter(|reference| { + reference + .get("snapshot-id") + .and_then(serde_json::Value::as_i64) + .is_some_and(|snapshot_id| Some(snapshot_id) != current_snapshot_id) + }) + .count(); + let user_defined_ref_count = refs.keys().filter(|name| name.as_str() != "main").count(); + + Ok(TableRefsResponse { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.to_string(), + current_metadata_location: entry.metadata_location, + current_snapshot_id, + protected_ref_count, + user_defined_ref_count, + refs, + }) +} + +fn external_catalog_bridge_response( + bucket: &str, + namespace: &crate::table_catalog::Namespace, + table: &str, +) -> ExternalCatalogBridgeResponse { + ExternalCatalogBridgeResponse { + table_bucket: bucket.to_string(), + namespace: namespace.public_name(), + table: table.to_string(), + status: "metadata-import-supported", + supported_import: "register/import an existing Iceberg metadata location into the RustFS catalog; online external catalog synchronization is not implemented", + unsupported_bridges: vec![ + ExternalCatalogBridgeCapability { + catalog: "polaris", + status: "unsupported", + reason: "online REST catalog bridge and policy synchronization are not implemented", + }, + ExternalCatalogBridgeCapability { + catalog: "glue", + status: "unsupported", + reason: "Glue catalog synchronization is not implemented", + }, + ExternalCatalogBridgeCapability { + catalog: "dlf", + status: "unsupported", + reason: "DLF catalog synchronization is not implemented", + }, + ExternalCatalogBridgeCapability { + catalog: "hive-metastore", + status: "unsupported", + reason: "Hive Metastore synchronization is not implemented", + }, + ], + } +} + async fn catalog_import_response( store: &crate::table_catalog::ObjectTableCatalogStore, metadata_backend: &B, @@ -2883,6 +3080,32 @@ impl Operation for RestRegisterTableHandler { } } +pub struct RestListViewsHandler {} + +#[async_trait::async_trait] +impl Operation for RestListViewsHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; + unsupported_catalog_feature_response("iceberg-views", "use Iceberg tables; view catalog routes are reserved") + } +} + +pub struct RestCreateViewHandler {} + +#[async_trait::async_trait] +impl Operation for RestCreateViewHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::CreateTableAction).await?; + unsupported_catalog_feature_response("iceberg-views", "create a table or register an existing Iceberg table") + } +} + pub struct RestLoadTableHandler {} #[async_trait::async_trait] @@ -2967,6 +3190,65 @@ impl Operation for RestDropTableHandler { } } +pub struct RestLoadViewHandler {} + +#[async_trait::async_trait] +impl Operation for RestLoadViewHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let _view = view_name_from_params(¶ms)?; + let resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableMetadataAction).await?; + unsupported_catalog_feature_response("iceberg-views", "use Iceberg tables; view load is reserved") + } +} + +pub struct RestReplaceViewHandler {} + +#[async_trait::async_trait] +impl Operation for RestReplaceViewHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let _view = view_name_from_params(¶ms)?; + let resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::CommitTableAction).await?; + unsupported_catalog_feature_response("iceberg-views", "commit table metadata updates; view replacement is reserved") + } +} + +pub struct RestDropViewHandler {} + +#[async_trait::async_trait] +impl Operation for RestDropViewHandler { + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + let warehouse = warehouse_from_params(¶ms)?; + let namespace = namespace_from_params(¶ms)?; + let _view = view_name_from_params(¶ms)?; + let resource = TableCatalogResource::namespace(&warehouse, &namespace); + authorize_table_catalog_resource_request(&req, &resource, AdminAction::DeleteTableAction).await?; + unsupported_catalog_feature_response("iceberg-views", "drop table resources; view deletion is reserved") + } +} + +pub struct ListTableRefsHandler {} + +#[async_trait::async_trait] +impl Operation for ListTableRefsHandler { + 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 metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let response = table_refs_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?; + build_json_response(StatusCode::OK, &response) + } +} + pub struct GetTableMetadataLocationHandler {} #[async_trait::async_trait] @@ -3126,6 +3408,20 @@ impl Operation for ImportTableCatalogHandler { } } +pub struct ExternalCatalogBridgeHandler {} + +#[async_trait::async_trait] +impl Operation for ExternalCatalogBridgeHandler { + 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?; + build_json_response(StatusCode::OK, &external_catalog_bridge_response(&warehouse, &namespace, &table)) + } +} + pub struct GetTableCatalogDiagnosticsHandler {} #[async_trait::async_trait] @@ -3246,6 +3542,8 @@ mod tests { .endpoints .contains(&"POST /{warehouse}/namespaces/{namespace}/tables") ); + assert!(response.endpoints.contains(&"GET /{warehouse}/namespaces/{namespace}/views")); + assert!(response.endpoints.contains(&"POST /{warehouse}/namespaces/{namespace}/views")); assert!( response .endpoints @@ -3266,6 +3564,21 @@ mod tests { .endpoints .contains(&"GET /{warehouse}/namespaces/{namespace}/tables/{table}/credentials") ); + assert!( + response + .endpoints + .contains(&"GET /{warehouse}/namespaces/{namespace}/views/{view}") + ); + assert!( + response + .endpoints + .contains(&"GET /{warehouse}/namespaces/{namespace}/tables/{table}/refs") + ); + assert!( + response + .endpoints + .contains(&"GET /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external") + ); assert!( response .endpoints @@ -3306,11 +3619,17 @@ mod tests { ("RestListTablesHandler", "AdminAction::GetTableAction"), ("RestCreateTableHandler", "AdminAction::CreateTableAction"), ("RestRegisterTableHandler", "AdminAction::RegisterTableAction"), + ("RestListViewsHandler", "AdminAction::GetTableMetadataAction"), + ("RestCreateViewHandler", "AdminAction::CreateTableAction"), ("RestLoadTableHandler", "AdminAction::GetTableMetadataAction"), ("RestTableExistsHandler", "AdminAction::GetTableAction"), ("RestLoadCredentialsHandler", "AdminAction::GetTableCredentialsAction"), ("RestCommitTableHandler", "AdminAction::CommitTableAction"), ("RestDropTableHandler", "AdminAction::DeleteTableAction"), + ("RestLoadViewHandler", "AdminAction::GetTableMetadataAction"), + ("RestReplaceViewHandler", "AdminAction::CommitTableAction"), + ("RestDropViewHandler", "AdminAction::DeleteTableAction"), + ("ListTableRefsHandler", "AdminAction::GetTableMetadataAction"), ("GetTableMetadataLocationHandler", "AdminAction::GetTableMetadataLocationAction"), ("UpdateTableMetadataLocationHandler", "AdminAction::SetTableMetadataLocationAction"), ("RestTableMetadataMaintenanceHandler", "AdminAction::RunTableMaintenanceAction"), @@ -3319,6 +3638,7 @@ mod tests { ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), + ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"), ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), ("RecoverTableCatalogHandler", "AdminAction::CommitTableAction"), ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), @@ -3344,6 +3664,7 @@ mod tests { ("RestLoadCredentialsHandler", "AdminAction::GetTableCredentialsAction"), ("RestCommitTableHandler", "AdminAction::CommitTableAction"), ("RestDropTableHandler", "AdminAction::DeleteTableAction"), + ("ListTableRefsHandler", "AdminAction::GetTableMetadataAction"), ("GetTableMetadataLocationHandler", "AdminAction::GetTableMetadataLocationAction"), ("UpdateTableMetadataLocationHandler", "AdminAction::SetTableMetadataLocationAction"), ("RestTableMetadataMaintenanceHandler", "AdminAction::RunTableMaintenanceAction"), @@ -3352,6 +3673,7 @@ mod tests { ("GetTableMaintenanceJobHandler", "AdminAction::GetTableLifecycleAction"), ("ExportTableCatalogHandler", "AdminAction::GetTableMetadataAction"), ("ImportTableCatalogHandler", "AdminAction::RegisterTableAction"), + ("ExternalCatalogBridgeHandler", "AdminAction::GetTableMetadataAction"), ("GetTableCatalogDiagnosticsHandler", "AdminAction::GetTableMetadataAction"), ("RecoverTableCatalogHandler", "AdminAction::CommitTableAction"), ("RollbackTableCatalogHandler", "AdminAction::CommitTableAction"), @@ -3412,11 +3734,17 @@ mod tests { let _: &RestListTablesHandler = &LIST_TABLES_HANDLER; let _: &RestCreateTableHandler = &CREATE_TABLE_HANDLER; let _: &RestRegisterTableHandler = ®ISTER_TABLE_HANDLER; + let _: &RestListViewsHandler = &LIST_VIEWS_HANDLER; + let _: &RestCreateViewHandler = &CREATE_VIEW_HANDLER; let _: &RestLoadTableHandler = &LOAD_TABLE_HANDLER; let _: &RestTableExistsHandler = &TABLE_EXISTS_HANDLER; let _: &RestLoadCredentialsHandler = &LOAD_CREDENTIALS_HANDLER; let _: &RestCommitTableHandler = &COMMIT_TABLE_HANDLER; let _: &RestDropTableHandler = &DROP_TABLE_HANDLER; + let _: &RestLoadViewHandler = &LOAD_VIEW_HANDLER; + let _: &RestReplaceViewHandler = &REPLACE_VIEW_HANDLER; + let _: &RestDropViewHandler = &DROP_VIEW_HANDLER; + let _: &ListTableRefsHandler = &LIST_TABLE_REFS_HANDLER; let _: &GetTableMetadataLocationHandler = &GET_TABLE_METADATA_LOCATION_HANDLER; let _: &UpdateTableMetadataLocationHandler = &UPDATE_TABLE_METADATA_LOCATION_HANDLER; let _: &RestTableMetadataMaintenanceHandler = &TABLE_METADATA_MAINTENANCE_HANDLER; @@ -3425,7 +3753,9 @@ mod tests { let _: &GetTableMaintenanceJobHandler = &GET_TABLE_MAINTENANCE_JOB_HANDLER; let _: &ExportTableCatalogHandler = &EXPORT_TABLE_CATALOG_HANDLER; let _: &ImportTableCatalogHandler = &IMPORT_TABLE_CATALOG_HANDLER; + let _: &ExternalCatalogBridgeHandler = &EXTERNAL_CATALOG_BRIDGE_HANDLER; let _: &GetTableCatalogDiagnosticsHandler = &GET_TABLE_CATALOG_DIAGNOSTICS_HANDLER; + let _: &RecoverTableCatalogHandler = &RECOVER_TABLE_CATALOG_HANDLER; let _: &RollbackTableCatalogHandler = &ROLLBACK_TABLE_CATALOG_HANDLER; assert_operation::(); @@ -3438,11 +3768,17 @@ 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::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -3451,7 +3787,9 @@ mod tests { assert_operation::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); } @@ -4407,6 +4745,7 @@ mod tests { retain_recent_metadata_files: 2, delete_enabled: true, background_enabled: false, + ..Default::default() }, ) .await @@ -4424,6 +4763,7 @@ mod tests { retain_recent_metadata_files: 2, delete_enabled: true, background_enabled: true, + ..Default::default() }, ) .await @@ -4831,6 +5171,62 @@ mod tests { assert!(result.is_err()); } + #[tokio::test] + async fn table_refs_response_reports_current_and_user_defined_refs() { + 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"); + let current = crate::table_catalog::default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + seed_object_table_for_metadata_maintenance(&store, &backend, bucket, &namespace, &table, current.clone()).await; + backend + .put_json_with_mod_time( + bucket, + ¤t, + serde_json::json!({ + "current-snapshot-id": 10, + "refs": { + "main": { + "snapshot-id": 10, + "type": "branch" + }, + "audit": { + "snapshot-id": 9, + "type": "tag" + } + } + }), + Some(OffsetDateTime::UNIX_EPOCH), + ) + .await; + + let response = table_refs_response(&store, &backend, bucket, &namespace, "events") + .await + .expect("refs response should load"); + + assert_eq!(response.current_snapshot_id, Some(10)); + assert_eq!(response.protected_ref_count, 1); + assert_eq!(response.user_defined_ref_count, 1); + assert!(response.refs.contains_key("main")); + assert!(response.refs.contains_key("audit")); + } + + #[test] + fn external_catalog_bridge_response_lists_unsupported_bridges() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let response = external_catalog_bridge_response("warehouse", &namespace, "events"); + + assert_eq!(response.status, "metadata-import-supported"); + assert_eq!(response.unsupported_bridges.len(), 4); + assert!( + response + .unsupported_bridges + .iter() + .any(|bridge| bridge.catalog == "polaris" && bridge.status == "unsupported") + ); + } + #[test] fn commit_requirements_reject_mismatched_table_uuid() { let metadata = serde_json::json!({ diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 81344b6b1..840fd8904 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -670,6 +670,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ REGISTER_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views", + CREATE_TABLE, + RouteRiskLevel::High, + ), admin( HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -700,6 +712,30 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ DELETE_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + COMMIT_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Delete, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + DELETE_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/refs", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", @@ -748,6 +784,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ REGISTER_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", @@ -827,6 +869,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ REGISTER_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views", + CREATE_TABLE, + RouteRiskLevel::High, + ), admin( HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -857,6 +911,30 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ DELETE_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + COMMIT_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Delete, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + DELETE_TABLE, + RouteRiskLevel::High, + ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/refs", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/metadata-location", @@ -905,6 +983,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ REGISTER_TABLE, RouteRiskLevel::High, ), + admin( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external", + GET_TABLE_METADATA, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", @@ -1106,7 +1190,7 @@ 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(), 54); + assert_eq!(table_specs.count(), 68); 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); @@ -1115,6 +1199,12 @@ mod tests { assert_action(HttpMethod::Head, "/_iceberg/v1/{warehouse}/namespaces/{namespace}", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE); assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE); + assert_action( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views", + GET_TABLE_METADATA, + ); + assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views", CREATE_TABLE); assert_action( HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -1125,6 +1215,26 @@ mod tests { "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/credentials", GET_TABLE_CREDENTIALS, ); + assert_action( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + GET_TABLE_METADATA, + ); + assert_action( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + COMMIT_TABLE, + ); + assert_action( + HttpMethod::Delete, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/views/{view}", + DELETE_TABLE, + ); + assert_action( + HttpMethod::Get, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/refs", + GET_TABLE_METADATA, + ); assert_action( HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -1195,6 +1305,11 @@ mod tests { "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/import", REGISTER_TABLE, ); + assert_action( + HttpMethod::Get, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/external", + GET_TABLE_METADATA, + ); assert_action( HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 1a3ed20df..db9aa3691 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -312,6 +312,16 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/register", "/analytics/namespaces/sales/register", ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/views", + "/analytics/namespaces/sales/views", + ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/views", + "/analytics/namespaces/sales/views", + ), table_route_sample( Method::GET, "/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -337,6 +347,26 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}", "/analytics/namespaces/sales/tables/orders", ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + table_route_sample( + Method::DELETE, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/refs", + "/analytics/namespaces/sales/tables/orders/refs", + ), table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", @@ -377,6 +407,11 @@ fn expected_admin_route_matrix() -> Vec { "/{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/external", + "/analytics/namespaces/sales/tables/orders/catalog/external", + ), table_route_sample( Method::GET, "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", @@ -415,6 +450,16 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/register", "/analytics/namespaces/sales/register", ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/views", + "/analytics/namespaces/sales/views", + ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/views", + "/analytics/namespaces/sales/views", + ), compat_table_route_sample( Method::GET, "/{warehouse}/namespaces/{namespace}/tables/{table}", @@ -440,6 +485,26 @@ fn expected_admin_route_matrix() -> Vec { "/{warehouse}/namespaces/{namespace}/tables/{table}", "/analytics/namespaces/sales/tables/orders", ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + compat_table_route_sample( + Method::DELETE, + "/{warehouse}/namespaces/{namespace}/views/{view}", + "/analytics/namespaces/sales/views/recent_orders", + ), + compat_table_route_sample( + Method::GET, + "/{warehouse}/namespaces/{namespace}/tables/{table}/refs", + "/analytics/namespaces/sales/tables/orders/refs", + ), compat_table_route_sample( Method::POST, "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata", @@ -480,6 +545,11 @@ fn expected_admin_route_matrix() -> Vec { "/{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/external", + "/analytics/namespaces/sales/tables/orders/catalog/external", + ), compat_table_route_sample( Method::GET, "/{warehouse}/namespaces/{namespace}/tables/{table}/catalog/diagnostics", @@ -614,6 +684,8 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables")); assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables")); assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/register")); + assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/views")); + assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/views")); assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); assert_route( &router, @@ -622,6 +694,26 @@ fn test_register_routes_cover_representative_admin_paths() { ); assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); assert_route(&router, Method::DELETE, &table_catalog_path("/analytics/namespaces/sales/tables/orders")); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::POST, + &table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::DELETE, + &table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::GET, + &table_catalog_path("/analytics/namespaces/sales/tables/orders/refs"), + ); assert_route( &router, Method::POST, @@ -662,6 +754,11 @@ fn test_register_routes_cover_representative_admin_paths() { 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/external"), + ); assert_route( &router, Method::GET, @@ -687,6 +784,8 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/tables")); assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/tables")); assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/register")); + assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/views")); + assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/views")); assert_route( &router, Method::GET, @@ -707,6 +806,26 @@ fn test_register_routes_cover_representative_admin_paths() { Method::DELETE, &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders"), ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::POST, + &compat_table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::DELETE, + &compat_table_catalog_path("/analytics/namespaces/sales/views/recent_orders"), + ); + assert_route( + &router, + Method::GET, + &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/refs"), + ); assert_route( &router, Method::POST, @@ -747,6 +866,11 @@ fn test_register_routes_cover_representative_admin_paths() { 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/external"), + ); assert_route( &router, Method::GET, diff --git a/rustfs/src/table_catalog.rs b/rustfs/src/table_catalog.rs index aaceba812..b614980e3 100644 --- a/rustfs/src/table_catalog.rs +++ b/rustfs/src/table_catalog.rs @@ -83,6 +83,7 @@ const MAINTENANCE_JOB_ALIAS_LATEST: &str = "latest"; const MAINTENANCE_JOB_ALIAS_CURRENT: &str = "current"; const TABLE_CATALOG_LIST_MAX_KEYS: i32 = 1000; const TABLE_METADATA_CLEANUP_SAFETY_WINDOW_SECONDS: i64 = 15 * 60; +const TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS: u64 = 24 * 60 * 60; const TABLE_COMMIT_SLOW_LOG_THRESHOLD: StdDuration = StdDuration::from_secs(2); const ICEBERG_MAIN_REF: &str = "main"; const ICEBERG_MIN_SNAPSHOTS_TO_KEEP_PROPERTY: &str = "history.expire.min-snapshots-to-keep"; @@ -273,6 +274,16 @@ pub(crate) struct TableMaintenanceConfig { pub delete_enabled: bool, #[serde(rename = "background-enabled")] pub background_enabled: bool, + #[serde(default, rename = "max-retry-attempts")] + pub max_retry_attempts: u16, + #[serde(default, rename = "retry-initial-backoff-seconds")] + pub retry_initial_backoff_seconds: u64, + #[serde(default, rename = "retry-max-backoff-seconds")] + pub retry_max_backoff_seconds: u64, + #[serde(default, rename = "quarantine-enabled")] + pub quarantine_enabled: bool, + #[serde(default, rename = "quarantine-retention-seconds")] + pub quarantine_retention_seconds: u64, } impl Default for TableMaintenanceConfig { @@ -282,6 +293,11 @@ impl Default for TableMaintenanceConfig { retain_recent_metadata_files: 0, delete_enabled: false, background_enabled: false, + max_retry_attempts: 0, + retry_initial_backoff_seconds: 5, + retry_max_backoff_seconds: 300, + quarantine_enabled: false, + quarantine_retention_seconds: 0, } } } @@ -322,6 +338,16 @@ pub(crate) struct TableMetadataMaintenanceJob { #[serde(default)] pub lease_id: String, #[serde(default)] + pub attempt: u16, + #[serde(default, rename = "max-retry-attempts")] + pub max_retry_attempts: u16, + #[serde(default, rename = "next-retry-after")] + pub next_retry_after: Option, + #[serde(default, rename = "quarantine-enabled")] + pub quarantine_enabled: bool, + #[serde(default, rename = "quarantine-retention-seconds")] + pub quarantine_retention_seconds: u64, + #[serde(default)] pub heartbeat_at: Option, #[serde(default)] pub started_at: Option, @@ -342,6 +368,8 @@ pub(crate) struct TableMetadataMaintenanceJob { pub deletable_metadata_file_count: usize, #[serde(default)] pub deleted_metadata_file_count: usize, + #[serde(default)] + pub quarantined_object_count: usize, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -355,12 +383,44 @@ pub(crate) struct TableMetadataMaintenanceReport { pub object_reports: Vec, #[serde(default)] pub referenced_object_reports: Vec, + #[serde(default, rename = "reachability-graph")] + pub reachability_graph: TableMaintenanceReachabilityGraphReport, #[serde(default, rename = "snapshot-expiration")] pub snapshot_expiration: Option, #[serde(default)] pub compaction: Option, } +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub(crate) struct TableMaintenanceReachabilityGraphReport { + pub status: TableMaintenanceReachabilityGraphStatus, + pub metadata_file_count: usize, + pub manifest_list_count: usize, + pub manifest_file_count: usize, + pub data_file_count: usize, + pub delete_file_count: usize, + pub manual_review_count: usize, + pub reasons: Vec, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableMaintenanceReachabilityGraphStatus { + Complete, + #[default] + ManualReviewRequired, +} + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub(crate) enum TableMaintenanceReachabilityGraphReason { + MetadataJsonParsed, + ManifestListAvroReferenced, + ManifestAvroReaderUnavailable, + DataFileCleanupDeferred, + DeleteFileCleanupDeferred, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct TableSnapshotExpirationConfig { @@ -518,6 +578,8 @@ pub(crate) enum TableMetadataMaintenanceReason { DeletedByMaintenance, ManifestList, UnsupportedManifestAvro, + QuarantineEnabled, + RetryScheduled, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -1372,9 +1434,13 @@ where 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()) + let config = self + .read_entry::(self.catalog_bucket(), &config_path) + .await? + .map(|(config, _)| config) + .unwrap_or_default(); + validate_table_maintenance_config(&config)?; + Ok(config) } pub(crate) async fn put_table_bucket_maintenance_config( @@ -1415,6 +1481,7 @@ where .read_entry::(self.catalog_bucket(), &table_config_path) .await? { + validate_table_maintenance_config(&config)?; return Ok(TableMaintenanceEffectiveConfig { config, source: TableMaintenanceConfigSource::TableOverride, @@ -1426,6 +1493,7 @@ where .read_entry::(self.catalog_bucket(), &bucket_config_path) .await? { + validate_table_maintenance_config(&config)?; return Ok(TableMaintenanceEffectiveConfig { config, source: TableMaintenanceConfigSource::TableBucketDefault, @@ -1893,6 +1961,8 @@ where let retained_metadata_locations = retained.into_iter().collect::>(); let object_reports = metadata_maintenance_object_reports(maintenance_reasons); let referenced_object_reports = metadata_maintenance_referenced_object_reports(¤t_metadata); + let reachability_graph = + metadata_maintenance_reachability_graph_report(planned_metadata_file_count, &referenced_object_reports); Ok(TableMetadataMaintenanceReport { job: TableMetadataMaintenanceJob { @@ -1907,6 +1977,11 @@ where config_source: TableMaintenanceConfigSource::Default, worker_id: None, lease_id: String::new(), + attempt: 0, + max_retry_attempts: 0, + next_retry_after: None, + quarantine_enabled: false, + quarantine_retention_seconds: 0, heartbeat_at: None, started_at: None, finished_at: None, @@ -1921,6 +1996,7 @@ where cleanup_candidate_count: cleanup_candidate_locations.len(), deletable_metadata_file_count: deletable_metadata_locations.len(), deleted_metadata_file_count: 0, + quarantined_object_count: 0, }, current_metadata_location, retained_metadata_locations, @@ -1928,6 +2004,7 @@ where deletable_metadata_locations, object_reports, referenced_object_reports, + reachability_graph, snapshot_expiration: None, compaction: None, }) @@ -2003,6 +2080,11 @@ where report.job.config_source = effective.source; report.job.worker_id = worker_id; report.job.lease_id = Uuid::new_v4().to_string(); + report.job.attempt = 1; + report.job.max_retry_attempts = effective.config.max_retry_attempts; + report.job.next_retry_after = None; + report.job.quarantine_enabled = effective.config.quarantine_enabled; + report.job.quarantine_retention_seconds = effective.config.quarantine_retention_seconds; report.job.heartbeat_at = Some(started_at.clone()); report.job.started_at = Some(started_at); report.job.finished_at = None; @@ -2012,6 +2094,7 @@ where let mut failed = report; failed.job.status = TableMetadataMaintenanceJobStatus::Failed; failed.job.failure_reason = Some("metadata delete is disabled by maintenance config".to_string()); + apply_maintenance_retry_after(&mut failed.job, &effective.config, OffsetDateTime::now_utc()); failed.job.finished_at = Some(maintenance_timestamp(OffsetDateTime::now_utc())); self.put_table_metadata_maintenance_report(&failed).await?; return Ok(failed); @@ -2028,6 +2111,7 @@ where let mut failed = running_report; failed.job.status = TableMetadataMaintenanceJobStatus::Failed; failed.job.failure_reason = Some(err.to_string()); + apply_maintenance_retry_after(&mut failed.job, &effective.config, OffsetDateTime::now_utc()); failed.job.finished_at = Some(maintenance_timestamp(OffsetDateTime::now_utc())); self.put_table_metadata_maintenance_report(&failed).await?; return Err(err); @@ -2159,6 +2243,7 @@ where deletable_metadata_locations: cleanup_candidate_locations, object_reports, referenced_object_reports, + reachability_graph: report.reachability_graph, snapshot_expiration: report.snapshot_expiration, compaction: report.compaction, }) @@ -2837,6 +2922,54 @@ fn metadata_maintenance_referenced_object_reports( .collect() } +fn metadata_maintenance_reachability_graph_report( + metadata_file_count: usize, + referenced_object_reports: &[TableMetadataMaintenanceReferencedObjectReport], +) -> TableMaintenanceReachabilityGraphReport { + let manifest_list_count = referenced_object_reports + .iter() + .filter(|report| report.object_kind == TableMetadataMaintenanceObjectKind::ManifestList) + .count(); + let manifest_file_count = referenced_object_reports + .iter() + .filter(|report| report.object_kind == TableMetadataMaintenanceObjectKind::ManifestFile) + .count(); + let data_file_count = referenced_object_reports + .iter() + .filter(|report| report.object_kind == TableMetadataMaintenanceObjectKind::DataFile) + .count(); + let delete_file_count = referenced_object_reports + .iter() + .filter(|report| report.object_kind == TableMetadataMaintenanceObjectKind::DeleteFile) + .count(); + let manual_review_count = referenced_object_reports + .iter() + .filter(|report| report.state == TableMetadataMaintenanceObjectState::ManualReviewRequired) + .count(); + let mut reasons = BTreeSet::from([TableMaintenanceReachabilityGraphReason::MetadataJsonParsed]); + if manifest_list_count > 0 { + reasons.insert(TableMaintenanceReachabilityGraphReason::ManifestListAvroReferenced); + reasons.insert(TableMaintenanceReachabilityGraphReason::ManifestAvroReaderUnavailable); + } + reasons.insert(TableMaintenanceReachabilityGraphReason::DataFileCleanupDeferred); + reasons.insert(TableMaintenanceReachabilityGraphReason::DeleteFileCleanupDeferred); + + TableMaintenanceReachabilityGraphReport { + status: if manual_review_count == 0 { + TableMaintenanceReachabilityGraphStatus::Complete + } else { + TableMaintenanceReachabilityGraphStatus::ManualReviewRequired + }, + metadata_file_count, + manifest_list_count, + manifest_file_count, + data_file_count, + delete_file_count, + manual_review_count, + reasons: reasons.into_iter().collect(), + } +} + fn mark_deleted_metadata_object_reports( object_reports: &mut [TableMetadataMaintenanceObjectReport], deleted_locations: &BTreeSet, @@ -3338,9 +3471,52 @@ fn validate_table_maintenance_config(config: &TableMaintenanceConfig) -> TableCa "background table maintenance is not supported".to_string(), )); } + if config.max_retry_attempts > 10 { + return Err(TableCatalogStoreError::Invalid("max-retry-attempts cannot exceed 10".to_string())); + } + if config.max_retry_attempts > 0 && config.retry_initial_backoff_seconds == 0 { + return Err(TableCatalogStoreError::Invalid( + "retry-initial-backoff-seconds must be greater than zero when retry is enabled".to_string(), + )); + } + if config.max_retry_attempts > 0 && config.retry_initial_backoff_seconds > TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS { + return Err(TableCatalogStoreError::Invalid(format!( + "retry-initial-backoff-seconds cannot exceed {TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS}" + ))); + } + if config.max_retry_attempts > 0 && config.retry_max_backoff_seconds > TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS { + return Err(TableCatalogStoreError::Invalid(format!( + "retry-max-backoff-seconds cannot exceed {TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS}" + ))); + } + if config.max_retry_attempts > 0 && config.retry_max_backoff_seconds < config.retry_initial_backoff_seconds { + return Err(TableCatalogStoreError::Invalid( + "retry-max-backoff-seconds must be greater than or equal to retry-initial-backoff-seconds".to_string(), + )); + } + if config.quarantine_enabled && config.quarantine_retention_seconds == 0 { + return Err(TableCatalogStoreError::Invalid( + "quarantine-retention-seconds must be greater than zero when quarantine is enabled".to_string(), + )); + } Ok(()) } +fn apply_maintenance_retry_after(job: &mut TableMetadataMaintenanceJob, config: &TableMaintenanceConfig, now: OffsetDateTime) { + if config.max_retry_attempts == 0 || job.attempt >= config.max_retry_attempts { + job.next_retry_after = None; + return; + } + let attempt_index = u32::from(job.attempt.saturating_sub(1)); + let multiplier = 1_u64.checked_shl(attempt_index).unwrap_or(u64::MAX); + let delay_seconds = config + .retry_initial_backoff_seconds + .saturating_mul(multiplier) + .min(config.retry_max_backoff_seconds); + let delay_seconds = i64::try_from(delay_seconds).unwrap_or(i64::MAX); + job.next_retry_after = Some(maintenance_timestamp(now.saturating_add(Duration::seconds(delay_seconds)))); +} + fn validate_table_snapshot_expiration_config(config: &TableSnapshotExpirationConfig) -> TableCatalogStoreResult<()> { if config.min_snapshots_to_keep == 0 { return Err(TableCatalogStoreError::Invalid( @@ -4938,6 +5114,7 @@ mod tests { retain_recent_metadata_files: 7, delete_enabled: true, background_enabled: false, + ..Default::default() }, ) .await @@ -4998,6 +5175,7 @@ mod tests { retain_recent_metadata_files: 1, delete_enabled: false, background_enabled: false, + ..Default::default() }, ) .await @@ -5024,6 +5202,7 @@ mod tests { retain_recent_metadata_files: 3, delete_enabled: true, background_enabled: false, + ..Default::default() }, ) .await @@ -5049,6 +5228,7 @@ mod tests { retain_recent_metadata_files: 1, delete_enabled: false, background_enabled: false, + ..Default::default() }, ) .await @@ -5084,6 +5264,7 @@ mod tests { retain_recent_metadata_files: 1, delete_enabled: false, background_enabled: true, + ..Default::default() }, ) .await @@ -5100,6 +5281,7 @@ mod tests { retain_recent_metadata_files: 1, delete_enabled: false, background_enabled: true, + ..Default::default() }, ) .await @@ -5107,6 +5289,136 @@ mod tests { assert_matches!(table_err, TableCatalogStoreError::Invalid(_)); } + #[tokio::test] + async fn maintenance_config_accepts_retry_and_quarantine_policy() { + 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 config = store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 2, + delete_enabled: false, + background_enabled: false, + max_retry_attempts: 3, + retry_initial_backoff_seconds: 10, + retry_max_backoff_seconds: 60, + quarantine_enabled: true, + quarantine_retention_seconds: 86_400, + }, + ) + .await + .expect("retry and quarantine maintenance config should persist"); + + assert_eq!(config.max_retry_attempts, 3); + assert_eq!(config.retry_initial_backoff_seconds, 10); + assert_eq!(config.retry_max_backoff_seconds, 60); + assert!(config.quarantine_enabled); + assert_eq!(config.quarantine_retention_seconds, 86_400); + } + + #[tokio::test] + async fn maintenance_config_rejects_retry_backoff_above_limit() { + 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 initial_err = store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 1, + delete_enabled: false, + background_enabled: false, + max_retry_attempts: 1, + retry_initial_backoff_seconds: TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS.saturating_add(1), + retry_max_backoff_seconds: TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS.saturating_add(1), + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert_matches!(initial_err, TableCatalogStoreError::Invalid(_)); + + let max_err = store + .put_table_maintenance_config( + bucket, + "sales", + "orders", + TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 1, + delete_enabled: false, + background_enabled: false, + max_retry_attempts: 1, + retry_initial_backoff_seconds: TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS, + retry_max_backoff_seconds: TABLE_MAINTENANCE_RETRY_BACKOFF_MAX_SECONDS.saturating_add(1), + ..Default::default() + }, + ) + .await + .unwrap_err(); + assert_matches!(max_err, TableCatalogStoreError::Invalid(_)); + } + + #[tokio::test] + async fn maintenance_run_rejects_existing_invalid_retry_config_before_scheduling() { + 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 config_path = store + .paths + .table_maintenance_config_path(bucket, &namespace, &table, "table-id"); + store + .write_entry( + store.catalog_bucket(), + &config_path, + &TableMaintenanceConfig { + version: TABLE_MAINTENANCE_CONFIG_VERSION, + retain_recent_metadata_files: 1, + delete_enabled: false, + background_enabled: false, + max_retry_attempts: 1, + retry_initial_backoff_seconds: u64::MAX, + retry_max_backoff_seconds: u64::MAX, + ..Default::default() + }, + TableCatalogPutPrecondition::Any, + ) + .await + .expect("invalid legacy maintenance config should be seeded"); + + let err = store + .run_table_metadata_maintenance(bucket, "sales", "orders", true, Some("worker-a".to_string())) + .await + .unwrap_err(); + + assert_matches!(err, TableCatalogStoreError::Invalid(_)); + } + #[tokio::test] async fn maintenance_run_persists_latest_job_alias_with_worker_and_lease_context() { let backend = TestCatalogObjectBackend::default(); @@ -5126,6 +5438,7 @@ mod tests { retain_recent_metadata_files: 0, delete_enabled: false, background_enabled: false, + ..Default::default() }, ) .await @@ -5144,6 +5457,9 @@ mod tests { assert_eq!(report.job.config_source, TableMaintenanceConfigSource::TableBucketDefault); assert_eq!(report.job.worker_id.as_deref(), Some("worker-a")); assert!(!report.job.lease_id.is_empty()); + assert_eq!(report.job.attempt, 1); + assert_eq!(report.job.max_retry_attempts, 0); + assert!(report.job.next_retry_after.is_none()); assert!(report.job.heartbeat_at.is_some()); assert!(report.job.started_at.is_some()); assert!(report.job.finished_at.is_some()); @@ -5184,6 +5500,11 @@ mod tests { retain_recent_metadata_files: 0, delete_enabled: false, background_enabled: false, + max_retry_attempts: 2, + retry_initial_backoff_seconds: 10, + retry_max_backoff_seconds: 30, + quarantine_enabled: true, + quarantine_retention_seconds: 86_400, }, ) .await @@ -5201,6 +5522,10 @@ mod tests { assert_eq!(report.job.operation, TableMetadataMaintenanceOperation::Delete); assert_eq!(report.job.status, TableMetadataMaintenanceJobStatus::Failed); assert_eq!(report.job.config_source, TableMaintenanceConfigSource::TableOverride); + assert_eq!(report.job.max_retry_attempts, 2); + assert!(report.job.next_retry_after.is_some()); + assert!(report.job.quarantine_enabled); + assert_eq!(report.job.quarantine_retention_seconds, 86_400); assert!( report .job @@ -5285,6 +5610,19 @@ mod tests { TableMetadataMaintenanceReason::UnsupportedManifestAvro, ] ); + assert_eq!( + report.reachability_graph.status, + TableMaintenanceReachabilityGraphStatus::ManualReviewRequired + ); + assert_eq!(report.reachability_graph.metadata_file_count, 2); + assert_eq!(report.reachability_graph.manifest_list_count, 1); + assert_eq!(report.reachability_graph.manual_review_count, 1); + assert!( + report + .reachability_graph + .reasons + .contains(&TableMaintenanceReachabilityGraphReason::ManifestAvroReaderUnavailable) + ); } #[tokio::test] diff --git a/scripts/table-catalog/README.md b/scripts/table-catalog/README.md index 0c7ce735e..b5c1d7068 100644 --- a/scripts/table-catalog/README.md +++ b/scripts/table-catalog/README.md @@ -149,12 +149,13 @@ Unsupported behavior is documented instead of hidden behind internal errors. The current unsupported inventory is: - credential vending: automated after table bootstrap with exact-prefix validation and a data-plane scope probe; full no-long-term-data-credential bootstrap is not claimed -- background maintenance worker: unsupported -- manifest/data reachability cleanup: unsupported +- background maintenance worker: unsupported; maintenance job reports expose retry/quarantine policy fields for future workers +- manifest/data reachability cleanup: fail-closed reachability graph reporting only; metadata cleanup must not delete manifest, data, or delete files - snapshot expiration dry-run planning and manual catalog commit: supported through metadata maintenance reports - automatic maintenance scheduling: unsupported - compaction rewrite: unsupported; planning reports fail closed until manifest Avro reading and rewrite support are implemented -- Iceberg views: unsupported +- Iceberg views: stable unsupported routes are registered and return explicit unsupported JSON +- external catalog bridges: metadata import/register is supported, but Polaris/Glue/DLF/Hive synchronization is unsupported - multi-table transactions: not a short-term production claim ## Credential Boundary diff --git a/scripts/table-catalog/pyiceberg_smoke.py b/scripts/table-catalog/pyiceberg_smoke.py index f058c2483..47f72a509 100755 --- a/scripts/table-catalog/pyiceberg_smoke.py +++ b/scripts/table-catalog/pyiceberg_smoke.py @@ -135,13 +135,13 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [ "capability": "background-maintenance-worker", "status": "unsupported", "roadmap_area": "maintenance-worker", - "expected_behavior": "manual maintenance APIs may exist, but background-enabled maintenance is rejected", + "expected_behavior": "manual maintenance APIs expose retry/quarantine policy fields in reports, but background-enabled maintenance is rejected until a worker owns scheduling", }, { "capability": "manifest-data-reachability-cleanup", - "status": "unsupported", + "status": "fail-closed-reporting", "roadmap_area": "reachability-cleanup", - "expected_behavior": "metadata-only cleanup must not delete manifest, data, or delete files", + "expected_behavior": "metadata maintenance reports expose reachability graph status and referenced manifest-list objects, but cleanup must not delete manifest, data, or delete files", }, { "capability": "snapshot-expiration", @@ -157,9 +157,15 @@ UNSUPPORTED_INVENTORY: list[dict[str, str]] = [ }, { "capability": "iceberg-views", - "status": "unsupported", + "status": "stable-unsupported-routes", "roadmap_area": "view-api", - "expected_behavior": "view routes should return a stable unsupported response until implemented", + "expected_behavior": "view routes are registered and should return a stable unsupported JSON response until implemented", + }, + { + "capability": "external-catalog-bridge", + "status": "metadata-import-only", + "roadmap_area": "external-catalog", + "expected_behavior": "catalog import/register supports an existing Iceberg metadata location, while Polaris, Glue, DLF, and Hive synchronization remain unsupported", }, { "capability": "multi-table-transactions",