feat(table-catalog): add REST catalog alias (#3316)

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

rustfs/src/admin/handlers/table_catalog.rs

@@ -17,7 +17,7 @@ use crate::admin::{
     router::{AdminOperation, Operation, S3Router},
 };
 use crate::auth::{check_key_valid, get_session_token};
-use crate::server::{RemoteAddr, TABLE_CATALOG_PREFIX};
+use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX};
 use crate::table_catalog::DEFAULT_WAREHOUSE_ID;
 use http::{HeaderMap, HeaderValue, StatusCode};
 use hyper::Method;
@@ -33,6 +33,8 @@ use uuid::Uuid;
 
 const JSON_CONTENT_TYPE: &str = "application/json";
 const WAREHOUSE_PROPERTY: &str = "warehouse";
+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 CREDENTIAL_VENDING_CONFIG_KEY: &str = "rustfs.credential-vending";
 const CREDENTIAL_VENDING_UNSUPPORTED: &str = "unsupported";
 const TABLE_CATALOG_NAMESPACE_RESOURCE_ROOT: &str = "namespaces";
@@ -188,64 +190,68 @@ struct RestCommitTableResponse {
 }
 
 pub fn register_table_catalog_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    for prefix in [TABLE_CATALOG_PREFIX, TABLE_CATALOG_COMPAT_PREFIX] {
+        register_table_catalog_prefix_routes(r, prefix)?;
+    }
+
+    Ok(())
+}
+
+fn register_table_catalog_prefix_routes(r: &mut S3Router<AdminOperation>, prefix: &str) -> std::io::Result<()> {
+    r.insert(Method::GET, format!("{prefix}/config").as_str(), AdminOperation(&GET_CONFIG_HANDLER))?;
     r.insert(
         Method::GET,
-        format!("{TABLE_CATALOG_PREFIX}/config").as_str(),
-        AdminOperation(&GET_CONFIG_HANDLER),
-    )?;
-    r.insert(
-        Method::GET,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces").as_str(),
         AdminOperation(&LIST_NAMESPACES_HANDLER),
     )?;
     r.insert(
         Method::POST,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces").as_str(),
         AdminOperation(&CREATE_NAMESPACE_HANDLER),
     )?;
     r.insert(
         Method::GET,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
         AdminOperation(&GET_NAMESPACE_HANDLER),
     )?;
     r.insert(
         Method::DELETE,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
         AdminOperation(&DROP_NAMESPACE_HANDLER),
     )?;
     r.insert(
         Method::GET,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(),
         AdminOperation(&LIST_TABLES_HANDLER),
     )?;
     r.insert(
         Method::POST,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables").as_str(),
         AdminOperation(&CREATE_TABLE_HANDLER),
     )?;
     r.insert(
         Method::POST,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/register").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/register").as_str(),
         AdminOperation(&REGISTER_TABLE_HANDLER),
     )?;
     r.insert(
         Method::GET,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
         AdminOperation(&LOAD_TABLE_HANDLER),
     )?;
     r.insert(
         Method::POST,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
         AdminOperation(&COMMIT_TABLE_HANDLER),
     )?;
     r.insert(
         Method::DELETE,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}").as_str(),
         AdminOperation(&DROP_TABLE_HANDLER),
     )?;
     r.insert(
         Method::POST,
-        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/metadata").as_str(),
+        format!("{prefix}/{{warehouse}}/namespaces/{{namespace}}/tables/{{table}}/maintenance/metadata").as_str(),
         AdminOperation(&TABLE_METADATA_MAINTENANCE_HANDLER),
     )?;
 
