diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index 0245cf41..b5a6f70c 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -489,6 +489,12 @@ impl TableSchema { &new_schema.primary_keys, &new_schema.fields, )?; + 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) } @@ -936,6 +942,7 @@ impl Schema { AggregationConfig::new(&options).validate_create_mode(&primary_keys, &fields)?; Self::validate_first_row_changelog_producer(&options)?; Self::validate_rowkind_field(&options, &primary_keys, &fields)?; + Self::validate_sequence_field(&options, &fields, &partition_keys, &primary_keys)?; Self::validate_read_batch_size(&options)?; Ok(Self { @@ -1367,6 +1374,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, + 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 @@ -2949,6 +3019,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(