From 7211f2949899fa4e914f9ec268f6afab77752481 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 6 Aug 2026 05:07:30 +0800 Subject: [PATCH] feat(table-catalog): complete namespace REST contracts (#5745) * feat(table-catalog): complete namespace REST contracts * fix(table-catalog): simplify namespace existence guard * fix(table-catalog): restore migration guard coverage * fix(table-catalog): preserve encoded namespace segments * fix(table-catalog): reject implicit namespace creates Co-Authored-By: heihutu --------- Co-authored-by: Henry Guo Co-authored-by: houseme Co-authored-by: heihutu --- docs/architecture/compat-cleanup-register.md | 1 + .../src/admin/handlers/table_catalog/mod.rs | 235 ++++- .../admin/handlers/table_catalog/namespace.rs | 41 +- .../admin/handlers/table_catalog/routes.rs | 5 + .../src/admin/handlers/table_catalog/tests.rs | 372 ++++++- rustfs/src/admin/route_policy.rs | 25 +- rustfs/src/admin/route_registration_test.rs | 10 + rustfs/src/table_catalog/error.rs | 2 + rustfs/src/table_catalog/iceberg/commit.rs | 1 + .../src/table_catalog/iceberg/validation.rs | 33 +- rustfs/src/table_catalog/identifier.rs | 22 +- rustfs/src/table_catalog/mod.rs | 2 +- rustfs/src/table_catalog/model.rs | 113 +++ rustfs/src/table_catalog/store/migration.rs | 312 +++--- rustfs/src/table_catalog/store/mod.rs | 164 +++ rustfs/src/table_catalog/store/object.rs | 583 +++++++++-- rustfs/src/table_catalog/store/strong.rs | 338 ++++++- rustfs/src/table_catalog/tests.rs | 949 +++++++++++++++++- scripts/check_architecture_migration_rules.sh | 4 +- 19 files changed, 2848 insertions(+), 364 deletions(-) diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 0ddd4fc13..4dd566e44 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/docs/architecture/compat-cleanup-register.md @@ -12,6 +12,7 @@ for later deletion. ## Open Items +- `table-catalog-dotted-namespace` Iceberg REST namespace path compatibility: existing RustFS clients use dotted namespace paths, while the standard multi-level contract uses the URL-encoded unit separator `%1F`. New servers accept both forms so a rolling upgrade does not invalidate existing catalog configuration. Remove the dotted fallback after the minimum supported RustFS release advertises `%1F` and all supported clients have refreshed their catalog configuration. - `rustfs-5509` FileInfo positional MessagePack decoding: beta.11 serialized 28 fields, while beta.12 inserted transition-version fields in the middle and serialized an incompatible 30-field array. New releases write named maps and retain readers for both shipped array layouts so direct and rolling upgrades can read either release. Remove the positional-array readers after every supported direct-upgrade release writes named maps and no retained RPC payload can contain a pre-map FileInfo array. - `rustfs-5416` Helm distributed startup wait setting: charts that predate explicit local endpoint identity expose startupWaitTimeoutSeconds for their peer DNS/TCP init gate. The new chart keeps the value accepted but ignores it after moving startup convergence into RustFS. Remove the value and its documentation after the minimum supported direct-upgrade chart includes localEndpointHost.autoInject and no longer renders the peer gate. - `rustfs-5416-wait-mode` startup wait-mode validation: releases before explicit local endpoint identity treat an unknown RUSTFS_STARTUP_TOPOLOGY_WAIT_MODE value as auto. New servers retain that fallback only when no explicit local endpoint host is configured; an anchor requires a recognized mode so a typo cannot bypass DNS locality. Remove the fallback and reject every unknown value after every supported direct-upgrade chart validates this setting before rollout. diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 2b43a79e5..5cce15684 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -27,6 +27,7 @@ use http::{HeaderMap, HeaderValue, StatusCode}; use hyper::Method; use matchit::Params; use metrics::{counter, histogram}; +use percent_encoding::percent_decode_str; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_iam::sys::SESSION_POLICY_NAME; use rustfs_policy::{ @@ -69,8 +70,11 @@ const ENV_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: &str = "RUSTFS_TABLE_CATALOG_CRE const DEFAULT_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 15 * 60; const MIN_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60; const MAX_TABLE_CATALOG_CREDENTIAL_TTL_SECONDS: i64 = 60 * 60; +const NAMESPACE_REQUEST_BODY_MAX_SIZE: usize = MAX_ADMIN_REQUEST_BODY_SIZE; +const NAMESPACE_REQUEST_BODY_TIMEOUT: StdDuration = StdDuration::from_secs(10); const WAREHOUSE_PROPERTY: &str = "warehouse"; const PREFIX_PROPERTY: &str = "prefix"; +const NAMESPACE_SEPARATOR_PROPERTY: &str = "namespace-separator"; const ICEBERG_ERROR_ALREADY_EXISTS: &str = "AlreadyExistsException"; const ICEBERG_ERROR_BAD_REQUEST: &str = "BadRequestException"; const ICEBERG_ERROR_COMMIT_FAILED: &str = "CommitFailedException"; @@ -80,12 +84,16 @@ const ICEBERG_ERROR_NO_SUCH_RESOURCE: &str = "NoSuchResourceException"; const ICEBERG_ERROR_NO_SUCH_TABLE: &str = "NoSuchTableException"; const ICEBERG_ERROR_NO_SUCH_VIEW: &str = "NoSuchViewException"; const ICEBERG_ERROR_REST: &str = "RESTException"; +const ICEBERG_ERROR_UNPROCESSABLE_ENTITY: &str = "UnprocessableEntityException"; +const ICEBERG_ERROR_UNSUPPORTED_OPERATION: &str = "UnsupportedOperationException"; const REST_PAGE_TOKEN_VERSION: u8 = 1; const REST_PAGE_TOKEN_MAX_LENGTH: usize = 16 * 1024; const REST_DEFAULT_PAGE_SIZE: usize = 1000; const REST_MAX_PAGE_SIZE: usize = 1000; const REST_PAGE_TOKEN_QUERY_PARAMETER: &str = "pageToken"; const REST_PAGE_SIZE_QUERY_PARAMETER: &str = "pageSize"; +const REST_NAMESPACE_SEPARATOR: char = '\u{1f}'; +const REST_NAMESPACE_SEPARATOR_URL_ENCODED: &str = "%1F"; const CATALOG_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-endpoint-prefix"; const CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY: &str = "rustfs.catalog-compat-endpoint-prefix"; const CATALOG_BACKING_CONFIG_KEY: &str = "rustfs.catalog-backing"; @@ -188,6 +196,7 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/recovery", "POST /{warehouse}/namespaces/{namespace}/tables/{table}/catalog/rollback", ]; +const TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS: &[&str] = &["POST /v1/{prefix}/namespaces/{namespace}/properties"]; static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {}; static ENABLE_TABLE_BUCKET_HANDLER: EnableTableBucketHandler = EnableTableBucketHandler {}; @@ -200,6 +209,7 @@ static LIST_NAMESPACES_HANDLER: RestListNamespacesHandler = RestListNamespacesHa static CREATE_NAMESPACE_HANDLER: RestCreateNamespaceHandler = RestCreateNamespaceHandler {}; static GET_NAMESPACE_HANDLER: RestGetNamespaceHandler = RestGetNamespaceHandler {}; static NAMESPACE_EXISTS_HANDLER: RestNamespaceExistsHandler = RestNamespaceExistsHandler {}; +static UPDATE_NAMESPACE_PROPERTIES_HANDLER: RestUpdateNamespacePropertiesHandler = RestUpdateNamespacePropertiesHandler {}; static DROP_NAMESPACE_HANDLER: RestDropNamespaceHandler = RestDropNamespaceHandler {}; static LIST_TABLES_HANDLER: RestListTablesHandler = RestListTablesHandler {}; static CREATE_TABLE_HANDLER: RestCreateTableHandler = RestCreateTableHandler {}; @@ -264,6 +274,15 @@ struct CreateNamespaceRequest { properties: BTreeMap, } +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct UpdateNamespacePropertiesRequest { + #[serde(default)] + removals: Vec, + #[serde(default)] + updates: BTreeMap, +} + #[derive(Debug, Deserialize)] #[serde(deny_unknown_fields)] struct RegisterTableRequest { @@ -888,7 +907,8 @@ struct TableMetadataLocationResponse { fn catalog_config_response(warehouse: Option<&str>) -> S3Result { let usecase = default_admin_usecase(); let backing_mode = crate::table_catalog::TableCatalogBackingMode::from_env().map_err(catalog_store_error)?; - let mut overrides = BTreeMap::new(); + let mut overrides = + BTreeMap::from([(NAMESPACE_SEPARATOR_PROPERTY.to_string(), REST_NAMESPACE_SEPARATOR_URL_ENCODED.to_string())]); if backing_mode != crate::table_catalog::TableCatalogBackingMode::ObjectBacked { overrides.insert(CATALOG_BACKING_CONFIG_KEY.to_string(), backing_mode.as_str().to_string()); } @@ -907,10 +927,14 @@ fn catalog_config_response(warehouse: Option<&str>) -> S3Result(mut input: Body) -> S3Result { serde_json::from_slice(&body).map_err(|err| s3_error!(InvalidRequest, "invalid JSON: {}", err)) } +async fn read_bounded_json_body( + headers: &HeaderMap, + mut input: Body, + max_size: usize, + timeout: StdDuration, + operation: &str, +) -> S3Result { + if let Some(content_length) = headers.get(http::header::CONTENT_LENGTH) { + let content_length = content_length + .to_str() + .map_err(|_| s3_error!(InvalidRequest, "Content-Length must be valid ASCII"))? + .parse::() + .map_err(|_| s3_error!(InvalidRequest, "Content-Length must be a non-negative integer"))?; + if content_length > max_size { + return Err(s3_error!(InvalidRequest, "{operation} request body is too large")); + } + } + let body = tokio::time::timeout(timeout, input.store_all_limited(max_size)) + .await + .map_err(|_| s3_error!(InvalidRequest, "timed out reading {operation} request body"))? + .map_err(|err| s3_error!(InvalidRequest, "failed to read request body: {}", err))?; + if body.is_empty() { + return Err(s3_error!(InvalidRequest, "request body is required")); + } + serde_json::from_slice(&body).map_err(|err| s3_error!(InvalidRequest, "invalid JSON: {}", err)) +} + async fn read_json_body_or_default(mut input: Body) -> S3Result where T: Default + DeserializeOwned, @@ -1294,7 +1345,69 @@ fn encode_rest_page_token(cursor: &str, context: &str) -> S3Result { fn namespace_from_params(params: &Params<'_, '_>) -> S3Result { let namespace = params.get("namespace").unwrap_or(""); - crate::table_catalog::Namespace::parse(namespace).map_err(|err| s3_error!(InvalidRequest, "invalid namespace: {}", err)) + namespace_from_path_value(namespace) +} + +fn namespace_from_path_value(value: &str) -> S3Result { + let legacy_dotted = value.contains('.') + && !value.contains(REST_NAMESPACE_SEPARATOR) + && !value.contains(REST_NAMESPACE_SEPARATOR_URL_ENCODED) + && !value.contains("%1f"); + // RUSTFS_COMPAT_TODO(table-catalog-dotted-namespace): Remove after the minimum supported release + // advertises %1F; until then, keep dotted paths for clients using the legacy namespace contract. + let segments = if legacy_dotted { + value + .split('.') + .map(|segment| { + percent_decode_str(segment) + .decode_utf8() + .map(|decoded| decoded.into_owned()) + .map_err(|_| s3_error!(InvalidRequest, "namespace path must be valid UTF-8")) + }) + .collect::>>()? + } else { + let decoded = percent_decode_str(value) + .decode_utf8() + .map_err(|_| s3_error!(InvalidRequest, "namespace path must be valid UTF-8"))?; + decoded.split(REST_NAMESPACE_SEPARATOR).map(str::to_string).collect() + }; + crate::table_catalog::Namespace::from_segments(segments) + .map_err(|err| s3_error!(InvalidRequest, "invalid namespace: {}", err)) +} + +fn rest_namespace_parent_from_query(uri: &http::Uri) -> S3Result> { + let mut parent = None; + let mut parent_seen = false; + if let Some(query) = uri.query() { + for (key, value) in url::form_urlencoded::parse(query.as_bytes()) { + if key != "parent" { + continue; + } + if parent_seen { + return Err(iceberg_rest_error( + ICEBERG_ERROR_BAD_REQUEST, + StatusCode::BAD_REQUEST, + "parent query parameter must not be repeated", + )); + } + parent_seen = true; + if !value.is_empty() { + parent = Some( + crate::table_catalog::Namespace::from_segments( + value.split(REST_NAMESPACE_SEPARATOR).map(str::to_string).collect(), + ) + .map_err(|err| { + iceberg_rest_error( + ICEBERG_ERROR_BAD_REQUEST, + StatusCode::BAD_REQUEST, + format!("invalid parent namespace: {err}"), + ) + })?, + ); + } + } + } + Ok(parent) } fn table_name_from_params(params: &Params<'_, '_>) -> S3Result { @@ -1460,12 +1573,8 @@ fn namespace_segments(namespace: &crate::table_catalog::Namespace) -> Vec S3Result { - if segments.is_empty() { - return Err(s3_error!(InvalidRequest, "namespace cannot be empty")); - } - - let namespace = segments.join("."); - crate::table_catalog::Namespace::parse(&namespace).map_err(|err| s3_error!(InvalidRequest, "invalid namespace: {}", err)) + crate::table_catalog::Namespace::from_segments(segments.to_vec()) + .map_err(|err| s3_error!(InvalidRequest, "invalid namespace: {}", err)) } fn namespace_response_from_entry(entry: crate::table_catalog::NamespaceEntry) -> S3Result { @@ -1477,24 +1586,6 @@ fn namespace_response_from_entry(entry: crate::table_catalog::NamespaceEntry) -> }) } -fn list_namespaces_response_from_entries( - entries: Vec, - next_page_token: Option, -) -> S3Result { - let namespaces = entries - .into_iter() - .map(|entry| { - let namespace = crate::table_catalog::Namespace::parse(&entry.namespace) - .map_err(|err| s3_error!(InternalError, "persisted namespace entry is invalid: {}", err))?; - Ok(namespace_segments(&namespace)) - }) - .collect::>>()?; - Ok(RestListNamespacesResponse { - namespaces, - next_page_token, - }) -} - fn list_tables_response_from_entries( entries: Vec, next_page_token: Option, @@ -3511,6 +3602,7 @@ fn namespace_entry_from_create_request( request: CreateNamespaceRequest, ) -> S3Result { let namespace = namespace_from_segments(&request.namespace)?; + crate::table_catalog::validate_namespace_properties(&request.properties).map_err(catalog_store_error)?; Ok(crate::table_catalog::NamespaceEntry { version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, table_bucket: bucket.to_string(), @@ -3540,6 +3632,9 @@ fn catalog_store_error(err: crate::table_catalog::TableCatalogStoreError) -> S3E crate::table_catalog::TableCatalogStoreError::Invalid(message) => { iceberg_rest_error(ICEBERG_ERROR_BAD_REQUEST, StatusCode::BAD_REQUEST, message) } + crate::table_catalog::TableCatalogStoreError::Unsupported(message) => { + iceberg_rest_error(ICEBERG_ERROR_UNSUPPORTED_OPERATION, StatusCode::NOT_ACCEPTABLE, message) + } crate::table_catalog::TableCatalogStoreError::Internal(message) => { iceberg_rest_error(ICEBERG_ERROR_REST, StatusCode::INTERNAL_SERVER_ERROR, message) } @@ -3586,28 +3681,61 @@ where namespace_response_from_entry(entry) } -async fn list_namespaces_response(store: &S, bucket: &str, uri: &http::Uri) -> S3Result +async fn list_namespaces_response( + store: &S, + bucket: &str, + parent: Option<&crate::table_catalog::Namespace>, + uri: &http::Uri, +) -> S3Result where S: crate::table_catalog::TableCatalogStore + ?Sized, { + let parent_name = parent.map(crate::table_catalog::Namespace::public_name); let context = RestPageContext { resource: TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT, warehouse: bucket, - namespace: None, + namespace: parent_name.as_deref(), }; let pagination = rest_pagination_from_query(uri, context)?; + let map_list_error = |err| match err { + crate::table_catalog::TableCatalogStoreError::NotFound(message) => { + iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_NAMESPACE, StatusCode::NOT_FOUND, message) + } + err => catalog_store_error(err), + }; let page = match pagination.page_request() { Some((cursor, limit)) => store - .list_namespaces_page(bucket, cursor, limit) + .list_namespace_children_page(bucket, parent_name.as_deref(), cursor, limit) .await - .map_err(catalog_store_error)?, + .map_err(map_list_error)?, None => crate::table_catalog::TableCatalogListPage { - entries: store.list_namespaces(bucket).await.map_err(catalog_store_error)?, + entries: store + .list_namespace_children(bucket, parent_name.as_deref()) + .await + .map_err(map_list_error)?, next_cursor: None, }, }; let next_page_token = pagination.next_page_token(page.next_cursor)?; - list_namespaces_response_from_entries(page.entries, next_page_token) + let namespaces = page + .entries + .into_iter() + .map(|entry| { + crate::table_catalog::Namespace::parse(&entry.namespace) + .map(|namespace| namespace_segments(&namespace)) + .map_err(|err| { + iceberg_rest_error( + ICEBERG_ERROR_REST, + StatusCode::INTERNAL_SERVER_ERROR, + format!("catalog namespace is invalid: {err}"), + ) + }) + }) + .collect::>>()?; + Ok(RestListNamespacesResponse { + namespaces, + next_page_token, + }) } async fn get_namespace_response( @@ -3622,6 +3750,7 @@ where .get_namespace(bucket, &namespace.public_name()) .await .map_err(catalog_store_error)? + .filter(|entry| entry.state == crate::table_catalog::TableCatalogEntryState::Active) else { return Err(iceberg_rest_error( ICEBERG_ERROR_NO_SUCH_NAMESPACE, @@ -3632,6 +3761,44 @@ where namespace_response_from_entry(entry) } +fn namespace_properties_update_from_request( + request: UpdateNamespacePropertiesRequest, +) -> S3Result { + crate::table_catalog::NamespacePropertiesUpdate::try_new(request.removals, request.updates).map_err(|err| match err { + crate::table_catalog::NamespacePropertiesUpdateError::DuplicateRemoval(key) => iceberg_rest_error( + ICEBERG_ERROR_BAD_REQUEST, + StatusCode::BAD_REQUEST, + format!("namespace property removal is repeated: {key}"), + ), + crate::table_catalog::NamespacePropertiesUpdateError::Overlap(key) => iceberg_rest_error( + ICEBERG_ERROR_UNPROCESSABLE_ENTITY, + StatusCode::UNPROCESSABLE_ENTITY, + format!("namespace property cannot be removed and updated in the same request: {key}"), + ), + }) +} + +async fn update_namespace_properties_response( + store: &S, + bucket: &str, + namespace: &crate::table_catalog::Namespace, + request: UpdateNamespacePropertiesRequest, +) -> S3Result +where + S: crate::table_catalog::TableCatalogStore + ?Sized, +{ + let update = namespace_properties_update_from_request(request)?; + store + .update_namespace_properties(bucket, &namespace.public_name(), update) + .await + .map_err(|err| match err { + crate::table_catalog::TableCatalogStoreError::NotFound(message) => { + iceberg_rest_error(ICEBERG_ERROR_NO_SUCH_NAMESPACE, StatusCode::NOT_FOUND, message) + } + err => catalog_store_error(err), + }) +} + async fn namespace_exists_status(store: &S, bucket: &str, namespace: &crate::table_catalog::Namespace) -> S3Result where S: crate::table_catalog::TableCatalogStore + ?Sized, @@ -3640,7 +3807,7 @@ where .get_namespace(bucket, &namespace.public_name()) .await .map_err(catalog_store_error)? - .is_some(); + .is_some_and(|entry| entry.state == crate::table_catalog::TableCatalogEntryState::Active); Ok(exists_status(exists)) } diff --git a/rustfs/src/admin/handlers/table_catalog/namespace.rs b/rustfs/src/admin/handlers/table_catalog/namespace.rs index 9a13ec36b..a956bde20 100644 --- a/rustfs/src/admin/handlers/table_catalog/namespace.rs +++ b/rustfs/src/admin/handlers/table_catalog/namespace.rs @@ -20,11 +20,15 @@ pub struct RestListNamespacesHandler {} impl Operation for RestListNamespacesHandler { async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { let warehouse = warehouse_from_params(¶ms)?; - let resource = TableCatalogResource::warehouse(&warehouse); + let parent = rest_namespace_parent_from_query(&req.uri)?; + let resource = match &parent { + Some(parent) => TableCatalogResource::namespace(&warehouse, parent), + None => TableCatalogResource::warehouse(&warehouse), + }; authorize_table_catalog_resource_request(&req, &resource, AdminAction::GetTableNamespaceAction).await?; ensure_table_bucket_enabled(&warehouse).await?; let store = table_catalog_store()?; - let response = list_namespaces_response(&store, &warehouse, &req.uri).await?; + let response = list_namespaces_response(&store, &warehouse, parent.as_ref(), &req.uri).await?; build_json_response(StatusCode::OK, &response) } } @@ -37,7 +41,14 @@ impl Operation for RestCreateNamespaceHandler { let warehouse = warehouse_from_params(¶ms)?; let resource = TableCatalogResource::warehouse(&warehouse); authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableNamespaceAction).await?; - let request = read_json_body::(req.input).await?; + let request = read_bounded_json_body::( + &req.headers, + req.input, + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace creation", + ) + .await?; let store = table_catalog_store()?; let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; let response = create_namespace_response(&store, &warehouse, request, table_bucket_enabled).await?; @@ -77,6 +88,30 @@ impl Operation for RestDropNamespaceHandler { } } +pub struct RestUpdateNamespacePropertiesHandler {} + +#[async_trait::async_trait] +impl Operation for RestUpdateNamespacePropertiesHandler { + 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::UpdateTableNamespacePropertiesAction).await?; + ensure_table_bucket_enabled(&warehouse).await?; + let request = read_bounded_json_body::( + &req.headers, + req.input, + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await?; + let store = table_catalog_store()?; + let response = update_namespace_properties_response(&store, &warehouse, &namespace, request).await?; + build_json_response(StatusCode::OK, &response) + } +} + pub struct RestNamespaceExistsHandler {} #[async_trait::async_trait] diff --git a/rustfs/src/admin/handlers/table_catalog/routes.rs b/rustfs/src/admin/handlers/table_catalog/routes.rs index f5faf157d..ceff0ca8b 100644 --- a/rustfs/src/admin/handlers/table_catalog/routes.rs +++ b/rustfs/src/admin/handlers/table_catalog/routes.rs @@ -69,6 +69,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}").as_str(), AdminOperation(&NAMESPACE_EXISTS_HANDLER), )?; + r.insert( + Method::POST, + format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/properties").as_str(), + AdminOperation(&UPDATE_NAMESPACE_PROPERTIES_HANDLER), + )?; r.insert( Method::DELETE, format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}").as_str(), diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index b83ae0e5d..fdfaf1db2 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -25,7 +25,15 @@ fn catalog_config_response_lists_standard_rest_endpoints() { Some(crate::table_catalog::TABLE_CATALOG_BACKING_OBJECT) ); assert!(!response.defaults.contains_key(PREFIX_PROPERTY)); - assert!(response.overrides.is_empty()); + assert_eq!( + response.overrides.get(NAMESPACE_SEPARATOR_PROPERTY).map(String::as_str), + Some(REST_NAMESPACE_SEPARATOR_URL_ENCODED) + ); + assert!( + !response + .endpoints + .contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties") + ); assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog"); @@ -145,6 +153,11 @@ fn catalog_config_response_reports_durable_strong_backing_override() { response.overrides.get(CATALOG_BACKING_CONFIG_KEY).map(String::as_str), Some(crate::table_catalog::TABLE_CATALOG_BACKING_DURABLE_STRONG) ); + assert!( + response + .endpoints + .contains(&"POST /v1/{prefix}/namespaces/{namespace}/properties") + ); } #[test] @@ -189,6 +202,12 @@ fn catalog_conflicts_use_operation_specific_iceberg_errors() { )); assert_eq!(namespace_not_found.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_NO_SUCH_NAMESPACE.into())); assert_eq!(namespace_not_found.status_code(), Some(StatusCode::NOT_FOUND)); + + let unsupported = catalog_store_error(crate::table_catalog::TableCatalogStoreError::Unsupported( + "operation is unavailable".to_string(), + )); + assert_eq!(unsupported.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNSUPPORTED_OPERATION.into())); + assert_eq!(unsupported.status_code(), Some(StatusCode::NOT_ACCEPTABLE)); } #[test] @@ -221,6 +240,10 @@ fn table_catalog_handlers_require_table_admin_actions() { ("RestCreateNamespaceHandler", "AdminAction::SetTableNamespaceAction"), ("RestGetNamespaceHandler", "AdminAction::GetTableNamespaceAction"), ("RestNamespaceExistsHandler", "AdminAction::GetTableNamespaceAction"), + ( + "RestUpdateNamespacePropertiesHandler", + "AdminAction::UpdateTableNamespacePropertiesAction", + ), ("RestDropNamespaceHandler", "AdminAction::DeleteTableNamespaceAction"), ("RestListTablesHandler", "AdminAction::GetTableAction"), ("RestCreateTableHandler", "AdminAction::CreateTableAction"), @@ -342,7 +365,7 @@ fn table_catalog_list_handlers_parse_standard_pagination() { for (handler, helper_call) in [ ( "RestListNamespacesHandler", - "list_namespaces_response(&store, &warehouse, &req.uri).await?", + "list_namespaces_response(&store, &warehouse, parent.as_ref(), &req.uri).await?", ), ( "RestListTablesHandler", @@ -361,6 +384,18 @@ fn table_catalog_list_handlers_parse_standard_pagination() { } } +#[test] +fn namespace_write_handlers_bound_request_bodies() { + let src = table_catalog_handler_source(); + for handler in ["RestCreateNamespaceHandler", "RestUpdateNamespacePropertiesHandler"] { + let block = operation_block(&src, handler); + assert!( + block.contains("read_bounded_json_body::<"), + "{handler} should enforce the namespace request body limit and timeout" + ); + } +} + #[test] fn table_catalog_handlers_require_enabled_table_bucket_marker_before_catalog_state() { let src = table_catalog_handler_source(); @@ -373,6 +408,7 @@ fn table_catalog_handlers_require_enabled_table_bucket_marker_before_catalog_sta "RestCreateNamespaceHandler", "RestGetNamespaceHandler", "RestNamespaceExistsHandler", + "RestUpdateNamespacePropertiesHandler", "RestDropNamespaceHandler", "RestListTablesHandler", "RestCreateTableHandler", @@ -504,6 +540,7 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() { let _: &RestCreateNamespaceHandler = &CREATE_NAMESPACE_HANDLER; let _: &RestGetNamespaceHandler = &GET_NAMESPACE_HANDLER; let _: &RestNamespaceExistsHandler = &NAMESPACE_EXISTS_HANDLER; + let _: &RestUpdateNamespacePropertiesHandler = &UPDATE_NAMESPACE_PROPERTIES_HANDLER; let _: &RestDropNamespaceHandler = &DROP_NAMESPACE_HANDLER; let _: &RestListTablesHandler = &LIST_TABLES_HANDLER; let _: &RestCreateTableHandler = &CREATE_TABLE_HANDLER; @@ -546,6 +583,7 @@ fn rest_catalog_mvp_routes_use_implemented_handlers() { assert_operation::(); assert_operation::(); assert_operation::(); + assert_operation::(); assert_operation::(); assert_operation::(); assert_operation::(); @@ -730,6 +768,13 @@ fn table_catalog_ingress_requests_reject_unknown_fields() { "unexpected": true }), ); + assert_rejects_unknown_field::( + "UpdateNamespacePropertiesRequest", + serde_json::json!({ + "updates": {}, + "unexpected": true + }), + ); assert_rejects_unknown_field::( "RegisterTableRequest", serde_json::json!({ @@ -849,6 +894,187 @@ fn create_namespace_request_uses_rest_namespace_segments_and_properties() { assert_eq!(namespace.public_name(), "analytics.daily_events"); assert_eq!(response.namespace, vec!["analytics".to_string(), "daily_events".to_string()]); assert_eq!(response.properties.get("owner").map(String::as_str), Some("lakehouse")); + assert!(namespace_from_segments(&["analytics.daily_events".to_string()]).is_err()); +} + +#[tokio::test] +async fn invalid_namespace_properties_fail_before_catalog_state_changes() { + let store = TestTableCatalogStore::default(); + let error = create_namespace_response( + &store, + "warehouse", + CreateNamespaceRequest { + namespace: vec!["analytics".to_string()], + properties: BTreeMap::from([( + "owner".to_string(), + "x".repeat(crate::table_catalog::NAMESPACE_PROPERTY_VALUE_MAX_LEN + 1), + )]), + }, + true, + ) + .await + .expect_err("oversized namespace property must fail"); + + assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into())); + assert!(store.table_buckets.lock().await.is_empty()); + assert!(store.namespaces.lock().await.is_empty()); +} + +#[test] +fn namespace_parent_query_accepts_standard_and_legacy_path_separators() { + let path_namespace = namespace_from_path_value("accounting%1Ftax").expect("encoded path namespace should parse"); + assert_eq!(path_namespace.public_name(), "accounting.tax"); + let lowercase_path_namespace = + namespace_from_path_value("accounting%1ftax").expect("lowercase encoded path namespace should parse"); + assert_eq!(lowercase_path_namespace.public_name(), "accounting.tax"); + let legacy_path_namespace = namespace_from_path_value("accounting.tax").expect("legacy dotted path namespace should parse"); + assert_eq!(legacy_path_namespace.public_name(), "accounting.tax"); + assert!(namespace_from_path_value("accounting%2Etax").is_err()); + assert!(namespace_from_path_value("accounting.tax%2Epaid").is_err()); + assert!(namespace_from_path_value("accounting%2Ftax").is_err()); + assert!(namespace_from_path_value("%FF").is_err()); + + let uri = "/iceberg/v1/analytics/namespaces?parent=accounting%1Ftax" + .parse() + .expect("URI"); + let parent = rest_namespace_parent_from_query(&uri) + .expect("parent query should parse") + .expect("parent should be present"); + assert_eq!(parent.public_name(), "accounting.tax"); + + let uri = "/iceberg/v1/analytics/namespaces?parent=".parse().expect("URI"); + assert!( + rest_namespace_parent_from_query(&uri) + .expect("empty parent should parse") + .is_none() + ); + + let uri = "/iceberg/v1/analytics/namespaces?parent=one&parent=two".parse().expect("URI"); + let error = rest_namespace_parent_from_query(&uri).expect_err("repeated parent should fail"); + assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into())); + assert_eq!(error.status_code(), Some(StatusCode::BAD_REQUEST)); +} + +#[test] +fn namespace_property_update_uses_standard_shape_and_rejects_invalid_key_sets() { + let request: UpdateNamespacePropertiesRequest = serde_json::from_value(serde_json::json!({ + "removals": ["retention"], + "updates": {"owner": "platform"} + })) + .expect("namespace property update should parse"); + let update = namespace_properties_update_from_request(request).expect("disjoint property update should validate"); + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let mut entry = crate::table_catalog::NamespaceEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "warehouse".to_string(), + namespace: namespace.public_name(), + namespace_id: namespace.storage_id(), + state: crate::table_catalog::TableCatalogEntryState::Active, + properties: BTreeMap::from([("retention".to_string(), "30d".to_string())]), + created_at: None, + updated_at: None, + }; + let result = update.apply_to(&mut entry); + assert_eq!(result.removed, vec!["retention".to_string()]); + assert_eq!(result.updated, vec!["owner".to_string()]); + assert_eq!(entry.properties.get("owner").map(String::as_str), Some("platform")); + + let duplicate = namespace_properties_update_from_request(UpdateNamespacePropertiesRequest { + removals: vec!["owner".to_string(), "owner".to_string()], + updates: BTreeMap::new(), + }) + .expect_err("duplicate removals should fail"); + assert_eq!(duplicate.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_BAD_REQUEST.into())); + assert_eq!(duplicate.status_code(), Some(StatusCode::BAD_REQUEST)); + + let overlap = namespace_properties_update_from_request(UpdateNamespacePropertiesRequest { + removals: vec!["owner".to_string()], + updates: BTreeMap::from([("owner".to_string(), "platform".to_string())]), + }) + .expect_err("overlapping property sets should fail"); + assert_eq!(overlap.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_UNPROCESSABLE_ENTITY.into())); + assert_eq!(overlap.status_code(), Some(StatusCode::UNPROCESSABLE_ENTITY)); +} + +#[tokio::test] +async fn namespace_property_update_body_is_bounded_and_required() { + let mut oversized_headers = HeaderMap::new(); + oversized_headers.insert( + http::header::CONTENT_LENGTH, + HeaderValue::from_str(&(NAMESPACE_REQUEST_BODY_MAX_SIZE + 1).to_string()).expect("content length should parse"), + ); + let error = read_bounded_json_body::( + &oversized_headers, + Body::empty(), + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await + .expect_err("oversized declared body should fail before reading"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + + let error = read_bounded_json_body::( + &HeaderMap::new(), + Body::from(vec![b' '; NAMESPACE_REQUEST_BODY_MAX_SIZE + 1]), + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await + .expect_err("oversized streamed body should fail"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + + let error = read_bounded_json_body::( + &HeaderMap::new(), + Body::empty(), + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await + .expect_err("empty body should fail"); + assert_eq!(error.code(), &S3ErrorCode::InvalidRequest); + + let request = read_bounded_json_body::( + &HeaderMap::new(), + Body::from(r#"{"updates":{"owner":"platform"}}"#.to_string()), + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await + .expect("bounded request should parse"); + assert_eq!(request.updates.get("owner").map(String::as_str), Some("platform")); + + let mut maximum_properties = BTreeMap::new(); + for index in 0..15 { + maximum_properties.insert(format!("k{index:02}"), "v".repeat(crate::table_catalog::NAMESPACE_PROPERTY_VALUE_MAX_LEN)); + } + let used = maximum_properties + .iter() + .map(|(key, value)| key.len() + value.len()) + .sum::(); + let final_key = "k15".to_string(); + maximum_properties.insert( + final_key.clone(), + "v".repeat(crate::table_catalog::NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES - used - final_key.len()), + ); + let maximum_body = serde_json::to_vec(&serde_json::json!({"updates": maximum_properties})) + .expect("maximum namespace properties should encode"); + assert!(maximum_body.len() > crate::table_catalog::NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES); + assert!(maximum_body.len() < NAMESPACE_REQUEST_BODY_MAX_SIZE); + let request = read_bounded_json_body::( + &HeaderMap::new(), + Body::from(maximum_body), + NAMESPACE_REQUEST_BODY_MAX_SIZE, + NAMESPACE_REQUEST_BODY_TIMEOUT, + "namespace properties", + ) + .await + .expect("maximum valid namespace properties body should parse"); + crate::table_catalog::validate_namespace_properties(&request.updates) + .expect("maximum valid namespace properties should remain within the domain limit"); } #[test] @@ -1044,6 +1270,84 @@ fn rest_pagination_rejects_malformed_token_payloads() { } } +#[tokio::test] +async fn namespace_listing_returns_direct_children_and_scopes_pagination_to_parent() { + let store = TestTableCatalogStore::default(); + for name in [ + "accounting", + "accounting.tax.paid", + "accounting.ledger", + "analytics", + "analytics.daily", + "sales", + ] { + let namespace = crate::table_catalog::Namespace::parse(name).expect("namespace should parse"); + store.namespaces.lock().await.push(crate::table_catalog::NamespaceEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "warehouse".to_string(), + namespace: namespace.public_name(), + namespace_id: namespace.storage_id(), + state: crate::table_catalog::TableCatalogEntryState::Active, + properties: BTreeMap::new(), + created_at: None, + updated_at: None, + }); + } + + let root_uri = "/".parse::().expect("root namespace URI should parse"); + let root = list_namespaces_response(&store, "warehouse", None, &root_uri) + .await + .expect("root namespace list should load"); + assert_eq!( + root.namespaces, + vec![ + vec!["accounting".to_string()], + vec!["analytics".to_string()], + vec!["sales".to_string()] + ] + ); + + let parent = crate::table_catalog::Namespace::parse("accounting").expect("parent namespace should parse"); + let first_uri = "/?pageSize=1".parse::().expect("first page URI should parse"); + let first = list_namespaces_response(&store, "warehouse", Some(&parent), &first_uri) + .await + .expect("first child page should load"); + assert_eq!(first.namespaces, vec![vec!["accounting".to_string(), "ledger".to_string()]]); + let token = first.next_page_token.expect("child continuation token should exist"); + let second_uri = format!("/?pageSize=1&pageToken={token}") + .parse::() + .expect("second page URI should parse"); + let second = list_namespaces_response(&store, "warehouse", Some(&parent), &second_uri) + .await + .expect("second child page should load"); + assert_eq!(second.namespaces, vec![vec!["accounting".to_string(), "tax".to_string()]]); + assert!(second.next_page_token.is_none()); + + let different_parent = crate::table_catalog::Namespace::parse("analytics").expect("parent namespace should parse"); + let mismatched_uri = format!("/?pageSize=1&pageToken={token}") + .parse::() + .expect("mismatched page URI should parse"); + assert!( + list_namespaces_response(&store, "warehouse", Some(&different_parent), &mismatched_uri) + .await + .is_err() + ); + + let missing = crate::table_catalog::Namespace::parse("missing").expect("missing namespace should parse"); + let error = list_namespaces_response(&store, "warehouse", Some(&missing), &root_uri) + .await + .expect_err("missing parent should return an Iceberg error"); + assert_eq!(error.code(), &S3ErrorCode::Custom(ICEBERG_ERROR_NO_SUCH_NAMESPACE.into())); + assert_eq!(error.status_code(), Some(StatusCode::NOT_FOUND)); + + let leaf = crate::table_catalog::Namespace::parse("sales").expect("leaf namespace should parse"); + let empty = list_namespaces_response(&store, "warehouse", Some(&leaf), &root_uri) + .await + .expect("existing leaf namespace should return an empty child list"); + assert!(empty.namespaces.is_empty()); + assert!(empty.next_page_token.is_none()); +} + #[tokio::test] async fn rest_list_pagination_covers_namespaces_tables_and_views() { let store = TestTableCatalogStore::default(); @@ -1098,7 +1402,7 @@ async fn rest_list_pagination_covers_namespaces_tables_and_views() { } let first_uri = "/?pageSize=1".parse::().expect("first page URI should parse"); - let namespaces = list_namespaces_response(&store, "warehouse", &first_uri) + let namespaces = list_namespaces_response(&store, "warehouse", None, &first_uri) .await .expect("namespace first page should load"); assert_eq!(namespaces.namespaces, vec![vec!["alpha".to_string()]]); @@ -1106,7 +1410,7 @@ async fn rest_list_pagination_covers_namespaces_tables_and_views() { let namespace_uri = format!("/?pageSize=1&pageToken={namespace_token}") .parse::() .expect("namespace continuation URI should parse"); - let namespaces = list_namespaces_response(&store, "warehouse", &namespace_uri) + let namespaces = list_namespaces_response(&store, "warehouse", None, &namespace_uri) .await .expect("namespace second page should load"); assert_eq!(namespaces.namespaces, vec![vec!["beta".to_string()]]); @@ -1146,7 +1450,7 @@ async fn rest_list_pagination_covers_namespaces_tables_and_views() { for uri in ["/", "/?pageSize=2"] { let uri = uri.parse::().expect("list URI should parse"); - let namespaces = list_namespaces_response(&store, "warehouse", &uri) + let namespaces = list_namespaces_response(&store, "warehouse", None, &uri) .await .expect("namespace exact page should load"); let tables = list_tables_response(&store, "warehouse", &namespace, &uri) @@ -5500,6 +5804,22 @@ impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore { .cloned()) } + async fn update_namespace_properties( + &self, + table_bucket: &str, + namespace: &str, + update: crate::table_catalog::NamespacePropertiesUpdate, + ) -> crate::table_catalog::TableCatalogStoreResult { + let mut namespaces = self.namespaces.lock().await; + let entry = namespaces + .iter_mut() + .find(|entry| entry.table_bucket == table_bucket && entry.namespace == namespace) + .ok_or_else(|| { + crate::table_catalog::TableCatalogStoreError::NotFound(format!("namespace {table_bucket}/{namespace}")) + })?; + Ok(update.apply_to(entry)) + } + async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> crate::table_catalog::TableCatalogStoreResult<()> { self.namespaces .lock() @@ -5557,6 +5877,20 @@ impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore { .collect()) } + async fn list_all_tables( + &self, + table_bucket: &str, + ) -> crate::table_catalog::TableCatalogStoreResult> { + Ok(self + .tables + .lock() + .await + .iter() + .filter(|entry| entry.table_bucket == table_bucket) + .cloned() + .collect()) + } + async fn load_table( &self, table_bucket: &str, @@ -5820,15 +6154,39 @@ async fn namespace_helpers_call_catalog_store() { assert_eq!(create.properties.get("owner").map(String::as_str), Some("lakehouse")); let unpaginated_uri = "/".parse::().expect("list URI should parse"); - let list = list_namespaces_response(&store, "warehouse", &unpaginated_uri) + let list = list_namespaces_response(&store, "warehouse", None, &unpaginated_uri) .await .expect("namespace list should load"); assert_eq!(list.namespaces, vec![vec!["analytics".to_string()]]); + let update = update_namespace_properties_response( + &store, + "warehouse", + &crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"), + UpdateNamespacePropertiesRequest { + removals: vec!["owner".to_string(), "missing".to_string()], + updates: BTreeMap::from([("retention".to_string(), "30d".to_string())]), + }, + ) + .await + .expect("namespace properties should update"); + assert_eq!(update.updated, vec!["retention".to_string()]); + assert_eq!(update.removed, vec!["owner".to_string()]); + assert_eq!(update.missing, vec!["missing".to_string()]); + let updated = get_namespace_response( + &store, + "warehouse", + &crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"), + ) + .await + .expect("updated namespace should load"); + assert_eq!(updated.properties.get("retention").map(String::as_str), Some("30d")); + assert!(!updated.properties.contains_key("owner")); + drop_namespace_in_store(&store, "warehouse", "analytics") .await .expect("namespace should drop"); - let list = list_namespaces_response(&store, "warehouse", &unpaginated_uri) + let list = list_namespaces_response(&store, "warehouse", None, &unpaginated_uri) .await .expect("namespace list should load after drop"); assert!(list.namespaces.is_empty()); diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 568e5d348..c493f2bca 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -90,6 +90,7 @@ const SET_TABLE_BUCKET: AdminActionRef = AdminActionRef::new("SetTableBucketActi const SET_TABLE_LIFECYCLE: AdminActionRef = AdminActionRef::new("SetTableLifecycleAction"); const SET_TABLE_METADATA_LOCATION: AdminActionRef = AdminActionRef::new("SetTableMetadataLocationAction"); const SET_TABLE_NAMESPACE: AdminActionRef = AdminActionRef::new("SetTableNamespaceAction"); +const UPDATE_TABLE_NAMESPACE_PROPERTIES: AdminActionRef = AdminActionRef::new("UpdateTableNamespacePropertiesAction"); const SET_TIER: AdminActionRef = AdminActionRef::new("SetTierAction"); const SITE_REPLICATION_ADD: AdminActionRef = AdminActionRef::new("SiteReplicationAddAction"); const SITE_REPLICATION_INFO: AdminActionRef = AdminActionRef::new("SiteReplicationInfoAction"); @@ -911,6 +912,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_NAMESPACE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/properties", + UPDATE_TABLE_NAMESPACE_PROPERTIES, + RouteRiskLevel::High, + ), admin( HttpMethod::Delete, "/iceberg/v1/{warehouse}/namespaces/{namespace}", @@ -1188,6 +1195,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ GET_TABLE_NAMESPACE, RouteRiskLevel::Sensitive, ), + admin( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/properties", + UPDATE_TABLE_NAMESPACE_PROPERTIES, + RouteRiskLevel::High, + ), admin( HttpMethod::Delete, "/_iceberg/v1/{warehouse}/namespaces/{namespace}", @@ -1641,13 +1654,23 @@ 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(), 94); + assert_eq!(table_specs.count(), 96); assert_action(HttpMethod::Put, "/iceberg/v1/buckets/{warehouse}", SET_TABLE_BUCKET); assert_action(HttpMethod::Get, "/_iceberg/v1/buckets/{warehouse}", GET_TABLE_BUCKET); assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Head, "/iceberg/v1/{warehouse}/namespaces/{namespace}", GET_TABLE_NAMESPACE); assert_action(HttpMethod::Head, "/_iceberg/v1/{warehouse}/namespaces/{namespace}", GET_TABLE_NAMESPACE); + assert_action( + HttpMethod::Post, + "/iceberg/v1/{warehouse}/namespaces/{namespace}/properties", + UPDATE_TABLE_NAMESPACE_PROPERTIES, + ); + assert_action( + HttpMethod::Post, + "/_iceberg/v1/{warehouse}/namespaces/{namespace}/properties", + UPDATE_TABLE_NAMESPACE_PROPERTIES, + ); 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( diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index acc1cbaab..bdb28c98c 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -383,6 +383,11 @@ fn expected_admin_route_matrix() -> Vec { table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"), table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), table_route_sample(Method::HEAD, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), + table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/properties", + "/analytics/namespaces/sales/properties", + ), table_route_sample(Method::DELETE, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), table_route_sample( Method::GET, @@ -574,6 +579,11 @@ fn expected_admin_route_matrix() -> Vec { compat_table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"), compat_table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), compat_table_route_sample(Method::HEAD, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), + compat_table_route_sample( + Method::POST, + "/{warehouse}/namespaces/{namespace}/properties", + "/analytics/namespaces/sales/properties", + ), compat_table_route_sample(Method::DELETE, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"), compat_table_route_sample( Method::GET, diff --git a/rustfs/src/table_catalog/error.rs b/rustfs/src/table_catalog/error.rs index 9b2fef4e1..26208789c 100644 --- a/rustfs/src/table_catalog/error.rs +++ b/rustfs/src/table_catalog/error.rs @@ -60,6 +60,7 @@ pub(crate) enum TableCatalogStoreError { NotFound(String), Conflict(String), Invalid(String), + Unsupported(String), Internal(String), } @@ -69,6 +70,7 @@ impl fmt::Display for TableCatalogStoreError { Self::NotFound(message) => write!(f, "table catalog entry not found: {message}"), Self::Conflict(message) => write!(f, "table catalog conflict: {message}"), Self::Invalid(message) => write!(f, "invalid table catalog entry: {message}"), + Self::Unsupported(message) => write!(f, "unsupported table catalog operation: {message}"), Self::Internal(message) => write!(f, "table catalog store error: {message}"), } } diff --git a/rustfs/src/table_catalog/iceberg/commit.rs b/rustfs/src/table_catalog/iceberg/commit.rs index 1bc90bb13..89756743e 100644 --- a/rustfs/src/table_catalog/iceberg/commit.rs +++ b/rustfs/src/table_catalog/iceberg/commit.rs @@ -196,6 +196,7 @@ fn table_catalog_store_result_label(result: &TableCatalogStoreResult) -> & Err(TableCatalogStoreError::Conflict(_)) => "conflict", Err(TableCatalogStoreError::Invalid(_)) => "invalid", Err(TableCatalogStoreError::NotFound(_)) => "not_found", + Err(TableCatalogStoreError::Unsupported(_)) => "unsupported", Err(TableCatalogStoreError::Internal(_)) => "failure", } } diff --git a/rustfs/src/table_catalog/iceberg/validation.rs b/rustfs/src/table_catalog/iceberg/validation.rs index 24ab10003..eb55a98ed 100644 --- a/rustfs/src/table_catalog/iceberg/validation.rs +++ b/rustfs/src/table_catalog/iceberg/validation.rs @@ -191,28 +191,23 @@ where } let mut matched: Option = None; - for namespace in store.list_namespaces(bucket).await? { - if namespace.state != TableCatalogEntryState::Active { + for table in store.list_all_tables(bucket).await? { + if table.state != TableCatalogEntryState::Active { continue; } - for table in store.list_tables(bucket, &namespace.namespace).await? { - if table.state != TableCatalogEntryState::Active { - continue; - } - let Ok(warehouse_object_prefix) = table_warehouse_object_prefix(&table) else { - continue; - }; - if !object.starts_with(&warehouse_object_prefix) { - continue; - } - if matched - .as_ref() - .is_some_and(|current| current.warehouse_object_prefix.len() >= warehouse_object_prefix.len()) - { - continue; - } - matched = Some(table_data_plane_resource_from_entry(table, warehouse_object_prefix)); + let Ok(warehouse_object_prefix) = table_warehouse_object_prefix(&table) else { + continue; + }; + if !object.starts_with(&warehouse_object_prefix) { + continue; } + if matched + .as_ref() + .is_some_and(|current| current.warehouse_object_prefix.len() >= warehouse_object_prefix.len()) + { + continue; + } + matched = Some(table_data_plane_resource_from_entry(table, warehouse_object_prefix)); } Ok(matched) diff --git a/rustfs/src/table_catalog/identifier.rs b/rustfs/src/table_catalog/identifier.rs index b4576a72a..e4bad17a3 100644 --- a/rustfs/src/table_catalog/identifier.rs +++ b/rustfs/src/table_catalog/identifier.rs @@ -40,18 +40,28 @@ impl Namespace { pub const MAX_LEN: usize = 512; pub fn parse(value: &str) -> Result { - if value.is_empty() { - return Err(CatalogIdentifierError::Empty); - } if value.len() > Self::MAX_LEN { return Err(CatalogIdentifierError::NamespaceTooLong { max: Self::MAX_LEN }); } + Self::from_segments(value.split('.').map(str::to_string).collect()) + } - let mut segments = Vec::new(); - for segment in value.split('.') { - segments.push(IdentifierSegment::parse(segment.to_string())?); + pub(crate) fn from_segments(values: Vec) -> Result { + if values.is_empty() { + return Err(CatalogIdentifierError::Empty); + } + let value_len = values + .iter() + .try_fold(values.len().saturating_sub(1), |length, value| length.checked_add(value.len())) + .ok_or(CatalogIdentifierError::NamespaceTooLong { max: Self::MAX_LEN })?; + if value_len > Self::MAX_LEN { + return Err(CatalogIdentifierError::NamespaceTooLong { max: Self::MAX_LEN }); } + let segments = values + .into_iter() + .map(IdentifierSegment::parse) + .collect::, _>>()?; Ok(Self { segments }) } diff --git a/rustfs/src/table_catalog/mod.rs b/rustfs/src/table_catalog/mod.rs index b15ab9ef6..585847250 100644 --- a/rustfs/src/table_catalog/mod.rs +++ b/rustfs/src/table_catalog/mod.rs @@ -275,7 +275,7 @@ where TableCatalogListPage { entries, next_cursor } } -fn catalog_list_page_from_entries( +pub(crate) fn catalog_list_page_from_entries( mut entries: Vec, cursor: Option<&str>, limit: NonZeroUsize, diff --git a/rustfs/src/table_catalog/model.rs b/rustfs/src/table_catalog/model.rs index efc13e685..242ec71ba 100644 --- a/rustfs/src/table_catalog/model.rs +++ b/rustfs/src/table_catalog/model.rs @@ -71,6 +71,119 @@ pub(crate) struct NamespaceEntry { pub updated_at: Option, } +pub(crate) const NAMESPACE_PROPERTIES_MAX_ENTRIES: usize = 256; +pub(crate) const NAMESPACE_PROPERTY_KEY_MAX_LEN: usize = 256; +pub(crate) const NAMESPACE_PROPERTY_VALUE_MAX_LEN: usize = 4096; +pub(crate) const NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES: usize = 64 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct NamespacePropertiesUpdate { + removals: Vec, + updates: BTreeMap, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum NamespacePropertiesUpdateError { + DuplicateRemoval(String), + Overlap(String), +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct NamespacePropertiesUpdateResult { + pub updated: Vec, + pub removed: Vec, + pub missing: Vec, +} + +impl NamespacePropertiesUpdate { + pub(crate) fn try_new( + removals: Vec, + updates: BTreeMap, + ) -> Result { + let mut removal_keys = BTreeSet::new(); + for key in &removals { + if !removal_keys.insert(key.as_str()) { + return Err(NamespacePropertiesUpdateError::DuplicateRemoval(key.clone())); + } + if updates.contains_key(key) { + return Err(NamespacePropertiesUpdateError::Overlap(key.clone())); + } + } + Ok(Self { removals, updates }) + } + + pub(crate) fn apply_to(self, entry: &mut NamespaceEntry) -> NamespacePropertiesUpdateResult { + let updated = self.updates.keys().cloned().collect::>(); + for (key, value) in self.updates { + entry.properties.insert(key, value); + } + + let mut removed = Vec::new(); + let mut missing = Vec::new(); + for key in self.removals { + if entry.properties.remove(&key).is_some() { + removed.push(key); + } else { + missing.push(key); + } + } + + NamespacePropertiesUpdateResult { + updated, + removed, + missing, + } + } +} + +pub(crate) fn validate_namespace_properties(properties: &BTreeMap) -> TableCatalogStoreResult<()> { + if properties.len() > NAMESPACE_PROPERTIES_MAX_ENTRIES { + return Err(TableCatalogStoreError::Invalid(format!( + "namespace properties exceed the maximum of {NAMESPACE_PROPERTIES_MAX_ENTRIES} entries" + ))); + } + + let mut total_bytes = 0usize; + for (key, value) in properties { + if key.is_empty() || key.len() > NAMESPACE_PROPERTY_KEY_MAX_LEN { + return Err(TableCatalogStoreError::Invalid(format!( + "namespace property key length must be between 1 and {NAMESPACE_PROPERTY_KEY_MAX_LEN} bytes" + ))); + } + if value.len() > NAMESPACE_PROPERTY_VALUE_MAX_LEN { + return Err(TableCatalogStoreError::Invalid(format!( + "namespace property value exceeds {NAMESPACE_PROPERTY_VALUE_MAX_LEN} bytes" + ))); + } + total_bytes += key.len() + value.len(); + if total_bytes > NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES { + return Err(TableCatalogStoreError::Invalid(format!( + "namespace properties exceed {NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES} bytes" + ))); + } + } + Ok(()) +} + +pub(crate) fn namespace_is_descendant(candidate: &str, parent: &str) -> bool { + candidate.strip_prefix(parent).is_some_and(|suffix| suffix.starts_with('.')) +} + +pub(crate) fn synthetic_namespace_entry(table_bucket: &str, namespace: &Namespace) -> NamespaceEntry { + let namespace_id = namespace.storage_id(); + let namespace = namespace.public_name(); + NamespaceEntry { + version: TABLE_CATALOG_ENTRY_VERSION, + table_bucket: table_bucket.to_string(), + namespace, + namespace_id, + state: TableCatalogEntryState::Active, + properties: BTreeMap::new(), + created_at: None, + updated_at: None, + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub(crate) struct TableEntry { diff --git a/rustfs/src/table_catalog/store/migration.rs b/rustfs/src/table_catalog/store/migration.rs index c94178134..ae949c65d 100644 --- a/rustfs/src/table_catalog/store/migration.rs +++ b/rustfs/src/table_catalog/store/migration.rs @@ -12,7 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::object::ObjectTableCatalogStore; +use super::object::{ + ObjectTableCatalogStore, validate_namespace_entry_object, validate_table_entry_object, validate_view_entry_object, +}; use super::strong::{ StrongCommitSnapshotRecord, StrongTableCatalogBucketSnapshot, StrongTableCatalogState, TableCatalogBackingMigrationFence, TableCatalogBackingMigrationFenceStatus, TableCatalogBackingMigrationGlobalFence, table_catalog_bucket_snapshot_fingerprint, @@ -233,16 +235,6 @@ where .backend .list_objects(self.catalog_bucket(), &self.paths.namespace_entries_prefix(table_bucket)) .await?; - let mut unmatched_table_objects = namespace_objects - .iter() - .filter(|object| object.ends_with(TABLE_ENTRY_FILE)) - .cloned() - .collect::>(); - let mut unmatched_view_objects = namespace_objects - .iter() - .filter(|object| object.ends_with(VIEW_ENTRY_FILE)) - .cloned() - .collect::>(); for namespace_object in namespace_objects .iter() .filter(|object| object.ends_with(NAMESPACE_ENTRY_FILE)) @@ -260,128 +252,92 @@ where "namespace changed while preparing durable strong snapshot: {namespace_object}" ))); }; - if namespace_entry.table_bucket != table_bucket { - return Err(TableCatalogStoreError::Invalid(format!( - "namespace {} belongs to a different table bucket", - namespace_entry.namespace - ))); - } - let namespace = parse_namespace_for_store(&namespace_entry.namespace)?; - - let table_objects = self - .backend - .list_objects(self.catalog_bucket(), &self.paths.table_entries_prefix(table_bucket, &namespace)) - .await?; - for table_object in table_objects.iter().filter(|object| object.ends_with(TABLE_ENTRY_FILE)) { - unmatched_table_objects.remove(table_object); - guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), table_object).await?); - let Some((table_entry, _)) = self - .read_entry_unlocked::(self.catalog_bucket(), table_object) - .await? - else { - return Err(TableCatalogStoreError::Conflict(format!( - "table changed while preparing durable strong snapshot: {table_object}" - ))); - }; - if table_entry.table_bucket != table_bucket || table_entry.namespace != namespace_entry.namespace { - return Err(TableCatalogStoreError::Invalid(format!( - "table {} does not match its catalog namespace", - table_entry.table - ))); - } - - for commit_object in self - .backend - .list_objects( - self.catalog_bucket(), - &self.paths.commit_log_entries_prefix(table_bucket, &table_entry.table_id), - ) - .await? - .into_iter() - .filter(|object| object.ends_with(".json")) - { - let Some((commit, _)) = self - .read_entry_unlocked::(self.catalog_bucket(), &commit_object) - .await? - else { - return Err(TableCatalogStoreError::Conflict(format!( - "commit log changed while preparing durable strong snapshot: {commit_object}" - ))); - }; - commits.push(StrongCommitSnapshotRecord { - table_bucket: table_bucket.to_string(), - table_id: table_entry.table_id.clone(), - lookup_key: commit.commit_id.clone(), - commit, - }); - } - for idempotency_object in self - .backend - .list_objects( - self.catalog_bucket(), - &self - .paths - .commit_idempotency_entries_prefix(table_bucket, &table_entry.table_id), - ) - .await? - .into_iter() - .filter(|object| object.ends_with(".json")) - { - let Some((commit, _)) = self - .read_entry_unlocked::(self.catalog_bucket(), &idempotency_object) - .await? - else { - return Err(TableCatalogStoreError::Conflict(format!( - "idempotency index changed while preparing durable strong snapshot: {idempotency_object}" - ))); - }; - let lookup_key = commit.idempotency_key.clone().ok_or_else(|| { - TableCatalogStoreError::Invalid(format!("idempotency index {idempotency_object} has no idempotency key")) - })?; - idempotency.push(StrongCommitSnapshotRecord { - table_bucket: table_bucket.to_string(), - table_id: table_entry.table_id.clone(), - lookup_key, - commit, - }); - } - tables.push(table_entry); - } - - let view_objects = self - .backend - .list_objects(self.catalog_bucket(), &self.paths.view_entries_prefix(table_bucket, &namespace)) - .await?; - for view_object in view_objects.iter().filter(|object| object.ends_with(VIEW_ENTRY_FILE)) { - unmatched_view_objects.remove(view_object); - guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), view_object).await?); - let Some((view_entry, _)) = self - .read_entry_unlocked::(self.catalog_bucket(), view_object) - .await? - else { - return Err(TableCatalogStoreError::Conflict(format!( - "view changed while preparing durable strong snapshot: {view_object}" - ))); - }; - if view_entry.table_bucket != table_bucket || view_entry.namespace != namespace_entry.namespace { - return Err(TableCatalogStoreError::Invalid(format!( - "view {} does not match its catalog namespace", - view_entry.view - ))); - } - views.push(view_entry); - } + validate_namespace_entry_object(&self.paths, namespace_object, &namespace_entry)?; namespaces.push(namespace_entry); } - if let Some(object) = unmatched_table_objects.first() { - return Err(TableCatalogStoreError::Invalid(format!( - "table entry has no namespace entry during durable strong migration: {object}" - ))); + + for table_object in namespace_objects.iter().filter(|object| object.ends_with(TABLE_ENTRY_FILE)) { + guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), table_object).await?); + let Some((table_entry, _)) = self + .read_entry_unlocked::(self.catalog_bucket(), table_object) + .await? + else { + return Err(TableCatalogStoreError::Conflict(format!( + "table changed while preparing durable strong snapshot: {table_object}" + ))); + }; + validate_table_entry_object(&self.paths, table_object, &table_entry)?; + + for commit_object in self + .backend + .list_objects( + self.catalog_bucket(), + &self.paths.commit_log_entries_prefix(table_bucket, &table_entry.table_id), + ) + .await? + .into_iter() + .filter(|object| object.ends_with(".json")) + { + let Some((commit, _)) = self + .read_entry_unlocked::(self.catalog_bucket(), &commit_object) + .await? + else { + return Err(TableCatalogStoreError::Conflict(format!( + "commit log changed while preparing durable strong snapshot: {commit_object}" + ))); + }; + commits.push(StrongCommitSnapshotRecord { + table_bucket: table_bucket.to_string(), + table_id: table_entry.table_id.clone(), + lookup_key: commit.commit_id.clone(), + commit, + }); + } + for idempotency_object in self + .backend + .list_objects( + self.catalog_bucket(), + &self + .paths + .commit_idempotency_entries_prefix(table_bucket, &table_entry.table_id), + ) + .await? + .into_iter() + .filter(|object| object.ends_with(".json")) + { + let Some((commit, _)) = self + .read_entry_unlocked::(self.catalog_bucket(), &idempotency_object) + .await? + else { + return Err(TableCatalogStoreError::Conflict(format!( + "idempotency index changed while preparing durable strong snapshot: {idempotency_object}" + ))); + }; + let lookup_key = commit.idempotency_key.clone().ok_or_else(|| { + TableCatalogStoreError::Invalid(format!("idempotency index {idempotency_object} has no idempotency key")) + })?; + idempotency.push(StrongCommitSnapshotRecord { + table_bucket: table_bucket.to_string(), + table_id: table_entry.table_id.clone(), + lookup_key, + commit, + }); + } + tables.push(table_entry); } - if let Some(object) = unmatched_view_objects.first() { - return Err(TableCatalogStoreError::Invalid(format!( - "view entry has no namespace entry during durable strong migration: {object}" - ))); + + for view_object in namespace_objects.iter().filter(|object| object.ends_with(VIEW_ENTRY_FILE)) { + guards.push(self.backend.acquire_write_lock(self.catalog_bucket(), view_object).await?); + let Some((view_entry, _)) = self + .read_entry_unlocked::(self.catalog_bucket(), view_object) + .await? + else { + return Err(TableCatalogStoreError::Conflict(format!( + "view changed while preparing durable strong snapshot: {view_object}" + ))); + }; + validate_view_entry_object(&self.paths, view_object, &view_entry)?; + views.push(view_entry); } namespaces.sort_by(|left, right| left.namespace.cmp(&right.namespace)); @@ -495,10 +451,8 @@ where } } - let mut state = StrongTableCatalogState { - hydrated: true, - ..StrongTableCatalogState::default() - }; + let mut state = StrongTableCatalogState::default(); + state.hydrated = true; StrongTableCatalogStore::::insert_bucket_snapshot_locked(&mut state, snapshot.clone())?; if state.namespaces.len() != snapshot.namespaces.len() || state.tables.len() != snapshot.tables.len() @@ -521,7 +475,11 @@ where return Err(TableCatalogStoreError::NotFound(format!("table bucket {table_bucket}"))); } - let namespaces = self.list_namespaces(table_bucket).await?; + let namespace_objects = self + .backend + .list_objects(self.catalog_bucket(), &self.paths.namespace_entries_prefix(table_bucket)) + .await?; + let mut namespace_count: usize = 0; let mut table_count: usize = 0; let mut view_count: usize = 0; let mut commit_log_count: usize = 0; @@ -530,38 +488,58 @@ where let mut manual_review_count: usize = 0; let mut warehouse_prefix_owners = BTreeMap::::new(); - for namespace in &namespaces { - let tables = self.list_tables(table_bucket, &namespace.namespace).await?; - for table in tables { - table_count += 1; - if table.state == TableCatalogEntryState::Active { - let warehouse_prefix = table_warehouse_object_prefix(&table)?; - warehouse_prefix_owners - .entry(warehouse_prefix) - .and_modify(|count| *count = count.saturating_add(1)) - .or_insert(1); - } - - let recovery = self.table_commit_recovery_report_for_entry(&table, 0).await?; - commit_log_count = commit_log_count.saturating_add(recovery.commits.len()); - idempotency_index_count = idempotency_index_count.saturating_add( - self.backend - .list_objects( - self.catalog_bucket(), - &self.paths.commit_idempotency_entries_prefix(table_bucket, &table.table_id), - ) - .await? - .into_iter() - .filter(|object| object.ends_with(".json")) - .count(), - ); - recovery_required_count = recovery_required_count - .saturating_add(recovery.staged_before_table_update_count) - .saturating_add(recovery.finalization_required_count) - .saturating_add(recovery.idempotency_repair_required_count); - manual_review_count = manual_review_count.saturating_add(recovery.manual_review_count); + for object in namespace_objects { + if object.ends_with(NAMESPACE_ENTRY_FILE) { + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? else { + continue; + }; + validate_namespace_entry_object(&self.paths, &object, &entry)?; + namespace_count = namespace_count.saturating_add(1); + continue; } - view_count = view_count.saturating_add(self.list_views(table_bucket, &namespace.namespace).await?.len()); + if object.ends_with(VIEW_ENTRY_FILE) { + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? else { + continue; + }; + validate_view_entry_object(&self.paths, &object, &entry)?; + view_count = view_count.saturating_add(1); + continue; + } + if !object.ends_with(TABLE_ENTRY_FILE) { + continue; + } + + let Some((table, _)) = self.read_entry::(self.catalog_bucket(), &object).await? else { + continue; + }; + validate_table_entry_object(&self.paths, &object, &table)?; + table_count = table_count.saturating_add(1); + if table.state == TableCatalogEntryState::Active { + let warehouse_prefix = table_warehouse_object_prefix(&table)?; + warehouse_prefix_owners + .entry(warehouse_prefix) + .and_modify(|count| *count = count.saturating_add(1)) + .or_insert(1); + } + + let recovery = self.table_commit_recovery_report_for_entry(&table, 0).await?; + commit_log_count = commit_log_count.saturating_add(recovery.commits.len()); + idempotency_index_count = idempotency_index_count.saturating_add( + self.backend + .list_objects( + self.catalog_bucket(), + &self.paths.commit_idempotency_entries_prefix(table_bucket, &table.table_id), + ) + .await? + .into_iter() + .filter(|object| object.ends_with(".json")) + .count(), + ); + recovery_required_count = recovery_required_count + .saturating_add(recovery.staged_before_table_update_count) + .saturating_add(recovery.finalization_required_count) + .saturating_add(recovery.idempotency_repair_required_count); + manual_review_count = manual_review_count.saturating_add(recovery.manual_review_count); } let warehouse_index_ready = self.warehouse_index_ready(table_bucket).await?; @@ -633,7 +611,7 @@ where source_kind: TableCatalogBackingKind::ObjectBacked, target_kind: TableCatalogBackingKind::StrongKvWal, status, - namespace_count: namespaces.len(), + namespace_count, table_count, view_count, commit_log_count, diff --git a/rustfs/src/table_catalog/store/mod.rs b/rustfs/src/table_catalog/store/mod.rs index 3b5c3559b..84f71bf6d 100644 --- a/rustfs/src/table_catalog/store/mod.rs +++ b/rustfs/src/table_catalog/store/mod.rs @@ -24,6 +24,56 @@ pub(crate) use object::ObjectTableCatalogStore; pub(super) use strong::StrongTableCatalogSnapshot; pub(crate) use strong::StrongTableCatalogStore; +fn validate_namespace_entry_identity(entry: &NamespaceEntry) -> TableCatalogStoreResult { + validate_catalog_entry_version("namespace", entry.version)?; + let namespace = parse_namespace_for_store(&entry.namespace)?; + if entry.namespace_id != namespace.storage_id() { + return Err(TableCatalogStoreError::Invalid( + "catalog namespace entry storage identity does not match its namespace".to_string(), + )); + } + Ok(namespace) +} + +fn direct_namespace_children( + table_bucket: &str, + parent: Option<&Namespace>, + entries: Vec, +) -> TableCatalogStoreResult> { + let parent_depth = parent.map_or(0, |parent| parent.segments().len()); + let mut children = BTreeMap::new(); + for entry in entries { + if entry.table_bucket != table_bucket { + return Err(TableCatalogStoreError::Invalid( + "catalog namespace entry belongs to a different table bucket".to_string(), + )); + } + let namespace = validate_namespace_entry_identity(&entry)?; + if entry.state != TableCatalogEntryState::Active + || parent.is_some_and(|parent| !namespace.segments().starts_with(parent.segments())) + || namespace.segments().len() <= parent_depth + { + continue; + } + let child = Namespace::from_segments( + namespace.segments()[..=parent_depth] + .iter() + .map(|segment| segment.as_str().to_string()) + .collect(), + ) + .map_err(|err| TableCatalogStoreError::Invalid(format!("invalid catalog namespace child: {err}")))?; + let child_name = child.public_name(); + if namespace == child { + children.insert(child_name, entry); + } else { + children + .entry(child_name) + .or_insert_with(|| synthetic_namespace_entry(table_bucket, &child)); + } + } + Ok(children.into_values().collect()) +} + #[async_trait::async_trait] pub(crate) trait TableCatalogStore: Send + Sync { async fn get_table_bucket(&self, table_bucket: &str) -> TableCatalogStoreResult>; @@ -34,6 +84,55 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult>; + async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult> { + let parent = parse_namespace_for_store(parent)?.public_name(); + Ok(self + .list_namespaces(table_bucket) + .await? + .into_iter() + .filter(|entry| entry.namespace == parent || namespace_is_descendant(&entry.namespace, &parent)) + .collect()) + } + + async fn list_namespace_children( + &self, + table_bucket: &str, + parent: Option<&str>, + ) -> TableCatalogStoreResult> { + let parent = parent.map(parse_namespace_for_store).transpose()?; + if let Some(parent) = parent.as_ref() + && self + .get_namespace(table_bucket, &parent.public_name()) + .await? + .is_none_or(|entry| entry.state != TableCatalogEntryState::Active) + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + parent.public_name() + ))); + } + let entries = match parent.as_ref() { + Some(parent) => self.list_namespaces_under(table_bucket, &parent.public_name()).await?, + None => self.list_namespaces(table_bucket).await?, + }; + direct_namespace_children(table_bucket, parent.as_ref(), entries) + } + + async fn list_namespace_children_page( + &self, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + Ok(catalog_list_page_from_entries( + self.list_namespace_children(table_bucket, parent).await?, + cursor, + limit, + |entry| &entry.namespace, + )) + } + async fn list_namespaces_page( &self, table_bucket: &str, @@ -50,6 +149,17 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult>; + async fn update_namespace_properties( + &self, + _table_bucket: &str, + _namespace: &str, + _update: NamespacePropertiesUpdate, + ) -> TableCatalogStoreResult { + Err(TableCatalogStoreError::Unsupported( + "namespace property updates are not supported by this catalog store".to_string(), + )) + } + async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()>; async fn create_table(&self, entry: TableEntry) -> TableCatalogStoreResult<()>; @@ -58,6 +168,8 @@ pub(crate) trait TableCatalogStore: Send + Sync { async fn list_tables(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult>; + async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult>; + async fn list_tables_page( &self, table_bucket: &str, @@ -548,6 +660,37 @@ where } } + async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_namespaces_under(table_bucket, parent).await, + Self::DurableStrong(store) => store.list_namespaces_under(table_bucket, parent).await, + } + } + + async fn list_namespace_children( + &self, + table_bucket: &str, + parent: Option<&str>, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_namespace_children(table_bucket, parent).await, + Self::DurableStrong(store) => store.list_namespace_children(table_bucket, parent).await, + } + } + + async fn list_namespace_children_page( + &self, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_namespace_children_page(table_bucket, parent, cursor, limit).await, + Self::DurableStrong(store) => store.list_namespace_children_page(table_bucket, parent, cursor, limit).await, + } + } + async fn list_namespaces_page( &self, table_bucket: &str, @@ -567,6 +710,20 @@ where } } + async fn update_namespace_properties( + &self, + table_bucket: &str, + namespace: &str, + update: NamespacePropertiesUpdate, + ) -> TableCatalogStoreResult { + match self { + Self::ObjectBacked(_) => Err(TableCatalogStoreError::Unsupported( + "namespace property updates require durable-strong catalog backing".to_string(), + )), + Self::DurableStrong(store) => store.update_namespace_properties(table_bucket, namespace, update).await, + } + } + async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> { match self { Self::ObjectBacked(store) => store.drop_namespace(table_bucket, namespace).await, @@ -595,6 +752,13 @@ where } } + async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult> { + match self { + Self::ObjectBacked(store) => store.list_all_tables(table_bucket).await, + Self::DurableStrong(store) => store.list_all_tables(table_bucket).await, + } + } + async fn list_tables_page( &self, table_bucket: &str, diff --git a/rustfs/src/table_catalog/store/object.rs b/rustfs/src/table_catalog/store/object.rs index 4acac2539..afc0470c6 100644 --- a/rustfs/src/table_catalog/store/object.rs +++ b/rustfs/src/table_catalog/store/object.rs @@ -14,6 +14,57 @@ use super::*; +pub(super) fn validate_namespace_entry_object( + paths: &TableCatalogObjectPaths, + object: &str, + entry: &NamespaceEntry, +) -> TableCatalogStoreResult<()> { + let namespace = validate_namespace_entry_identity(entry)?; + if paths.namespace_entry_path(&entry.table_bucket, &namespace) != object { + return Err(TableCatalogStoreError::Invalid( + "catalog namespace entry identity does not match its object path".to_string(), + )); + } + Ok(()) +} + +pub(super) fn validate_table_entry_object( + paths: &TableCatalogObjectPaths, + object: &str, + entry: &TableEntry, +) -> TableCatalogStoreResult { + validate_catalog_entry_version("table", entry.version)?; + let namespace = parse_namespace_for_store(&entry.namespace)?; + let table = parse_table_for_store(&entry.table)?; + if paths.table_entry_path(&entry.table_bucket, &namespace, &table) != object { + return Err(TableCatalogStoreError::Invalid( + "catalog table entry identity does not match its object path".to_string(), + )); + } + Ok(namespace) +} + +pub(super) fn validate_view_entry_object( + paths: &TableCatalogObjectPaths, + object: &str, + entry: &ViewEntry, +) -> TableCatalogStoreResult { + validate_catalog_entry_version("view", entry.version)?; + let namespace = parse_namespace_for_store(&entry.namespace)?; + let view = parse_table_for_store(&entry.view)?; + if paths.view_entry_path(&entry.table_bucket, &namespace, &view) != object { + return Err(TableCatalogStoreError::Invalid( + "catalog view entry identity does not match its object path".to_string(), + )); + } + Ok(namespace) +} + +struct ActiveNamespaceEvidence { + namespace: Namespace, + explicit_entry: Option, +} + #[derive(Clone)] pub(crate) struct ObjectTableCatalogStore { pub(in crate::table_catalog) backend: B, @@ -35,15 +86,19 @@ where RUSTFS_META_BUCKET } - async fn list_entry_page( + async fn list_entry_page( &self, prefix: &str, entry_file: &str, cursor: Option<&str>, limit: NonZeroUsize, + include: P, + validate: V, ) -> TableCatalogStoreResult> where T: DeserializeOwned, + P: Fn(&T) -> bool, + V: Fn(&str, &T) -> TableCatalogStoreResult<()>, { let cursor = catalog_list_cursor(cursor, OBJECT_CATALOG_LIST_CURSOR_PREFIX)?; if cursor.is_some_and(|cursor| !cursor.starts_with(prefix)) { @@ -88,6 +143,10 @@ where let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? else { continue; }; + validate(&object, &entry)?; + if !include(&entry) { + continue; + } last_entry_path = Some(object); entries.push(entry); } @@ -100,6 +159,281 @@ where Ok(TableCatalogListPage { entries, next_cursor }) } + async fn read_active_namespace_evidence(&self, object: &str) -> TableCatalogStoreResult> { + if object.ends_with(NAMESPACE_ENTRY_FILE) { + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), object).await? else { + return Ok(None); + }; + validate_namespace_entry_object(&self.paths, object, &entry)?; + if entry.state != TableCatalogEntryState::Active { + return Ok(None); + } + return Ok(Some(ActiveNamespaceEvidence { + namespace: parse_namespace_for_store(&entry.namespace)?, + explicit_entry: Some(entry), + })); + } + if object.ends_with(TABLE_ENTRY_FILE) { + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), object).await? else { + return Ok(None); + }; + let namespace = validate_table_entry_object(&self.paths, object, &entry)?; + return Ok((entry.state == TableCatalogEntryState::Active).then_some(ActiveNamespaceEvidence { + namespace, + explicit_entry: None, + })); + } + if object.ends_with(VIEW_ENTRY_FILE) { + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), object).await? else { + return Ok(None); + }; + let namespace = validate_view_entry_object(&self.paths, object, &entry)?; + return Ok((entry.state == TableCatalogEntryState::Active).then_some(ActiveNamespaceEvidence { + namespace, + explicit_entry: None, + })); + } + Ok(None) + } + + async fn has_active_namespace_object(&self, table_bucket: &str, namespace: &Namespace) -> TableCatalogStoreResult { + let scan_limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS) + .ok_or_else(|| TableCatalogStoreError::Internal("catalog object scan limit must be positive".to_string()))?; + for prefix in [ + self.paths.table_entries_prefix(table_bucket, namespace), + self.paths.view_entries_prefix(table_bucket, namespace), + ] { + let mut cursor = None; + loop { + let page = self + .backend + .list_objects_page(self.catalog_bucket(), &prefix, cursor.as_deref(), scan_limit) + .await?; + let last_scanned = page.objects.last().cloned(); + for object in page.objects { + if self + .read_active_namespace_evidence(&object) + .await? + .is_some_and(|evidence| evidence.namespace == *namespace) + { + return Ok(true); + } + } + if !page.is_truncated { + break; + } + let next = last_scanned.ok_or_else(|| { + TableCatalogStoreError::Internal("catalog namespace object scan made no progress".to_string()) + })?; + if cursor.as_deref().is_some_and(|cursor| next.as_str() <= cursor) { + return Err(TableCatalogStoreError::Internal( + "catalog namespace object scan did not advance".to_string(), + )); + } + cursor = Some(next); + } + } + Ok(false) + } + + async fn has_active_namespace_descendant(&self, table_bucket: &str, namespace: &Namespace) -> TableCatalogStoreResult { + let parent = namespace.public_name(); + let descendant_prefix = format!("{}{}/", self.paths.namespace_entries_prefix(table_bucket), namespace.storage_id()); + let scan_limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS) + .ok_or_else(|| TableCatalogStoreError::Internal("catalog object scan limit must be positive".to_string()))?; + let mut cursor = None; + loop { + let page = self + .backend + .list_objects_page(self.catalog_bucket(), &descendant_prefix, cursor.as_deref(), scan_limit) + .await?; + let last_scanned = page.objects.last().cloned(); + for object in page.objects { + if self + .read_active_namespace_evidence(&object) + .await? + .is_some_and(|evidence| namespace_is_descendant(&evidence.namespace.public_name(), &parent)) + { + return Ok(true); + } + } + if !page.is_truncated { + return Ok(false); + } + let next = last_scanned.ok_or_else(|| { + TableCatalogStoreError::Internal("catalog namespace descendant scan made no progress".to_string()) + })?; + if cursor.as_deref().is_some_and(|cursor| next.as_str() <= cursor) { + return Err(TableCatalogStoreError::Internal( + "catalog namespace descendant scan did not advance".to_string(), + )); + } + cursor = Some(next); + } + } + + async fn require_active_namespace_unlocked( + &self, + table_bucket: &str, + namespace: &Namespace, + namespace_path: &str, + ) -> TableCatalogStoreResult<()> { + let current = self + .read_entry_unlocked::(self.catalog_bucket(), namespace_path) + .await?; + if let Some((entry, _)) = current.as_ref() { + validate_namespace_entry_object(&self.paths, namespace_path, entry)?; + if entry.state == TableCatalogEntryState::Active { + return Ok(()); + } + } + if !self.has_active_namespace_object(table_bucket, namespace).await? + && !self.has_active_namespace_descendant(table_bucket, namespace).await? + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + namespace.public_name() + ))); + } + Ok(()) + } + + async fn list_namespace_children_page_inner( + &self, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + let parent = parent.map(parse_namespace_for_store).transpose()?; + if let Some(parent) = parent.as_ref() + && self + .get_namespace(table_bucket, &parent.public_name()) + .await? + .is_none_or(|entry| entry.state != TableCatalogEntryState::Active) + { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + parent.public_name() + ))); + } + + let namespace_prefix = self.paths.namespace_entries_prefix(table_bucket); + let scan_prefix = match parent.as_ref() { + Some(parent) => format!("{namespace_prefix}{}/", parent.storage_id()), + None => namespace_prefix, + }; + let mut scan_cursor = catalog_list_cursor(cursor, OBJECT_CATALOG_LIST_CURSOR_PREFIX)?.map(str::to_string); + if scan_cursor.as_ref().is_some_and(|cursor| !cursor.starts_with(&scan_prefix)) { + return Err(TableCatalogStoreError::Invalid( + "page cursor does not match this namespace child list operation".to_string(), + )); + } + let scan_limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS) + .ok_or_else(|| TableCatalogStoreError::Internal("catalog object scan limit must be positive".to_string()))?; + let mut children = Vec::with_capacity(limit.get().saturating_add(1)); + + loop { + let page = self + .backend + .list_objects_page(self.catalog_bucket(), &scan_prefix, scan_cursor.as_deref(), scan_limit) + .await?; + let last_scanned = page.objects.last().cloned(); + let mut current_segment = None; + let mut current_segment_visible = false; + + for object in page.objects { + let Some(relative) = object.strip_prefix(&scan_prefix) else { + return Err(TableCatalogStoreError::Invalid( + "catalog namespace child object is outside its list prefix".to_string(), + )); + }; + let Some((segment, _)) = relative.split_once('/') else { + continue; + }; + if current_segment.as_deref() != Some(segment) { + current_segment = Some(segment.to_string()); + current_segment_visible = false; + } + if current_segment_visible { + continue; + } + let Some(evidence) = self.read_active_namespace_evidence(&object).await? else { + continue; + }; + let namespace = evidence.namespace; + if parent + .as_ref() + .is_some_and(|parent| !namespace.segments().starts_with(parent.segments())) + { + continue; + } + let parent_depth = parent.as_ref().map_or(0, |parent| parent.segments().len()); + if namespace.segments().len() <= parent_depth || namespace.segments()[parent_depth].as_str() != segment { + continue; + } + let child = Namespace::from_segments( + namespace.segments()[..=parent_depth] + .iter() + .map(|segment| segment.as_str().to_string()) + .collect(), + ) + .map_err(|err| TableCatalogStoreError::Invalid(format!("invalid catalog namespace child: {err}")))?; + let child_entry = evidence + .explicit_entry + .filter(|_| namespace == child) + .unwrap_or_else(|| synthetic_namespace_entry(table_bucket, &child)); + let child_cursor = format!("{OBJECT_CATALOG_LIST_CURSOR_PREFIX}{scan_prefix}{segment}/\u{10ffff}"); + children.push((child_entry, child_cursor)); + current_segment_visible = true; + if children.len() > limit.get() { + let next_cursor = children.get(limit.get().saturating_sub(1)).map(|(_, cursor)| cursor.clone()); + children.truncate(limit.get()); + return Ok(TableCatalogListPage { + entries: children.into_iter().map(|(entry, _)| entry).collect(), + next_cursor, + }); + } + } + + if !page.is_truncated { + return Ok(TableCatalogListPage { + entries: children.into_iter().map(|(entry, _)| entry).collect(), + next_cursor: None, + }); + } + let next = match (current_segment_visible, current_segment) { + (true, Some(segment)) => format!("{scan_prefix}{segment}/\u{10ffff}"), + _ => last_scanned.ok_or_else(|| { + TableCatalogStoreError::Internal("catalog namespace child scan made no progress".to_string()) + })?, + }; + if scan_cursor.as_deref().is_some_and(|cursor| next.as_str() <= cursor) { + return Err(TableCatalogStoreError::Internal( + "catalog namespace child scan did not advance".to_string(), + )); + } + scan_cursor = Some(next); + } + } + + async fn list_active_namespaces_with_prefix(&self, prefix: &str) -> TableCatalogStoreResult> { + let mut entries = Vec::new(); + for object in self.backend.list_objects(self.catalog_bucket(), prefix).await? { + if !object.ends_with(NAMESPACE_ENTRY_FILE) { + continue; + } + if let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? { + validate_namespace_entry_object(&self.paths, &object, &entry)?; + if entry.state == TableCatalogEntryState::Active { + entries.push(entry); + } + } + } + entries.sort_by(|left, right| left.namespace.cmp(&right.namespace)); + Ok(entries) + } + pub(in crate::table_catalog) async fn read_entry( &self, bucket: &str, @@ -534,31 +868,26 @@ where if self.read_warehouse_index_state_unlocked(table_bucket).await? { return Ok(()); } - for namespace in self.list_namespaces(table_bucket).await? { - if namespace.state != TableCatalogEntryState::Active { + for table in self.list_all_tables(table_bucket).await? { + if table.state != TableCatalogEntryState::Active { continue; } - for table in self.list_tables(table_bucket, &namespace.namespace).await? { - if table.state != TableCatalogEntryState::Active { + if let Err(err) = self + .backfill_active_table_warehouse_index(&table.table_bucket, &table.namespace, &table.table) + .await + { + if matches!(&err, TableCatalogStoreError::Invalid(_)) { + tracing::warn!( + table_bucket = %table.table_bucket, + namespace = %table.namespace, + table = %table.table, + table_id = %table.table_id, + error = %err, + "skipping invalid table warehouse location while backfilling warehouse index" + ); continue; } - if let Err(err) = self - .backfill_active_table_warehouse_index(&table.table_bucket, &table.namespace, &table.table) - .await - { - if matches!(&err, TableCatalogStoreError::Invalid(_)) { - tracing::warn!( - table_bucket = %table.table_bucket, - namespace = %table.namespace, - table = %table.table, - table_id = %table.table_id, - error = %err, - "skipping invalid table warehouse location while backfilling warehouse index" - ); - continue; - } - return Err(err); - } + return Err(err); } } self.write_warehouse_index_state_unlocked(table_bucket).await @@ -641,16 +970,8 @@ where .backend .acquire_write_lock(self.catalog_bucket(), &namespace_path) .await?; - if self - .read_entry_unlocked::(self.catalog_bucket(), &namespace_path) - .await? - .is_none() - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } + self.require_active_namespace_unlocked(&entry.table_bucket, &namespace, &namespace_path) + .await?; let table_path = self.paths.table_entry_path(&entry.table_bucket, &namespace, &table); let _table_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &table_path).await?; let reservation = self.reserve_table_warehouse_index(&entry).await?; @@ -676,16 +997,8 @@ where .backend .acquire_write_lock(self.catalog_bucket(), &namespace_path) .await?; - if self - .read_entry_unlocked::(self.catalog_bucket(), &namespace_path) - .await? - .is_none() - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } + self.require_active_namespace_unlocked(&entry.table_bucket, &namespace, &namespace_path) + .await?; let view_path = self.paths.view_entry_path(&entry.table_bucket, &namespace, &view); let _view_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &view_path).await?; self.write_entry_unlocked(self.catalog_bucket(), &view_path, &entry, precondition) @@ -3164,33 +3477,48 @@ where } async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> { - validate_catalog_entry_version("namespace", entry.version)?; + let namespace = validate_namespace_entry_identity(&entry)?; + validate_namespace_properties(&entry.properties)?; self.require_table_bucket(&entry.table_bucket).await?; - let namespace = parse_namespace_for_store(&entry.namespace)?; let _migration_guard = self.acquire_object_backed_catalog_write_permit(&entry.table_bucket).await?; let bucket_path = self.paths.table_bucket_entry_path(&entry.table_bucket); let _bucket_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &bucket_path).await?; let object = self.paths.namespace_entry_path(&entry.table_bucket, &namespace); - self.write_entry(self.catalog_bucket(), &object, &entry, TableCatalogPutPrecondition::IfAbsent) + let _namespace_guard = self.backend.acquire_write_lock(self.catalog_bucket(), &object).await?; + let precondition = match self + .read_entry_unlocked::(self.catalog_bucket(), &object) + .await? + { + Some((current, etag)) => { + validate_namespace_entry_object(&self.paths, &object, ¤t)?; + if current.state == TableCatalogEntryState::Active { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: namespace {}/{}", + entry.table_bucket, entry.namespace + ))); + } + etag.map(TableCatalogPutPrecondition::IfMatch) + .ok_or_else(|| TableCatalogStoreError::Internal(format!("catalog namespace entry has no etag: {object}")))? + } + None => { + if self.has_active_namespace_object(&entry.table_bucket, &namespace).await? + || self.has_active_namespace_descendant(&entry.table_bucket, &namespace).await? + { + return Err(TableCatalogStoreError::Conflict(format!( + "catalog object already exists: namespace {}/{}", + entry.table_bucket, entry.namespace + ))); + } + TableCatalogPutPrecondition::IfAbsent + } + }; + self.write_entry_unlocked(self.catalog_bucket(), &object, &entry, precondition) .await } async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult> { - let mut entries = Vec::new(); - for object in self - .backend - .list_objects(self.catalog_bucket(), &self.paths.namespace_entries_prefix(table_bucket)) - .await? - { - if !object.ends_with(NAMESPACE_ENTRY_FILE) { - continue; - } - if let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? { - entries.push(entry); - } - } - entries.sort_by(|left, right| left.namespace.cmp(&right.namespace)); - Ok(entries) + self.list_active_namespaces_with_prefix(&self.paths.namespace_entries_prefix(table_bucket)) + .await } async fn list_namespaces_page( @@ -3199,15 +3527,82 @@ where cursor: Option<&str>, limit: NonZeroUsize, ) -> TableCatalogStoreResult> { - self.list_entry_page(&self.paths.namespace_entries_prefix(table_bucket), NAMESPACE_ENTRY_FILE, cursor, limit) - .await + self.list_entry_page( + &self.paths.namespace_entries_prefix(table_bucket), + NAMESPACE_ENTRY_FILE, + cursor, + limit, + |entry: &NamespaceEntry| entry.state == TableCatalogEntryState::Active, + |object, entry: &NamespaceEntry| validate_namespace_entry_object(&self.paths, object, entry), + ) + .await } async fn get_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult> { let namespace = parse_namespace_for_store(namespace)?; - self.read_entry::(self.catalog_bucket(), &self.paths.namespace_entry_path(table_bucket, &namespace)) + let namespace_path = self.paths.namespace_entry_path(table_bucket, &namespace); + let exact = self + .read_entry::(self.catalog_bucket(), &namespace_path) + .await? + .map(|(entry, _)| entry); + if let Some(entry) = exact.as_ref() { + validate_namespace_entry_object(&self.paths, &namespace_path, entry)?; + } + if exact + .as_ref() + .is_some_and(|entry| entry.state == TableCatalogEntryState::Active) + { + return Ok(exact); + } + if self.has_active_namespace_object(table_bucket, &namespace).await? + || self.has_active_namespace_descendant(table_bucket, &namespace).await? + { + return Ok(Some(synthetic_namespace_entry(table_bucket, &namespace))); + } + Ok(None) + } + + async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult> { + let parent = parse_namespace_for_store(parent)?; + let prefix = format!("{}{}/", self.paths.namespace_entries_prefix(table_bucket), parent.storage_id()); + self.list_active_namespaces_with_prefix(&prefix).await + } + + async fn list_namespace_children( + &self, + table_bucket: &str, + parent: Option<&str>, + ) -> TableCatalogStoreResult> { + let limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS) + .ok_or_else(|| TableCatalogStoreError::Internal("catalog namespace list limit must be positive".to_string()))?; + let mut entries = Vec::new(); + let mut cursor = None; + loop { + let page = self + .list_namespace_children_page_inner(table_bucket, parent, cursor.as_deref(), limit) + .await?; + entries.extend(page.entries); + let Some(next_cursor) = page.next_cursor else { + return Ok(entries); + }; + if cursor.as_deref() == Some(next_cursor.as_str()) { + return Err(TableCatalogStoreError::Internal( + "catalog namespace child pagination did not advance".to_string(), + )); + } + cursor = Some(next_cursor); + } + } + + async fn list_namespace_children_page( + &self, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + self.list_namespace_children_page_inner(table_bucket, parent, cursor, limit) .await - .map(|entry| entry.map(|(namespace, _)| namespace)) } async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> { @@ -3221,11 +3616,30 @@ where .backend .acquire_write_lock(self.catalog_bucket(), &namespace_path) .await?; - if self + if self.has_active_namespace_descendant(table_bucket, &namespace).await? { + return Err(TableCatalogStoreError::Conflict(format!( + "namespace {table_bucket}/{} has child namespaces", + namespace.public_name() + ))); + } + if self.has_active_namespace_object(table_bucket, &namespace).await? { + return Err(TableCatalogStoreError::Conflict(format!( + "namespace {table_bucket}/{} is not empty", + namespace.public_name() + ))); + } + let current = self .read_entry_unlocked::(self.catalog_bucket(), &namespace_path) - .await? - .is_none() - { + .await?; + let Some((current, _)) = current else { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {}/{}", + table_bucket, + namespace.public_name() + ))); + }; + validate_namespace_entry_object(&self.paths, &namespace_path, ¤t)?; + if current.state != TableCatalogEntryState::Active { return Err(TableCatalogStoreError::NotFound(format!( "namespace {}/{}", table_bucket, @@ -3278,6 +3692,26 @@ where Ok(entries) } + async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult> { + let mut entries = Vec::new(); + for object in self + .backend + .list_objects(self.catalog_bucket(), &self.paths.namespace_entries_prefix(table_bucket)) + .await? + { + if !object.ends_with(TABLE_ENTRY_FILE) { + continue; + } + let Some((entry, _)) = self.read_entry::(self.catalog_bucket(), &object).await? else { + continue; + }; + validate_table_entry_object(&self.paths, &object, &entry)?; + entries.push(entry); + } + entries.sort_by(|left, right| (&left.namespace, &left.table).cmp(&(&right.namespace, &right.table))); + Ok(entries) + } + async fn list_tables_page( &self, table_bucket: &str, @@ -3291,6 +3725,8 @@ where TABLE_ENTRY_FILE, cursor, limit, + |_: &TableEntry| true, + |_, _: &TableEntry| Ok(()), ) .await } @@ -3698,8 +4134,15 @@ where limit: NonZeroUsize, ) -> TableCatalogStoreResult> { let namespace = parse_namespace_for_store(namespace)?; - self.list_entry_page(&self.paths.view_entries_prefix(table_bucket, &namespace), VIEW_ENTRY_FILE, cursor, limit) - .await + self.list_entry_page( + &self.paths.view_entries_prefix(table_bucket, &namespace), + VIEW_ENTRY_FILE, + cursor, + limit, + |_: &ViewEntry| true, + |_, _: &ViewEntry| Ok(()), + ) + .await } async fn load_view(&self, table_bucket: &str, namespace: &str, view: &str) -> TableCatalogStoreResult> { diff --git a/rustfs/src/table_catalog/store/strong.rs b/rustfs/src/table_catalog/store/strong.rs index acc0627ac..48c1c505d 100644 --- a/rustfs/src/table_catalog/store/strong.rs +++ b/rustfs/src/table_catalog/store/strong.rs @@ -17,6 +17,7 @@ use super::*; pub(in crate::table_catalog) type StrongNamespaceKey = (String, String); type StrongResourceKey = (String, String, String); type StrongCommitKey = (String, String, String); +type StrongNamespaceChildKey = (String, String, String); type StrongWarehouseIndex = BTreeMap>; #[derive(Clone, Default)] @@ -25,6 +26,8 @@ pub(in crate::table_catalog) struct StrongTableCatalogState { pub(super) snapshot_etag: Option, pub(super) table_buckets: BTreeMap, pub(in crate::table_catalog) namespaces: BTreeMap, + namespace_children: BTreeMap, + namespace_objects: BTreeSet, pub(super) tables: BTreeMap, pub(super) views: BTreeMap, pub(super) commits: BTreeMap, @@ -126,6 +129,93 @@ where (table_bucket.to_string(), namespace.public_name()) } + fn has_active_namespace_descendant_locked(state: &StrongTableCatalogState, table_bucket: &str, parent: &str) -> bool { + let range_start = (table_bucket.to_string(), parent.to_string(), String::new()); + state + .namespace_children + .range(range_start..) + .next() + .is_some_and(|((bucket, candidate_parent, _), _)| bucket == table_bucket && candidate_parent == parent) + } + + fn namespace_exists_locked(state: &StrongTableCatalogState, table_bucket: &str, namespace: &Namespace) -> bool { + let key = Self::namespace_key(table_bucket, namespace); + state + .namespaces + .get(&key) + .is_some_and(|entry| entry.state == TableCatalogEntryState::Active) + || state.namespace_objects.contains(&key) + || Self::has_active_namespace_descendant_locked(state, table_bucket, &namespace.public_name()) + } + + fn require_active_namespace_locked( + state: &StrongTableCatalogState, + table_bucket: &str, + namespace: &Namespace, + ) -> TableCatalogStoreResult<()> { + if Self::namespace_exists_locked(state, table_bucket, namespace) { + return Ok(()); + } + Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + namespace.public_name() + ))) + } + + fn list_namespace_children_page_locked( + state: &StrongTableCatalogState, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + let parent = parent.map(parse_namespace_for_store).transpose()?; + let parent_name = parent.as_ref().map_or_else(String::new, Namespace::public_name); + if let Some(parent) = parent.as_ref() + && !Self::namespace_exists_locked(state, table_bucket, parent) + { + return Err(TableCatalogStoreError::NotFound(format!("namespace {table_bucket}/{parent_name}"))); + } + + let cursor = catalog_list_cursor(cursor, STRONG_CATALOG_LIST_CURSOR_PREFIX)?; + let start = match cursor { + Some(cursor) => { + let child = parse_namespace_for_store(cursor)?; + let parent_depth = parent.as_ref().map_or(0, |parent| parent.segments().len()); + if child.segments().len() != parent_depth.saturating_add(1) + || parent + .as_ref() + .is_some_and(|parent| !child.segments().starts_with(parent.segments())) + { + return Err(TableCatalogStoreError::Invalid( + "page cursor does not match this namespace child list operation".to_string(), + )); + } + let sort_key = format!("{}/", child.segments()[parent_depth].as_str()); + Bound::Excluded((table_bucket.to_string(), parent_name.clone(), sort_key)) + } + None => Bound::Included((table_bucket.to_string(), parent_name.clone(), String::new())), + }; + let entries = state + .namespace_children + .range((start, Bound::Unbounded)) + .take_while(|((bucket, candidate_parent, _), _)| bucket == table_bucket && candidate_parent == &parent_name) + .take(limit.get().saturating_add(1)) + .map(|(_, child_name)| { + let child = parse_namespace_for_store(child_name)?; + Ok(state + .namespaces + .get(&Self::namespace_key(table_bucket, &child)) + .filter(|entry| entry.state == TableCatalogEntryState::Active) + .cloned() + .unwrap_or_else(|| synthetic_namespace_entry(table_bucket, &child))) + }) + .collect::>>()?; + Ok(finish_catalog_list_page(entries, limit, STRONG_CATALOG_LIST_CURSOR_PREFIX, |entry| { + &entry.namespace + })) + } + fn table_key(table_bucket: &str, namespace: &Namespace, table: &IdentifierSegment) -> StrongResourceKey { (table_bucket.to_string(), namespace.public_name(), table.as_str().to_string()) } @@ -225,6 +315,12 @@ where fn remove_bucket_from_state_locked(state: &mut StrongTableCatalogState, table_bucket: &str) { state.table_buckets.remove(table_bucket); state.namespaces.retain(|(entry_bucket, _), _| entry_bucket != table_bucket); + state + .namespace_children + .retain(|(entry_bucket, _, _), _| entry_bucket != table_bucket); + state + .namespace_objects + .retain(|(entry_bucket, _)| entry_bucket != table_bucket); state.tables.retain(|(entry_bucket, _, _), _| entry_bucket != table_bucket); state.views.retain(|(entry_bucket, _, _), _| entry_bucket != table_bucket); state.commits.retain(|(entry_bucket, _, _), _| entry_bucket != table_bucket); @@ -242,7 +338,7 @@ where Self::remove_bucket_from_state_locked(state, &table_bucket); state.table_buckets.insert(table_bucket.clone(), snapshot.table_bucket); for entry in snapshot.namespaces { - let namespace = parse_namespace_for_store(&entry.namespace)?; + let namespace = validate_namespace_entry_identity(&entry)?; state.namespaces.insert(Self::namespace_key(&table_bucket, &namespace), entry); } for entry in snapshot.tables { @@ -267,9 +363,65 @@ where record.commit, ); } + Self::rebuild_namespace_indexes_locked(state)?; Self::rebuild_warehouse_index_locked(state) } + fn index_namespace_children( + children: &mut BTreeMap, + table_bucket: &str, + namespace: &Namespace, + ) { + for depth in 0..namespace.segments().len() { + let parent = namespace.segments()[..depth] + .iter() + .map(IdentifierSegment::as_str) + .collect::>() + .join("."); + let child = namespace.segments()[..=depth] + .iter() + .map(IdentifierSegment::as_str) + .collect::>() + .join("."); + let sort_key = format!("{}/", namespace.segments()[depth].as_str()); + children.insert((table_bucket.to_string(), parent, sort_key), child); + } + } + + fn rebuild_namespace_indexes_locked(state: &mut StrongTableCatalogState) -> TableCatalogStoreResult<()> { + let mut children = BTreeMap::new(); + for entry in state + .namespaces + .values() + .filter(|entry| entry.state == TableCatalogEntryState::Active) + { + let namespace = validate_namespace_entry_identity(entry)?; + Self::index_namespace_children(&mut children, &entry.table_bucket, &namespace); + } + + let mut objects = BTreeSet::new(); + for (table_bucket, namespace_name) in state + .tables + .values() + .filter(|entry| entry.state == TableCatalogEntryState::Active) + .map(|entry| (&entry.table_bucket, &entry.namespace)) + .chain( + state + .views + .values() + .filter(|entry| entry.state == TableCatalogEntryState::Active) + .map(|entry| (&entry.table_bucket, &entry.namespace)), + ) + { + let namespace = parse_namespace_for_store(namespace_name)?; + objects.insert(Self::namespace_key(table_bucket, &namespace)); + Self::index_namespace_children(&mut children, table_bucket, &namespace); + } + state.namespace_children = children; + state.namespace_objects = objects; + Ok(()) + } + fn rebuild_warehouse_index_locked(state: &mut StrongTableCatalogState) -> TableCatalogStoreResult<()> { let mut warehouse_index: StrongWarehouseIndex = BTreeMap::new(); for ((table_bucket, namespace, table), entry) in &state.tables { @@ -283,11 +435,8 @@ where { continue; } - if !state - .namespaces - .get(&(table_bucket.clone(), namespace.clone())) - .is_some_and(|entry| entry.state == TableCatalogEntryState::Active) - { + let namespace_identity = parse_namespace_for_store(namespace)?; + if !Self::namespace_exists_locked(state, table_bucket, &namespace_identity) { continue; } let Ok(warehouse_object_prefix) = table_warehouse_object_prefix(entry) else { @@ -312,6 +461,7 @@ where fn snapshot_from_mutated_state_locked( state: &mut StrongTableCatalogState, ) -> TableCatalogStoreResult { + Self::rebuild_namespace_indexes_locked(state)?; Self::rebuild_warehouse_index_locked(state)?; Ok(Self::snapshot_from_state_locked(state)) } @@ -336,7 +486,7 @@ where state.table_buckets.insert(entry.table_bucket.clone(), entry); } for entry in snapshot.namespaces { - let namespace = parse_namespace_for_store(&entry.namespace)?; + let namespace = validate_namespace_entry_identity(&entry)?; state .namespaces .insert(Self::namespace_key(&entry.table_bucket, &namespace), entry); @@ -367,6 +517,7 @@ where record.commit, ); } + Self::rebuild_namespace_indexes_locked(&mut state)?; Self::rebuild_warehouse_index_locked(&mut state)?; Ok(state) } @@ -812,13 +963,16 @@ where async fn create_namespace(&self, entry: NamespaceEntry) -> TableCatalogStoreResult<()> { let _write_guard = self.write_lock.lock().await; self.hydrate_state().await?; - validate_catalog_entry_version("namespace", entry.version)?; - let namespace = parse_namespace_for_store(&entry.namespace)?; + let namespace = validate_namespace_entry_identity(&entry)?; + validate_namespace_properties(&entry.properties)?; let key = Self::namespace_key(&entry.table_bucket, &namespace); let (snapshot, precondition) = { let state = self.state.lock().await; Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if state.namespaces.contains_key(&key) { + let existing = state.namespaces.get(&key); + if existing.is_some_and(|entry| entry.state == TableCatalogEntryState::Active) + || (existing.is_none() && Self::namespace_exists_locked(&state, &entry.table_bucket, &namespace)) + { return Err(TableCatalogStoreError::Conflict(format!( "catalog object already exists: namespace {}/{}", entry.table_bucket, entry.namespace @@ -837,13 +991,71 @@ where let mut entries = state .namespaces .iter() - .filter(|((bucket, _), _)| bucket == table_bucket) + .filter(|((bucket, _), entry)| bucket == table_bucket && entry.state == TableCatalogEntryState::Active) .map(|(_, entry)| entry.clone()) .collect::>(); entries.sort_by(|left, right| left.namespace.cmp(&right.namespace)); Ok(entries) } + async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; + let parent = parse_namespace_for_store(parent)?.public_name(); + let state = self.state.lock().await; + let exact = state + .namespaces + .get(&(table_bucket.to_string(), parent.clone())) + .filter(|entry| entry.state == TableCatalogEntryState::Active) + .cloned(); + let descendant_start = (table_bucket.to_string(), format!("{parent}.")); + let descendants = state + .namespaces + .range(descendant_start..) + .take_while(|((bucket, namespace), _)| bucket == table_bucket && namespace_is_descendant(namespace, &parent)) + .filter(|(_, entry)| entry.state == TableCatalogEntryState::Active) + .map(|(_, entry)| entry.clone()) + .collect::>(); + Ok(exact.into_iter().chain(descendants).collect()) + } + + async fn list_namespace_children( + &self, + table_bucket: &str, + parent: Option<&str>, + ) -> TableCatalogStoreResult> { + let limit = NonZeroUsize::new(TABLE_CATALOG_LIST_MAX_KEYS) + .ok_or_else(|| TableCatalogStoreError::Internal("catalog namespace list limit must be positive".to_string()))?; + let mut entries = Vec::new(); + let mut cursor = None; + loop { + let page = self + .list_namespace_children_page(table_bucket, parent, cursor.as_deref(), limit) + .await?; + entries.extend(page.entries); + let Some(next_cursor) = page.next_cursor else { + return Ok(entries); + }; + if cursor.as_deref() == Some(next_cursor.as_str()) { + return Err(TableCatalogStoreError::Internal( + "catalog namespace child pagination did not advance".to_string(), + )); + } + cursor = Some(next_cursor); + } + } + + async fn list_namespace_children_page( + &self, + table_bucket: &str, + parent: Option<&str>, + cursor: Option<&str>, + limit: NonZeroUsize, + ) -> TableCatalogStoreResult> { + self.hydrate_state().await?; + let state = self.state.lock().await; + Self::list_namespace_children_page_locked(&state, table_bucket, parent, cursor, limit) + } + async fn list_namespaces_page( &self, table_bucket: &str, @@ -865,6 +1077,7 @@ where .namespaces .range((start, Bound::Unbounded)) .take_while(|((bucket, _), _)| bucket == table_bucket) + .filter(|(_, entry)| entry.state == TableCatalogEntryState::Active) .take(limit.get().saturating_add(1)) .map(|(_, entry)| entry.clone()) .collect(); @@ -877,7 +1090,60 @@ where self.hydrate_state().await?; let namespace = parse_namespace_for_store(namespace)?; let state = self.state.lock().await; - Ok(state.namespaces.get(&Self::namespace_key(table_bucket, &namespace)).cloned()) + let exact = state.namespaces.get(&Self::namespace_key(table_bucket, &namespace)).cloned(); + if exact + .as_ref() + .is_some_and(|entry| entry.state == TableCatalogEntryState::Active) + { + return Ok(exact); + } + if Self::namespace_exists_locked(&state, table_bucket, &namespace) { + return Ok(Some(synthetic_namespace_entry(table_bucket, &namespace))); + } + Ok(None) + } + + async fn update_namespace_properties( + &self, + table_bucket: &str, + namespace: &str, + update: NamespacePropertiesUpdate, + ) -> TableCatalogStoreResult { + let _write_guard = self.write_lock.lock().await; + self.hydrate_state().await?; + let namespace = parse_namespace_for_store(namespace)?; + let key = Self::namespace_key(table_bucket, &namespace); + let (snapshot, precondition, result) = { + let state = self.state.lock().await; + let current = state + .namespaces + .get(&key) + .filter(|entry| entry.state == TableCatalogEntryState::Active); + let mut next = match current { + Some(current) => current.clone(), + None if Self::namespace_exists_locked(&state, table_bucket, &namespace) => { + synthetic_namespace_entry(table_bucket, &namespace) + } + None => { + return Err(TableCatalogStoreError::NotFound(format!( + "namespace {table_bucket}/{}", + namespace.public_name() + ))); + } + }; + let result = update.apply_to(&mut next); + validate_namespace_properties(&next.properties)?; + let unchanged = + current.map_or_else(|| next == synthetic_namespace_entry(table_bucket, &namespace), |current| &next == current); + if unchanged { + return Ok(result); + } + let (precondition, mut draft_state) = Self::snapshot_draft_context_locked(&state); + draft_state.namespaces.insert(key, next); + (Self::snapshot_from_mutated_state_locked(&mut draft_state)?, precondition, result) + }; + self.finalize_snapshot_write(snapshot, precondition).await?; + Ok(result) } async fn drop_namespace(&self, table_bucket: &str, namespace: &str) -> TableCatalogStoreResult<()> { @@ -887,7 +1153,22 @@ where let key = Self::namespace_key(table_bucket, &namespace); let (snapshot, precondition) = { let state = self.state.lock().await; - if !state.namespaces.contains_key(&key) { + let parent = namespace.public_name(); + if Self::has_active_namespace_descendant_locked(&state, table_bucket, &parent) { + return Err(TableCatalogStoreError::Conflict(format!( + "namespace {table_bucket}/{parent} has child namespaces" + ))); + } + if state.namespace_objects.contains(&key) { + return Err(TableCatalogStoreError::Conflict(format!( + "namespace {table_bucket}/{parent} is not empty" + ))); + } + if !state + .namespaces + .get(&key) + .is_some_and(|entry| entry.state == TableCatalogEntryState::Active) + { return Err(TableCatalogStoreError::NotFound(format!( "namespace {}/{}", table_bucket, @@ -931,15 +1212,7 @@ where let (snapshot, precondition) = { let state = self.state.lock().await; Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if !state - .namespaces - .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } + Self::require_active_namespace_locked(&state, &entry.table_bucket, &namespace)?; if state.tables.contains_key(&key) { return Err(TableCatalogStoreError::Conflict(format!( "catalog object already exists: table {}/{}/{}", @@ -968,6 +1241,17 @@ where Ok(entries) } + async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult> { + self.hydrate_state().await?; + let state = self.state.lock().await; + Ok(state + .tables + .range((table_bucket.to_string(), String::new(), String::new())..) + .take_while(|((bucket, _, _), _)| bucket == table_bucket) + .map(|(_, entry)| entry.clone()) + .collect()) + } + async fn list_tables_page( &self, table_bucket: &str, @@ -1177,15 +1461,7 @@ where let (snapshot, precondition) = { let state = self.state.lock().await; Self::require_table_bucket_in_state(&state, &entry.table_bucket)?; - if !state - .namespaces - .contains_key(&Self::namespace_key(&entry.table_bucket, &namespace)) - { - return Err(TableCatalogStoreError::NotFound(format!( - "namespace {}/{}", - entry.table_bucket, entry.namespace - ))); - } + Self::require_active_namespace_locked(&state, &entry.table_bucket, &namespace)?; if state.views.contains_key(&key) { return Err(TableCatalogStoreError::Conflict(format!( "catalog object already exists: view {}/{}/{}", diff --git a/rustfs/src/table_catalog/tests.rs b/rustfs/src/table_catalog/tests.rs index 740661da7..019755c0a 100644 --- a/rustfs/src/table_catalog/tests.rs +++ b/rustfs/src/table_catalog/tests.rs @@ -234,6 +234,10 @@ impl TableCatalogStore for NoopTableCatalogStore { Ok(Vec::new()) } + async fn list_all_tables(&self, _table_bucket: &str) -> TableCatalogStoreResult> { + Ok(Vec::new()) + } + async fn load_table( &self, _table_bucket: &str, @@ -1457,6 +1461,50 @@ async fn table_data_plane_resource_skips_stale_deeper_index_and_matches_parent() assert_eq!(backend.list_call_count().await, 0); } +#[tokio::test] +async fn object_catalog_rebuilds_warehouse_index_for_table_backed_namespace() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let current = default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"); + let object = "tables/table-id/data/part-00001.parquet"; + + seed_table_for_metadata_maintenance(&store, bucket, &namespace, &table, current).await; + backend + .delete_object(RUSTFS_META_BUCKET, &store.paths.namespace_entry_path(bucket, &namespace)) + .await + .expect("namespace marker should be removed"); + + let scanned = scan_table_data_plane_resource_for_object(&store, bucket, object) + .await + .expect("resource scan should succeed") + .expect("table entry should keep its namespace discoverable"); + assert_eq!(scanned.table, "orders"); + + backend + .delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_entry_path(bucket, "tables/table-id/")) + .await + .expect("warehouse index entry should be removed"); + backend + .delete_object(RUSTFS_META_BUCKET, &store.paths.warehouse_index_state_path(bucket)) + .await + .expect("warehouse index state should be removed"); + + let rebuilt = table_data_plane_resource_for_object(&store, bucket, object) + .await + .expect("resource lookup should rebuild the index") + .expect("rebuilt index should resolve the table"); + assert_eq!(rebuilt.table, "orders"); + assert!( + store + .warehouse_index_ready(bucket) + .await + .expect("warehouse index state should load") + ); +} + #[tokio::test] async fn object_table_catalog_store_rejects_duplicate_warehouse_prefix() { let backend = TestCatalogObjectBackend::default(); @@ -6055,7 +6103,7 @@ async fn durable_strong_migration_dry_run_reports_ready_catalog_inventory() { } #[tokio::test] -async fn durable_strong_migration_rejects_orphan_table_without_namespace() { +async fn durable_strong_migration_preserves_table_backed_namespace() { let backend = TestCatalogObjectBackend::default(); let object_store = ObjectTableCatalogStore::new(backend.clone()); let bucket = "analytics"; @@ -6073,27 +6121,51 @@ async fn durable_strong_migration_rejects_orphan_table_without_namespace() { object_store .load_table(bucket, &namespace.public_name(), table.as_str()) .await - .unwrap() + .expect("table-backed namespace lookup should succeed") .is_some() ); + let dry_run = object_store + .plan_durable_strong_backing_migration(bucket) + .await + .expect("table-backed namespace dry run should succeed"); + assert_eq!(dry_run.namespace_count, 0); + assert_eq!(dry_run.table_count, 1); - let error = object_store + let materialized = object_store .materialize_durable_strong_backing_migration(bucket) .await - .unwrap_err(); - assert_matches!( - error, - TableCatalogStoreError::Invalid(message) if message.contains("table entry has no namespace entry") + .expect("table-backed namespace should migrate"); + assert_eq!(materialized.namespace_count, 0); + assert_eq!(materialized.table_count, 1); + let strong_store = StrongTableCatalogStore::new(backend); + assert!( + strong_store + .get_namespace(bucket, &namespace.public_name()) + .await + .expect("table-backed namespace should load") + .is_some() + ); + assert!( + strong_store + .load_table(bucket, &namespace.public_name(), table.as_str()) + .await + .expect("migrated table should load") + .is_some() + ); + assert_eq!( + strong_store + .list_namespace_children(bucket, None) + .await + .expect("table-backed namespace children should list") + .iter() + .map(|entry| entry.namespace.as_str()) + .collect::>(), + ["sales"] ); - object_store - .create_namespace(test_namespace_entry(bucket, &namespace)) - .await - .expect("failed migration must leave source catalog writable"); - object_store.put_table_bucket(test_bucket_entry("research")).await.unwrap(); } #[tokio::test] -async fn durable_strong_migration_rejects_orphan_view_without_namespace() { +async fn durable_strong_migration_preserves_view_backed_namespace() { let backend = TestCatalogObjectBackend::default(); let object_store = ObjectTableCatalogStore::new(backend.clone()); let bucket = "analytics"; @@ -6121,22 +6193,37 @@ async fn durable_strong_migration_rejects_orphan_view_without_namespace() { object_store .load_view(bucket, &namespace.public_name(), view.as_str()) .await - .unwrap() + .expect("view-backed namespace lookup should succeed") .is_some() ); + let dry_run = object_store + .plan_durable_strong_backing_migration(bucket) + .await + .expect("view-backed namespace dry run should succeed"); + assert_eq!(dry_run.namespace_count, 0); + assert_eq!(dry_run.view_count, 1); - let error = object_store + let materialized = object_store .materialize_durable_strong_backing_migration(bucket) .await - .unwrap_err(); - assert_matches!( - error, - TableCatalogStoreError::Invalid(message) if message.contains("view entry has no namespace entry") + .expect("view-backed namespace should migrate"); + assert_eq!(materialized.namespace_count, 0); + assert_eq!(materialized.view_count, 1); + let strong_store = StrongTableCatalogStore::new(backend); + assert!( + strong_store + .get_namespace(bucket, &namespace.public_name()) + .await + .expect("view-backed namespace should load") + .is_some() + ); + assert!( + strong_store + .load_view(bucket, &namespace.public_name(), view.as_str()) + .await + .expect("migrated view should load") + .is_some() ); - object_store - .create_namespace(test_namespace_entry(bucket, &namespace)) - .await - .expect("failed migration must leave source catalog writable"); } #[tokio::test] @@ -6558,10 +6645,16 @@ async fn durable_strong_migration_dry_run_reports_recovery_blockers() { .paths .commit_idempotency_entry_path(bucket, "table-id", "client-request"); backend.delete_object(RUSTFS_META_BUCKET, &idempotency_path).await.unwrap(); + backend + .delete_object(RUSTFS_META_BUCKET, &store.paths.namespace_entry_path(bucket, &namespace)) + .await + .unwrap(); let report = store.plan_durable_strong_backing_migration(bucket).await.unwrap(); assert_eq!(report.status, TableCatalogBackingMigrationStatus::RecoveryRequired); + assert_eq!(report.namespace_count, 0); + assert_eq!(report.table_count, 1); assert_eq!(report.commit_log_count, 1); assert_eq!(report.idempotency_index_count, 0); assert!( @@ -8560,6 +8653,816 @@ fn namespace_length_is_bounded_for_catalog_paths_and_page_tokens() { ); } +#[test] +fn namespace_from_segments_preserves_rest_boundaries_and_length_limit() { + let namespace = Namespace::from_segments(vec!["analytics".to_string(), "daily_events".to_string()]) + .expect("multipart namespace should parse"); + assert_eq!(namespace.public_name(), "analytics.daily_events"); + assert_eq!(synthetic_namespace_entry("warehouse", &namespace).namespace_id, "analytics/daily_events"); + assert!(Namespace::from_segments(vec!["analytics.daily_events".to_string()]).is_err()); + + let mut segments = vec!["a".repeat(63); 8]; + segments[0].push('a'); + Namespace::from_segments(segments.clone()).expect("namespace at the maximum length should parse"); + segments.push("a".to_string()); + assert_eq!( + Namespace::from_segments(segments), + Err(CatalogIdentifierError::NamespaceTooLong { max: Namespace::MAX_LEN }) + ); +} + +#[test] +fn namespace_property_update_and_limits_reject_ambiguous_or_oversized_state() { + let overlap = NamespacePropertiesUpdate::try_new( + vec!["owner".to_string()], + BTreeMap::from([("owner".to_string(), "platform".to_string())]), + ) + .expect_err("overlapping update should fail"); + assert_matches!(overlap, NamespacePropertiesUpdateError::Overlap(key) if key == "owner"); + + let duplicate = NamespacePropertiesUpdate::try_new(vec!["owner".to_string(), "owner".to_string()], BTreeMap::new()) + .expect_err("duplicate removal should fail"); + assert_matches!(duplicate, NamespacePropertiesUpdateError::DuplicateRemoval(key) if key == "owner"); + + let mut exact_total = BTreeMap::new(); + for index in 0..15 { + exact_total.insert(format!("k{index:02}"), "v".repeat(NAMESPACE_PROPERTY_VALUE_MAX_LEN)); + } + let used = exact_total.iter().map(|(key, value)| key.len() + value.len()).sum::(); + let final_key = "k15".to_string(); + exact_total.insert( + final_key.clone(), + "v".repeat(NAMESPACE_PROPERTIES_MAX_TOTAL_BYTES - used - final_key.len()), + ); + assert!(validate_namespace_properties(&exact_total).is_ok()); + + exact_total + .get_mut(&final_key) + .expect("final property should exist") + .push('v'); + assert_matches!(validate_namespace_properties(&exact_total), Err(TableCatalogStoreError::Invalid(_))); + assert_matches!( + validate_namespace_properties(&BTreeMap::from([(String::new(), "value".to_string())])), + Err(TableCatalogStoreError::Invalid(_)) + ); + assert_matches!( + validate_namespace_properties(&BTreeMap::from([("k".repeat(NAMESPACE_PROPERTY_KEY_MAX_LEN + 1), "value".to_string(),)])), + Err(TableCatalogStoreError::Invalid(_)) + ); + assert_matches!( + validate_namespace_properties(&BTreeMap::from([ + ("owner".to_string(), "v".repeat(NAMESPACE_PROPERTY_VALUE_MAX_LEN + 1),) + ])), + Err(TableCatalogStoreError::Invalid(_)) + ); + let too_many = (0..=NAMESPACE_PROPERTIES_MAX_ENTRIES) + .map(|index| (format!("key{index}"), "value".to_string())) + .collect(); + assert_matches!(validate_namespace_properties(&too_many), Err(TableCatalogStoreError::Invalid(_))); +} + +#[tokio::test] +async fn configured_object_catalog_rejects_namespace_property_update_without_mutation() { + let backend = TestCatalogObjectBackend::default(); + let store = ConfiguredTableCatalogStore::new(backend, TableCatalogBackingMode::ObjectBacked); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let mut entry = test_namespace_entry(bucket, &namespace); + entry.properties = BTreeMap::from([("owner".to_string(), "lakehouse".to_string())]); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + store.create_namespace(entry).await.expect("namespace should be created"); + + let result = store + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect_err("object-backed namespace property update should be unsupported"); + assert_matches!(result, TableCatalogStoreError::Unsupported(_)); + let stored = store + .get_namespace(bucket, "sales") + .await + .expect("namespace lookup should succeed") + .expect("namespace should remain"); + assert_eq!(stored.properties.get("owner").map(String::as_str), Some("lakehouse")); +} + +#[tokio::test] +async fn strong_catalog_namespace_property_update_survives_restart_and_failed_write() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket entry should be seeded"); + let mut entry = test_namespace_entry(bucket, &namespace); + entry.properties.insert("owner".to_string(), "lakehouse".to_string()); + store.create_namespace(entry).await.expect("namespace should be created"); + + let result = store + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new( + vec!["missing".to_string()], + BTreeMap::from([("owner".to_string(), "platform".to_string())]), + ) + .expect("namespace update should validate"), + ) + .await + .expect("namespace properties should update"); + assert_eq!(result.updated, vec!["owner".to_string()]); + assert_eq!(result.missing, vec!["missing".to_string()]); + + let restarted = StrongTableCatalogStore::new(backend.clone()); + let stored = restarted + .get_namespace(bucket, "sales") + .await + .expect("namespace lookup after restart should succeed") + .expect("namespace should survive restart"); + assert_eq!(stored.properties.get("owner").map(String::as_str), Some("platform")); + + backend + .fail_next_put( + RUSTFS_META_BUCKET, + &StrongTableCatalogStore::::snapshot_object_path(), + ) + .await; + let no_op = restarted + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect("unchanged namespace properties should not write a snapshot"); + assert_eq!(no_op.updated, vec!["owner".to_string()]); + let error = restarted + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "failed-update".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect_err("failed snapshot write should fail namespace update"); + assert_matches!(error, TableCatalogStoreError::Internal(_)); + + let after_failure = StrongTableCatalogStore::new(backend) + .get_namespace(bucket, "sales") + .await + .expect("namespace lookup after failed write should succeed") + .expect("namespace should remain"); + assert_eq!(after_failure.properties.get("owner").map(String::as_str), Some("platform")); +} + +#[tokio::test] +async fn namespace_properties_load_legacy_values_but_reject_oversized_writes() { + let backend = TestCatalogObjectBackend::default(); + let object_store = ObjectTableCatalogStore::new(backend.clone()); + let strong_store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + object_store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("object-backed bucket should be created"); + strong_store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("strong bucket should be created"); + + let mut oversized = test_namespace_entry(bucket, &namespace); + oversized + .properties + .insert("owner".to_string(), "x".repeat(NAMESPACE_PROPERTY_VALUE_MAX_LEN + 1)); + assert_matches!( + object_store.create_namespace(oversized.clone()).await, + Err(TableCatalogStoreError::Invalid(_)) + ); + assert_matches!( + strong_store.create_namespace(oversized.clone()).await, + Err(TableCatalogStoreError::Invalid(_)) + ); + + backend + .seed_object( + RUSTFS_META_BUCKET, + &object_store.paths.namespace_entry_path(bucket, &namespace), + serde_json::to_vec(&oversized).expect("legacy namespace should serialize"), + ) + .await; + let legacy = object_store + .get_namespace(bucket, "sales") + .await + .expect("legacy namespace lookup should succeed") + .expect("legacy namespace should remain readable"); + assert_eq!( + legacy.properties.get("owner").map(String::len), + Some(NAMESPACE_PROPERTY_VALUE_MAX_LEN + 1) + ); + + strong_store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + assert_matches!( + strong_store + .update_namespace_properties( + bucket, + "sales", + NamespacePropertiesUpdate::try_new(Vec::new(), oversized.properties) + .expect("property update request should validate structurally"), + ) + .await, + Err(TableCatalogStoreError::Invalid(_)) + ); + let stored = strong_store + .get_namespace(bucket, "sales") + .await + .expect("namespace lookup should succeed") + .expect("namespace should remain"); + assert!(stored.properties.is_empty()); +} + +#[tokio::test] +async fn object_catalog_rejects_mismatched_namespace_storage_identity() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales.daily").expect("namespace should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("object-backed bucket should be created"); + let mut entry = test_namespace_entry(bucket, &namespace); + entry.namespace_id = "sales.daily".to_string(); + backend + .seed_object( + RUSTFS_META_BUCKET, + &store.paths.namespace_entry_path(bucket, &namespace), + serde_json::to_vec(&entry).expect("namespace should serialize"), + ) + .await; + + assert_matches!(store.get_namespace(bucket, "sales.daily").await, Err(TableCatalogStoreError::Invalid(_))); +} + +#[tokio::test] +async fn strong_catalog_rejects_mismatched_namespace_storage_identity() { + let backend = TestCatalogObjectBackend::default(); + let store = StrongTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales.daily").expect("namespace should parse"); + let mut entry = test_namespace_entry(bucket, &namespace); + entry.namespace_id = "sales.daily".to_string(); + let snapshot = StrongTableCatalogSnapshot { + version: STRONG_TABLE_CATALOG_SNAPSHOT_VERSION, + table_buckets: vec![test_bucket_entry(bucket)], + namespaces: vec![entry], + tables: Vec::new(), + views: Vec::new(), + commits: Vec::new(), + idempotency: Vec::new(), + }; + backend + .seed_object( + RUSTFS_META_BUCKET, + &StrongTableCatalogStore::::snapshot_object_path(), + serde_json::to_vec(&snapshot).expect("strong snapshot should encode"), + ) + .await; + + assert_matches!(store.get_namespace(bucket, "sales.daily").await, Err(TableCatalogStoreError::Invalid(_))); +} + +#[tokio::test] +async fn catalog_backings_expose_implicit_parents_and_protect_child_namespaces() { + let bucket = "analytics"; + let parent = Namespace::parse("sales").expect("parent namespace should parse"); + let child = Namespace::parse("sales.daily").expect("child namespace should parse"); + + let object_backend = TestCatalogObjectBackend::default(); + let object_store = ObjectTableCatalogStore::new(object_backend.clone()); + object_store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("object-backed bucket should be created"); + object_store + .create_namespace(test_namespace_entry(bucket, &child)) + .await + .expect("object-backed child namespace should be created"); + let implicit_parent = object_store + .get_namespace(bucket, &parent.public_name()) + .await + .expect("object-backed parent lookup should succeed") + .expect("implicit object-backed parent should exist"); + assert_eq!(implicit_parent.namespace, parent.public_name()); + assert!(implicit_parent.properties.is_empty()); + assert_matches!( + object_store.drop_namespace(bucket, &parent.public_name()).await, + Err(TableCatalogStoreError::Conflict(_)) + ); + assert_matches!( + object_store.create_namespace(test_namespace_entry(bucket, &parent)).await, + Err(TableCatalogStoreError::Conflict(_)) + ); + let mut inactive_parent = test_namespace_entry(bucket, &parent); + inactive_parent.state = TableCatalogEntryState::Deleted; + object_backend + .seed_object( + RUSTFS_META_BUCKET, + &object_store.paths.namespace_entry_path(bucket, &parent), + serde_json::to_vec(&inactive_parent).expect("inactive parent should serialize"), + ) + .await; + object_store + .create_namespace(test_namespace_entry(bucket, &parent)) + .await + .expect("inactive object-backed parent should be replaceable"); + assert_eq!( + object_store + .list_namespaces_under(bucket, &parent.public_name()) + .await + .expect("object-backed descendants should list") + .len(), + 2 + ); + + let strong_backend = TestCatalogObjectBackend::default(); + let strong_store = StrongTableCatalogStore::new(strong_backend.clone()); + strong_store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("strong bucket should be created"); + strong_store + .create_namespace(test_namespace_entry(bucket, &child)) + .await + .expect("strong child namespace should be created"); + let implicit_parent = strong_store + .get_namespace(bucket, &parent.public_name()) + .await + .expect("strong parent lookup should succeed") + .expect("implicit strong parent should exist"); + assert_eq!(implicit_parent.namespace, parent.public_name()); + assert_matches!( + strong_store.create_namespace(test_namespace_entry(bucket, &parent)).await, + Err(TableCatalogStoreError::Conflict(_)) + ); + strong_store + .update_namespace_properties( + bucket, + &parent.public_name(), + NamespacePropertiesUpdate::try_new(Vec::new(), BTreeMap::from([("owner".to_string(), "platform".to_string())])) + .expect("namespace update should validate"), + ) + .await + .expect("implicit strong parent should materialize on property update"); + let restarted = StrongTableCatalogStore::new(strong_backend); + let materialized_parent = restarted + .get_namespace(bucket, &parent.public_name()) + .await + .expect("materialized parent lookup should succeed") + .expect("materialized parent should exist"); + assert_eq!(materialized_parent.properties.get("owner").map(String::as_str), Some("platform")); + assert_matches!( + restarted.drop_namespace(bucket, &parent.public_name()).await, + Err(TableCatalogStoreError::Conflict(_)) + ); +} + +#[tokio::test] +async fn object_catalog_namespace_replacement_is_fenced_by_observed_etag() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + + let recreated = Namespace::parse("recreated").expect("namespace should parse"); + let recreated_path = store.paths.namespace_entry_path(bucket, &recreated); + let mut inactive = test_namespace_entry(bucket, &recreated); + inactive.state = TableCatalogEntryState::Deleted; + backend + .seed_object( + RUSTFS_META_BUCKET, + &recreated_path, + serde_json::to_vec(&inactive).expect("inactive namespace should serialize"), + ) + .await; + let pause = backend.pause_next_put(RUSTFS_META_BUCKET, &recreated_path).await; + let stale_store = store.clone(); + let stale_entry = test_namespace_entry(bucket, &recreated); + let stale_recreate = tokio::spawn(async move { stale_store.create_namespace(stale_entry).await }); + pause.wait_started().await; + + let mut winner = test_namespace_entry(bucket, &recreated); + winner.properties.insert("owner".to_string(), "winner".to_string()); + backend + .seed_object( + RUSTFS_META_BUCKET, + &recreated_path, + serde_json::to_vec(&winner).expect("winning namespace should serialize"), + ) + .await; + pause.release(); + assert_matches!( + stale_recreate.await.expect("stale namespace recreation task should finish"), + Err(TableCatalogStoreError::Conflict(_)) + ); + let stored = store + .get_namespace(bucket, &recreated.public_name()) + .await + .expect("winning namespace should load") + .expect("winning namespace should remain"); + assert_eq!(stored.properties.get("owner").map(String::as_str), Some("winner")); +} + +async fn assert_direct_namespace_child_contract(store: &S, bucket: &str, cursor_prefix: &str) +where + S: TableCatalogStore + ?Sized, +{ + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + for name in ["alpha.deep.leaf", "alpha-beta", "beta", "sales.daily"] { + let namespace = Namespace::parse(name).expect("namespace should parse"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + } + + let one = NonZeroUsize::new(1).expect("page size should be non-zero"); + let first = store + .list_namespace_children_page(bucket, None, None, one) + .await + .expect("first root child page should load"); + assert_eq!( + first.entries.iter().map(|entry| entry.namespace.as_str()).collect::>(), + ["alpha-beta"] + ); + let cursor = first.next_cursor.expect("first root child page should continue"); + assert!(cursor.starts_with(cursor_prefix)); + let second = store + .list_namespace_children_page(bucket, None, Some(&cursor), one) + .await + .expect("second root child page should load"); + assert_eq!( + second + .entries + .iter() + .map(|entry| entry.namespace.as_str()) + .collect::>(), + ["alpha"] + ); + + let exact = store + .list_namespace_children_page(bucket, None, None, NonZeroUsize::new(4).expect("page size should be non-zero")) + .await + .expect("exact root child page should load"); + assert_eq!( + exact.entries.iter().map(|entry| entry.namespace.as_str()).collect::>(), + ["alpha-beta", "alpha", "beta", "sales"] + ); + assert!(exact.next_cursor.is_none()); + + let truncated = store + .list_namespace_children_page(bucket, None, None, NonZeroUsize::new(3).expect("page size should be non-zero")) + .await + .expect("truncated root child page should load"); + assert_eq!(truncated.entries.len(), 3); + let final_page = store + .list_namespace_children_page(bucket, None, truncated.next_cursor.as_deref(), one) + .await + .expect("final root child page should load"); + assert_eq!( + final_page + .entries + .iter() + .map(|entry| entry.namespace.as_str()) + .collect::>(), + ["sales"] + ); + assert!(final_page.next_cursor.is_none()); + + let children = store + .list_namespace_children(bucket, Some("alpha")) + .await + .expect("implicit parent children should list"); + assert_eq!(children.iter().map(|entry| entry.namespace.as_str()).collect::>(), ["alpha.deep"]); + let exact_child = store + .list_namespace_children_page(bucket, Some("alpha"), None, one) + .await + .expect("exact parent child page should load"); + assert_eq!(exact_child.entries.len(), 1); + assert!(exact_child.next_cursor.is_none()); + assert_matches!( + store + .list_namespace_children_page(bucket, Some("alpha"), Some(&cursor), one) + .await, + Err(TableCatalogStoreError::Invalid(_)) + ); + assert_matches!( + store.list_namespace_children(bucket, Some("missing")).await, + Err(TableCatalogStoreError::NotFound(_)) + ); +} + +#[tokio::test] +async fn catalog_backings_page_direct_namespace_children_with_scoped_cursors() { + assert_direct_namespace_child_contract( + &ObjectTableCatalogStore::new(TestCatalogObjectBackend::default()), + "object-catalog", + OBJECT_CATALOG_LIST_CURSOR_PREFIX, + ) + .await; + assert_direct_namespace_child_contract( + &StrongTableCatalogStore::new(TestCatalogObjectBackend::default()), + "strong-catalog", + STRONG_CATALOG_LIST_CURSOR_PREFIX, + ) + .await; +} + +#[tokio::test] +async fn object_catalog_direct_child_page_skips_the_rest_of_a_visible_subtree() { + let backend = TestCatalogObjectBackend::default(); + let store = ObjectTableCatalogStore::new(backend.clone()); + let bucket = "analytics"; + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + + for index in 0..=TABLE_CATALOG_LIST_MAX_KEYS { + let namespace = Namespace::parse(&format!("alpha.n{index:04}.leaf")).expect("namespace should parse"); + let path = store.paths.namespace_entry_path(bucket, &namespace); + backend + .seed_object( + RUSTFS_META_BUCKET, + &path, + serde_json::to_vec(&test_namespace_entry(bucket, &namespace)).expect("namespace should serialize"), + ) + .await; + } + let beta = Namespace::parse("beta").expect("namespace should parse"); + backend + .seed_object( + RUSTFS_META_BUCKET, + &store.paths.namespace_entry_path(bucket, &beta), + serde_json::to_vec(&test_namespace_entry(bucket, &beta)).expect("namespace should serialize"), + ) + .await; + backend.reset_call_counts().await; + + let page = store + .list_namespace_children_page(bucket, None, None, NonZeroUsize::new(1).expect("page size should be non-zero")) + .await + .expect("direct child page should load"); + assert_eq!(page.entries.iter().map(|entry| entry.namespace.as_str()).collect::>(), ["alpha"]); + assert!(page.next_cursor.is_some()); + assert_eq!(backend.read_call_count().await, 2); + assert_eq!(backend.list_call_count().await, 2); +} + +async fn assert_implicit_parent_resources(store: &S, bucket: &str) +where + S: TableCatalogStore + ?Sized, +{ + let table_parent = Namespace::parse("sales").expect("table parent should parse"); + let view_parent = Namespace::parse("reports").expect("view parent should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let view = IdentifierSegment::parse("recent_orders").expect("view should parse"); + let namespaces = store.list_namespaces(bucket).await.expect("explicit namespaces should list"); + assert!(!namespaces.iter().any(|entry| entry.namespace == table_parent.public_name())); + assert!(!namespaces.iter().any(|entry| entry.namespace == view_parent.public_name())); + assert!( + store + .get_namespace(bucket, &table_parent.public_name()) + .await + .expect("implicit table parent should load") + .is_some() + ); + assert!( + store + .get_namespace(bucket, &view_parent.public_name()) + .await + .expect("implicit view parent should load") + .is_some() + ); + assert!( + store + .load_table(bucket, &table_parent.public_name(), table.as_str()) + .await + .expect("implicit parent table should load") + .is_some() + ); + assert!( + store + .load_view(bucket, &view_parent.public_name(), view.as_str()) + .await + .expect("implicit parent view should load") + .is_some() + ); + assert_matches!( + store.create_namespace(test_namespace_entry(bucket, &table_parent)).await, + Err(TableCatalogStoreError::Conflict(_)) + ); + assert_matches!( + store.create_namespace(test_namespace_entry(bucket, &view_parent)).await, + Err(TableCatalogStoreError::Conflict(_)) + ); + let tables = store + .list_all_tables(bucket) + .await + .expect("table-backed namespaces should not hide tables"); + assert_eq!(tables.len(), 1); + assert_eq!(tables[0].namespace, table_parent.public_name()); + assert_eq!(tables[0].table, table.as_str()); +} + +async fn create_resources_in_implicit_parents(store: &S, bucket: &str) +where + S: TableCatalogStore + ?Sized, +{ + let table_parent = Namespace::parse("sales").expect("table parent should parse"); + let table_child = Namespace::parse("sales.daily").expect("table child should parse"); + let view_parent = Namespace::parse("reports").expect("view parent should parse"); + let view_child = Namespace::parse("reports.daily").expect("view child should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let view = IdentifierSegment::parse("recent_orders").expect("view should parse"); + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &table_child)) + .await + .expect("table child namespace should be created"); + store + .create_namespace(test_namespace_entry(bucket, &view_child)) + .await + .expect("view child namespace should be created"); + store + .create_table(test_table_entry( + bucket, + &table_parent, + &table, + default_table_metadata_file_path(&table_parent, &table, "00001.metadata.json"), + )) + .await + .expect("table creation should accept its implicit parent"); + store + .create_view(test_view_entry( + bucket, + &view_parent, + &view, + default_view_metadata_file_path(&view_parent, &view, "00001.view.json"), + )) + .await + .expect("view creation should accept its implicit parent"); + assert_implicit_parent_resources(store, bucket).await; + + store + .drop_namespace(bucket, &table_child.public_name()) + .await + .expect("table child namespace should be removable"); + store + .drop_namespace(bucket, &view_child.public_name()) + .await + .expect("view child namespace should be removable"); + assert_implicit_parent_resources(store, bucket).await; + let roots = store + .list_namespace_children(bucket, None) + .await + .expect("resource-backed root namespaces should list"); + assert_eq!( + roots.iter().map(|entry| entry.namespace.as_str()).collect::>(), + ["reports", "sales"] + ); + assert_matches!( + store.drop_namespace(bucket, &table_parent.public_name()).await, + Err(TableCatalogStoreError::Conflict(_)) + ); +} + +async fn assert_inactive_namespace_rejects_resource_creation(store: &S, bucket: &str) +where + S: TableCatalogStore + ?Sized, +{ + let namespace = Namespace::parse("inactive").expect("inactive namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + let view = IdentifierSegment::parse("recent_orders").expect("view should parse"); + let mut entry = test_namespace_entry(bucket, &namespace); + entry.state = TableCatalogEntryState::Deleted; + + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(entry) + .await + .expect("inactive namespace marker should be seeded"); + let table_error = store + .create_table(test_table_entry( + bucket, + &namespace, + &table, + default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"), + )) + .await + .expect_err("inactive namespace must reject table creation"); + assert_matches!(table_error, TableCatalogStoreError::NotFound(_)); + let view_error = store + .create_view(test_view_entry( + bucket, + &namespace, + &view, + default_view_metadata_file_path(&namespace, &view, "00001.view.json"), + )) + .await + .expect_err("inactive namespace must reject view creation"); + assert_matches!(view_error, TableCatalogStoreError::NotFound(_)); + assert!( + store + .list_namespaces(bucket) + .await + .expect("namespaces should list") + .is_empty() + ); +} + +#[tokio::test] +async fn catalog_backings_keep_implicit_parents_implicit_during_resource_creation() { + let bucket = "analytics"; + let object_store = ObjectTableCatalogStore::new(TestCatalogObjectBackend::default()); + create_resources_in_implicit_parents(&object_store, bucket).await; + + let strong_backend = TestCatalogObjectBackend::default(); + let strong_store = StrongTableCatalogStore::new(strong_backend.clone()); + create_resources_in_implicit_parents(&strong_store, bucket).await; + let restarted = StrongTableCatalogStore::new(strong_backend); + assert_implicit_parent_resources(&restarted, bucket).await; +} + +#[tokio::test] +async fn strong_catalog_namespace_drop_rejects_retained_inactive_resources() { + let store = StrongTableCatalogStore::new(TestCatalogObjectBackend::default()); + let bucket = "analytics"; + let namespace = Namespace::parse("sales").expect("namespace should parse"); + let table = IdentifierSegment::parse("orders").expect("table should parse"); + store + .put_table_bucket(test_bucket_entry(bucket)) + .await + .expect("table bucket should be created"); + store + .create_namespace(test_namespace_entry(bucket, &namespace)) + .await + .expect("namespace should be created"); + let mut entry = test_table_entry( + bucket, + &namespace, + &table, + default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"), + ); + entry.state = TableCatalogEntryState::Deleted; + store.create_table(entry).await.expect("inactive table should be retained"); + + assert_matches!( + store.drop_namespace(bucket, &namespace.public_name()).await, + Err(TableCatalogStoreError::Conflict(_)) + ); +} + +#[tokio::test] +async fn catalog_backings_reject_resource_creation_in_inactive_namespace() { + assert_inactive_namespace_rejects_resource_creation( + &ObjectTableCatalogStore::new(TestCatalogObjectBackend::default()), + "object-catalog", + ) + .await; + assert_inactive_namespace_rejects_resource_creation( + &StrongTableCatalogStore::new(TestCatalogObjectBackend::default()), + "strong-catalog", + ) + .await; +} + #[test] fn resolver_builds_paths_under_reserved_table_boundary() { let table = TableIdentifier::new( diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index b9d676e62..843c2407c 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -1856,7 +1856,7 @@ fi rg -n --with-filename 'crate::storage::.*ecstore_|^\s*ecstore_[a-z_]+(?:::|,|\})' \ rustfs/src/startup_bucket_metadata.rs \ rustfs/src/startup_shutdown.rs \ - rustfs/src/table_catalog.rs \ + rustfs/src/table_catalog \ rustfs/src/storage/s3_api/bucket.rs \ rustfs/src/storage/s3_api/multipart.rs \ rustfs/src/config/config_test.rs || true @@ -4515,7 +4515,7 @@ fi rustfs/src/init.rs \ rustfs/src/runtime_capabilities.rs \ rustfs/src/workload_admission.rs \ - rustfs/src/table_catalog.rs \ + rustfs/src/table_catalog \ rustfs/src/error.rs || true ) >"$RUSTFS_ROOT_COMPAT_RELATIVE_CONSUMER_HITS_FILE"