fix(spec): reject reserved system field names at create and alter time

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()

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<String, String> = 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"),