@@ -254,7 +260,11 @@ pub fn register_table_catalog_route(r: &mut S3Router<AdminOperation>) -> std::io
 
 fn catalog_config_response() -> CatalogConfigResponse {
     CatalogConfigResponse {
-        defaults: BTreeMap::from([(WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID)]),
+        defaults: BTreeMap::from([
+            (WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID),
+            (CATALOG_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_PREFIX),
+            (CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY, TABLE_CATALOG_COMPAT_PREFIX),
+        ]),
         overrides: BTreeMap::new(),
         endpoints: TABLE_CATALOG_ENDPOINTS.to_vec(),
     }
@@ -1765,6 +1775,11 @@ mod tests {
         let response = catalog_config_response();
 
         assert_eq!(response.defaults.get(WAREHOUSE_PROPERTY), Some(&DEFAULT_WAREHOUSE_ID));
+        assert_eq!(response.defaults.get(CATALOG_ENDPOINT_PREFIX_CONFIG_KEY), Some(&TABLE_CATALOG_PREFIX));
+        assert_eq!(
+            response.defaults.get(CATALOG_COMPAT_ENDPOINT_PREFIX_CONFIG_KEY),
+            Some(&TABLE_CATALOG_COMPAT_PREFIX)
+        );
         assert!(response.overrides.is_empty());
         assert!(response.endpoints.contains(&"GET /{warehouse}/namespaces"));
         assert!(response.endpoints.contains(&"POST /{warehouse}/namespaces"));

rustfs/src/admin/route_policy.rs

@@ -656,6 +656,73 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         RUN_TABLE_MAINTENANCE,
         RouteRiskLevel::High,
     ),
+    admin(HttpMethod::Get, "/_iceberg/v1/config", GET_TABLE_CATALOG, RouteRiskLevel::Sensitive),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/namespaces",
+        GET_TABLE_NAMESPACE,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces",
+        SET_TABLE_NAMESPACE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}",
+        GET_TABLE_NAMESPACE,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Delete,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}",
+        DELETE_TABLE_NAMESPACE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables",
+        GET_TABLE,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables",
+        CREATE_TABLE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/register",
+        REGISTER_TABLE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Get,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
+        GET_TABLE_METADATA,
+        RouteRiskLevel::Sensitive,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
+        COMMIT_TABLE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Delete,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
+        DELETE_TABLE,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata",
+        RUN_TABLE_MAINTENANCE,
+        RouteRiskLevel::High,
+    ),
 ];
 
 pub const DEFERRED_ADMIN_ROUTE_POLICIES: &[DeferredAdminRoutePolicy] = &[
@@ -828,25 +895,42 @@ mod tests {
     fn route_policy_keeps_table_catalog_outside_admin_prefix() {
         let table_specs = ADMIN_ROUTE_POLICY_SPECS
             .iter()
-            .filter(|spec| spec.path().starts_with("/iceberg/v1"));
-        assert_eq!(table_specs.count(), 12);
+            .filter(|spec| spec.path().starts_with("/iceberg/v1") || spec.path().starts_with("/_iceberg/v1"));
+        assert_eq!(table_specs.count(), 24);
         assert_action(HttpMethod::Get, "/iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
+        assert_action(HttpMethod::Get, "/_iceberg/v1/{warehouse}/namespaces", GET_TABLE_NAMESPACE);
         assert_action(HttpMethod::Post, "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE);
+        assert_action(HttpMethod::Post, "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables", CREATE_TABLE);
         assert_action(
             HttpMethod::Get,
             "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
             GET_TABLE_METADATA,
         );
+        assert_action(
+            HttpMethod::Get,
+            "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
+            GET_TABLE_METADATA,
+        );
         assert_action(
             HttpMethod::Post,
             "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
             COMMIT_TABLE,
         );
+        assert_action(
+            HttpMethod::Post,
+            "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}",
+            COMMIT_TABLE,
+        );
         assert_action(
             HttpMethod::Post,
             "/iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata",
             RUN_TABLE_MAINTENANCE,
         );
+        assert_action(
+            HttpMethod::Post,
+            "/_iceberg/v1/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata",
+            RUN_TABLE_MAINTENANCE,
+        );
     }
 
     #[test]

rustfs/src/admin/route_registration_test.rs

@@ -18,7 +18,7 @@ use crate::admin::{
 };
 use crate::server::{
     ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH,
-    TABLE_CATALOG_PREFIX,
+    TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX,
 };
 use hyper::Method;
 use serial_test::serial;
@@ -37,6 +37,10 @@ fn table_catalog_path(path: &str) -> String {
     format!("{}{}", TABLE_CATALOG_PREFIX, path)
 }
 
+fn compat_table_catalog_path(path: &str) -> String {
+    format!("{}{}", TABLE_CATALOG_COMPAT_PREFIX, path)
+}
+
 #[derive(Debug)]
 struct RouteMatrixEntry {
     method: Method,
@@ -77,6 +81,14 @@ fn table_route_sample(method: Method, pattern: &str, sample: &str) -> RouteMatri
     route_sample(method, table_catalog_path(pattern), table_catalog_path(sample))
 }
 
+fn compat_table_route(method: Method, pattern: &str) -> RouteMatrixEntry {
+    route(method, compat_table_catalog_path(pattern))
+}
+
+fn compat_table_route_sample(method: Method, pattern: &str, sample: &str) -> RouteMatrixEntry {
+    route_sample(method, compat_table_catalog_path(pattern), compat_table_catalog_path(sample))
+}
+
 fn route_key(method: &Method, path: &str) -> String {
     format!("{}|{}", method.as_str(), path)
 }
@@ -308,6 +320,46 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
             "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata",
             "/analytics/namespaces/sales/tables/orders/maintenance/metadata",
         ),
+        compat_table_route(Method::GET, "/config"),
+        compat_table_route_sample(Method::GET, "/{warehouse}/namespaces", "/analytics/namespaces"),
+        compat_table_route_sample(Method::POST, "/{warehouse}/namespaces", "/analytics/namespaces"),
+        compat_table_route_sample(Method::GET, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"),
+        compat_table_route_sample(Method::DELETE, "/{warehouse}/namespaces/{namespace}", "/analytics/namespaces/sales"),
+        compat_table_route_sample(
+            Method::GET,
+            "/{warehouse}/namespaces/{namespace}/tables",
+            "/analytics/namespaces/sales/tables",
+        ),
+        compat_table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/tables",
+            "/analytics/namespaces/sales/tables",
+        ),
+        compat_table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/register",
+            "/analytics/namespaces/sales/register",
+        ),
+        compat_table_route_sample(
+            Method::GET,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}",
+            "/analytics/namespaces/sales/tables/orders",
+        ),
+        compat_table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}",
+            "/analytics/namespaces/sales/tables/orders",
+        ),
+        compat_table_route_sample(
+            Method::DELETE,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}",
+            "/analytics/namespaces/sales/tables/orders",
+        ),
+        compat_table_route_sample(
+            Method::POST,
+            "/{warehouse}/namespaces/{namespace}/tables/{table}/maintenance/metadata",
+            "/analytics/namespaces/sales/tables/orders/maintenance/metadata",
+        ),
     ]
 }
 
