diff --git a/crates/paimon/src/spec/core_options.rs b/crates/paimon/src/spec/core_options.rs index 1ac9dcd1..e4ce0ca0 100644 --- a/crates/paimon/src/spec/core_options.rs +++ b/crates/paimon/src/spec/core_options.rs @@ -78,6 +78,8 @@ const READ_BATCH_SIZE_OPTION: &str = "read.batch-size"; const PARQUET_ROW_GROUP_PARALLELISM_OPTION: &str = "read.parquet.row-group.parallelism"; const PARQUET_ROW_GROUP_MAX_INFLIGHT_BYTES_OPTION: &str = "read.parquet.row-group.max-inflight-bytes"; +pub(crate) const TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION: &str = + "table-read.sequence-number.enabled"; pub(crate) const SEQUENCE_FIELD_OPTION: &str = "sequence.field"; pub(crate) const DISABLE_EXPLICIT_TYPE_CASTING_OPTION: &str = "disable-explicit-type-casting"; pub(crate) const DISABLE_ALTER_COLUMN_NULL_TO_NOT_NULL_OPTION: &str = @@ -466,6 +468,18 @@ impl<'a> CoreOptions<'a> { .unwrap_or(false) } + /// Whether reads expose the `_SEQUENCE_NUMBER` system column + /// (`table-read.sequence-number.enabled`, default `false`). + /// + /// Only meaningful for primary-key tables: the sequence number lives in the + /// merge key, so an append table has no such column to project. + pub fn table_read_sequence_number_enabled(&self) -> bool { + self.options + .get(TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION) + .map(|value| value.eq_ignore_ascii_case("true")) + .unwrap_or(false) + } + /// Whether `deletion-vectors.merge-on-read` is set (default `false`, matching /// Java `CoreOptions.DELETION_VECTORS_MERGE_ON_READ`). When true, uncompacted /// (level-0) data is made visible by merging on read; when false, deletion diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs index a154414b..24a7b498 100644 --- a/crates/paimon/src/spec/schema.rs +++ b/crates/paimon/src/spec/schema.rs @@ -19,6 +19,7 @@ use crate::spec::core_options::{ first_row_supports_changelog_producer, ChangelogProducer, CoreOptions, MergeEngine, BLOB_DESCRIPTOR_FIELD_OPTION, BLOB_FIELD_OPTION, BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, CHANGELOG_PRODUCER_OPTION, POSTPONE_BUCKET, QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION, + TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION, }; use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType, VarCharType}; use crate::spec::{ @@ -1141,6 +1142,7 @@ impl Schema { AggregationConfig::new(options).validate_create_mode(primary_keys, fields)?; Self::validate_first_row_changelog_producer(options)?; Self::validate_changelog_producer_requires_primary_keys(options, primary_keys)?; + Self::validate_read_sequence_number_requires_primary_keys(options, primary_keys)?; Self::validate_rowkind_field(options, primary_keys, fields)?; Self::validate_deletion_vectors(options)?; Self::validate_bucket_keys(options, fields, partition_keys, primary_keys)?; @@ -1531,6 +1533,32 @@ impl Schema { }) } + /// Reject `table-read.sequence-number.enabled` on a table without primary keys, + /// mirroring Java `SchemaValidation#validateChangelogReadSequenceNumber`. + /// + /// The sequence number is part of the merge key, so an append table has no such + /// column. Enabling the option there is accepted today and the read path then + /// projects a field id that no data file carries, so the column comes back + /// entirely NULL instead of raising. + fn validate_read_sequence_number_requires_primary_keys( + options: &HashMap, + primary_keys: &[String], + ) -> crate::Result<()> { + if !primary_keys.is_empty() + || !CoreOptions::new(options).table_read_sequence_number_enabled() + { + return Ok(()); + } + + Err(crate::Error::ConfigInvalid { + message: format!( + "Cannot enable '{TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION}' for \ + non-primary-key table. Sequence number is only available for \ + primary key tables." + ), + }) + } + fn validate_deletion_vectors(options: &HashMap) -> crate::Result<()> { let core = CoreOptions::new(options); if !core.deletion_vectors_enabled() { @@ -3094,6 +3122,62 @@ mod tests { ); } + #[test] + fn test_create_schema_rejects_read_sequence_number_without_primary_keys() { + // Java `validateChangelogReadSequenceNumber`: the sequence number lives in + // the merge key, so an append table has no such column to project. + for value in ["true", "TRUE"] { + assert_config_invalid( + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .option("table-read.sequence-number.enabled", value) + .build(), + "non-primary-key table", + ); + } + } + + #[test] + fn test_create_schema_accepts_read_sequence_number_with_primary_keys() { + // Guard against over-rejecting: a primary-key table may enable it, and an + // append table may still spell the default out explicitly. + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .primary_key(["id"]) + .option("table-read.sequence-number.enabled", "true") + .build() + .unwrap(); + + Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .option("table-read.sequence-number.enabled", "false") + .build() + .unwrap(); + } + + #[test] + fn test_alter_set_read_sequence_number_without_primary_keys_rejected() { + let table_schema = TableSchema::new( + 0, + &Schema::builder() + .column("id", DataType::Int(IntType::new())) + .column("value", DataType::Int(IntType::new())) + .build() + .unwrap(), + ); + + assert_config_invalid( + table_schema.apply_changes(vec![crate::spec::SchemaChange::set_option( + "table-read.sequence-number.enabled".to_string(), + "true".to_string(), + )]), + "non-primary-key table", + ); + } + fn cast_test_schema(options: &[(&str, &str)]) -> TableSchema { let mut builder = Schema::builder() .column("a", DataType::Int(IntType::new()))