mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-16 18:08:21 +00:00
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 <marshawcoco@users.noreply.github.com>
This commit is contained in:
Henry Guo
@@ -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(),
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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<usize>) -> TableCatalogStoreResult<String> {
|
||||
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<i32, i32>,
|
||||
}
|
||||
|
||||
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<String, serde_json::Value>,
|
||||
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<IcebergSchemaFields> {
|
||||
let schema = schema
|
||||
.as_object()
|
||||
|
||||
Reference in New Issue
Block a user