@@ -430,6 +482,34 @@ fn test_register_routes_cover_representative_admin_paths() {
         Method::POST,
         &table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/metadata"),
     );
+    assert_route(&router, Method::GET, &compat_table_catalog_path("/config"));
+    assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces"));
+    assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces"));
+    assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales"));
+    assert_route(&router, Method::DELETE, &compat_table_catalog_path("/analytics/namespaces/sales"));
+    assert_route(&router, Method::GET, &compat_table_catalog_path("/analytics/namespaces/sales/tables"));
+    assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/tables"));
+    assert_route(&router, Method::POST, &compat_table_catalog_path("/analytics/namespaces/sales/register"));
+    assert_route(
+        &router,
+        Method::GET,
+        &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders"),
+    );
+    assert_route(
+        &router,
+        Method::POST,
+        &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders"),
+    );
+    assert_route(
+        &router,
+        Method::DELETE,
+        &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders"),
+    );
+    assert_route(
+        &router,
+        Method::POST,
+        &compat_table_catalog_path("/analytics/namespaces/sales/tables/orders/maintenance/metadata"),
+    );
 
     assert_route(&router, Method::POST, &admin_path("/v3/service"));
     assert_route(&router, Method::GET, &admin_path("/v3/info"));

rustfs/src/admin/router.rs

