feat(table-catalog): add REST catalog route surface (#3211)

* feat(table-catalog): add REST catalog route surface

* fix(table-catalog): avoid test-only router import

* fix(table-catalog): align unsupported REST responses

---------

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

rustfs/src/admin/handlers/mod.rs

@@ -45,6 +45,7 @@ pub mod service_account;
 pub mod site_replication;
 pub mod sts;
 pub mod system;
+pub mod table_catalog;
 mod target_descriptor;
 pub mod tier;
 pub mod tls_debug;
@@ -95,6 +96,7 @@ mod tests {
         let _site_replication_add_handler = site_replication::SiteReplicationAddHandler {};
         let _site_replication_info_handler = site_replication::SiteReplicationInfoHandler {};
         let _site_replication_status_handler = site_replication::SiteReplicationStatusHandler {};
+        let _table_catalog_config_handler = table_catalog::GetCatalogConfigHandler {};
 
         // Just verify they can be created without panicking
         // Test passes if we reach this point without panicking

rustfs/src/admin/handlers/table_catalog.rs

@@ -0,0 +1,260 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::admin::{
+    auth::validate_admin_request,
+    router::{AdminOperation, Operation, S3Router},
+};
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{RemoteAddr, TABLE_CATALOG_PREFIX};
+use crate::table_catalog::DEFAULT_WAREHOUSE_ID;
+use http::{HeaderMap, HeaderValue, StatusCode};
+use hyper::Method;
+use matchit::Params;
+use rustfs_policy::policy::action::{Action, AdminAction};
+use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
+use serde::Serialize;
+use std::collections::BTreeMap;
+
+const JSON_CONTENT_TYPE: &str = "application/json";
+const WAREHOUSE_PROPERTY: &str = "warehouse";
+const UNSUPPORTED_OPERATION_TYPE: &str = "UnsupportedOperationException";
+const UNSUPPORTED_OPERATION_STATUS: StatusCode = StatusCode::NOT_ACCEPTABLE;
+
+static GET_CONFIG_HANDLER: GetCatalogConfigHandler = GetCatalogConfigHandler {};
+static LIST_NAMESPACES_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::GetTableNamespaceAction,
+    operation: "list namespaces",
+};
+static CREATE_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::SetTableNamespaceAction,
+    operation: "create namespace",
+};
+static GET_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::GetTableNamespaceAction,
+    operation: "get namespace",
+};
+static DROP_NAMESPACE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::DeleteTableNamespaceAction,
+    operation: "drop namespace",
+};
+static LIST_TABLES_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::GetTableAction,
+    operation: "list tables",
+};
+static CREATE_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::CreateTableAction,
+    operation: "create table",
+};
+static REGISTER_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::RegisterTableAction,
+    operation: "register table",
+};
+static LOAD_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::GetTableAction,
+    operation: "load table",
+};
+static COMMIT_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::CommitTableAction,
+    operation: "commit table",
+};
+static DROP_TABLE_HANDLER: UnsupportedCatalogOperationHandler = UnsupportedCatalogOperationHandler {
+    action: AdminAction::DeleteTableAction,
+    operation: "drop table",
+};
+
+#[derive(Debug, Serialize)]
+struct CatalogConfigResponse {
+    defaults: BTreeMap<&'static str, &'static str>,
+    overrides: BTreeMap<&'static str, &'static str>,
+    endpoints: Vec<&'static str>,
+}
+
+#[derive(Debug, Serialize)]
+struct ErrorResponse {
+    error: RestCatalogError,
+}
+
+#[derive(Debug, Serialize)]
+struct RestCatalogError {
+    message: String,
+    #[serde(rename = "type")]
+    error_type: &'static str,
+    code: u16,
+}
+
+pub fn register_table_catalog_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
+    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(),
+        AdminOperation(&LIST_NAMESPACES_HANDLER),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces").as_str(),
+        AdminOperation(&CREATE_NAMESPACE_HANDLER),
+    )?;
+    r.insert(
+        Method::GET,
+        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
+        AdminOperation(&GET_NAMESPACE_HANDLER),
+    )?;
+    r.insert(
+        Method::DELETE,
+        format!("{TABLE_CATALOG_PREFIX}/{{warehouse}}/namespaces/{{namespace}}").as_str(),
+        AdminOperation(&DROP_NAMESPACE_HANDLER),
+    )?;
+    r.insert(
+        Method::GET,
+        format!("{TABLE_CATALOG_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(),
+        AdminOperation(&CREATE_TABLE_HANDLER),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{TABLE_CATALOG_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(),
+        AdminOperation(&LOAD_TABLE_HANDLER),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{TABLE_CATALOG_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(),
+        AdminOperation(&DROP_TABLE_HANDLER),
+    )?;
+
+    Ok(())
+}
+
+fn catalog_config_response() -> CatalogConfigResponse {
+    CatalogConfigResponse {
+        defaults: BTreeMap::from([(WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID)]),
+        overrides: BTreeMap::new(),
+        endpoints: Vec::new(),
+    }
+}
+
+fn build_json_response<T: Serialize>(status: StatusCode, body: &T) -> S3Result<S3Response<(StatusCode, Body)>> {
+    let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
+    let mut headers = HeaderMap::new();
+    headers.insert(CONTENT_TYPE, HeaderValue::from_static(JSON_CONTENT_TYPE));
+    Ok(S3Response::with_headers((status, Body::from(data)), headers))
+}
+
+async fn authorize_table_catalog_request(req: &S3Request<Body>, action: AdminAction) -> S3Result<()> {
+    let Some(input_cred) = &req.credentials else {
+        return Err(s3_error!(InvalidRequest, "authentication required"));
+    };
+
+    let (cred, owner) =
+        check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?;
+
+    validate_admin_request(
+        &req.headers,
+        &cred,
+        owner,
+        false,
+        vec![Action::AdminAction(action)],
+        req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
+}
+
+fn unsupported_response(operation: &str) -> S3Result<S3Response<(StatusCode, Body)>> {
+    build_json_response(
+        UNSUPPORTED_OPERATION_STATUS,
+        &ErrorResponse {
+            error: RestCatalogError {
+                message: format!("Iceberg REST Catalog {operation} is not implemented yet"),
+                error_type: UNSUPPORTED_OPERATION_TYPE,
+                code: UNSUPPORTED_OPERATION_STATUS.as_u16(),
+            },
+        },
+    )
+}
+
+pub struct GetCatalogConfigHandler {}
+
+#[async_trait::async_trait]
+impl Operation for GetCatalogConfigHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_table_catalog_request(&req, AdminAction::GetTableCatalogAction).await?;
+        build_json_response(StatusCode::OK, &catalog_config_response())
+    }
+}
+
+pub struct UnsupportedCatalogOperationHandler {
+    action: AdminAction,
+    operation: &'static str,
+}
+
+#[async_trait::async_trait]
+impl Operation for UnsupportedCatalogOperationHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_table_catalog_request(&req, self.action).await?;
+        unsupported_response(self.operation)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn catalog_config_response_uses_default_warehouse_without_unsupported_endpoints() {
+        let response = catalog_config_response();
+
+        assert_eq!(response.defaults.get(WAREHOUSE_PROPERTY), Some(&DEFAULT_WAREHOUSE_ID));
+        assert!(response.overrides.is_empty());
+        assert!(response.endpoints.is_empty());
+    }
+
+    #[test]
+    fn unsupported_response_uses_rest_catalog_unsupported_status() {
+        let response = unsupported_response("create table").expect("unsupported response should build");
+
+        assert_eq!(response.output.0, UNSUPPORTED_OPERATION_STATUS);
+    }
+
+    #[test]
+    fn table_catalog_handlers_require_table_admin_actions() {
+        let src = include_str!("table_catalog.rs");
+
+        assert!(src.contains("AdminAction::GetTableCatalogAction"));
+        assert!(src.contains("AdminAction::GetTableNamespaceAction"));
+        assert!(src.contains("AdminAction::SetTableNamespaceAction"));
+        assert!(src.contains("AdminAction::DeleteTableNamespaceAction"));
+        assert!(src.contains("AdminAction::CreateTableAction"));
+        assert!(src.contains("AdminAction::RegisterTableAction"));
+        assert!(src.contains("AdminAction::CommitTableAction"));
+        assert!(src.contains("AdminAction::DeleteTableAction"));
+    }
+}

