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 4a947ecd Expand native table sources and nullable primary-key writes
(#1155)
4a947ecd is described below
commit 4a947ecdf80ac6e856e0ab99a0963027099dc39e
Author: Jingsong Lee <[email protected]>
AuthorDate: Sun Oct 18 16:02:03 2026 +0800
Expand native table sources and nullable primary-key writes (#1155)
---
crates/paimon/src/catalog/filesystem.rs | 130 +++++++-
crates/paimon/src/spec/core_options.rs | 128 +++++++-
crates/paimon/src/spec/schema.rs | 228 +++++++++++++--
crates/paimon/src/table/pk_vector_scan.rs | 26 +-
crates/paimon/src/table/snapshot_scan_tests.rs | 101 +++++++
crates/paimon/src/table/stream_table_scan.rs | 19 ++
crates/paimon/src/table/table_merge_input.rs | 39 ++-
crates/paimon/src/table/table_merge_into_tests.rs | 50 ++++
crates/paimon/src/table/table_scan.rs | 19 ++
crates/paimon/tests/nullable_primary_key_test.rs | 342 ++++++++++++++++++++++
crates/paimon/tests/rest_catalog/merge_sources.rs | 285 ++++++++++++++++++
crates/paimon/tests/rest_catalog_test.rs | 4 +
12 files changed, 1338 insertions(+), 33 deletions(-)
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index a957c500..0689545b 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -29,7 +29,7 @@ use crate::io::cache::{create_local_cache, LocalCache};
use crate::io::FileIO;
use crate::spec::{
CoreOptions, Schema, Snapshot, TableSchema, TableType,
INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
- TABLE_TYPE_OPTION,
+ PRIMARY_KEY_NULLABLE_OPTION, TABLE_TYPE_OPTION,
};
use crate::table::{ObjectTable, SchemaManager, Table};
use async_trait::async_trait;
@@ -554,6 +554,26 @@ impl Catalog for FileSystemCatalog {
})?;
reject_immutable_option_changes(current.options(), &changes)?;
+ // Java permits changing the key policy before the first snapshot and
+ // normalizes the fields again. Once data exists the option is
immutable.
+ // Avoid snapshot IO for unrelated schema changes or exact SET no-ops.
+ let changes_key_nullability = changes.iter().any(|change| match change
{
+ crate::spec::SchemaChange::SetOption { key, value } => {
+ key == PRIMARY_KEY_NULLABLE_OPTION &&
current.options().get(key) != Some(value)
+ }
+ crate::spec::SchemaChange::RemoveOption { key } => key ==
PRIMARY_KEY_NULLABLE_OPTION,
+ _ => false,
+ });
+ if changes_key_nullability
+ && crate::table::SnapshotManager::new(self.file_io.clone(),
table_path.clone())
+ .get_latest_snapshot()
+ .await?
+ .is_some()
+ {
+ return Err(Error::Unsupported {
+ message: format!("Change '{PRIMARY_KEY_NULLABLE_OPTION}' is
not supported yet."),
+ });
+ }
let new_schema = current
.apply_changes(changes)
@@ -1086,6 +1106,114 @@ mod tests {
catalog.get_table(&identifier).await.unwrap();
}
+ #[tokio::test]
+ async fn nullable_key_policy_changes_only_before_first_snapshot() {
+ use crate::spec::{DataType, IntType, SchemaChange};
+
+ let catalog = create_memory_catalog();
+ catalog
+ .create_database("db", false, HashMap::new())
+ .await
+ .unwrap();
+ for initial in [None, Some("false"), Some("true")] {
+ let identifier = Identifier::new("db",
initial.unwrap_or("default"));
+ let mut schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1");
+ if let Some(initial) = initial {
+ schema = schema.option("primary-key.nullable", initial);
+ }
+ catalog
+ .create_table(&identifier, schema.build().unwrap(), false)
+ .await
+ .unwrap();
+ // As in Java, an empty table can change and reset this option.
+ for value in ["true", "false", "true"] {
+ catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::set_option(
+ "primary-key.nullable".into(),
+ value.into(),
+ )],
+ false,
+ )
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert_eq!(
+ table.schema().fields()[0].data_type().is_nullable(),
+ value == "true"
+ );
+ }
+ catalog
+ .alter_table(
+ &identifier,
+
vec![SchemaChange::remove_option("primary-key.nullable".into())],
+ false,
+ )
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert!(!table.schema().fields()[0].data_type().is_nullable());
+ if let Some(initial) = initial {
+ catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::set_option(
+ "primary-key.nullable".into(),
+ initial.into(),
+ )],
+ false,
+ )
+ .await
+ .unwrap();
+ }
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+ let before = catalog
+ .get_table(&identifier)
+ .await
+ .unwrap()
+ .schema()
+ .clone();
+ for value in ["false", "true", "TRUE"] {
+ let result = catalog
+ .alter_table(
+ &identifier,
+ vec![SchemaChange::set_option(
+ "primary-key.nullable".into(),
+ value.into(),
+ )],
+ false,
+ )
+ .await;
+ if Some(value) == initial {
+ result.unwrap();
+ } else {
+ assert!(
+ matches!(result, Err(Error::Unsupported { .. })),
+ "{initial:?}/{value}: {result:?}"
+ );
+ }
+ }
+ let result = catalog
+ .alter_table(
+ &identifier,
+
vec![SchemaChange::remove_option("primary-key.nullable".into())],
+ false,
+ )
+ .await;
+ assert!(
+ matches!(result, Err(Error::Unsupported { .. })),
+ "{initial:?}: {result:?}"
+ );
+ let after = catalog.get_table(&identifier).await.unwrap();
+ assert_eq!(after.schema().options(), before.options());
+ assert_eq!(after.schema().fields(), before.fields());
+ }
+ }
+
/// Give the table a first snapshot, so a test can assert that a rejection
does
/// not depend on whether one exists.
async fn give_the_table_a_snapshot(catalog: &FileSystemCatalog,
identifier: &Identifier) {
diff --git a/crates/paimon/src/spec/core_options.rs
b/crates/paimon/src/spec/core_options.rs
index 0b6c8b29..cba2dab6 100644
--- a/crates/paimon/src/spec/core_options.rs
+++ b/crates/paimon/src/spec/core_options.rs
@@ -58,6 +58,7 @@ const MAX_FORMAT_TABLE_SCAN_LIST_PARALLELISM: i64 = 1000;
pub(crate) const BUCKET_KEY_OPTION: &str = "bucket-key";
const BUCKET_FUNCTION_TYPE_OPTION: &str = "bucket-function.type";
const BUCKET_OPTION: &str = "bucket";
+pub(crate) const PRIMARY_KEY_NULLABLE_OPTION: &str = "primary-key.nullable";
const DEFAULT_BUCKET: i32 = -1;
/// Postpone bucket mode: data is written to `bucket-postpone` directory
/// and is invisible to readers until compaction assigns real bucket numbers.
@@ -527,9 +528,10 @@ impl<'a> CoreOptions<'a> {
/// they surface as `Error::Unsupported` (mapped to `NotImplementedError`
at
/// the Python boundary). Explicit `scan.mode=from-snapshot` /
/// `from-timestamp` are the modes Java's `CoreOptions.setDefaultValues()`
- /// writes next to the corresponding selector, so they are accepted when
- /// that selector is present (the batch-read semantics are identical to
- /// leaving the mode at `default`); an explicit mode without its selector
+ /// writes next to the corresponding selector. `latest` / `latest-full`
+ /// read the latest full state in batch mode; `from-snapshot-full` reads
+ /// the selected full snapshot, as Java's batch starting scanners do.
+ /// An explicit point-in-time mode without its selector
/// is malformed input (`Error::DataInvalid`), mirroring Java's
/// `SchemaValidation`. Incremental timestamp windows reuse the batch
/// Delta/Changelog scanner; snapshot/tag ranges remain unimplemented here.
@@ -564,6 +566,15 @@ impl<'a> CoreOptions<'a> {
}
}
if let Some(mode) = self.options.get(SCAN_MODE_OPTION) {
+ if self.is_format_table()
+ && ["latest", "latest-full", "from-snapshot-full"]
+ .iter()
+ .any(|value| mode.eq_ignore_ascii_case(value))
+ {
+ return Err(crate::Error::Unsupported {
+ message: "Full-snapshot startup modes require a Paimon
table".into(),
+ });
+ }
if incremental
&& !mode.eq_ignore_ascii_case("default")
&& !mode.eq_ignore_ascii_case("incremental")
@@ -577,6 +588,10 @@ impl<'a> CoreOptions<'a> {
}
let selector_keys: &[&str] = if
mode.eq_ignore_ascii_case("default") {
return Ok(());
+ } else if mode.eq_ignore_ascii_case("latest")
+ || mode.eq_ignore_ascii_case("latest-full")
+ {
+ &[]
} else if mode.eq_ignore_ascii_case("incremental") {
if incremental {
return Ok(());
@@ -594,6 +609,8 @@ impl<'a> CoreOptions<'a> {
]
} else if mode.eq_ignore_ascii_case("from-timestamp") {
&[SCAN_TIMESTAMP_MILLIS_OPTION, SCAN_TIMESTAMP_OPTION]
+ } else if mode.eq_ignore_ascii_case("from-snapshot-full") {
+ &[SCAN_SNAPSHOT_ID_OPTION]
} else {
return Err(crate::Error::Unsupported {
message: format!(
@@ -601,9 +618,10 @@ impl<'a> CoreOptions<'a> {
),
});
};
- if !selector_keys
- .iter()
- .any(|key| self.options.contains_key(*key))
+ if !selector_keys.is_empty()
+ && !selector_keys
+ .iter()
+ .any(|key| self.options.contains_key(*key))
{
return Err(crate::Error::DataInvalid {
message: format!(
@@ -613,6 +631,18 @@ impl<'a> CoreOptions<'a> {
source: None,
});
}
+ if self
+ .configured_time_travel_selectors()
+ .iter()
+ .any(|key| !selector_keys.contains(key))
+ || self.options.contains_key("scan.file-creation-time-millis")
+ || self.options.contains_key("scan.creation-time-millis")
+ {
+ return Err(crate::Error::DataInvalid {
+ message: format!("scan.mode={mode} cannot be used with the
configured point-in-time scan options"),
+ source: None,
+ });
+ }
}
Ok(())
}
@@ -782,6 +812,14 @@ impl<'a> CoreOptions<'a> {
self.options.get(ROWKIND_FIELD_OPTION).map(String::as_str)
}
+ /// Whether primary keys allow null values. Null keys use null-safe
equality
+ /// when records are merged. Disabled by default, as in Java.
+ pub fn primary_key_nullable(&self) -> bool {
+ self.options
+ .get(PRIMARY_KEY_NULLABLE_OPTION)
+ .is_some_and(|v| v.eq_ignore_ascii_case("true"))
+ }
+
/// Whether to ignore delete records (and all retracts when used by
`RowKindFilter`).
pub fn ignore_delete(&self) -> bool {
for key in
@@ -3947,7 +3985,7 @@ mod tests {
let ok = HashMap::from([("scan.mode".to_string(),
"default".to_string())]);
assert!(CoreOptions::new(&ok).validate_scan_options().is_ok());
// unimplemented modes Unsupported
- for mode in ["compacted-full", "latest", "latest-full"] {
+ for mode in ["compacted-full", "latest-delta", "unknown"] {
let bad = HashMap::from([("scan.mode".to_string(),
mode.to_string())]);
let err =
CoreOptions::new(&bad).validate_scan_options().unwrap_err();
assert!(
@@ -3957,6 +3995,82 @@ mod tests {
}
}
+ #[test]
+ fn test_full_startup_modes_validate_batch_options() {
+ for mode in ["latest", "latest-full", "LATEST-FULL"] {
+ let options = HashMap::from([("scan.mode".into(), mode.into())]);
+ assert!(
+ CoreOptions::new(&options).validate_scan_options().is_ok(),
+ "{mode}"
+ );
+ }
+ let options = HashMap::from([
+ ("scan.mode".into(), "from-snapshot-full".into()),
+ ("scan.snapshot-id".into(), "1".into()),
+ ]);
+ assert!(CoreOptions::new(&options).validate_scan_options().is_ok());
+ }
+
+ #[test]
+ fn test_full_startup_modes_reject_conflicting_selectors() {
+ for mode in ["latest", "latest-full", "from-snapshot-full"] {
+ for key in [
+ "scan.version",
+ "scan.snapshot-id",
+ "scan.tag-name",
+ "scan.watermark",
+ "scan.timestamp",
+ "scan.timestamp-millis",
+ "scan.file-creation-time-millis",
+ "scan.creation-time-millis",
+ "incremental-between-timestamp",
+ ] {
+ if mode == "from-snapshot-full" && key == "scan.snapshot-id" {
+ continue;
+ }
+ let mut options =
+ HashMap::from([("scan.mode".into(), mode.into()),
(key.into(), "1".into())]);
+ if mode == "from-snapshot-full" {
+ options.insert("scan.snapshot-id".into(), "1".into());
+ }
+ assert!(
+ matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::DataInvalid { .. })
+ ),
+ "{mode} / {key}"
+ );
+ }
+ }
+ for selector in [None, Some("scan.tag-name"), Some("scan.version")] {
+ let mut options = HashMap::from([("scan.mode".into(),
"from-snapshot-full".into())]);
+ if let Some(selector) = selector {
+ options.insert(selector.into(), "1".into());
+ }
+ assert!(matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::DataInvalid { .. })
+ ));
+ }
+ }
+
+ #[test]
+ fn test_full_startup_modes_do_not_enable_format_table_snapshots() {
+ for mode in ["latest", "latest-full", "from-snapshot-full"] {
+ let mut options = HashMap::from([
+ ("type".into(), "format-table".into()),
+ ("scan.mode".into(), mode.into()),
+ ]);
+ if mode == "from-snapshot-full" {
+ options.insert("scan.snapshot-id".into(), "1".into());
+ }
+ assert!(matches!(
+ CoreOptions::new(&options).validate_scan_options(),
+ Err(crate::Error::Unsupported { .. })
+ ));
+ }
+ }
+
#[test]
fn test_validate_scan_options_explicit_mode_with_matching_selector() {
// Java's CoreOptions.setDefaultValues() writes scan.mode=from-snapshot
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index 8970213b..9e54e08d 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -19,8 +19,9 @@ use crate::spec::core_options::{
first_row_supports_changelog_producer, ChangelogProducer, CoreOptions,
MergeEngine,
BLOB_DESCRIPTOR_FIELD_FALLBACK, BLOB_DESCRIPTOR_FIELD_OPTION,
BLOB_FIELD_OPTION,
BLOB_VIEW_FIELD_OPTION, BUCKET_KEY_OPTION, CHANGELOG_PRODUCER_OPTION,
- INDEX_FILE_IN_DATA_FILE_DIR_OPTION, POSTPONE_BUCKET,
QUERY_AUTH_ENABLED_OPTION,
- SEQUENCE_FIELD_OPTION, TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION,
TABLE_TYPE_OPTION,
+ INDEX_FILE_IN_DATA_FILE_DIR_OPTION, POSTPONE_BUCKET,
PRIMARY_KEY_NULLABLE_OPTION,
+ QUERY_AUTH_ENABLED_OPTION, SEQUENCE_FIELD_OPTION,
TABLE_READ_SEQUENCE_NUMBER_ENABLED_OPTION,
+ TABLE_TYPE_OPTION,
};
use crate::spec::types::{ArrayType, DataType, MapType, MultisetType, RowType,
VarCharType};
use crate::spec::{
@@ -245,6 +246,7 @@ impl TableSchema {
for key in [
INDEX_FILE_IN_DATA_FILE_DIR_OPTION,
"data-file.path-directory",
+ PRIMARY_KEY_NULLABLE_OPTION,
] {
match self.options.get(key) {
Some(stored) => extra.insert(key.to_string(), stored.clone()),
@@ -351,6 +353,8 @@ impl TableSchema {
/// Column errors ([`crate::Error::ColumnNotExist`] /
/// [`crate::Error::ColumnAlreadyExist`]) are returned with an empty table
/// name; the calling catalog fills in the table's full name.
+ /// The catalog also checks snapshot-dependent immutable options before
+ /// applying changes; this schema transformation has no snapshot context.
pub fn apply_changes(&self, changes: Vec<crate::spec::SchemaChange>) ->
crate::Result<Self> {
use crate::spec::SchemaChange;
@@ -680,8 +684,8 @@ impl TableSchema {
new_nullability,
} => {
let name = top_level_field(&field_names)?;
- // Primary keys are normalized to NOT NULL at create time;
- // a nullable key column would break key/bucket semantics.
+ // Key nullability is selected at create time by the
immutable
+ // primary-key.nullable option, not by ALTER COLUMN.
if new_nullability &&
new_schema.primary_keys.iter().any(|k| k == name) {
return Err(crate::Error::Unsupported {
message: "Cannot change nullability of primary
key".to_string(),
@@ -700,8 +704,9 @@ impl TableSchema {
disable_null_to_not_null,
)?;
let nt =
old.data_type().copy_with_nullable(new_nullability)?;
- fields[idx] = DataField::new(old.id(),
old.name().to_string(), nt)
- .with_description(old.description().map(|s|
s.to_string()));
+ let mut field = old.clone();
+ field.typ = nt;
+ fields[idx] = field;
}
SchemaChange::UpdateColumnComment {
field_names,
@@ -721,6 +726,13 @@ impl TableSchema {
}
}
+ // Java rebuilds Schema after ALTER, reapplying the key option even if
+ // an individual column change requested a different nullability.
+ Schema::normalize_primary_key_nullability(
+ &mut fields,
+ &new_schema.primary_keys,
+ new_schema.core_options().primary_key_nullable(),
+ )?;
new_schema.fields = fields;
new_schema.highest_field_id =
highest_field_id.max(Self::current_highest_field_id(&new_schema.fields));
@@ -1377,6 +1389,13 @@ impl Schema {
) -> crate::Result<()> {
// Create and alter share this, so an unparsable type never lands.
CoreOptions::new(options).table_type()?;
+ if primary_keys.is_empty() &&
CoreOptions::new(options).primary_key_nullable() {
+ return Err(crate::Error::ConfigInvalid {
+ message: format!(
+ "Option '{PRIMARY_KEY_NULLABLE_OPTION}' can only be
enabled for a table with primary keys."
+ ),
+ });
+ }
validate_no_reserved_field_names(fields)?;
Self::validate_key_field_types(fields, partition_keys, primary_keys,
options)?;
Self::validate_bucket_count(options)?;
@@ -1521,7 +1540,7 @@ impl Schema {
}
/// Normalize fields: validate duplicate/subset checks, promote configured
- /// binary BLOB fields, and make primary key columns non-nullable.
+ /// binary BLOB fields, and apply the configured primary key nullability.
/// Corresponds to Java `normalizeFields`.
fn normalize_fields(
fields: &[DataField],
@@ -1535,7 +1554,8 @@ impl Schema {
Self::validate_primary_keys(&field_names, primary_keys)?;
Self::validate_primary_keys_not_partition_only(partition_keys,
primary_keys)?;
- let blob_field_names = CoreOptions::new(options).blob_fields();
+ let core_options = CoreOptions::new(options);
+ let blob_field_names = core_options.blob_fields();
for name in &blob_field_names {
if !field_names.iter().any(|field| field == name) {
return Err(crate::Error::ConfigInvalid {
@@ -1548,25 +1568,42 @@ impl Schema {
return Ok(fields.to_vec());
}
- let pk_set: HashSet<&str> =
primary_keys.iter().map(String::as_str).collect();
let mut new_fields = Vec::with_capacity(fields.len());
for f in fields {
- let mut data_type = if blob_field_names.contains(f.name()) {
+ let data_type = if blob_field_names.contains(f.name()) {
normalize_blob_field_type(f.name(), f.data_type().clone())?
} else {
f.data_type().clone()
};
- if pk_set.contains(f.name()) && data_type.is_nullable() {
- data_type = data_type.copy_with_nullable(false)?;
- }
- new_fields.push(
- DataField::new(f.id(), f.name().to_string(), data_type)
- .with_description(f.description().map(|s| s.to_string())),
- );
+ let mut field = f.clone();
+ field.typ = data_type;
+ new_fields.push(field);
}
+ Self::normalize_primary_key_nullability(
+ &mut new_fields,
+ primary_keys,
+ core_options.primary_key_nullable(),
+ )?;
Ok(new_fields)
}
+ fn normalize_primary_key_nullability(
+ fields: &mut [DataField],
+ primary_keys: &[String],
+ nullable: bool,
+ ) -> crate::Result<()> {
+ if primary_keys.is_empty() {
+ return Ok(());
+ }
+ let primary_keys: HashSet<&str> =
primary_keys.iter().map(String::as_str).collect();
+ for field in fields {
+ if primary_keys.contains(field.name()) && field.typ.is_nullable()
!= nullable {
+ field.typ = field.typ.copy_with_nullable(nullable)?;
+ }
+ }
+ Ok(())
+ }
+
/// Table columns must not contain duplicate field names.
fn validate_no_duplicate_fields(field_names: &[String]) ->
crate::Result<()> {
let duplicates = Self::duplicate_fields(field_names);
@@ -2561,7 +2598,8 @@ impl SchemaBuilder {
self
}
- /// Set primary key columns. They must not be nullable.
+ /// Set primary key columns. Their nullability follows
`primary-key.nullable`
+ /// (false by default), regardless of the declared column nullability.
pub fn primary_key(mut self, names: impl IntoIterator<Item = impl
Into<String>>) -> Self {
self.primary_keys = names.into_iter().map(Into::into).collect();
self
@@ -2860,6 +2898,160 @@ mod tests {
}
}
+ #[test]
+ fn nullable_primary_key_option_normalizes_declared_not_null_keys() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("bucket", "1")
+ .option("primary-key.nullable", "true")
+ .build()
+ .unwrap();
+ assert!(schema.fields()[0].data_type().is_nullable());
+ }
+
+ #[test]
+ fn primary_key_normalization_preserves_field_metadata() {
+ for nullable in [false, true] {
+ let fields = vec![
+ DataField::new(
+ 7,
+ "id".into(),
+ DataType::Int(IntType::with_nullable(!nullable)),
+ )
+ .with_description(Some("identifier".into()))
+ .with_default_value(Some("42".into())),
+ DataField::new(11, "value".into(),
DataType::Int(IntType::new()))
+ .with_default_value(Some("1".into())),
+ ];
+ let schema = Schema::new(
+ fields.clone(),
+ vec![],
+ vec!["id".into()],
+ HashMap::from([(PRIMARY_KEY_NULLABLE_OPTION.into(),
nullable.to_string())]),
+ None,
+ )
+ .unwrap();
+ let mut expected = fields;
+ expected[0].typ = DataType::Int(IntType::with_nullable(nullable));
+ assert_eq!(schema.fields(), expected.as_slice());
+ let stored = TableSchema::new(0,
&schema).copy_with_options(HashMap::from([(
+ "alter-column-null-to-not-null.disabled".into(),
+ "false".into(),
+ )]));
+ let changed = stored
+
.apply_changes(vec![crate::spec::SchemaChange::UpdateColumnNullability {
+ field_names: vec!["id".into()],
+ new_nullability: false,
+ }])
+ .unwrap();
+ assert_eq!(changed.fields(), expected.as_slice());
+ }
+ }
+
+ #[test]
+ fn nullable_primary_keys_include_partition_columns() {
+ let schema = Schema::builder()
+ .column("pt", DataType::Int(IntType::with_nullable(false)))
+ .column("id", DataType::Int(IntType::with_nullable(false)))
+ .column("value", DataType::Int(IntType::with_nullable(false)))
+ .primary_key(["pt", "id"])
+ .partition_keys(["pt"])
+ .option(PRIMARY_KEY_NULLABLE_OPTION, "TRUE")
+ .build()
+ .unwrap();
+ assert!(schema.fields()[0].data_type().is_nullable());
+ assert!(schema.fields()[1].data_type().is_nullable());
+ assert!(!schema.fields()[2].data_type().is_nullable());
+ }
+
+ #[test]
+ fn nullable_primary_key_option_requires_primary_keys() {
+ let error = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .option(PRIMARY_KEY_NULLABLE_OPTION, "true")
+ .build()
+ .unwrap_err();
+ assert!(error
+ .to_string()
+ .contains("can only be enabled for a table with primary keys"));
+ }
+
+ #[test]
+ fn dynamic_options_preserve_primary_key_nullability() {
+ for nullable in [false, true] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option(PRIMARY_KEY_NULLABLE_OPTION, nullable.to_string())
+ .build()
+ .unwrap();
+ let stored = TableSchema::new(0, &schema);
+ let copied = stored.copy_with_options(HashMap::from([(
+ PRIMARY_KEY_NULLABLE_OPTION.into(),
+ (!nullable).to_string(),
+ )]));
+ assert_eq!(copied.core_options().primary_key_nullable(), nullable);
+ assert_eq!(copied.fields(), stored.fields());
+ let replaced =
stored.copy_with_replaced_dynamic_options(HashMap::new());
+ assert_eq!(replaced.core_options().primary_key_nullable(),
nullable);
+ }
+ }
+
+ #[test]
+ fn nullable_key_option_wins_over_alter_column_not_null() {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("value", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option(PRIMARY_KEY_NULLABLE_OPTION, "true")
+ .option("alter-column-null-to-not-null.disabled", "false")
+ .build()
+ .unwrap();
+ let stored = TableSchema::new(0, &schema);
+ let changed = stored
+
.apply_changes(vec![crate::spec::SchemaChange::UpdateColumnNullability {
+ field_names: vec!["id".into()],
+ new_nullability: false,
+ }])
+ .unwrap();
+ // Java reconstructs Schema after ALTER and reapplies the key option.
+ assert!(changed.fields()[0].data_type().is_nullable());
+ assert!(changed.core_options().primary_key_nullable());
+ }
+
+ #[test]
+ fn schema_changes_normalize_primary_key_nullability_option() {
+ use crate::spec::SchemaChange;
+
+ for nullable in [false, true] {
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option(PRIMARY_KEY_NULLABLE_OPTION, nullable.to_string())
+ .build()
+ .unwrap();
+ let stored = TableSchema::new(0, &schema);
+ let changed = stored
+ .apply_changes(vec![SchemaChange::SetOption {
+ key: PRIMARY_KEY_NULLABLE_OPTION.into(),
+ value: (!nullable).to_string(),
+ }])
+ .unwrap();
+ assert_eq!(changed.core_options().primary_key_nullable(),
!nullable);
+ assert_eq!(changed.fields()[0].data_type().is_nullable(),
!nullable);
+ assert_eq!(stored.fields()[0].data_type().is_nullable(), nullable);
+ let removed = stored
+ .apply_changes(vec![SchemaChange::RemoveOption {
+ key: PRIMARY_KEY_NULLABLE_OPTION.into(),
+ }])
+ .unwrap();
+ assert!(!removed.core_options().primary_key_nullable());
+ assert!(!removed.fields()[0].data_type().is_nullable());
+ }
+ }
+
#[test]
fn test_apply_changes_rejects_invalid_read_batch_size() {
let schema = Schema::builder()
diff --git a/crates/paimon/src/table/pk_vector_scan.rs
b/crates/paimon/src/table/pk_vector_scan.rs
index 0f6ad524..8d3d033a 100644
--- a/crates/paimon/src/table/pk_vector_scan.rs
+++ b/crates/paimon/src/table/pk_vector_scan.rs
@@ -1187,7 +1187,7 @@ mod tests {
async fn
plan_validates_original_scan_mode_and_rejects_nonempty_incremental_search() {
let (_tmp, table) = build_pruning_test_table(false).await;
for extra in [
- HashMap::from([("scan.mode".to_string(),
"latest-full".to_string())]),
+ HashMap::from([("scan.mode".to_string(),
"compacted-full".to_string())]),
HashMap::from([(
"incremental-between-timestamp".to_string(),
"0,9999999999999".to_string(),
@@ -1221,6 +1221,30 @@ mod tests {
.is_empty());
}
+ #[tokio::test]
+ async fn full_startup_modes_preserve_vector_scan_snapshot() {
+ let (_tmp, table) = build_pruning_test_table(false).await;
+ let snapshot =
table.resolve_read_snapshot().await.unwrap().unwrap();
+ for mode in ["latest", "latest-full", "from-snapshot-full"] {
+ let mut options = HashMap::from([("scan.mode".into(),
mode.into())]);
+ if mode == "from-snapshot-full" {
+ options.insert("scan.snapshot-id".into(),
snapshot.id().to_string());
+ }
+ let view = table.copy_with_options(options);
+ let plan = PkVectorScan::new(
+ &view,
+ prune_vector_field_id(&view),
+ PRUNE_INDEX_TYPE.to_string(),
+ None,
+ )
+ .plan()
+ .await
+ .unwrap();
+ assert_eq!(plan.snapshot_id, snapshot.id());
+ assert!(!plan.splits.is_empty());
+ }
+ }
+
#[tokio::test]
async fn
plan_uses_complete_retained_tag_metadata_and_explicit_empty_view() {
let (_tmp, table) = build_pruning_test_table(false).await;
diff --git a/crates/paimon/src/table/snapshot_scan_tests.rs
b/crates/paimon/src/table/snapshot_scan_tests.rs
index 50910988..c535dc2a 100644
--- a/crates/paimon/src/table/snapshot_scan_tests.rs
+++ b/crates/paimon/src/table/snapshot_scan_tests.rs
@@ -174,6 +174,107 @@ async fn
explicit_snapshot_overrides_startup_options_without_mutating_table() {
assert_eq!(rows(&selected, &batch).await, vec![(1, 10)]);
}
+#[tokio::test]
+async fn full_startup_modes_read_complete_batch_snapshots() {
+ for engine in [
+ "append",
+ "evolution",
+ "deduplicate",
+ "first-row",
+ "partial-update",
+ "aggregation",
+ ] {
+ let table = table(&format!("startup_modes_{engine}"), engine, "none",
1).await;
+ write_batch(&table, &make_batch(vec![1, 2], vec![10, 20])).await;
+ write_batch(&table, &make_batch(vec![1, 3], vec![11, 30])).await;
+ let expected = match engine {
+ "append" | "evolution" => vec![(1, 10), (1, 11), (2, 20), (3, 30)],
+ // Ordinary first-row scans skip uncompacted L0 files.
+ "first-row" => vec![],
+ "aggregation" => vec![(1, 21), (2, 20), (3, 30)],
+ _ => vec![(1, 11), (2, 20), (3, 30)],
+ };
+ for mode in ["latest", "latest-full", "from-snapshot-full"] {
+ let mut options = HashMap::from([("scan.mode".into(),
mode.into())]);
+ if mode == "from-snapshot-full" {
+ options.insert("scan.snapshot-id".into(), "1".into());
+ }
+ let selected = table.copy_with_options(options.clone());
+ let plan =
selected.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(
+ plan.snapshot_id(),
+ Some(if mode == "from-snapshot-full" { 1 } else { 2 })
+ );
+ assert!(plan.splits().iter().all(|split| !split.is_streaming()));
+ let want = if engine == "first-row" {
+ vec![]
+ } else if mode == "from-snapshot-full" {
+ vec![(1, 10), (2, 20)]
+ } else {
+ expected.clone()
+ };
+ assert_eq!(rows(&selected, &plan).await, want, "{engine}/{mode}");
+ assert_eq!(
+ selected.schema().options().get("scan.mode"),
+ options.get("scan.mode")
+ );
+ let mut builder = selected.new_read_builder();
+ builder.with_projection(&["value"]).unwrap();
+ builder.with_filter(
+ crate::spec::PredicateBuilder::new(selected.schema().fields())
+ .equal("id", crate::spec::Datum::Int(2))
+ .unwrap(),
+ );
+ builder.with_limit(1);
+ let plan = builder.new_scan().plan().await.unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ assert_eq!(
+ batches.iter().map(RecordBatch::num_rows).sum::<usize>(),
+ usize::from(engine != "first-row")
+ );
+ assert!(batches.iter().all(|batch| batch.num_columns() == 1));
+ }
+ }
+}
+
+#[tokio::test]
+async fn full_startup_modes_empty_tables_and_stream_boundaries() {
+ let table = table("startup_modes_empty", "append", "none", 1).await;
+ for mode in ["latest", "latest-full"] {
+ let selected =
table.copy_with_options(HashMap::from([("scan.mode".into(), mode.into())]));
+ let plan =
selected.new_read_builder().new_scan().plan().await.unwrap();
+ assert_eq!(plan.snapshot_id(), None);
+ assert!(plan.splits().is_empty());
+ }
+ for mode in ["latest", "from-snapshot-full"] {
+ let selected =
table.copy_with_options(HashMap::from([("scan.mode".into(), mode.into())]));
+ assert!(matches!(
+ selected.new_read_builder().new_stream_scan(),
+ Err(paimon::Error::Unsupported { .. })
+ ));
+ }
+ let selected = table.copy_with_options(HashMap::from([
+ ("scan.mode".into(), "from-snapshot-full".into()),
+ ("scan.snapshot-id".into(), "1".into()),
+ ]));
+ assert!(selected.new_read_builder().new_scan().plan().await.is_err());
+ let selected =
+ table.copy_with_options(HashMap::from([("scan.mode".into(),
"latest-full".into())]));
+ let mut stream = selected.new_read_builder().new_stream_scan().unwrap();
+ assert!(stream.plan().await.unwrap().is_none());
+ write_batch(&table, &make_batch(vec![1], vec![10])).await;
+ let plan = stream.plan().await.unwrap().unwrap();
+ assert_eq!(rows(&selected, &plan).await, vec![(1, 10)]);
+ assert_eq!(stream.checkpoint(), Some(2));
+}
+
#[tokio::test]
async fn changelog_reads_overwrite_while_batch_range_still_skips_it() {
let table = table("overwrite", "deduplicate", "input", 1).await;
diff --git a/crates/paimon/src/table/stream_table_scan.rs
b/crates/paimon/src/table/stream_table_scan.rs
index 96cadba6..2052d3ba 100644
--- a/crates/paimon/src/table/stream_table_scan.rs
+++ b/crates/paimon/src/table/stream_table_scan.rs
@@ -44,6 +44,25 @@ impl StreamTableScan {
pub(super) fn new(scan: TableScan<'_>) -> Result<Self> {
let mut scan = scan.into_owned_paimon_scan()?;
let options =
CoreOptions::new(scan.snapshot_table().schema().options());
+ // These aliases are supported by the batch starting scanner. A
+ // continuous reader needs a different starting position: latest must
+ // skip the initial state, and from-snapshot-full must start at that
+ // snapshot. Keep rejecting them until those stream policies exist.
+ if scan
+ .snapshot_table()
+ .schema()
+ .options()
+ .get("scan.mode")
+ .is_some_and(|mode| {
+ mode.eq_ignore_ascii_case("latest")
+ || mode.eq_ignore_ascii_case("from-snapshot-full")
+ })
+ {
+ return Err(Error::Unsupported {
+ message: "Stream scans do not yet support scan.mode=latest or
from-snapshot-full"
+ .into(),
+ });
+ }
let producer = options.try_changelog_producer()?;
let follow_up_mode = if producer == ChangelogProducer::None {
SnapshotScanMode::Delta
diff --git a/crates/paimon/src/table/table_merge_input.rs
b/crates/paimon/src/table/table_merge_input.rs
index 7512718b..2283a23f 100644
--- a/crates/paimon/src/table/table_merge_input.rs
+++ b/crates/paimon/src/table/table_merge_input.rs
@@ -118,23 +118,50 @@ impl MergeInput {
} else {
super::time_travel::resolve_snapshot(table).await?
};
- // Pin the source independently, keeping branch/time-travel resolution
- // and the caller's resolved schema. Include L0 as Python
plan_for_write.
- let table = if incremental {
+ // Pin the source independently, preserving its resolved read schema.
+ // Authorized sources keep their original catalog identity and plan the
+ // selected snapshot explicitly below. Include L0 as plan_for_write.
+ let query_auth = table.schema().core_options().query_auth_enabled();
+ let table = if incremental || query_auth {
table.clone()
} else {
table.copy_with_pinned_snapshot(snapshot.as_ref())
};
let mut builder = table.new_read_builder();
builder.with_projection(&names)?;
- let plan = if incremental || snapshot.is_some() {
- builder.new_scan().with_scan_all_files().plan().await?
+ let scan = if query_auth {
+ // Python materializes the entire source before matching. Ask for
+ // all source columns even when this MERGE only needs its keys,
+ // otherwise an unused column's mask/denial could be bypassed.
+ table
+ .new_read_builder()
+ .new_scan()
+ .authorize_merge_source()
+ .await?
+ } else {
+ builder.new_scan()
+ };
+ let plan = if query_auth && !incremental {
+ if let Some(snapshot) = snapshot {
+ scan.plan_selected_snapshot(snapshot,
super::table_scan::SnapshotScanMode::All)
+ .await?
+ } else {
+ super::source::Plan::new(Vec::new())
+ }
+ } else if incremental || snapshot.is_some() {
+ scan.with_scan_all_files().plan().await?
} else {
// Run the normal reader's policy checks even with no snapshot, but
// do not observe files from a commit arriving after resolution.
super::source::Plan::new(Vec::new())
};
- let mut stream = builder.new_read()?.to_arrow(plan.splits())?;
+ let read = builder.new_read()?;
+ if query_auth && plan.splits().is_empty() {
+ // Empty input carries no splits on which to stamp the live grant.
+ // Authorization above still runs for empty tables/windows.
+ return Self::new(vec![RecordBatch::new_empty(projected)]);
+ }
+ let mut stream = read.to_arrow(plan.splits())?;
let mut batches = Vec::new();
while let Some(batch) = stream.try_next().await? {
if batch.num_rows() > 0 {
diff --git a/crates/paimon/src/table/table_merge_into_tests.rs
b/crates/paimon/src/table/table_merge_into_tests.rs
index 2c520d21..6de570e2 100644
--- a/crates/paimon/src/table/table_merge_into_tests.rs
+++ b/crates/paimon/src/table/table_merge_into_tests.rs
@@ -703,6 +703,56 @@ async fn
core_merge_empty_table_source_keeps_no_match_assignments_lazy() {
assert_eq!(rows(&target).await, vec![(Some(1), Some(10))]);
}
+#[tokio::test]
+async fn core_merge_full_startup_modes_select_source_snapshot() {
+ for mode in ["latest", "latest-full", "from-snapshot-full"] {
+ let target = table_at(&format!("target_full_{mode}"));
+ seed(&target, &batch(vec![Some(1)], vec![Some(10)])).await;
+ let source = table_at(&format!("source_full_{mode}"));
+ seed(&source, &batch(vec![Some(1)], vec![Some(11)])).await;
+ seed(&source, &batch(vec![Some(2)], vec![Some(22)])).await;
+ let mut options =
std::collections::HashMap::from([("scan.mode".into(), mode.into())]);
+ if mode == "from-snapshot-full" {
+ options.insert("scan.snapshot-id".into(), "1".into());
+ }
+ let selected = source.copy_with_options(options);
+ let messages = target
+ .new_write_builder()
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Table(Arc::new(selected)),
+ vec![("id".into(), "id".into())],
+ vec![update(None)],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: vec![
+ ("id".into(),
MergeAssignment::SourceColumn("id".into())),
+ (
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ ),
+ ],
+ }],
+ )
+ .await
+ .unwrap();
+ TableCommit::new(target.clone(), "merge".into())
+ .commit(messages)
+ .await
+ .unwrap();
+ let mut expected = vec![(Some(1), Some(11))];
+ if mode != "from-snapshot-full" {
+ expected.push((Some(2), Some(22)));
+ }
+ assert_eq!(rows(&target).await, expected, "{mode}");
+ assert_eq!(
+ rows(&source).await,
+ vec![(Some(1), Some(11)), (Some(2), Some(22))]
+ );
+ }
+}
+
#[tokio::test]
async fn core_merge_empty_table_source_obeys_reader_authorization() {
let target = table();
diff --git a/crates/paimon/src/table/table_scan.rs
b/crates/paimon/src/table/table_scan.rs
index 4e89f05d..9e60f035 100644
--- a/crates/paimon/src/table/table_scan.rs
+++ b/crates/paimon/src/table/table_scan.rs
@@ -1144,6 +1144,25 @@ impl<'a> TableScan<'a> {
}
}
+ /// MERGE table sources follow plan_for_write: authorize the whole table,
+ /// then include L0 only when no row filter or column mask was returned.
+ /// Keep this scope internal; public metadata scans must still fail closed.
+ pub(super) async fn authorize_merge_source(&self) -> crate::Result<Self> {
+ let scan = self.authorize_snapshot_read().await?;
+ if let TableScanKind::Paimon(source) = &scan.0 {
+ if source
+ .query_auth_grant
+ .as_ref()
+ .is_some_and(|grant| !grant.is_unrestricted())
+ {
+ return Err(super::query_auth::unsupported(
+ "MERGE table sources require unrestricted query
authorization",
+ ));
+ }
+ }
+ Ok(scan)
+ }
+
pub(super) fn validate_snapshot_read(&self) -> crate::Result<()> {
let TableScanKind::Paimon(scan) = &self.0 else {
unreachable!("snapshot scans only contain Paimon tables")
diff --git a/crates/paimon/tests/nullable_primary_key_test.rs
b/crates/paimon/tests/nullable_primary_key_test.rs
new file mode 100644
index 00000000..cb69bf50
--- /dev/null
+++ b/crates/paimon/tests/nullable_primary_key_test.rs
@@ -0,0 +1,342 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+mod common;
+
+use arrow_array::{Array, ArrayRef, Int32Array, Int8Array, RecordBatch};
+use common::incremental_helpers::{memory_table, persist_table_schema,
setup_dirs, write_batch};
+use futures::TryStreamExt;
+use paimon::spec::{DataType, IntType, Predicate, PredicateBuilder, Schema,
TableSchema};
+use paimon::table::{MergeAssignment, MergeSource, Table, WhenMatched,
WhenNotMatched};
+use std::collections::HashMap;
+use std::sync::Arc;
+
+type Row = (Option<i32>, Option<i32>, Option<i32>);
+
+async fn table(name: &str, options: &[(&str, &str)], primary_key: bool) ->
Table {
+ let mut schema = Schema::builder()
+ // Engines may declare keys NOT NULL before applying the table option.
+ .column("pt", DataType::Int(IntType::with_nullable(!primary_key)))
+ .column("id", DataType::Int(IntType::with_nullable(!primary_key)))
+ .column("value", DataType::Int(IntType::new()));
+ if primary_key {
+ schema = schema
+ .primary_key(["pt", "id"])
+ .partition_keys(["pt"])
+ .option("primary-key.nullable", "true");
+ } else {
+ schema = schema
+ .option("data-evolution.enabled", "true")
+ .option("row-tracking.enabled", "true")
+ .option("deletion-vectors.enabled", "true");
+ }
+ for (key, value) in options {
+ schema = schema.option(*key, *value);
+ }
+ if options
+ .iter()
+ .any(|(key, value)| *key == "merge-engine" && *value == "aggregation")
+ {
+ schema = schema.option("fields.value.aggregate-function", "sum");
+ }
+ let path = format!("memory:/nullable/{name}");
+ let (io, table) = memory_table(&path, TableSchema::new(0,
&schema.build().unwrap()));
+ setup_dirs(&io, &path).await;
+ persist_table_schema(&io, &path, table.schema()).await;
+ table
+}
+
+fn batch(rows: &[Row]) -> RecordBatch {
+ RecordBatch::try_from_iter(["pt", "id",
"value"].into_iter().enumerate().map(
+ |(column, name)| {
+ let values: Int32Array = rows
+ .iter()
+ .map(|row| match column {
+ 0 => row.0,
+ 1 => row.1,
+ _ => row.2,
+ })
+ .collect();
+ (name, Arc::new(values) as ArrayRef)
+ },
+ ))
+ .unwrap()
+}
+
+async fn read(table: &Table, predicate: Option<Predicate>) -> Vec<Row> {
+ let mut builder = table.new_read_builder();
+ if let Some(predicate) = predicate {
+ builder.with_filter(predicate);
+ }
+ // First-row source reads include L0, as Python plan_for_write does.
+ let plan = builder
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .unwrap();
+ let batches: Vec<RecordBatch> = builder
+ .new_read()
+ .unwrap()
+ .to_arrow(plan.splits())
+ .unwrap()
+ .try_collect()
+ .await
+ .unwrap();
+ let mut rows = vec![];
+ for batch in batches {
+ let columns: Vec<_> = batch
+ .columns()
+ .iter()
+ .map(|array| array.as_any().downcast_ref::<Int32Array>().unwrap())
+ .collect();
+ for row in 0..batch.num_rows() {
+ let get =
+ |column: usize| (!columns[column].is_null(row)).then(||
columns[column].value(row));
+ rows.push((get(0), get(1), get(2)));
+ }
+ }
+ rows.sort_unstable();
+ rows
+}
+
+#[tokio::test]
+async fn
nullable_composite_keys_merge_across_writer_restarts_and_bucket_modes() {
+ for bucket in ["1", "3", "-1"] {
+ for engine in ["deduplicate", "partial-update", "aggregation",
"first-row"] {
+ let table = table(
+ &format!("{bucket}-{engine}"),
+ &[("bucket", bucket), ("merge-engine", engine)],
+ true,
+ )
+ .await;
+ let first = [
+ (None, None, Some(10)),
+ (Some(1), None, Some(100)),
+ (Some(1), Some(0), Some(1)),
+ ];
+ write_batch(&table, &batch(&first)).await;
+ // A new writer must restore dynamic bucket routing for null keys.
+ write_batch(
+ &table,
+ &batch(&[
+ (None, None, Some(20)),
+ (Some(1), None, Some(200)),
+ (None, Some(0), Some(2)),
+ ]),
+ )
+ .await;
+ let values = match engine {
+ "aggregation" => (30, 300),
+ "first-row" => (10, 100),
+ _ => (20, 200),
+ };
+ let expected = vec![
+ (None, None, Some(values.0)),
+ (None, Some(0), Some(2)),
+ (Some(1), None, Some(values.1)),
+ (Some(1), Some(0), Some(1)),
+ ];
+ assert_eq!(read(&table, None).await, expected,
"{bucket}/{engine}");
+ let pb = PredicateBuilder::new(table.schema().fields());
+ assert_eq!(
+ read(&table, Some(pb.is_null("id").unwrap())).await,
+ vec![expected[0], expected[2]],
+ "{bucket}/{engine}"
+ );
+ assert_eq!(
+ read(&table, Some(pb.is_null("pt").unwrap())).await,
+ expected[..2],
+ "{bucket}/{engine}"
+ );
+ let first_view = table.copy_with_options(HashMap::from([
+ ("scan.mode".into(), "from-snapshot-full".into()),
+ ("scan.snapshot-id".into(), "1".into()),
+ ]));
+ assert_eq!(read(&first_view, None).await, first);
+ }
+ }
+}
+
+#[tokio::test]
+async fn null_key_restores_its_nonzero_dynamic_bucket() {
+ let table = table(
+ "null-nonzero-bucket",
+ &[("bucket", "-1"), ("dynamic-bucket.target-row-num", "1")],
+ true,
+ )
+ .await;
+ write_batch(
+ &table,
+ &batch(&[(None, Some(0), Some(100)), (None, None, Some(10))]),
+ )
+ .await;
+ for value in [20, 30] {
+ let builder = table.new_write_builder();
+ let mut writer = builder.new_write().unwrap();
+ writer
+ .write_arrow_batch(&batch(&[(None, None, Some(value))]))
+ .await
+ .unwrap();
+ let messages = writer.prepare_commit().await.unwrap();
+ assert_eq!(messages.len(), 1);
+ assert_eq!(messages[0].bucket, 1);
+ builder.new_commit().commit(messages).await.unwrap();
+ }
+ assert_eq!(
+ read(&table, None).await,
+ vec![(None, None, Some(30)), (None, Some(0), Some(100))]
+ );
+}
+
+#[tokio::test]
+async fn null_key_retract_does_not_delete_zero_or_other_partition_keys() {
+ for bucket in ["3", "-1"] {
+ let table = table(&format!("delete-{bucket}"), &[("bucket", bucket)],
true).await;
+ write_batch(
+ &table,
+ &batch(&[
+ (None, None, Some(10)),
+ (None, Some(0), Some(20)),
+ (Some(1), None, Some(30)),
+ ]),
+ )
+ .await;
+ let mut retract = batch(&[(None, None, Some(10))]);
+ let mut fields = retract.schema().fields().to_vec();
+ fields.push(Arc::new(arrow_schema::Field::new(
+ "_VALUE_KIND",
+ arrow_schema::DataType::Int8,
+ false,
+ )));
+ let mut columns = retract.columns().to_vec();
+ columns.push(Arc::new(Int8Array::from(vec![3])));
+ retract =
+ RecordBatch::try_new(Arc::new(arrow_schema::Schema::new(fields)),
columns).unwrap();
+ write_batch(&table, &retract).await;
+ assert_eq!(
+ read(&table, None).await,
+ vec![(None, Some(0), Some(20)), (Some(1), None, Some(30))]
+ );
+ write_batch(&table, &batch(&[(None, None, Some(40))])).await;
+ assert_eq!(
+ read(&table, None).await,
+ vec![
+ (None, None, Some(40)),
+ (None, Some(0), Some(20)),
+ (Some(1), None, Some(30)),
+ ]
+ );
+ }
+}
+
+#[tokio::test]
+async fn nullable_primary_key_merge_source_keeps_sql_null_join_semantics() {
+ for engine in ["deduplicate", "partial-update", "aggregation",
"first-row"] {
+ let source = table(
+ &format!("source-{engine}"),
+ &[("bucket", "-1"), ("merge-engine", engine)],
+ true,
+ )
+ .await;
+ write_batch(
+ &source,
+ &batch(&[
+ (None, None, Some(10)),
+ (None, Some(1), Some(11)),
+ (Some(1), None, Some(12)),
+ (Some(1), Some(1), Some(100)),
+ ]),
+ )
+ .await;
+ write_batch(
+ &source,
+ &batch(&[
+ (None, None, Some(20)),
+ (None, Some(1), Some(21)),
+ (Some(1), None, Some(22)),
+ (Some(1), Some(1), Some(200)),
+ ]),
+ )
+ .await;
+ let target = table(&format!("target-{engine}"), &[], false).await;
+ write_batch(
+ &target,
+ &batch(&[
+ (None, None, Some(-1)),
+ (None, Some(1), Some(-3)),
+ (Some(1), None, Some(-4)),
+ (Some(1), Some(1), Some(-2)),
+ ]),
+ )
+ .await;
+ let builder = target.new_write_builder();
+ let messages = builder
+ .new_update()
+ .unwrap()
+ .merge_into(
+ MergeSource::Table(Arc::new(source.clone())),
+ vec![("pt".into(), "pt".into()), ("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete: false,
+ assignments: vec![(
+ "value".into(),
+ MergeAssignment::SourceColumn("value".into()),
+ )],
+ }],
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: ["pt", "id", "value"]
+ .into_iter()
+ .map(|name| (name.into(),
MergeAssignment::SourceColumn(name.into())))
+ .collect(),
+ }],
+ )
+ .await
+ .unwrap();
+ builder.new_commit().commit(messages).await.unwrap();
+ let values = match engine {
+ "aggregation" => (30, 32, 34, 300),
+ "first-row" => (10, 11, 12, 100),
+ _ => (20, 21, 22, 200),
+ };
+ // PK merging is null-safe, but MERGE's SQL equality must not match
nulls.
+ assert_eq!(
+ read(&target, None).await,
+ vec![
+ (None, None, Some(-1)),
+ (None, None, Some(values.0)),
+ (None, Some(1), Some(-3)),
+ (None, Some(1), Some(values.1)),
+ (Some(1), None, Some(-4)),
+ (Some(1), None, Some(values.2)),
+ (Some(1), Some(1), Some(values.3)),
+ ]
+ );
+ assert_eq!(
+ source
+ .snapshot_manager()
+ .get_latest_snapshot()
+ .await
+ .unwrap()
+ .unwrap()
+ .id(),
+ 2
+ );
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog/merge_sources.rs
b/crates/paimon/tests/rest_catalog/merge_sources.rs
new file mode 100644
index 00000000..1fcf5bb1
--- /dev/null
+++ b/crates/paimon/tests/rest_catalog/merge_sources.rs
@@ -0,0 +1,285 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use super::*;
+use paimon::table::{MergeAssignment, MergeSource, TableCommit, WhenMatched,
WhenNotMatched};
+
+fn target() -> Table {
+ Table::new(
+ paimon::io::FileIOBuilder::new("memory").build().unwrap(),
+ Identifier::new("default", "merge_target"),
+ "memory:/merge_target".into(),
+ paimon::spec::TableSchema::new(
+ 0,
+ &people(&[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ("deletion-vectors.enabled", "true"),
+ ]),
+ ),
+ None,
+ )
+}
+
+async fn merge(target: &Table, source: &Table, delete: bool) ->
paimon::Result<()> {
+ let messages = target
+ .new_write_builder()
+ .new_update()?
+ .merge_into(
+ MergeSource::Table(Arc::new(source.clone())),
+ vec![("id".into(), "id".into())],
+ vec![WhenMatched {
+ condition: None,
+ delete,
+ assignments: if delete {
+ Vec::new()
+ } else {
+ vec![("name".into(),
MergeAssignment::SourceColumn("name".into()))]
+ },
+ }],
+ if delete {
+ Vec::new()
+ } else {
+ vec![WhenNotMatched {
+ condition: None,
+ assignments: vec![
+ ("id".into(),
MergeAssignment::SourceColumn("id".into())),
+ ("name".into(),
MergeAssignment::SourceColumn("name".into())),
+ ],
+ }]
+ },
+ )
+ .await?;
+ TableCommit::new(target.clone(), "authorized-merge".into())
+ .commit(messages)
+ .await?;
+ Ok(())
+}
+
+#[tokio::test]
+async fn unrestricted_source_supports_merge_and_keeps_metadata_scans_closed() {
+ let g = written("merge_source", people, vec![people_rows(1..=2)]).await;
+ g.set_rules(&[], &[]);
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old"), (3, "keep")]),
"seed").await;
+ merge(&target, &g.table, false).await.unwrap();
+ assert_eq!(
+ id_names(&read_all(&target.new_read_builder()).await.unwrap()),
+ [
+ (1, Some("alice".into())),
+ (2, Some("bob".into())),
+ (3, Some("keep".into())),
+ ]
+ );
+ assert_eq!(g.ctx.server.auth_selects("default", "merge_source"), [None]);
+ // An unrestricted data-read grant must not expose a public metadata
bypass.
+ assert!(g
+ .table
+ .new_read_builder()
+ .new_scan()
+ .with_scan_all_files()
+ .plan()
+ .await
+ .is_err());
+}
+
+#[tokio::test]
+async fn
unrestricted_source_authorizes_empty_tables_and_keeps_historical_guard() {
+ let g = written(
+ "merge_history",
+ people,
+ vec![people_rows(1..=1), people_rows(2..=2)],
+ )
+ .await;
+ g.set_rules(&[], &[]);
+ let snapshot = g.table.snapshot_manager().get_snapshot(1).await.unwrap();
+ for view in [
+ g.table.copy_with_pinned_snapshot(None),
+ g.table.copy_with_pinned_snapshot(Some(&snapshot)),
+ ] {
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old")]), "seed").await;
+ let error = merge(&target, &view, false).await.unwrap_err();
+ assert!(error.to_string().contains("time-travelled"), "{error}");
+ assert_eq!(
+ id_names(&read_all(&target.new_read_builder()).await.unwrap()),
+ [(1, Some("old".into()))]
+ );
+ }
+ let empty = written("merge_empty", people, vec![]).await;
+ empty.set_rules(&[], &[]);
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old")]), "seed").await;
+ merge(&target, &empty.table, true).await.unwrap();
+ assert_eq!(
+ ids(&read_all(&target.new_read_builder()).await.unwrap()),
+ [1]
+ );
+ assert_eq!(
+ empty.ctx.server.auth_selects("default", "merge_empty"),
+ [None]
+ );
+}
+
+#[tokio::test]
+async fn
merge_rejects_restricted_sources_even_for_unused_fields_or_empty_views() {
+ for empty in [false, true] {
+ let source = if empty {
+ // The empty branch here is a genuinely empty current table, not
+ // a historical handle whose authorization is separately refused.
+ written("merge_rules_empty", people, vec![]).await
+ } else {
+ written("merge_rules_full", people, vec![people_rows(1..=2)]).await
+ };
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old")]), "seed").await;
+ for mask in [false, true] {
+ if mask {
+ source.set_rules(&[], &[("name", upper(1, "name"))]);
+ } else {
+ source.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 1)], &[]);
+ }
+ // DELETE matches only `id`: an unused column's mask still makes
the
+ // source unsuitable for Python's existing plan_for_write contract.
+ let error = merge(&target, &source.table, true).await.unwrap_err();
+ assert!(error.to_string().contains("unrestricted"), "{error}");
+ assert_eq!(
+ ids(&read_all(&target.new_read_builder()).await.unwrap()),
+ [1]
+ );
+ assert_eq!(
+ target
+ .snapshot_manager()
+ .get_latest_snapshot_id()
+ .await
+ .unwrap(),
+ Some(1)
+ );
+ }
+ assert!(source
+ .ctx
+ .server
+ .auth_selects("default", source.identifier.object())
+ .iter()
+ .all(Option::is_none));
+ }
+}
+
+#[tokio::test]
+async fn merge_authorizes_all_source_columns_and_refreshes_each_operation() {
+ let g = written("merge_columns", people, vec![people_rows(1..=2)]).await;
+ g.set_rules(&[], &[]);
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old")]), "seed").await;
+ g.ctx
+ .server
+ .set_column_auth("default", "merge_columns", vec!["id".into()]);
+ // Key-only matching must not reduce a table source to a different select.
+ assert_forbidden(merge(&target, &g.table, true).await.unwrap_err());
+ assert_eq!(
+ ids(&read_all(&target.new_read_builder()).await.unwrap()),
+ [1]
+ );
+ g.ctx
+ .server
+ .set_column_auth("default", "merge_columns", vec!["id".into(),
"name".into()]);
+ merge(&target, &g.table, false).await.unwrap();
+ g.set_rules(&[int_leaf(0, "id", "GREATER_THAN", 1)], &[]);
+ let error = merge(&target, &g.table, true).await.unwrap_err();
+ assert!(error.to_string().contains("unrestricted"), "{error}");
+ assert_eq!(
+ ids(&read_all(&target.new_read_builder()).await.unwrap()),
+ [1, 2]
+ );
+}
+
+#[tokio::test]
+async fn unrestricted_pk_source_merges_versions_including_level_zero() {
+ let g = written(
+ "merge_pk_source",
+ |options| {
+ let mut builder = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name",
DataType::VarChar(VarCharType::new(255).unwrap()))
+ .primary_key(["id"])
+ .option("bucket", "1");
+ for (key, value) in options {
+ builder = builder.option(*key, *value);
+ }
+ builder.build().unwrap()
+ },
+ vec![people_batch(&[(1, "old"), (2, "bob")]), people_rows(1..=1)],
+ )
+ .await;
+ g.set_rules(&[], &[]);
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "target")]), "seed").await;
+ merge(&target, &g.table, false).await.unwrap();
+ assert_eq!(
+ id_names(&read_all(&target.new_read_builder()).await.unwrap()),
+ [(1, Some("alice".into())), (2, Some("bob".into()))]
+ );
+}
+
+#[tokio::test]
+async fn
unrestricted_de_and_timestamp_window_sources_preserve_reader_semantics() {
+ for data_evolution in [false, true] {
+ let g = written(
+ "merge_window_source",
+ |options| {
+ if data_evolution {
+ people(
+ &[
+ options,
+ &[
+ ("data-evolution.enabled", "true"),
+ ("row-tracking.enabled", "true"),
+ ("deletion-vectors.enabled", "true"),
+ ],
+ ]
+ .concat(),
+ )
+ } else {
+ people(options)
+ }
+ },
+ vec![people_rows(1..=1), people_rows(2..=2)],
+ )
+ .await;
+ g.set_rules(&[], &[]);
+ let source = if data_evolution {
+ g.table.clone()
+ } else {
+ g.table.copy_with_options(HashMap::from([(
+ "incremental-between-timestamp".into(),
+ "0,9223372036854775807".into(),
+ )]))
+ };
+ let target = target();
+ write_batch(&target, people_batch(&[(1, "old")]), "seed").await;
+ merge(&target, &source, false).await.unwrap();
+ assert_eq!(
+ id_names(&read_all(&target.new_read_builder()).await.unwrap()),
+ [(1, Some("alice".into())), (2, Some("bob".into()))]
+ );
+ // Combined timestamp planning shares the same live grant.
+ assert_eq!(
+ g.ctx.server.auth_selects("default", "merge_window_source"),
+ [None]
+ );
+ }
+}
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index d51532cd..8f0e95b8 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -5820,3 +5820,7 @@ async fn test_query_auth_explicit_partition_selector() {
#[cfg(not(windows))]
#[path = "rest_catalog/authorized_snapshot_scans.rs"]
mod authorized_snapshot_scans;
+
+#[cfg(not(windows))]
+#[path = "rest_catalog/merge_sources.rs"]
+mod merge_sources;
