From dd0ef1f7665f453abf9cf19d4657d8d207a910dd Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Sun, 7 Jun 2026 21:54:15 +0800 Subject: [PATCH] feat(table-catalog): tighten REST load/register compatibility (#3245) Co-authored-by: Henry Guo Co-authored-by: cxymds --- rustfs/src/admin/handlers/table_catalog.rs | 246 ++++++++++++++++++--- 1 file changed, 218 insertions(+), 28 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 51138624b..9a14db665 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -38,7 +38,6 @@ const TABLE_CATALOG_ENDPOINTS: &[&str] = &[ "GET /{warehouse}/namespaces/{namespace}", "DELETE /{warehouse}/namespaces/{namespace}", "GET /{warehouse}/namespaces/{namespace}/tables", - "POST /{warehouse}/namespaces/{namespace}/tables", "POST /{warehouse}/namespaces/{namespace}/register", "GET /{warehouse}/namespaces/{namespace}/tables/{table}", "DELETE /{warehouse}/namespaces/{namespace}/tables/{table}", @@ -78,7 +77,7 @@ struct RegisterTableRequest { #[serde(rename = "metadata-location")] metadata_location: String, #[serde(default)] - properties: BTreeMap, + overwrite: bool, } #[derive(Debug, Deserialize)] @@ -277,11 +276,14 @@ fn table_name_from_params(params: &Params<'_, '_>) -> S3Result { Ok(table.to_string()) } -fn table_catalog_store() -> S3Result> { +fn table_catalog_backend() -> S3Result> { let store = new_object_layer_fn().ok_or_else(|| s3_error!(InternalError, "object store not initialized"))?; - Ok(crate::table_catalog::ObjectTableCatalogStore::new( - crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store), - )) + Ok(crate::table_catalog::EcStoreTableCatalogObjectBackend::new(store)) +} + +fn table_catalog_store() -> S3Result> { + let backend = table_catalog_backend()?; + Ok(crate::table_catalog::ObjectTableCatalogStore::new(backend)) } async fn table_bucket_enabled_from_metadata(bucket: &str) -> S3Result { @@ -375,10 +377,10 @@ fn list_tables_response_from_entries(entries: Vec RestLoadTableResponse { +fn load_table_response_from_entry(entry: crate::table_catalog::TableEntry, metadata: serde_json::Value) -> RestLoadTableResponse { RestLoadTableResponse { metadata_location: entry.metadata_location, - metadata: serde_json::Value::Null, + metadata, config: BTreeMap::from([("warehouse-location".to_string(), entry.warehouse_location)]), } } @@ -418,6 +420,9 @@ fn table_entry_from_register_request( namespace: &crate::table_catalog::Namespace, request: RegisterTableRequest, ) -> S3Result { + if request.overwrite { + return Err(s3_error!(NotImplemented, "register table overwrite is not supported")); + } let table = crate::table_catalog::IdentifierSegment::parse(request.name) .map_err(|err| s3_error!(InvalidRequest, "invalid table name: {}", err))?; if !crate::table_catalog::is_valid_table_metadata_location(namespace, &table, &request.metadata_location) { @@ -439,7 +444,7 @@ fn table_entry_from_register_request( version_token: format!("token-{}", Uuid::new_v4()), generation: 1, state: crate::table_catalog::TableCatalogEntryState::Active, - properties: request.properties, + properties: BTreeMap::new(), created_at: None, updated_at: None, }) @@ -450,15 +455,18 @@ fn table_entry_from_create_table_request( namespace: &crate::table_catalog::Namespace, request: CreateTableRequest, ) -> S3Result { - table_entry_from_register_request( + let properties = request.properties; + let mut entry = table_entry_from_register_request( bucket, namespace, RegisterTableRequest { name: request.name, metadata_location: request.metadata_location, - properties: request.properties, + overwrite: false, }, - ) + )?; + entry.properties = properties; + Ok(entry) } fn namespace_entry_from_create_request( @@ -545,6 +553,7 @@ where async fn register_table_response( store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, bucket: &str, namespace: &crate::table_catalog::Namespace, request: RegisterTableRequest, @@ -555,12 +564,14 @@ where { let entry = table_entry_from_register_request(bucket, namespace, request)?; ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; store.register_table(entry.clone()).await.map_err(catalog_store_error)?; - Ok(load_table_response_from_entry(entry)) + Ok(load_table_response_from_entry(entry, metadata)) } async fn create_table_response( store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, bucket: &str, namespace: &crate::table_catalog::Namespace, request: CreateTableRequest, @@ -571,8 +582,29 @@ where { let entry = table_entry_from_create_table_request(bucket, namespace, request)?; ensure_table_bucket_entry(store, bucket, table_bucket_enabled).await?; + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; store.create_table(entry.clone()).await.map_err(catalog_store_error)?; - Ok(load_table_response_from_entry(entry)) + Ok(load_table_response_from_entry(entry, metadata)) +} + +async fn read_table_metadata_json( + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, + bucket: &str, + metadata_location: &str, +) -> S3Result { + let Some(object) = metadata_backend + .read_object(bucket, metadata_location) + .await + .map_err(catalog_store_error)? + else { + return Err(s3_error!(InvalidRequest, "table metadata object not found: {metadata_location}")); + }; + let metadata = serde_json::from_slice::(&object.data) + .map_err(|err| s3_error!(InvalidRequest, "failed to parse table metadata JSON: {}", err))?; + if !metadata.is_object() { + return Err(s3_error!(InvalidRequest, "table metadata JSON must be an object")); + } + Ok(metadata) } async fn list_tables_response( @@ -592,6 +624,7 @@ where async fn load_table_response( store: &S, + metadata_backend: &impl crate::table_catalog::TableCatalogObjectBackend, bucket: &str, namespace: &crate::table_catalog::Namespace, table: &str, @@ -606,7 +639,8 @@ where else { return Err(s3_error!(InvalidRequest, "table not found")); }; - Ok(load_table_response_from_entry(entry)) + let metadata = read_table_metadata_json(metadata_backend, bucket, &entry.metadata_location).await?; + Ok(load_table_response_from_entry(entry, metadata)) } async fn commit_table_response( @@ -723,9 +757,11 @@ impl Operation for RestCreateTableHandler { let warehouse = warehouse_from_params(¶ms)?; let namespace = namespace_from_params(¶ms)?; let request = read_json_body::(req.input).await?; - let store = table_catalog_store()?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; - let response = create_table_response(&store, &warehouse, &namespace, request, table_bucket_enabled).await?; + let response = + create_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?; build_json_response(StatusCode::OK, &response) } } @@ -739,9 +775,11 @@ impl Operation for RestRegisterTableHandler { let warehouse = warehouse_from_params(¶ms)?; let namespace = namespace_from_params(¶ms)?; let request = read_json_body::(req.input).await?; - let store = table_catalog_store()?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); let table_bucket_enabled = table_bucket_enabled_from_metadata(&warehouse).await?; - let response = register_table_response(&store, &warehouse, &namespace, request, table_bucket_enabled).await?; + let response = + register_table_response(&store, &metadata_backend, &warehouse, &namespace, request, table_bucket_enabled).await?; build_json_response(StatusCode::OK, &response) } } @@ -755,8 +793,9 @@ impl Operation for RestLoadTableHandler { let warehouse = warehouse_from_params(¶ms)?; let namespace = namespace_from_params(¶ms)?; let table = table_name_from_params(¶ms)?; - let store = table_catalog_store()?; - let response = load_table_response(&store, &warehouse, &namespace, &table).await?; + let metadata_backend = table_catalog_backend()?; + let store = crate::table_catalog::ObjectTableCatalogStore::new(metadata_backend.clone()); + let response = load_table_response(&store, &metadata_backend, &warehouse, &namespace, &table).await?; build_json_response(StatusCode::OK, &response) } } @@ -796,6 +835,7 @@ impl Operation for RestDropTableHandler { mod tests { use super::*; use crate::table_catalog::TableCatalogStore; + use std::sync::Arc; #[test] fn catalog_config_response_lists_mvp_rest_endpoints() { @@ -810,6 +850,11 @@ mod tests { .endpoints .contains(&"POST /{warehouse}/namespaces/{namespace}/register") ); + assert!( + !response + .endpoints + .contains(&"POST /{warehouse}/namespaces/{namespace}/tables") + ); assert!( response .endpoints @@ -928,9 +973,7 @@ mod tests { let request: RegisterTableRequest = serde_json::from_value(serde_json::json!({ "name": "events", "metadata-location": ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json", - "properties": { - "write.format.default": "parquet" - } + "overwrite": false })) .expect("request should parse"); @@ -943,11 +986,44 @@ mod tests { entry.metadata_location, ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json" ); - assert_eq!(entry.properties.get("write.format.default").map(String::as_str), Some("parquet")); + assert!(entry.properties.is_empty()); assert_eq!(entry.generation, 1); assert!(!entry.version_token.is_empty()); } + #[test] + fn load_table_response_includes_rest_metadata_payload() { + let metadata = serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id" + }); + let response = load_table_response_from_entry( + crate::table_catalog::TableEntry { + version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, + table_bucket: "warehouse".to_string(), + namespace: "analytics".to_string(), + table: "events".to_string(), + table_id: "table-id".to_string(), + table_uuid: "table-uuid".to_string(), + format: "ICEBERG".to_string(), + format_version: 2, + warehouse_location: "s3://warehouse/tables/table-id".to_string(), + metadata_location: + ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json".to_string(), + version_token: "token-v1".to_string(), + generation: 1, + state: crate::table_catalog::TableCatalogEntryState::Active, + properties: BTreeMap::new(), + created_at: None, + updated_at: None, + }, + metadata.clone(), + ); + + assert_eq!(response.metadata, metadata); + } + #[test] fn commit_table_request_uses_rest_commit_fields() { let request: RestCommitTableRequest = serde_json::from_value(serde_json::json!({ @@ -987,6 +1063,89 @@ mod tests { commits: tokio::sync::Mutex>, } + #[derive(Clone, Default)] + struct TestTableCatalogObjectBackend { + objects: Arc>>, + } + + impl TestTableCatalogObjectBackend { + async fn put_json(&self, bucket: &str, object: &str, value: serde_json::Value) { + let data = serde_json::to_vec(&value).expect("metadata JSON should serialize"); + self.objects.lock().await.insert( + (bucket.to_string(), object.to_string()), + crate::table_catalog::TableCatalogObject { + data, + etag: Some("etag".to_string()), + }, + ); + } + } + + #[async_trait::async_trait] + impl crate::table_catalog::TableCatalogObjectBackend for TestTableCatalogObjectBackend { + async fn read_object( + &self, + bucket: &str, + object: &str, + ) -> crate::table_catalog::TableCatalogStoreResult> { + Ok(self + .objects + .lock() + .await + .get(&(bucket.to_string(), object.to_string())) + .cloned()) + } + + async fn object_exists(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult { + Ok(self + .objects + .lock() + .await + .contains_key(&(bucket.to_string(), object.to_string()))) + } + + async fn put_object( + &self, + bucket: &str, + object: &str, + data: Vec, + _precondition: crate::table_catalog::TableCatalogPutPrecondition, + ) -> crate::table_catalog::TableCatalogStoreResult<()> { + self.objects.lock().await.insert( + (bucket.to_string(), object.to_string()), + crate::table_catalog::TableCatalogObject { + data, + etag: Some("etag".to_string()), + }, + ); + Ok(()) + } + + async fn delete_object(&self, bucket: &str, object: &str) -> crate::table_catalog::TableCatalogStoreResult<()> { + self.objects.lock().await.remove(&(bucket.to_string(), object.to_string())); + Ok(()) + } + + async fn list_objects(&self, bucket: &str, prefix: &str) -> crate::table_catalog::TableCatalogStoreResult> { + Ok(self + .objects + .lock() + .await + .keys() + .filter(|(object_bucket, object)| object_bucket == bucket && object.starts_with(prefix)) + .map(|(_, object)| object.clone()) + .collect()) + } + + async fn acquire_write_lock( + &self, + _bucket: &str, + _object: &str, + ) -> crate::table_catalog::TableCatalogStoreResult> { + Ok(Box::new(())) + } + } + #[async_trait::async_trait] impl crate::table_catalog::TableCatalogStore for TestTableCatalogStore { async fn get_table_bucket( @@ -1288,6 +1447,7 @@ mod tests { #[tokio::test] async fn table_helpers_call_catalog_store() { let store = TestTableCatalogStore::default(); + let metadata_backend = TestTableCatalogObjectBackend::default(); let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); ensure_table_bucket_entry(&store, "warehouse", true) .await @@ -1306,14 +1466,26 @@ mod tests { let metadata_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00001.metadata.json"; + metadata_backend + .put_json( + "warehouse", + metadata_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id" + }), + ) + .await; let register = register_table_response( &store, + &metadata_backend, "warehouse", &namespace, RegisterTableRequest { name: "events".to_string(), metadata_location: metadata_location.to_string(), - properties: BTreeMap::new(), + overwrite: false, }, true, ) @@ -1321,16 +1493,18 @@ mod tests { .expect("table should register"); assert_eq!(register.metadata_location, metadata_location); + assert_eq!(register.metadata["format-version"], 2); let list = list_tables_response(&store, "warehouse", &namespace) .await .expect("table list should load"); assert_eq!(list.identifiers[0].name, "events"); - let load = load_table_response(&store, "warehouse", &namespace, "events") + let load = load_table_response(&store, &metadata_backend, "warehouse", &namespace, "events") .await .expect("table should load"); assert_eq!(load.metadata_location, metadata_location); + assert_eq!(load.metadata["table-uuid"], "table-uuid"); let current = store .load_table("warehouse", "analytics", "events") @@ -1339,6 +1513,18 @@ mod tests { .expect("table should exist"); let next_metadata_location = ".rustfs-table/warehouses/default/namespaces/analytics/tables/events/metadata/00002.metadata.json"; + metadata_backend + .put_json( + "warehouse", + next_metadata_location, + serde_json::json!({ + "format-version": 2, + "table-uuid": "table-uuid", + "location": "s3://warehouse/tables/table-id", + "last-sequence-number": 2 + }), + ) + .await; let commit = commit_table_response( &store, "warehouse", @@ -1372,6 +1558,10 @@ mod tests { drop_table_in_store(&store, "warehouse", &namespace, "events") .await .expect("table should drop"); - assert!(load_table_response(&store, "warehouse", &namespace, "events").await.is_err()); + assert!( + load_table_response(&store, &metadata_backend, "warehouse", &namespace, "events") + .await + .is_err() + ); } }