fix(spec): validate sequence.field against table schema

Creating a table with `sequence.field` referencing a missing column was
silently accepted, then the write path resolved sequence fields with a
lenient lookup and fell back to the auto-increment sequence, ignoring the
user's ordering intent. Repeated fields, `merge-engine=first-row` and
cross-partition update tables were likewise unchecked.

Validate `sequence.field` at create and alter time, mirroring Java
`SchemaValidation#validateSequenceField`: every listed field must exist,
must not repeat, first-row merge engine rejects user-defined sequence
fields, and cross-partition update tables (primary key constraint not
including all partition fields) reject `sequence.field` because partition
migration retracts old rows with generated DELETEs whose ordering a
user-provided sequence value could break.

Author: jackylee-ch

crates/paimon/src/spec/schema.rs

@@ -502,6 +502,12 @@ impl TableSchema {
             &new_schema.partition_keys,
             &new_schema.primary_keys,
         )?;
+        Schema::validate_sequence_field(
+            &new_schema.options,
+            &new_schema.fields,
+            &new_schema.partition_keys,
+            &new_schema.primary_keys,
+        )?;
         Schema::validate_read_batch_size(&new_schema.options)?;
         Ok(new_schema)
     }
@@ -979,6 +985,7 @@ impl Schema {
         Self::validate_first_row_changelog_producer(&options)?;
         Self::validate_rowkind_field(&options, &primary_keys, &fields)?;
         Self::validate_bucket_keys(&options, &fields, &partition_keys, &primary_keys)?;
+        Self::validate_sequence_field(&options, &fields, &partition_keys, &primary_keys)?;
         Self::validate_read_batch_size(&options)?;
 
         Ok(Self {
@@ -1461,6 +1468,69 @@ impl Schema {
             .map_err(Self::options_error_to_config_invalid)
     }
 
+    /// Validate the `sequence.field` option against the schema, mirroring
+    /// Java `SchemaValidation#validateSequenceField`:
+    /// * every listed field must exist in the table schema — otherwise the
+    ///   write path silently falls back to the auto-increment sequence
+    ///   (`TableWrite` resolves sequence fields with a lenient lookup) and
+    ///   merge results would ignore the user's ordering intent;
+    /// * a field must not be listed more than once;
+    /// * `merge-engine=first-row` does not support user-defined sequence
+    ///   fields;
+    /// * cross-partition update tables (primary key constraint not including
+    ///   all partition fields) do not support user-defined sequence fields,
+    ///   because partition migration retracts old rows with generated DELETEs
+    ///   whose ordering a user-provided sequence value could break.
+    fn validate_sequence_field(
+        options: &HashMap<String, String>,
+        fields: &[DataField],
+        partition_keys: &[String],
+        primary_keys: &[String],
+    ) -> crate::Result<()> {
+        let core = CoreOptions::new(options);
+        let sequence_fields = core.sequence_fields();
+        if sequence_fields.is_empty() {
+            return Ok(());
+        }
+
+        let mut seen: HashSet<&str> = HashSet::new();
+        for name in &sequence_fields {
+            if fields.iter().all(|f| f.name() != *name) {
+                return Err(crate::Error::ConfigInvalid {
+                    message: format!("Sequence field '{name}' can not be found in table schema."),
+                });
+            }
+            if !seen.insert(name) {
+                return Err(crate::Error::ConfigInvalid {
+                    message: format!("Sequence field '{name}' is defined repeatedly."),
+                });
+            }
+        }
+
+        let merge_engine = core
+            .merge_engine()
+            .map_err(Self::options_error_to_config_invalid)?;
+        if merge_engine == MergeEngine::FirstRow {
+            return Err(crate::Error::ConfigInvalid {
+                message: "Do not support use sequence field on FIRST_ROW merge engine.".to_string(),
+            });
+        }
+
+        let cross_partition_update = !primary_keys.is_empty()
+            && !partition_keys.is_empty()
+            && partition_keys.iter().any(|pt| !primary_keys.contains(pt));
+        if cross_partition_update {
+            return Err(crate::Error::ConfigInvalid {
+                message: format!(
+                    "You can not use sequence.field in cross partition update case \
+                     (Primary key constraint '{primary_keys:?}' not include all partition fields '{partition_keys:?}')."
+                ),
+            });
+        }
+
+        Ok(())
+    }
+
     /// Returns top-level Blob field names for create-time Blob contract checks.
     fn top_level_blob_field_names(fields: &[DataField]) -> Vec<&str> {
         fields
@@ -3273,6 +3343,151 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_create_schema_rejects_unknown_sequence_field() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("ts", DataType::Int(IntType::new()))
+            .primary_key(["id"])
+            .option("sequence.field", "no_such_col")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("no_such_col") && message.contains("can not be found")),
+            "sequence.field referencing a missing column should be rejected, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_create_schema_rejects_repeated_sequence_field() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("ts", DataType::Int(IntType::new()))
+            .primary_key(["id"])
+            .option("sequence.field", "ts,ts")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("ts") && message.contains("repeatedly")),
+            "repeated sequence.field should be rejected, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_create_schema_rejects_sequence_field_with_first_row() {
+        let err = Schema::builder()
+            .column("id", DataType::Int(IntType::new()))
+            .column("ts", DataType::Int(IntType::new()))
+            .primary_key(["id"])
+            .option("merge-engine", "first-row")
+            .option("sequence.field", "ts")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("FIRST_ROW")),
+            "sequence.field with merge-engine=first-row should be rejected, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_alter_set_unknown_sequence_field_rejected() {
+        let table_schema = TableSchema::new(
+            0,
+            &Schema::builder()
+                .column("id", DataType::Int(IntType::new()))
+                .column("ts", DataType::Int(IntType::new()))
+                .primary_key(["id"])
+                .build()
+                .unwrap(),
+        );
+
+        let err = table_schema
+            .apply_changes(vec![crate::spec::SchemaChange::set_option(
+                "sequence.field".to_string(),
+                "no_such_col".to_string(),
+            )])
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("no_such_col") && message.contains("can not be found")),
+            "alter setting sequence.field to a missing column should be rejected, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_create_schema_rejects_sequence_field_with_cross_partition_update() {
+        // PK (id) does not include the partition field (pt): cross-partition
+        // update case, where user-defined sequence fields are not supported.
+        let err = Schema::builder()
+            .column("pt", DataType::Int(IntType::new()))
+            .column("id", DataType::Int(IntType::new()))
+            .column("ts", DataType::Int(IntType::new()))
+            .partition_keys(["pt"])
+            .primary_key(["id"])
+            .option("sequence.field", "ts")
+            .build()
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("cross partition update")),
+            "sequence.field with cross-partition update should be rejected, got {err:?}"
+        );
+    }
+
+    #[test]
+    fn test_create_schema_accepts_sequence_field_when_pk_covers_partitions() {
+        // PK includes the partition field: not a cross-partition update case.
+        let schema = Schema::builder()
+            .column("pt", DataType::Int(IntType::new()))
+            .column("id", DataType::Int(IntType::new()))
+            .column("ts", DataType::Int(IntType::new()))
+            .partition_keys(["pt"])
+            .primary_key(["pt", "id"])
+            .option("sequence.field", "ts")
+            .build();
+
+        assert!(
+            schema.is_ok(),
+            "sequence.field with PK covering partition fields should be accepted, got {schema:?}"
+        );
+    }
+
+    #[test]
+    fn test_alter_set_sequence_field_with_cross_partition_update_rejected() {
+        let table_schema = TableSchema::new(
+            0,
+            &Schema::builder()
+                .column("pt", DataType::Int(IntType::new()))
+                .column("id", DataType::Int(IntType::new()))
+                .column("ts", DataType::Int(IntType::new()))
+                .partition_keys(["pt"])
+                .primary_key(["id"])
+                .build()
+                .unwrap(),
+        );
+
+        let err = table_schema
+            .apply_changes(vec![crate::spec::SchemaChange::set_option(
+                "sequence.field".to_string(),
+                "ts".to_string(),
+            )])
+            .unwrap_err();
+
+        assert!(
+            matches!(err, crate::Error::ConfigInvalid { ref message }
+                if message.contains("cross partition update")),
+            "alter setting sequence.field on cross-partition update table should be rejected, got {err:?}"
+        );
+    }
+
     #[test]
     fn test_drop_column_referenced_by_sequence_field_rejected() {
         let table_schema = TableSchema::new(