From 9e6e02ea09c86bedf44c7bd64a74ea02a0cff1de Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Mon, 17 Aug 2026 01:06:25 +0800 Subject: [PATCH] fix(table-catalog): assign fresh schema IDs on create (#6146) * fix(table-catalog): assign fresh schema IDs on create * fix(table-catalog): accept negative create schema IDs --------- Co-authored-by: Henry Guo --- .../src/admin/handlers/table_catalog/mod.rs | 9 +- .../src/admin/handlers/table_catalog/tests.rs | 202 +++++++++++++++++- .../src/table_catalog/iceberg/validation.rs | 183 ++++++++++++++++ 3 files changed, 390 insertions(+), 4 deletions(-) diff --git a/rustfs/src/admin/handlers/table_catalog/mod.rs b/rustfs/src/admin/handlers/table_catalog/mod.rs index 1dd03db42..d3138e96e 100644 --- a/rustfs/src/admin/handlers/table_catalog/mod.rs +++ b/rustfs/src/admin/handlers/table_catalog/mod.rs @@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request( let CreateTableRequest { name, location, - schema, - partition_spec, - write_order, + mut schema, + mut partition_spec, + mut write_order, stage_create, mut properties, } = request; @@ -3031,6 +3031,9 @@ fn table_entry_from_create_table_request( let metadata_location = crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id)); + crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut()) + .map_err(catalog_store_error)?; + let entry = crate::table_catalog::TableEntry { version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION, table_bucket: bucket.to_string(), diff --git a/rustfs/src/admin/handlers/table_catalog/tests.rs b/rustfs/src/admin/handlers/table_catalog/tests.rs index 013ae9304..969378424 100644 --- a/rustfs/src/admin/handlers/table_catalog/tests.rs +++ b/rustfs/src/admin/handlers/table_catalog/tests.rs @@ -1873,6 +1873,66 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() { assert_eq!(request.name, "events"); } +#[test] +fn create_table_assigns_positive_ids_to_spark_schema() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let request: CreateTableRequest = serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": { + "type": "struct", + "schema-id": 0, + "fields": [ + {"id": 0, "name": "id", "required": false, "type": "long"}, + {"id": 1, "name": "payload", "required": false, "type": "string"} + ] + }, + "partition-spec": {"spec-id": 0, "fields": []}, + "properties": {"owner": "spark"} + })) + .expect("Spark create table request should parse"); + + let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request) + .expect("catalog should assign positive field IDs"); + + assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1); + assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2); + assert_eq!(metadata["last-column-id"], 2); +} + +#[test] +fn create_table_assigns_fresh_id_to_negative_temporary_field_id() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let request: CreateTableRequest = serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": { + "type": "struct", + "identifier-field-ids": [-1], + "fields": [{"id": -1, "name": "id", "required": true, "type": "long"}] + }, + "partition-spec": { + "fields": [{"source-id": -1, "name": "id", "transform": "identity"}] + }, + "write-order": { + "fields": [{ + "source-id": -1, + "transform": "identity", + "direction": "asc", + "null-order": "nulls-first" + }] + } + })) + .expect("create table request with a negative temporary field ID should parse"); + + let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request) + .expect("catalog should replace the negative temporary field ID"); + + assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1); + assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1])); + assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1); + assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1); + assert_eq!(metadata["last-column-id"], 1); +} + #[test] fn create_table_request_honors_supported_format_version_property() { let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); @@ -1990,6 +2050,142 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() { assert_eq!(updated["default-sort-order-id"], 0); } +#[test] +fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let request: CreateTableRequest = serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": { + "type": "struct", + "schema-id": 41, + "identifier-field-ids": [0], + "fields": [ + {"id": 0, "name": "id", "required": true, "type": "long"}, + { + "id": 10, + "name": "details", + "required": false, + "type": { + "type": "struct", + "fields": [{"id": 11, "name": "category", "required": false, "type": "string"}] + } + }, + { + "id": 20, + "name": "tags", + "required": false, + "type": { + "type": "list", + "element-id": 21, + "element-required": false, + "element": "string" + } + }, + { + "id": 30, + "name": "attributes", + "required": false, + "type": { + "type": "map", + "key-id": 31, + "key": "string", + "value-id": 32, + "value-required": false, + "value": { + "type": "struct", + "fields": [{"id": 33, "name": "score", "required": false, "type": "int"}] + } + } + } + ] + }, + "partition-spec": { + "spec-id": 42, + "fields": [{"source-id": 0, "name": "id", "transform": "identity"}] + }, + "write-order": { + "order-id": 43, + "fields": [{ + "source-id": 11, + "transform": "identity", + "direction": "asc", + "null-order": "nulls-first" + }] + } + })) + .expect("create table request should parse"); + + let (_, metadata) = + table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs"); + + let schema = &metadata["schemas"][0]; + assert_eq!(schema["fields"][0]["id"], 1); + assert_eq!(schema["fields"][1]["id"], 2); + assert_eq!(schema["fields"][2]["id"], 3); + assert_eq!(schema["fields"][3]["id"], 4); + assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5); + assert_eq!(schema["fields"][2]["type"]["element-id"], 6); + assert_eq!(schema["fields"][3]["type"]["key-id"], 7); + assert_eq!(schema["fields"][3]["type"]["value-id"], 8); + assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9); + assert_eq!(schema["identifier-field-ids"], serde_json::json!([1])); + assert_eq!(metadata["last-column-id"], 9); + assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1); + assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5); +} + +#[test] +fn create_table_rejects_duplicate_temporary_schema_field_ids() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let request: CreateTableRequest = serde_json::from_value(serde_json::json!({ + "name": "events", + "schema": { + "type": "struct", + "fields": [ + {"id": 0, "name": "id", "required": false, "type": "long"}, + {"id": 0, "name": "payload", "required": false, "type": "string"} + ] + } + })) + .expect("create table request should parse"); + + let error = table_entry_from_create_table_request("warehouse", &namespace, request) + .expect_err("duplicate temporary field IDs must be rejected"); + + assert_eq!(error.message(), Some("duplicate create schema field id 0")); +} + +#[test] +fn create_table_rejects_excessive_schema_nesting() { + let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); + let mut field_type = serde_json::Value::from("long"); + for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 { + field_type = serde_json::json!({ + "type": "list", + "element-id": element_id, + "element-required": false, + "element": field_type + }); + } + let request = CreateTableRequest { + name: "events".to_string(), + location: None, + schema: serde_json::json!({ + "type": "struct", + "fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}] + }), + partition_spec: None, + write_order: None, + stage_create: false, + properties: BTreeMap::new(), + }; + + let error = table_entry_from_create_table_request("warehouse", &namespace, request) + .expect_err("excessively nested create schemas must be rejected"); + + assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth")); +} + #[test] fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() { let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse"); @@ -2247,7 +2443,11 @@ fn create_table_counts_collection_ids_in_last_column_id() { let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created"); - assert_eq!(metadata["last-column-id"], 9); + let schema = &metadata["schemas"][0]; + assert_eq!(schema["fields"][0]["type"]["element-id"], 3); + assert_eq!(schema["fields"][1]["type"]["key-id"], 4); + assert_eq!(schema["fields"][1]["type"]["value-id"], 5); + assert_eq!(metadata["last-column-id"], 5); } #[test] diff --git a/rustfs/src/table_catalog/iceberg/validation.rs b/rustfs/src/table_catalog/iceberg/validation.rs index 0783b5f78..8e2455b19 100644 --- a/rustfs/src/table_catalog/iceberg/validation.rs +++ b/rustfs/src/table_catalog/iceberg/validation.rs @@ -19,6 +19,7 @@ use futures::{StreamExt, TryStreamExt, stream}; use super::super::*; const ICEBERG_MAX_USER_FIELD_ID: i32 = i32::MAX - 200; +pub(crate) const ICEBERG_MAX_SCHEMA_NESTING_DEPTH: usize = 128; fn normalize_warehouse_object_prefix(object_prefix: &str, max_prefix_depth: Option) -> TableCatalogStoreResult { let object_prefix = object_prefix.strip_suffix('/').unwrap_or(object_prefix); @@ -1361,6 +1362,188 @@ fn validate_iceberg_schema(schema: &serde_json::Value, label: &str) -> TableCata Ok(validate_iceberg_schema_fields(schema, label)?.field_ids) } +pub(crate) fn assign_fresh_create_schema_ids( + schema: &mut serde_json::Value, + partition_spec: Option<&mut serde_json::Value>, + sort_order: Option<&mut serde_json::Value>, +) -> TableCatalogStoreResult<()> { + let mut assigner = FreshCreateSchemaIdAssigner::new(); + assigner.assign_schema(schema)?; + assigner.remap_identifier_field_ids(schema)?; + if let Some(partition_spec) = partition_spec { + assigner.remap_source_ids(partition_spec, "partition spec")?; + } + if let Some(sort_order) = sort_order { + assigner.remap_source_ids(sort_order, "sort order")?; + } + Ok(()) +} + +struct FreshCreateSchemaIdAssigner { + next_id: i32, + old_to_new: BTreeMap, +} + +impl FreshCreateSchemaIdAssigner { + fn new() -> Self { + Self { + next_id: 1, + old_to_new: BTreeMap::new(), + } + } + + fn assign_schema(&mut self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> { + let schema = schema + .as_object_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema must be a JSON object".to_string()))?; + if schema.get("type").and_then(serde_json::Value::as_str) != Some("struct") { + return Err(TableCatalogStoreError::Invalid("create schema type must be struct".to_string())); + } + let fields = schema + .get_mut("fields") + .and_then(serde_json::Value::as_array_mut) + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be an array".to_string()))?; + self.assign_struct_fields(fields, 0) + } + + fn assign_struct_fields(&mut self, fields: &mut [serde_json::Value], depth: usize) -> TableCatalogStoreResult<()> { + for field in fields.iter_mut() { + let field = field + .as_object_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?; + self.assign_object_id(field, "id", "create schema field id")?; + } + for field in fields { + let field = field + .as_object_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema fields must be JSON objects".to_string()))?; + let field_type = field + .get_mut("type") + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema field type is required".to_string()))?; + self.assign_type_ids(field_type, depth)?; + } + Ok(()) + } + + fn assign_type_ids(&mut self, field_type: &mut serde_json::Value, depth: usize) -> TableCatalogStoreResult<()> { + if field_type.is_string() { + return Ok(()); + } + if depth >= ICEBERG_MAX_SCHEMA_NESTING_DEPTH { + return Err(TableCatalogStoreError::Invalid( + "create schema exceeds the maximum nesting depth".to_string(), + )); + } + let nested_depth = depth + 1; + let field_type = field_type.as_object_mut().ok_or_else(|| { + TableCatalogStoreError::Invalid("create schema field type must be a string or JSON object".to_string()) + })?; + match field_type.get("type").and_then(serde_json::Value::as_str) { + Some("struct") => { + let fields = field_type + .get_mut("fields") + .and_then(serde_json::Value::as_array_mut) + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema struct fields must be an array".to_string()))?; + self.assign_struct_fields(fields, nested_depth) + } + Some("list") => { + self.assign_object_id(field_type, "element-id", "create schema list element-id")?; + let element = field_type + .get_mut("element") + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema list element is required".to_string()))?; + self.assign_type_ids(element, nested_depth) + } + Some("map") => { + self.assign_object_id(field_type, "key-id", "create schema map key-id")?; + self.assign_object_id(field_type, "value-id", "create schema map value-id")?; + let key = field_type + .get_mut("key") + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema map key is required".to_string()))?; + self.assign_type_ids(key, nested_depth)?; + let value = field_type + .get_mut("value") + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema map value is required".to_string()))?; + self.assign_type_ids(value, nested_depth) + } + _ => Err(TableCatalogStoreError::Invalid( + "create schema contains an unsupported field type".to_string(), + )), + } + } + + fn assign_object_id( + &mut self, + object: &mut serde_json::Map, + field: &str, + label: &str, + ) -> TableCatalogStoreResult<()> { + let old_id = required_i32_value(object, field, label)?; + let entry = match self.old_to_new.entry(old_id) { + std::collections::btree_map::Entry::Occupied(_) => { + return Err(TableCatalogStoreError::Invalid(format!("duplicate create schema field id {old_id}"))); + } + std::collections::btree_map::Entry::Vacant(entry) => entry, + }; + let new_id = self.next_id; + if new_id > ICEBERG_MAX_USER_FIELD_ID { + return Err(TableCatalogStoreError::Invalid( + "create schema exceeds the available Iceberg field ID range".to_string(), + )); + } + self.next_id = new_id.checked_add(1).ok_or_else(|| { + TableCatalogStoreError::Invalid("create schema exceeds the available Iceberg field ID range".to_string()) + })?; + entry.insert(new_id); + object.insert(field.to_string(), serde_json::Value::from(new_id)); + Ok(()) + } + + fn remap_identifier_field_ids(&self, schema: &mut serde_json::Value) -> TableCatalogStoreResult<()> { + let Some(identifier_field_ids) = schema + .as_object_mut() + .and_then(|schema| schema.get_mut("identifier-field-ids")) + else { + return Ok(()); + }; + let identifier_field_ids = identifier_field_ids + .as_array_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid("create schema identifier-field-ids must be an array".to_string()))?; + for field_id in identifier_field_ids { + let old_id = required_i32(field_id, "create schema identifier field id")?; + let new_id = self.old_to_new.get(&old_id).ok_or_else(|| { + TableCatalogStoreError::Invalid(format!( + "create schema identifier field id {old_id} does not reference a schema field" + )) + })?; + *field_id = serde_json::Value::from(*new_id); + } + Ok(()) + } + + fn remap_source_ids(&self, value: &mut serde_json::Value, label: &str) -> TableCatalogStoreResult<()> { + let value = value + .as_object_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} must be a JSON object")))?; + let Some(fields) = value.get_mut("fields") else { + return Ok(()); + }; + let fields = fields + .as_array_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be an array")))?; + for field in fields { + let field = field + .as_object_mut() + .ok_or_else(|| TableCatalogStoreError::Invalid(format!("{label} fields must be JSON objects")))?; + let old_id = required_i32_value(field, "source-id", &format!("{label} source-id"))?; + let new_id = self.old_to_new.get(&old_id).ok_or_else(|| { + TableCatalogStoreError::Invalid(format!("{label} source-id {old_id} does not reference the create schema")) + })?; + field.insert("source-id".to_string(), serde_json::Value::from(*new_id)); + } + Ok(()) + } +} + fn validate_iceberg_schema_fields(schema: &serde_json::Value, label: &str) -> TableCatalogStoreResult { let schema = schema .as_object()