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 <heihutu@gmail.com>

---------

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: heihutu <heihutu@gmail.com>

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.

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<String, String>,
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct UpdateNamespacePropertiesRequest {
+    #[serde(default)]
+    removals: Vec<String>,
+    #[serde(default)]
+    updates: BTreeMap<String, String>,
+}
+
 #[derive(Debug, Deserialize)]
 #[serde(deny_unknown_fields)]
 struct RegisterTableRequest {
@@ -888,7 +907,8 @@ struct TableMetadataLocationResponse {
 fn catalog_config_response(warehouse: Option<&str>) -> S3Result<CatalogConfigResponse> {
     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<CatalogConfigRes
     if let Some(warehouse) = warehouse {
         defaults.insert(PREFIX_PROPERTY.to_string(), warehouse.to_string());
     }
+    let mut endpoints = TABLE_CATALOG_ENDPOINTS.to_vec();
+    if backing_mode == crate::table_catalog::TableCatalogBackingMode::DurableStrong {
+        endpoints.extend_from_slice(TABLE_CATALOG_DURABLE_STRONG_ENDPOINTS);
+    }
     Ok(CatalogConfigResponse {
         defaults,
         overrides,
-        endpoints: TABLE_CATALOG_ENDPOINTS.to_vec(),
+        endpoints,
         admin_discovery: CatalogAdminDiscovery {
             runtime_capabilities: usecase.runtime_capabilities_route(),
             cluster_snapshot: usecase.cluster_snapshot_route(),
@@ -1114,6 +1138,33 @@ async fn read_json_body<T: DeserializeOwned>(mut input: Body) -> S3Result<T> {
     serde_json::from_slice(&body).map_err(|err| s3_error!(InvalidRequest, "invalid JSON: {}", err))
 }
 
+async fn read_bounded_json_body<T: DeserializeOwned>(
+    headers: &HeaderMap,
+    mut input: Body,
+    max_size: usize,
+    timeout: StdDuration,
+    operation: &str,
+) -> S3Result<T> {
+    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::<usize>()
+            .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<T>(mut input: Body) -> S3Result<T>
 where
     T: Default + DeserializeOwned,
@@ -1294,7 +1345,69 @@ fn encode_rest_page_token(cursor: &str, context: &str) -> S3Result<String> {
 
 fn namespace_from_params(params: &Params<'_, '_>) -> S3Result<crate::table_catalog::Namespace> {
     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<crate::table_catalog::Namespace> {
+    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::<S3Result<Vec<_>>>()?
+    } 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<Option<crate::table_catalog::Namespace>> {
+    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<String> {
@@ -1460,12 +1573,8 @@ fn namespace_segments(namespace: &crate::table_catalog::Namespace) -> Vec<String
 }
 
 fn namespace_from_segments(segments: &[String]) -> S3Result<crate::table_catalog::Namespace> {
-    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<RestNamespaceResponse> {
@@ -1477,24 +1586,6 @@ fn namespace_response_from_entry(entry: crate::table_catalog::NamespaceEntry) ->
     })
 }
 
-fn list_namespaces_response_from_entries(
-    entries: Vec<crate::table_catalog::NamespaceEntry>,
-    next_page_token: Option<String>,
-) -> S3Result<RestListNamespacesResponse> {
-    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::<S3Result<Vec<_>>>()?;
-    Ok(RestListNamespacesResponse {
-        namespaces,
-        next_page_token,
-    })
-}
-
 fn list_tables_response_from_entries(
     entries: Vec<crate::table_catalog::TableEntry>,
     next_page_token: Option<String>,
@@ -3511,6 +3602,7 @@ fn namespace_entry_from_create_request(
     request: CreateNamespaceRequest,
 ) -> S3Result<crate::table_catalog::NamespaceEntry> {
     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<S>(store: &S, bucket: &str, uri: &http::Uri) -> S3Result<RestListNamespacesResponse>
+async fn list_namespaces_response<S>(
+    store: &S,
+    bucket: &str,
+    parent: Option<&crate::table_catalog::Namespace>,
+    uri: &http::Uri,
+) -> S3Result<RestListNamespacesResponse>
 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::<S3Result<Vec<_>>>()?;
+    Ok(RestListNamespacesResponse {
+        namespaces,
+        next_page_token,
+    })
 }
 
 async fn get_namespace_response<S>(
@@ -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> {
+    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<S>(
+    store: &S,
+    bucket: &str,
+    namespace: &crate::table_catalog::Namespace,
+    request: UpdateNamespacePropertiesRequest,
+) -> S3Result<crate::table_catalog::NamespacePropertiesUpdateResult>
+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<S>(store: &S, bucket: &str, namespace: &crate::table_catalog::Namespace) -> S3Result<StatusCode>
 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))
 }
 

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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         let warehouse = warehouse_from_params(&params)?;
-        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(&params)?;
         let resource = TableCatalogResource::warehouse(&warehouse);
         authorize_table_catalog_resource_request(&req, &resource, AdminAction::SetTableNamespaceAction).await?;
-        let request = read_json_body::<CreateNamespaceRequest>(req.input).await?;
+        let request = read_bounded_json_body::<CreateNamespaceRequest>(
+            &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<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        let warehouse = warehouse_from_params(&params)?;
+        let namespace = namespace_from_params(&params)?;
+        let resource = TableCatalogResource::namespace(&warehouse, &namespace);
+        authorize_table_catalog_resource_request(&req, &resource, AdminAction::UpdateTableNamespacePropertiesAction).await?;
+        ensure_table_bucket_enabled(&warehouse).await?;
+        let request = read_bounded_json_body::<UpdateNamespacePropertiesRequest>(
+            &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]

rustfs/src/admin/handlers/table_catalog/routes.rs

@@ -69,6 +69,11 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, 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(),

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::<RestCreateNamespaceHandler>();
     assert_operation::<RestGetNamespaceHandler>();
     assert_operation::<RestNamespaceExistsHandler>();
+    assert_operation::<RestUpdateNamespacePropertiesHandler>();
     assert_operation::<RestDropNamespaceHandler>();
     assert_operation::<RestListTablesHandler>();
     assert_operation::<RestCreateTableHandler>();
@@ -730,6 +768,13 @@ fn table_catalog_ingress_requests_reject_unknown_fields() {
             "unexpected": true
         }),
     );
+    assert_rejects_unknown_field::<UpdateNamespacePropertiesRequest>(
+        "UpdateNamespacePropertiesRequest",
+        serde_json::json!({
+            "updates": {},
+            "unexpected": true
+        }),
+    );
     assert_rejects_unknown_field::<RegisterTableRequest>(
         "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::<UpdateNamespacePropertiesRequest>(
+        &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::<UpdateNamespacePropertiesRequest>(
+        &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::<UpdateNamespacePropertiesRequest>(
+        &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::<UpdateNamespacePropertiesRequest>(
+        &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::<usize>();
+    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::<UpdateNamespacePropertiesRequest>(
+        &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::<http::Uri>().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::<http::Uri>().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::<http::Uri>()
+        .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::<http::Uri>()
+        .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::<http::Uri>().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::<http::Uri>()
         .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::<http::Uri>().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<crate::table_catalog::NamespacePropertiesUpdateResult> {
+        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<Vec<crate::table_catalog::TableEntry>> {
+        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::<http::Uri>().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());

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(

rustfs/src/admin/route_registration_test.rs

@@ -383,6 +383,11 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         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<RouteMatrixEntry> {
         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,

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}"),
         }
     }

rustfs/src/table_catalog/iceberg/commit.rs

@@ -196,6 +196,7 @@ fn table_catalog_store_result_label<T>(result: &TableCatalogStoreResult<T>) -> &
         Err(TableCatalogStoreError::Conflict(_)) => "conflict",
         Err(TableCatalogStoreError::Invalid(_)) => "invalid",
         Err(TableCatalogStoreError::NotFound(_)) => "not_found",
+        Err(TableCatalogStoreError::Unsupported(_)) => "unsupported",
         Err(TableCatalogStoreError::Internal(_)) => "failure",
     }
 }

rustfs/src/table_catalog/iceberg/validation.rs

@@ -191,28 +191,23 @@ where
     }
 
     let mut matched: Option<TableDataPlaneResource> = 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)

rustfs/src/table_catalog/identifier.rs

@@ -40,18 +40,28 @@ impl Namespace {
     pub const MAX_LEN: usize = 512;
 
     pub fn parse(value: &str) -> Result<Self, CatalogIdentifierError> {
-        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<String>) -> Result<Self, CatalogIdentifierError> {
+        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::<Result<Vec<_>, _>>()?;
         Ok(Self { segments })
     }
 

rustfs/src/table_catalog/mod.rs

@@ -275,7 +275,7 @@ where
     TableCatalogListPage { entries, next_cursor }
 }
 
-fn catalog_list_page_from_entries<T, F>(
+pub(crate) fn catalog_list_page_from_entries<T, F>(
     mut entries: Vec<T>,
     cursor: Option<&str>,
     limit: NonZeroUsize,

rustfs/src/table_catalog/model.rs

@@ -71,6 +71,119 @@ pub(crate) struct NamespaceEntry {
     pub updated_at: Option<String>,
 }
 
+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<String>,
+    updates: BTreeMap<String, String>,
+}
+
+#[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<String>,
+    pub removed: Vec<String>,
+    pub missing: Vec<String>,
+}
+
+impl NamespacePropertiesUpdate {
+    pub(crate) fn try_new(
+        removals: Vec<String>,
+        updates: BTreeMap<String, String>,
+    ) -> Result<Self, NamespacePropertiesUpdateError> {
+        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::<Vec<_>>();
+        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<String, String>) -> 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 {

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::<BTreeSet<_>>();
-        let mut unmatched_view_objects = namespace_objects
-            .iter()
-            .filter(|object| object.ends_with(VIEW_ENTRY_FILE))
-            .cloned()
-            .collect::<BTreeSet<_>>();
         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::<TableEntry>(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::<CommitLogEntry>(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::<CommitLogEntry>(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::<ViewEntry>(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::<TableEntry>(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::<CommitLogEntry>(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::<CommitLogEntry>(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::<ViewEntry>(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::<B>::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::<String, usize>::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::<NamespaceEntry>(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::<ViewEntry>(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::<TableEntry>(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,

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<Namespace> {
+    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<NamespaceEntry>,
+) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
+    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<Option<TableBucketEntry>>;
@@ -34,6 +84,55 @@ pub(crate) trait TableCatalogStore: Send + Sync {
 
     async fn list_namespaces(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>>;
 
+    async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
+        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<Vec<NamespaceEntry>> {
+        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<TableCatalogListPage<NamespaceEntry>> {
+        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<Option<NamespaceEntry>>;
 
+    async fn update_namespace_properties(
+        &self,
+        _table_bucket: &str,
+        _namespace: &str,
+        _update: NamespacePropertiesUpdate,
+    ) -> TableCatalogStoreResult<NamespacePropertiesUpdateResult> {
+        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<Vec<TableEntry>>;
 
+    async fn list_all_tables(&self, table_bucket: &str) -> TableCatalogStoreResult<Vec<TableEntry>>;
+
     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<Vec<NamespaceEntry>> {
+        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<Vec<NamespaceEntry>> {
+        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<TableCatalogListPage<NamespaceEntry>> {
+        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<NamespacePropertiesUpdateResult> {
+        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<Vec<TableEntry>> {
+        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,

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<Namespace> {
+    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<Namespace> {
+    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<NamespaceEntry>,
+}
+
 #[derive(Clone)]
 pub(crate) struct ObjectTableCatalogStore<B> {
     pub(in crate::table_catalog) backend: B,
@@ -35,15 +86,19 @@ where
         RUSTFS_META_BUCKET
     }
 
-    async fn list_entry_page<T>(
+    async fn list_entry_page<T, P, V>(
         &self,
         prefix: &str,
         entry_file: &str,
         cursor: Option<&str>,
         limit: NonZeroUsize,
+        include: P,
+        validate: V,
     ) -> TableCatalogStoreResult<TableCatalogListPage<T>>
     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::<T>(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<Option<ActiveNamespaceEvidence>> {
+        if object.ends_with(NAMESPACE_ENTRY_FILE) {
+            let Some((entry, _)) = self.read_entry::<NamespaceEntry>(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::<TableEntry>(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::<ViewEntry>(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<bool> {
+        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<bool> {
+        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::<NamespaceEntry>(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<TableCatalogListPage<NamespaceEntry>> {
+        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<Vec<NamespaceEntry>> {
+        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::<NamespaceEntry>(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<T>(
         &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::<NamespaceEntry>(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::<NamespaceEntry>(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::<NamespaceEntry>(self.catalog_bucket(), &object)
+            .await?
+        {
+            Some((current, etag)) => {
+                validate_namespace_entry_object(&self.paths, &object, &current)?;
+                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<Vec<NamespaceEntry>> {
-        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::<NamespaceEntry>(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<TableCatalogListPage<NamespaceEntry>> {
-        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<Option<NamespaceEntry>> {
         let namespace = parse_namespace_for_store(namespace)?;
-        self.read_entry::<NamespaceEntry>(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::<NamespaceEntry>(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<Vec<NamespaceEntry>> {
+        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<Vec<NamespaceEntry>> {
+        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<TableCatalogListPage<NamespaceEntry>> {
+        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::<NamespaceEntry>(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, &current)?;
+        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<Vec<TableEntry>> {
+        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::<TableEntry>(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<TableCatalogListPage<ViewEntry>> {
         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<Option<ViewEntry>> {

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<String, BTreeMap<String, StrongResourceKey>>;
 
 #[derive(Clone, Default)]
@@ -25,6 +26,8 @@ pub(in crate::table_catalog) struct StrongTableCatalogState {
     pub(super) snapshot_etag: Option<String>,
     pub(super) table_buckets: BTreeMap<String, TableBucketEntry>,
     pub(in crate::table_catalog) namespaces: BTreeMap<StrongNamespaceKey, NamespaceEntry>,
+    namespace_children: BTreeMap<StrongNamespaceChildKey, String>,
+    namespace_objects: BTreeSet<StrongNamespaceKey>,
     pub(super) tables: BTreeMap<StrongResourceKey, TableEntry>,
     pub(super) views: BTreeMap<StrongResourceKey, ViewEntry>,
     pub(super) commits: BTreeMap<StrongCommitKey, CommitLogEntry>,
@@ -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<TableCatalogListPage<NamespaceEntry>> {
+        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::<TableCatalogStoreResult<Vec<_>>>()?;
+        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<StrongNamespaceChildKey, String>,
+        table_bucket: &str,
+        namespace: &Namespace,
+    ) {
+        for depth in 0..namespace.segments().len() {
+            let parent = namespace.segments()[..depth]
+                .iter()
+                .map(IdentifierSegment::as_str)
+                .collect::<Vec<_>>()
+                .join(".");
+            let child = namespace.segments()[..=depth]
+                .iter()
+                .map(IdentifierSegment::as_str)
+                .collect::<Vec<_>>()
+                .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<StrongTableCatalogSnapshot> {
+        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::<Vec<_>>();
         entries.sort_by(|left, right| left.namespace.cmp(&right.namespace));
         Ok(entries)
     }
 
+    async fn list_namespaces_under(&self, table_bucket: &str, parent: &str) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
+        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::<Vec<_>>();
+        Ok(exact.into_iter().chain(descendants).collect())
+    }
+
+    async fn list_namespace_children(
+        &self,
+        table_bucket: &str,
+        parent: Option<&str>,
+    ) -> TableCatalogStoreResult<Vec<NamespaceEntry>> {
+        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<TableCatalogListPage<NamespaceEntry>> {
+        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<NamespacePropertiesUpdateResult> {
+        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<Vec<TableEntry>> {
+        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 {}/{}/{}",

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"