@@ -2475,6 +2475,7 @@ mod tests {
         assert!(is_admin_path("/rustfs/admin/v3/info"));
         assert!(is_admin_path("/minio/admin/v3/info"));
         assert!(is_admin_path(&format!("{}/config", crate::server::TABLE_CATALOG_PREFIX)));
+        assert!(is_admin_path("/_iceberg/v1/config"));
         assert!(!is_admin_path("/bucket/object"));
         assert!(!is_admin_path("/rustfs/administrator/object"));
         assert!(!is_admin_path("/minio/administrator/object"));

rustfs/src/server/layer.rs

@@ -19,8 +19,8 @@ use crate::server::cors;
 use crate::server::hybrid::HybridBody;
 use crate::server::{
     ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX,
-    MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TABLE_CATALOG_PREFIX, active_http_requests,
-    collect_dependency_readiness_report, has_path_prefix, is_admin_path,
+    MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report,
+    has_path_prefix, is_admin_path, is_table_catalog_path,
 };
 use crate::storage::apply_cors_headers;
 use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
@@ -864,7 +864,7 @@ fn is_object_attributes_request(req: &HttpRequest<Incoming>) -> bool {
         || has_path_prefix(path, MINIO_ADMIN_PREFIX)
         || has_path_prefix(path, RUSTFS_ADMIN_PREFIX)
         || has_path_prefix(path, MINIO_ADMIN_V3_PREFIX)
-        || has_path_prefix(path, TABLE_CATALOG_PREFIX)
+        || is_table_catalog_path(path)
         || has_path_prefix(path, CONSOLE_PREFIX)
         || has_path_prefix(path, RPC_PREFIX)
     {
@@ -909,7 +909,7 @@ impl ConditionalCorsLayer {
         // Exclude Admin, Console, RPC, and configured special paths
         !has_path_prefix(path, ADMIN_PREFIX)
             && !has_path_prefix(path, MINIO_ADMIN_PREFIX)
-            && !has_path_prefix(path, TABLE_CATALOG_PREFIX)
+            && !is_table_catalog_path(path)
             && !has_path_prefix(path, RPC_PREFIX)
             && !is_console_path(path)
             && !Self::EXCLUDED_EXACT_PATHS.contains(&path)
@@ -1960,7 +1960,11 @@ mod tests {
         assert!(ConditionalCorsLayer::is_s3_path("/"));
         assert!(!ConditionalCorsLayer::is_s3_path("/rustfs/admin/v3/info"));
         assert!(!ConditionalCorsLayer::is_s3_path("/minio/admin/v3/info"));
-        assert!(!ConditionalCorsLayer::is_s3_path(&format!("{TABLE_CATALOG_PREFIX}/config")));
+        assert!(!ConditionalCorsLayer::is_s3_path(&format!(
+            "{}/config",
+            crate::server::TABLE_CATALOG_PREFIX
+        )));
+        assert!(!ConditionalCorsLayer::is_s3_path("/_iceberg/v1/config"));
         assert!(ConditionalCorsLayer::is_s3_path("/minio/adminx/object"));
         assert!(!ConditionalCorsLayer::is_s3_path("/health"));
         assert!(!ConditionalCorsLayer::is_s3_path("/health/ready"));

rustfs/src/server/mod.rs

@@ -49,7 +49,8 @@ pub(crate) use module_switch::{
 pub(crate) use prefix::{
     ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE,
     MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX,
-    TABLE_CATALOG_PREFIX, TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path,
+    TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX, TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path,
+    is_table_catalog_path,
 };
 pub(crate) use readiness::DependencyReadiness;
 pub(crate) use readiness::DependencyReadinessReport;

rustfs/src/server/prefix.rs

@@ -46,11 +46,16 @@ pub(crate) const MINIO_ADMIN_PREFIX: &str = "/minio/admin";
 /// Iceberg REST Catalog prefix for RustFS S3 Tables control-plane routes.
 pub(crate) const TABLE_CATALOG_PREFIX: &str = "/iceberg/v1";
 
+/// MinIO AIStor-compatible Iceberg REST Catalog prefix alias.
+pub(crate) const TABLE_CATALOG_COMPAT_PREFIX: &str = "/_iceberg/v1";
+
 /// Returns true for the admin prefix itself or slash-delimited children.
 pub(crate) fn is_admin_path(path: &str) -> bool {
-    has_path_prefix(path, ADMIN_PREFIX)
-        || has_path_prefix(path, MINIO_ADMIN_PREFIX)
-        || has_path_prefix(path, TABLE_CATALOG_PREFIX)
+    has_path_prefix(path, ADMIN_PREFIX) || has_path_prefix(path, MINIO_ADMIN_PREFIX) || is_table_catalog_path(path)
+}
+
+pub(crate) fn is_table_catalog_path(path: &str) -> bool {
+    has_path_prefix(path, TABLE_CATALOG_PREFIX) || has_path_prefix(path, TABLE_CATALOG_COMPAT_PREFIX)
 }
 
 pub(crate) fn has_path_prefix(path: &str, prefix: &str) -> bool {

rustfs/src/server/readiness.rs

@@ -12,8 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::server::has_path_prefix;
 use crate::server::{ServiceState, ServiceStateManager};
+use crate::server::{has_path_prefix, is_table_catalog_path};
 use bytes::Bytes;
 use http::{Request as HttpRequest, Response, StatusCode};
 use http_body::Body;
@@ -140,7 +140,7 @@ fn is_probe_path(path: &str) -> bool {
 
     let is_prefix_probe = has_path_prefix(path, crate::server::RUSTFS_ADMIN_PREFIX)
         || has_path_prefix(path, crate::server::MINIO_ADMIN_V3_PREFIX)
-        || has_path_prefix(path, crate::server::TABLE_CATALOG_PREFIX)
+        || is_table_catalog_path(path)
         || has_path_prefix(path, crate::server::CONSOLE_PREFIX)
         || has_path_prefix(path, crate::server::RPC_PREFIX)
         || has_path_prefix(path, crate::server::ADMIN_PREFIX)
@@ -721,6 +721,7 @@ mod tests {
         assert!(is_probe_path("/minio/admin/v3/info"));
         assert!(is_probe_path("/rustfs/admin/v3/info"));
         assert!(is_probe_path(&format!("{}/config", crate::server::TABLE_CATALOG_PREFIX)));
+        assert!(is_probe_path("/_iceberg/v1/config"));
         assert!(is_probe_path("/rustfs/console/"));
         assert!(!is_probe_path("/minio/adminx/object"));
         assert!(!is_probe_path("/rustfs/adminx/object"));