diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index b3c48d13f..00abf3e35 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -203,31 +203,17 @@ impl TableSchema { /// colliding with a system field (e.g. `_ROW_ID`) is otherwise excluded /// from the physical read and silently filled with the system value. fn validate_no_reserved_fields(&self) -> crate::Result<()> { - // Java SpecialFields.SYSTEM_FIELD_NAMES. - const SYSTEM_FIELD_NAMES: [&str; 5] = [ - SEQUENCE_NUMBER_FIELD_NAME, - VALUE_KIND_FIELD_NAME, - "_LEVEL", - ROW_KIND_FIELD_NAME, - ROW_ID_FIELD_NAME, - ]; - const KEY_FIELD_PREFIX: &str = "_KEY_"; // Java SpecialFields.SYSTEM_FIELD_ID_START = Integer.MAX_VALUE / 2. const SYSTEM_FIELD_ID_START: i32 = i32::MAX / 2; + validate_no_reserved_field_names(&self.fields)?; + for field in &self.fields { - let name = field.name(); - if name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) { - return Err(crate::Error::ConfigInvalid { - message: format!( - "Field name '{name}' is reserved for system use and cannot be used in a table schema" - ), - }); - } if field.id() >= SYSTEM_FIELD_ID_START { return Err(crate::Error::DataInvalid { message: format!( - "Field '{name}' uses reserved system field id {}", + "Field '{}' uses reserved system field id {}", + field.name(), field.id() ), source: None, @@ -562,6 +548,7 @@ impl TableSchema { // Re-run create-time validations on the final schema, mirroring Java // `SchemaValidation.validateTableSchema` after applying changes. + validate_no_reserved_field_names(&new_schema.fields)?; Schema::validate_key_field_types( &new_schema.fields, &new_schema.primary_keys, @@ -625,6 +612,35 @@ impl TableSchema { } } +/// Reject column names reserved for system use, mirroring Java `SpecialFields`: +/// the five `SYSTEM_FIELD_NAMES` and the `_KEY_` key-field prefix. +/// +/// A user column colliding with a system field is otherwise excluded from the +/// physical read and silently filled with the system value. +fn validate_no_reserved_field_names(fields: &[DataField]) -> crate::Result<()> { + // Java SpecialFields.SYSTEM_FIELD_NAMES. + const SYSTEM_FIELD_NAMES: [&str; 5] = [ + SEQUENCE_NUMBER_FIELD_NAME, + VALUE_KIND_FIELD_NAME, + "_LEVEL", + ROW_KIND_FIELD_NAME, + ROW_ID_FIELD_NAME, + ]; + const KEY_FIELD_PREFIX: &str = "_KEY_"; + + for field in fields { + let name = field.name(); + if name.starts_with(KEY_FIELD_PREFIX) || SYSTEM_FIELD_NAMES.contains(&name) { + return Err(crate::Error::ConfigInvalid { + message: format!( + "Field name '{name}' is reserved for system use and cannot be used in a table schema" + ), + }); + } + } + Ok(()) +} + /// Extract the single top-level column name from a `field_names` path. /// /// Nested struct field paths (length > 1) are not yet supported. @@ -1084,6 +1100,7 @@ impl Schema { let partition_keys = Self::normalize_partition_keys(&partition_keys, &mut options)?; Self::normalize_blob_comment_directives(&mut fields, &mut options)?; let fields = Self::normalize_fields(&fields, &partition_keys, &primary_keys, &options)?; + validate_no_reserved_field_names(&fields)?; Self::validate_key_field_types(&fields, &primary_keys, &options)?; Self::validate_blob_fields(&fields, &partition_keys, &options)?; Self::validate_vector_store_fields(&fields, &partition_keys, &options)?; @@ -3393,6 +3410,99 @@ mod tests { assert_primary_key_index_column_changes_rejected(&table_schema, "embedding", vector_type); } + #[test] + fn test_create_schema_rejects_reserved_field_names() { + // Java `SpecialFields.SYSTEM_FIELD_NAMES` plus the `_KEY_` prefix. A + // user column with one of these names is excluded from the physical + // read and silently filled with the system value. + for name in [ + "_SEQUENCE_NUMBER", + "_VALUE_KIND", + "_LEVEL", + "rowkind", + "_ROW_ID", + "_KEY_id", + ] { + let err = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column(name, DataType::Int(IntType::new())) + .build() + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains(name) && message.contains("reserved")), + "reserved field name '{name}' should be rejected at create time, got {err:?}" + ); + } + } + + #[test] + fn test_create_schema_accepts_names_that_only_look_reserved() { + // Guard against over-rejecting: only exact system names and the + // `_KEY_` prefix are reserved. + let schema = Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("_sequence_number", DataType::Int(IntType::new())) + .column("row_kind", DataType::Int(IntType::new())) + .column("_KEY", DataType::Int(IntType::new())) + .build(); + + assert!( + schema.is_ok(), + "names that merely resemble system fields should be accepted, got {schema:?}" + ); + } + + #[test] + fn test_alter_add_reserved_field_name_rejected() { + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .build() + .unwrap(), + ); + + let err = table_schema + .apply_changes(vec![crate::spec::SchemaChange::add_column( + "_ROW_ID".to_string(), + DataType::Int(IntType::new()), + )]) + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("_ROW_ID") && message.contains("reserved")), + "adding a reserved column name should be rejected, got {err:?}" + ); + } + + #[test] + fn test_alter_rename_to_reserved_field_name_rejected() { + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("v", DataType::Int(IntType::new())) + .build() + .unwrap(), + ); + + let err = table_schema + .apply_changes(vec![crate::spec::SchemaChange::rename_column( + "v".to_string(), + "_VALUE_KIND".to_string(), + )]) + .unwrap_err(); + + assert!( + matches!(err, crate::Error::ConfigInvalid { ref message } + if message.contains("_VALUE_KIND") && message.contains("reserved")), + "renaming a column to a reserved name should be rejected, got {err:?}" + ); + } + #[test] fn test_create_schema_rejects_unknown_bucket_key() { let err = Schema::builder() diff --git a/crates/paimon/src/table/hybrid_search_builder.rs b/crates/paimon/src/table/hybrid_search_builder.rs index ab2908206..fef41aa74 100644 --- a/crates/paimon/src/table/hybrid_search_builder.rs +++ b/crates/paimon/src/table/hybrid_search_builder.rs @@ -1638,24 +1638,33 @@ mod pk_hybrid_tests { /// A primary-key hybrid table whose user schema carries an extra column named /// `reserved`, used to prove the materialized read rejects a reserved metadata /// name arriving via the default (all-columns) projection. + /// + /// Deserialized from JSON rather than built through `Schema::builder`, because + /// `Schema::new` rejects reserved system field names at create time. The read + /// guard exists for exactly this case: metadata written by another engine. fn pk_hybrid_table_with_reserved_column(reserved: &str) -> Table { let file_io = FileIOBuilder::new("memory").build().unwrap(); - let mut builder = Schema::builder() - .column("id", DataType::Int(IntType::new())) - .column( - VECTOR_COLUMN, - DataType::Vector( - VectorType::try_new(true, DIM as u32, DataType::Float(FloatType::new())) - .unwrap(), - ), - ) - .column(TEXT_COLUMN, DataType::VarChar(VarCharType::string_type())) - .column(reserved, DataType::VarChar(VarCharType::string_type())) - .primary_key(["id"]); - for (k, v) in table_options() { - builder = builder.option(k, v); - } - let schema = TableSchema::new(0, &builder.build().unwrap()); + let options: HashMap = table_options().into_iter().collect(); + let schema: TableSchema = serde_json::from_value(serde_json::json!({ + "version": TableSchema::CURRENT_VERSION, + "id": 0, + "fields": [ + {"id": 0, "name": "id", "type": "INT NOT NULL"}, + { + "id": 1, + "name": VECTOR_COLUMN, + "type": {"type": "VECTOR", "element": "FLOAT", "length": DIM}, + }, + {"id": 2, "name": TEXT_COLUMN, "type": "STRING"}, + {"id": 3, "name": reserved, "type": "STRING"}, + ], + "highestFieldId": 3, + "partitionKeys": [], + "primaryKeys": ["id"], + "options": options, + "timeMillis": 0, + })) + .expect("reserved-column schema should deserialize"); Table::new( file_io, Identifier::new("default", "pk_hybrid_reserved"),