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 80d8c4e9 fix(spec): reject table-read.sequence-number.enabled without
primary keys (#684)
80d8c4e9 is described below
commit 80d8c4e93eebc56c9b2360be9d6ad8eb833376f0
Author: jackylee <[email protected]>
AuthorDate: Thu Aug 6 20:52:54 2026 +0800
fix(spec): reject table-read.sequence-number.enabled without primary keys
(#684)
---
crates/paimon/src/spec/core_options.rs | 14 ++++++
crates/paimon/src/spec/schema.rs | 84 ++++++++++++++++++++++++++++++++++
2 files changed, 98 insertions(+)
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<String, String>,
+ 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<String, String>) ->
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()))
