This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 28971b56 fix(spec): validate sequence.field against table schema (#615)
28971b56 is described below
commit 28971b568d7073c509c7ccc1c02f4882b41f930f
Author: jackylee <[email protected]>
AuthorDate: Tue Aug 4 21:57:05 2026 +0800
fix(spec): validate sequence.field against table schema (#615)
---
crates/paimon/src/spec/aggregation.rs | 93 +++++++++---
crates/paimon/src/spec/mod.rs | 3 +-
crates/paimon/src/spec/schema.rs | 268 +++++++++++++++++++++++++++++++++-
3 files changed, 337 insertions(+), 27 deletions(-)
diff --git a/crates/paimon/src/spec/aggregation.rs
b/crates/paimon/src/spec/aggregation.rs
index 194d7952..7885a86a 100644
--- a/crates/paimon/src/spec/aggregation.rs
+++ b/crates/paimon/src/spec/aggregation.rs
@@ -185,9 +185,6 @@ impl<'a> AggregationConfig<'a> {
fields: &[DataField],
primary_keys: &[String],
) -> crate::Result<()> {
- // Same source as the read path: `sequence.field` parsed by
CoreOptions.
- let core_options = CoreOptions::new(self.options);
- let sequence_fields = core_options.sequence_fields();
for (key, value) in self.options {
let Some((col, kind)) = parse_field_scoped_option_key(key) else {
continue;
@@ -204,14 +201,6 @@ impl<'a> AggregationConfig<'a> {
});
};
if matches!(kind, FieldScopedOptionKind::AggregateFunction) {
- if sequence_fields.contains(&col) {
- return Err(crate::Error::ConfigInvalid {
- message: format!(
- "Should not define aggregation on sequence field:
'{col}'."
- ),
- });
- }
-
if primary_keys.iter().any(|pk| pk == col) {
if !is_known_aggregator_name(value) {
return Err(crate::Error::ConfigInvalid {
@@ -460,6 +449,31 @@ pub(crate) fn validate_aggregator_for_type(
}
}
+/// Reject `fields.<col>.aggregate-function` on a column listed in
+/// `sequence.field`, mirroring Java `SchemaValidation#validateSequenceField`,
+/// which checks `options.fieldAggFunc(field) == null` for every sequence field
+/// regardless of the configured merge engine.
+pub(crate) fn validate_no_aggregation_on_sequence_field(
+ options: &HashMap<String, String>,
+) -> crate::Result<()> {
+ let core_options = CoreOptions::new(options);
+ let sequence_fields = core_options.sequence_fields();
+ if sequence_fields.is_empty() {
+ return Ok(());
+ }
+
+ for field in sequence_fields {
+ let key = format!("{FIELDS_PREFIX}{field}{AGG_FUNCTION_SUFFIX}");
+ if options.contains_key(&key) {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!("Should not define aggregation on sequence
field: '{field}'."),
+ });
+ }
+ }
+
+ Ok(())
+}
+
fn is_unsupported_aggregation_option(key: &str) -> bool {
key == IGNORE_DELETE_OPTION
|| key.ends_with(IGNORE_DELETE_SUFFIX)
@@ -664,22 +678,53 @@ mod tests {
}
#[test]
- fn test_validate_create_mode_rejects_aggregation_on_sequence_field() {
- // Java rejects aggregation definitions on sequence fields during
- // schema validation; the runtime still forces sequence fields to
- // last_value when reading old or externally-created metadata.
+ fn test_rejects_aggregation_on_sequence_field_for_every_merge_engine() {
+ // Java rejects aggregation definitions on sequence fields inside
+ // `validateSequenceField`, which runs for every merge engine; the
+ // runtime still forces sequence fields to last_value when reading old
+ // or externally-created metadata.
+ for engine in [
+ None,
+ Some("deduplicate"),
+ Some("first-row"),
+ Some("partial-update"),
+ Some("aggregation"),
+ ] {
+ let mut options = HashMap::from([
+ ("sequence.field".to_string(), "amount".to_string()),
+ (
+ "fields.amount.aggregate-function".to_string(),
+ "listagg".to_string(),
+ ),
+ ]);
+ if let Some(engine) = engine {
+ options.insert(MERGE_ENGINE_OPTION.to_string(),
engine.to_string());
+ }
+
+ let err =
validate_no_aggregation_on_sequence_field(&options).unwrap_err();
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sequence field") &&
message.contains("amount")),
+ "merge-engine={engine:?} should reject aggregation on a
sequence field, got {err:?}"
+ );
+ }
+ }
+
+ #[test]
+ fn test_accepts_aggregation_on_a_non_sequence_field() {
+ // Guard against over-rejecting: only the sequence field itself is
+ // off limits, other columns may carry an aggregate function.
let options = aggregation_options(&[
("sequence.field", "amount"),
- ("fields.amount.aggregate-function", "listagg"),
+ ("fields.price.aggregate-function", "sum"),
]);
- let err = AggregationConfig::new(&options)
- .validate_create_mode(&pk(), &sample_fields())
- .unwrap_err();
- assert!(
- matches!(err, crate::Error::ConfigInvalid { ref message }
- if message.contains("sequence field") &&
message.contains("amount")),
- "expected sequence-field aggregation rejection, got {err:?}"
- );
+ assert!(validate_no_aggregation_on_sequence_field(&options).is_ok());
+ }
+
+ #[test]
+ fn test_accepts_options_without_a_sequence_field() {
+ let options =
aggregation_options(&[("fields.amount.aggregate-function", "sum")]);
+ assert!(validate_no_aggregation_on_sequence_field(&options).is_ok());
}
#[test]
diff --git a/crates/paimon/src/spec/mod.rs b/crates/paimon/src/spec/mod.rs
index d210beed..22780b91 100644
--- a/crates/paimon/src/spec/mod.rs
+++ b/crates/paimon/src/spec/mod.rs
@@ -40,7 +40,8 @@ pub(crate) use partial_update::PartialUpdateConfig;
mod aggregation;
pub(crate) use aggregation::{
- remove_field_scoped_options, rename_field_scoped_options,
AggregationConfig,
+ remove_field_scoped_options, rename_field_scoped_options,
+ validate_no_aggregation_on_sequence_field, AggregationConfig,
};
mod data_type_casts;
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index ce7d1c40..c1c7bb83 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -22,8 +22,9 @@ use crate::spec::core_options::{
};
use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType,
VarCharType};
use crate::spec::{
- remove_field_scoped_options, rename_field_scoped_options,
AggregationConfig, BlobType,
- ColumnMove, ColumnMoveType, PartialUpdateConfig,
+ remove_field_scoped_options, rename_field_scoped_options,
+ validate_no_aggregation_on_sequence_field, AggregationConfig, BlobType,
ColumnMove,
+ ColumnMoveType, PartialUpdateConfig,
};
use serde::{Deserialize, Serialize};
use serde_with::serde_as;
@@ -1136,11 +1137,13 @@ impl Schema {
Self::validate_blob_fields(fields, partition_keys, options)?;
Self::validate_vector_store_fields(fields, partition_keys, options)?;
PartialUpdateConfig::new(options).validate_create_mode(!primary_keys.is_empty())?;
+ validate_no_aggregation_on_sequence_field(options)?;
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_deletion_vectors(options)?;
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)?;
Self::validate_primary_key_vector_index(fields, primary_keys,
options)?;
Self::validate_primary_key_full_text_index(fields, primary_keys,
options)?;
@@ -1647,6 +1650,73 @@ impl Schema {
.map_err(Self::options_error_to_config_invalid)
}
+ /// Validate the `sequence.field` option against the schema, mirroring four
+ /// of the five checks in 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.
+ ///
+ /// The fifth check — Java's `options.fieldAggFunc(field) == null` — is
+ /// [`validate_no_aggregation_on_sequence_field`], which is keyed only on
the
+ /// option map so it applies to every merge engine.
+ 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(())
+ }
+
fn validate_primary_key_vector_index(
fields: &[DataField],
primary_keys: &[String],
@@ -4382,6 +4452,55 @@ mod tests {
));
}
+ #[test]
+ fn
test_create_schema_rejects_aggregation_on_sequence_field_without_agg_engine() {
+ // Java `validateSequenceField` checks `fieldAggFunc(field) == null`
for
+ // every merge engine, so the default (deduplicate) engine must reject
+ // this too, not just `merge-engine=aggregation`.
+ let err = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("ts", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("sequence.field", "ts")
+ .option("fields.ts.aggregate-function", "sum")
+ .build()
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sequence field") &&
message.contains("ts")),
+ "aggregation on a sequence field should be rejected on the default
\
+ merge engine, got {err:?}"
+ );
+ }
+
+ #[test]
+ fn
test_alter_set_aggregation_on_sequence_field_rejected_without_agg_engine() {
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("ts", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("sequence.field", "ts")
+ .build()
+ .unwrap(),
+ );
+
+ let err = table_schema
+ .apply_changes(vec![crate::spec::SchemaChange::set_option(
+ "fields.ts.aggregate-function".to_string(),
+ "sum".to_string(),
+ )])
+ .unwrap_err();
+
+ assert!(
+ matches!(err, crate::Error::ConfigInvalid { ref message }
+ if message.contains("sequence field") &&
message.contains("ts")),
+ "alter adding aggregation on a sequence field should be rejected,
got {err:?}"
+ );
+ }
+
#[test]
fn test_create_schema_rejects_reserved_field_names() {
// Java `SpecialFields.SYSTEM_FIELD_NAMES` plus the `_KEY_` prefix. A
@@ -4689,6 +4808,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(