rustfs/src/admin/mod.rs

@@ -28,7 +28,7 @@ mod route_registration_test;
 
 use handlers::{
     audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools,
-    profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user,
+    profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user,
 };
 use router::{AdminOperation, S3Router};
 use s3s::route::S3Route;
@@ -70,6 +70,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result<impl S3Route>
     tls_debug::register_tls_debug_route(&mut r)?;
     kms::register_kms_route(&mut r)?;
     oidc::register_oidc_route(&mut r)?;
+    table_catalog::register_table_catalog_route(&mut r)?;
 
     Ok(r)
 }

rustfs/src/admin/route_registration_test.rs

@@ -15,11 +15,15 @@
 use crate::admin::{
     handlers::{
         audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools,
-        profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user,
+        profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, table_catalog, tier, tls_debug,
+        user,
     },
     router::{AdminOperation, S3Router},
 };
-use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH};
+use crate::server::{
+    ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH,
+    TABLE_CATALOG_PREFIX,
+};
 use hyper::Method;
 use serial_test::serial;
 use temp_env::with_var;
@@ -32,6 +36,10 @@ fn compat_admin_alias_path(path: &str) -> String {
     format!("{}{}", MINIO_ADMIN_PREFIX, path)
 }
 
+fn table_catalog_path(path: &str) -> String {
+    format!("{}{}", TABLE_CATALOG_PREFIX, path)
+}
+
 fn assert_route(router: &S3Router<AdminOperation>, method: Method, path: &str) {
     assert!(
         router.contains_route(method.clone(), path),
@@ -64,6 +72,7 @@ fn register_admin_routes(router: &mut S3Router<AdminOperation>) {
     tls_debug::register_tls_debug_route(router).expect("register tls debug route");
     kms::register_kms_route(router).expect("register kms route");
     oidc::register_oidc_route(router).expect("register oidc route");
+    table_catalog::register_table_catalog_route(router).expect("register table catalog route");
 }
 
 // register_admin_routes reads ENV_HEALTH_ENDPOINT_ENABLE to decide whether
@@ -126,6 +135,18 @@ fn test_register_routes_cover_representative_admin_paths() {
     assert_route(&router, Method::PUT, &admin_path("/v3/config"));
     assert_route(&router, Method::GET, &admin_path("/v3/scanner/status"));
 
+    assert_route(&router, Method::GET, &table_catalog_path("/config"));
+    assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces"));
+    assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces"));
+    assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales"));
+    assert_route(&router, Method::DELETE, &table_catalog_path("/analytics/namespaces/sales"));
+    assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables"));
+    assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables"));
+    assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/register"));
+    assert_route(&router, Method::GET, &table_catalog_path("/analytics/namespaces/sales/tables/orders"));
+    assert_route(&router, Method::POST, &table_catalog_path("/analytics/namespaces/sales/tables/orders"));
+    assert_route(&router, Method::DELETE, &table_catalog_path("/analytics/namespaces/sales/tables/orders"));
+
     assert_route(&router, Method::POST, &admin_path("/v3/service"));
     assert_route(&router, Method::GET, &admin_path("/v3/info"));
     assert_route(&router, Method::GET, &admin_path("/v3/storageinfo"));

rustfs/src/admin/router.rs

@@ -2458,6 +2458,7 @@ mod tests {
     fn is_admin_path_accepts_rustfs_and_compat_prefixes() {
         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("/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, active_http_requests, collect_dependency_readiness_report,
-    has_path_prefix, is_admin_path,
+    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,
 };
 use crate::storage::apply_cors_headers;
 use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
@@ -864,6 +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)
         || has_path_prefix(path, CONSOLE_PREFIX)
         || has_path_prefix(path, RPC_PREFIX)
     {
@@ -908,6 +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)
             && !has_path_prefix(path, RPC_PREFIX)
             && !is_console_path(path)
             && !Self::EXCLUDED_EXACT_PATHS.contains(&path)
@@ -1958,6 +1960,7 @@ 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("/minio/adminx/object"));
         assert!(!ConditionalCorsLayer::is_s3_path("/health"));
         assert!(!ConditionalCorsLayer::is_s3_path("/health/ready"));

rustfs/src/server/mod.rs

@@ -49,7 +49,7 @@ 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,
-    TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path,
+    TABLE_CATALOG_PREFIX, TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path,
 };
 pub(crate) use readiness::DependencyReadiness;
 pub(crate) use readiness::DependencyReadinessReport;

rustfs/src/server/prefix.rs

@@ -43,9 +43,14 @@ pub(crate) const ADMIN_PREFIX: &str = "/rustfs/admin";
 /// This alias allows stock MinIO admin tooling to reach RustFS handlers.
 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";
+
 /// 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, ADMIN_PREFIX)
+        || has_path_prefix(path, MINIO_ADMIN_PREFIX)
+        || has_path_prefix(path, TABLE_CATALOG_PREFIX)
 }
 
 pub(crate) fn has_path_prefix(path: &str, prefix: &str) -> bool {

rustfs/src/server/readiness.rs

@@ -140,6 +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)
         || has_path_prefix(path, crate::server::CONSOLE_PREFIX)
         || has_path_prefix(path, crate::server::RPC_PREFIX)
         || has_path_prefix(path, crate::server::ADMIN_PREFIX)
@@ -719,6 +720,7 @@ mod tests {
     fn probe_path_checks_admin_boundaries() {
         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("/rustfs/console/"));
         assert!(!is_probe_path("/minio/adminx/object"));
         assert!(!is_probe_path("/rustfs/adminx/object"));