Skip to content

Commit 3f063d0

Browse files
committed
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.
1 parent 27dea79 commit 3f063d0

1 file changed

Lines changed: 215 additions & 0 deletions

File tree

crates/paimon/src/spec/schema.rs

Lines changed: 215 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -489,6 +489,12 @@ impl TableSchema {
489489
&new_schema.primary_keys,
490490
&new_schema.fields,
491491
)?;
492+
Schema::validate_sequence_field(
493+
&new_schema.options,
494+
&new_schema.fields,
495+
&new_schema.partition_keys,
496+
&new_schema.primary_keys,
497+
)?;
492498
Schema::validate_read_batch_size(&new_schema.options)?;
493499
Ok(new_schema)
494500
}
@@ -936,6 +942,7 @@ impl Schema {
936942
AggregationConfig::new(&options).validate_create_mode(&primary_keys, &fields)?;
937943
Self::validate_first_row_changelog_producer(&options)?;
938944
Self::validate_rowkind_field(&options, &primary_keys, &fields)?;
945+
Self::validate_sequence_field(&options, &fields, &partition_keys, &primary_keys)?;
939946
Self::validate_read_batch_size(&options)?;
940947

941948
Ok(Self {
@@ -1367,6 +1374,69 @@ impl Schema {
13671374
.map_err(Self::options_error_to_config_invalid)
13681375
}
13691376

1377+
/// Validate the `sequence.field` option against the schema, mirroring
1378+
/// Java `SchemaValidation#validateSequenceField`:
1379+
/// * every listed field must exist in the table schema — otherwise the
1380+
/// write path silently falls back to the auto-increment sequence
1381+
/// (`TableWrite` resolves sequence fields with a lenient lookup) and
1382+
/// merge results would ignore the user's ordering intent;
1383+
/// * a field must not be listed more than once;
1384+
/// * `merge-engine=first-row` does not support user-defined sequence
1385+
/// fields;
1386+
/// * cross-partition update tables (primary key constraint not including
1387+
/// all partition fields) do not support user-defined sequence fields,
1388+
/// because partition migration retracts old rows with generated DELETEs
1389+
/// whose ordering a user-provided sequence value could break.
1390+
fn validate_sequence_field(
1391+
options: &HashMap<String, String>,
1392+
fields: &[DataField],
1393+
partition_keys: &[String],
1394+
primary_keys: &[String],
1395+
) -> crate::Result<()> {
1396+
let core = CoreOptions::new(options);
1397+
let sequence_fields = core.sequence_fields();
1398+
if sequence_fields.is_empty() {
1399+
return Ok(());
1400+
}
1401+
1402+
let mut seen: HashSet<&str> = HashSet::new();
1403+
for name in &sequence_fields {
1404+
if fields.iter().all(|f| f.name() != *name) {
1405+
return Err(crate::Error::ConfigInvalid {
1406+
message: format!("Sequence field '{name}' can not be found in table schema."),
1407+
});
1408+
}
1409+
if !seen.insert(name) {
1410+
return Err(crate::Error::ConfigInvalid {
1411+
message: format!("Sequence field '{name}' is defined repeatedly."),
1412+
});
1413+
}
1414+
}
1415+
1416+
let merge_engine = core
1417+
.merge_engine()
1418+
.map_err(Self::options_error_to_config_invalid)?;
1419+
if merge_engine == MergeEngine::FirstRow {
1420+
return Err(crate::Error::ConfigInvalid {
1421+
message: "Do not support use sequence field on FIRST_ROW merge engine.".to_string(),
1422+
});
1423+
}
1424+
1425+
let cross_partition_update = !primary_keys.is_empty()
1426+
&& !partition_keys.is_empty()
1427+
&& partition_keys.iter().any(|pt| !primary_keys.contains(pt));
1428+
if cross_partition_update {
1429+
return Err(crate::Error::ConfigInvalid {
1430+
message: format!(
1431+
"You can not use sequence.field in cross partition update case \
1432+
(Primary key constraint '{primary_keys:?}' not include all partition fields '{partition_keys:?}')."
1433+
),
1434+
});
1435+
}
1436+
1437+
Ok(())
1438+
}
1439+
13701440
/// Returns top-level Blob field names for create-time Blob contract checks.
13711441
fn top_level_blob_field_names(fields: &[DataField]) -> Vec<&str> {
13721442
fields
@@ -2949,6 +3019,151 @@ mod tests {
29493019
);
29503020
}
29513021

3022+
#[test]
3023+
fn test_create_schema_rejects_unknown_sequence_field() {
3024+
let err = Schema::builder()
3025+
.column("id", DataType::Int(IntType::new()))
3026+
.column("ts", DataType::Int(IntType::new()))
3027+
.primary_key(["id"])
3028+
.option("sequence.field", "no_such_col")
3029+
.build()
3030+
.unwrap_err();
3031+
3032+
assert!(
3033+
matches!(err, crate::Error::ConfigInvalid { ref message }
3034+
if message.contains("no_such_col") && message.contains("can not be found")),
3035+
"sequence.field referencing a missing column should be rejected, got {err:?}"
3036+
);
3037+
}
3038+
3039+
#[test]
3040+
fn test_create_schema_rejects_repeated_sequence_field() {
3041+
let err = Schema::builder()
3042+
.column("id", DataType::Int(IntType::new()))
3043+
.column("ts", DataType::Int(IntType::new()))
3044+
.primary_key(["id"])
3045+
.option("sequence.field", "ts,ts")
3046+
.build()
3047+
.unwrap_err();
3048+
3049+
assert!(
3050+
matches!(err, crate::Error::ConfigInvalid { ref message }
3051+
if message.contains("ts") && message.contains("repeatedly")),
3052+
"repeated sequence.field should be rejected, got {err:?}"
3053+
);
3054+
}
3055+
3056+
#[test]
3057+
fn test_create_schema_rejects_sequence_field_with_first_row() {
3058+
let err = Schema::builder()
3059+
.column("id", DataType::Int(IntType::new()))
3060+
.column("ts", DataType::Int(IntType::new()))
3061+
.primary_key(["id"])
3062+
.option("merge-engine", "first-row")
3063+
.option("sequence.field", "ts")
3064+
.build()
3065+
.unwrap_err();
3066+
3067+
assert!(
3068+
matches!(err, crate::Error::ConfigInvalid { ref message }
3069+
if message.contains("FIRST_ROW")),
3070+
"sequence.field with merge-engine=first-row should be rejected, got {err:?}"
3071+
);
3072+
}
3073+
3074+
#[test]
3075+
fn test_alter_set_unknown_sequence_field_rejected() {
3076+
let table_schema = TableSchema::new(
3077+
0,
3078+
&Schema::builder()
3079+
.column("id", DataType::Int(IntType::new()))
3080+
.column("ts", DataType::Int(IntType::new()))
3081+
.primary_key(["id"])
3082+
.build()
3083+
.unwrap(),
3084+
);
3085+
3086+
let err = table_schema
3087+
.apply_changes(vec![crate::spec::SchemaChange::set_option(
3088+
"sequence.field".to_string(),
3089+
"no_such_col".to_string(),
3090+
)])
3091+
.unwrap_err();
3092+
3093+
assert!(
3094+
matches!(err, crate::Error::ConfigInvalid { ref message }
3095+
if message.contains("no_such_col") && message.contains("can not be found")),
3096+
"alter setting sequence.field to a missing column should be rejected, got {err:?}"
3097+
);
3098+
}
3099+
3100+
#[test]
3101+
fn test_create_schema_rejects_sequence_field_with_cross_partition_update() {
3102+
// PK (id) does not include the partition field (pt): cross-partition
3103+
// update case, where user-defined sequence fields are not supported.
3104+
let err = Schema::builder()
3105+
.column("pt", DataType::Int(IntType::new()))
3106+
.column("id", DataType::Int(IntType::new()))
3107+
.column("ts", DataType::Int(IntType::new()))
3108+
.partition_keys(["pt"])
3109+
.primary_key(["id"])
3110+
.option("sequence.field", "ts")
3111+
.build()
3112+
.unwrap_err();
3113+
3114+
assert!(
3115+
matches!(err, crate::Error::ConfigInvalid { ref message }
3116+
if message.contains("cross partition update")),
3117+
"sequence.field with cross-partition update should be rejected, got {err:?}"
3118+
);
3119+
}
3120+
3121+
#[test]
3122+
fn test_create_schema_accepts_sequence_field_when_pk_covers_partitions() {
3123+
// PK includes the partition field: not a cross-partition update case.
3124+
let schema = Schema::builder()
3125+
.column("pt", DataType::Int(IntType::new()))
3126+
.column("id", DataType::Int(IntType::new()))
3127+
.column("ts", DataType::Int(IntType::new()))
3128+
.partition_keys(["pt"])
3129+
.primary_key(["pt", "id"])
3130+
.option("sequence.field", "ts")
3131+
.build();
3132+
3133+
assert!(
3134+
schema.is_ok(),
3135+
"sequence.field with PK covering partition fields should be accepted, got {schema:?}"
3136+
);
3137+
}
3138+
3139+
#[test]
3140+
fn test_alter_set_sequence_field_with_cross_partition_update_rejected() {
3141+
let table_schema = TableSchema::new(
3142+
0,
3143+
&Schema::builder()
3144+
.column("pt", DataType::Int(IntType::new()))
3145+
.column("id", DataType::Int(IntType::new()))
3146+
.column("ts", DataType::Int(IntType::new()))
3147+
.partition_keys(["pt"])
3148+
.primary_key(["id"])
3149+
.build()
3150+
.unwrap(),
3151+
);
3152+
3153+
let err = table_schema
3154+
.apply_changes(vec![crate::spec::SchemaChange::set_option(
3155+
"sequence.field".to_string(),
3156+
"ts".to_string(),
3157+
)])
3158+
.unwrap_err();
3159+
3160+
assert!(
3161+
matches!(err, crate::Error::ConfigInvalid { ref message }
3162+
if message.contains("cross partition update")),
3163+
"alter setting sequence.field on cross-partition update table should be rejected, got {err:?}"
3164+
);
3165+
}
3166+
29523167
#[test]
29533168
fn test_drop_column_referenced_by_sequence_field_rejected() {
29543169
let table_schema = TableSchema::new(

0 commit comments

Comments
 (0)