gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(table-catalog): add production iceberg surfaces (#3473)

Browse Source
This commit is contained in:
Henry Guo
2026-06-15 16:53:14 +08:00
committed by GitHub
parent d28aedfbb3
commit 307d788da1
6 changed files with 992 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/table_catalog.rs
+396
View File
@@ -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<String>,
}
#[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<i64>,
protected_ref_count: usize,
user_defined_ref_count: usize,
refs: BTreeMap<String, serde_json::Value>,
}
#[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<ExternalCatalogBridgeCapability>,
}
#[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<AdminOperation>, prefix
format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/register").as_str(),
AdminOperation(&REGISTER_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<AdminOperation>, 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<AdminOperation>, 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<S3Response<(StatusCode, Body)>> {
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<String> {
Ok(table.to_string())
}
fn view_name_from_params(params: &Params<'_, '_>) -> S3Result<String> {
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<String> {
let job = params.get("job").unwrap_or("");
if job.is_empty() {
@@ -2623,6 +2735,91 @@ where
Ok(report)
}
async fn table_refs_response<S>(
store: &S,
metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend,
bucket: &str,
namespace: &crate::table_catalog::Namespace,
table: &str,
) -> S3Result<TableRefsResponse>
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::<BTreeMap<_, _>>();
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<B>(
store: &crate::table_catalog::ObjectTableCatalogStore<B>,
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let _view = view_name_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let _view = view_name_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let _view = view_name_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let table = table_name_from_params(&params)?;
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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
let warehouse = warehouse_from_params(&params)?;
let namespace = namespace_from_params(&params)?;
let table = table_name_from_params(&params)?;
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 = &REGISTER_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::<EnableTableBucketHandler>();
@@ -3438,11 +3768,17 @@ mod tests {
assert_operation::<RestListTablesHandler>();
assert_operation::<RestCreateTableHandler>();
assert_operation::<RestRegisterTableHandler>();
assert_operation::<RestListViewsHandler>();
assert_operation::<RestCreateViewHandler>();
assert_operation::<RestLoadTableHandler>();
assert_operation::<RestTableExistsHandler>();
assert_operation::<RestLoadCredentialsHandler>();
assert_operation::<RestCommitTableHandler>();
assert_operation::<RestDropTableHandler>();
assert_operation::<RestLoadViewHandler>();
assert_operation::<RestReplaceViewHandler>();
assert_operation::<RestDropViewHandler>();
assert_operation::<ListTableRefsHandler>();
assert_operation::<GetTableMetadataLocationHandler>();
assert_operation::<UpdateTableMetadataLocationHandler>();
assert_operation::<RestTableMetadataMaintenanceHandler>();
@@ -3451,7 +3787,9 @@ mod tests {
assert_operation::<GetTableMaintenanceJobHandler>();
assert_operation::<ExportTableCatalogHandler>();
assert_operation::<ImportTableCatalogHandler>();
assert_operation::<ExternalCatalogBridgeHandler>();
assert_operation::<GetTableCatalogDiagnosticsHandler>();
assert_operation::<RecoverTableCatalogHandler>();
assert_operation::<RollbackTableCatalogHandler>();
}
@@ -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,
&current,
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!({
rustfs/src/admin/route_policy.rs
+116 -1
View File
@@ -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",
rustfs/src/admin/route_registration_test.rs
+124
View File
@@ -312,6 +312,16 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
"/{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<RouteMatrixEntry> {
"/{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<RouteMatrixEntry> {
"/{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<RouteMatrixEntry> {
"/{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<RouteMatrixEntry> {
"/{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<RouteMatrixEntry> {
"/{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,
rustfs/src/table_catalog.rs
+341 -3
View File
@@ -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<String>,
#[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<String>,
#[serde(default)]
pub started_at: Option<String>,
@@ -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<TableMetadataMaintenanceObjectReport>,
#[serde(default)]
pub referenced_object_reports: Vec<TableMetadataMaintenanceReferencedObjectReport>,
#[serde(default, rename = "reachability-graph")]
pub reachability_graph: TableMaintenanceReachabilityGraphReport,
#[serde(default, rename = "snapshot-expiration")]
pub snapshot_expiration: Option<TableSnapshotExpirationReport>,
#[serde(default)]
pub compaction: Option<TableCompactionPlanningReport>,
}
#[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<TableMaintenanceReachabilityGraphReason>,
}
#[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::<TableMaintenanceConfig>(self.catalog_bucket(), &config_path)
.await
.map(|entry| entry.map(|(config, _)| config).unwrap_or_default())
let config = self
.read_entry::<TableMaintenanceConfig>(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::<TableMaintenanceConfig>(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::<TableMaintenanceConfig>(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::<Vec<_>>();
let object_reports = metadata_maintenance_object_reports(maintenance_reasons);
let referenced_object_reports = metadata_maintenance_referenced_object_reports(&current_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<String>,
@@ -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]
scripts/table-catalog/README.md
+4 -3
View File
@@ -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
scripts/table-catalog/pyiceberg_smoke.py
+11 -5
View File
@@ -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",