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 09a0668c fix(spec): rewrite field-scoped options on column rename
(#635)
09a0668c is described below
commit 09a0668c9cffbf89c0b6821268bf027ded5e6550
Author: kid <[email protected]>
AuthorDate: Mon Aug 3 11:41:32 2026 +0800
fix(spec): rewrite field-scoped options on column rename (#635)
---
crates/paimon/src/spec/aggregation.rs | 162 ++++++++++++++++++++++++++--
crates/paimon/src/spec/schema.rs | 192 +++++++++++++++++++++++++++++++++-
2 files changed, 343 insertions(+), 11 deletions(-)
diff --git a/crates/paimon/src/spec/aggregation.rs
b/crates/paimon/src/spec/aggregation.rs
index 0de0f86e..194d7952 100644
--- a/crates/paimon/src/spec/aggregation.rs
+++ b/crates/paimon/src/spec/aggregation.rs
@@ -33,6 +33,10 @@ const DISTINCT_SUFFIX: &str = ".distinct";
const SEQUENCE_GROUP_SUFFIX: &str = ".sequence-group";
const NESTED_KEY_SUFFIX: &str = ".nested-key";
const COUNT_LIMIT_SUFFIX: &str = ".count-limit";
+const MAP_STORAGE_LAYOUT_SUFFIX: &str = ".map.storage-layout";
+const MAP_SHARED_SHREDDING_MAX_COLUMNS_SUFFIX: &str =
".map.shared-shredding.max-columns";
+const MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY_SUFFIX: &str =
+ ".map.shared-shredding.column-placement-policy";
/// Minimal aggregation mode recognized by the current Rust implementation.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -278,14 +282,34 @@ fn parse_field_scoped_option_key(key: &str) ->
Option<(&str, FieldScopedOptionKi
None
}
-/// Field-scoped aggregation option suffixes whose key names a single column,
-/// so the column rename/drop path can keep them in sync with the schema.
-const FIELD_SCOPED_RENAMEABLE_SUFFIXES: [&str; 2] =
- [AGG_FUNCTION_SUFFIX, LIST_AGG_DELIMITER_SUFFIX];
-
-/// Rename a column inside field-scoped aggregation option KEYS, mirroring
-/// Java `SchemaManager.applyRenameColumnsToOptions` (case 2): the value is
-/// unchanged, only `fields.<old>.<suffix>` -> `fields.<new>.<suffix>`.
+/// Field-scoped option suffixes whose key names a single column
+/// (`fields.<col>.<suffix>`) while the value carries no column names, so the
+/// column rename/drop path can keep them in sync with the schema by rewriting
+/// the key only. Mirrors case 2 of Java
+/// `SchemaManager.applyRenameColumnsToOptions` (Java's case-2 list; the
+/// aggregation/partial-update engines do not accept every entry at create
+/// time, but the keys can still arrive in Java-written metadata).
+const FIELD_SCOPED_RENAMEABLE_SUFFIXES: [&str; 7] = [
+ AGG_FUNCTION_SUFFIX,
+ IGNORE_RETRACT_SUFFIX,
+ DISTINCT_SUFFIX,
+ LIST_AGG_DELIMITER_SUFFIX,
+ MAP_STORAGE_LAYOUT_SUFFIX,
+ MAP_SHARED_SHREDDING_MAX_COLUMNS_SUFFIX,
+ MAP_SHARED_SHREDDING_COLUMN_PLACEMENT_POLICY_SUFFIX,
+];
+
+/// Field-scoped option suffixes where BOTH the key's comma-separated field
+/// list and the value may name columns (`fields.<col,col,...>.<suffix>`).
+/// Mirrors case 3 of Java `SchemaManager.applyRenameColumnsToOptions`.
+const FIELD_SCOPED_FIELD_LIST_SUFFIXES: [&str; 2] = [SEQUENCE_GROUP_SUFFIX,
NESTED_KEY_SUFFIX];
+
+/// Rename a column inside field-scoped option keys (and, for field-list
+/// options, values), mirroring Java
`SchemaManager.applyRenameColumnsToOptions`:
+///
+/// - case 2: `fields.<old>.<suffix>` -> `fields.<new>.<suffix>`, value kept
+/// - case 3 (`sequence-group`, `nested-key`): every comma-separated entry
+/// equal to `old` is renamed in BOTH the key's field list and the value
pub(crate) fn rename_field_scoped_options(
options: &mut HashMap<String, String>,
old: &str,
@@ -297,6 +321,41 @@ pub(crate) fn rename_field_scoped_options(
options.insert(format!("{FIELDS_PREFIX}{new}{suffix}"), value);
}
}
+
+ // Collect first: mutating the map while iterating it is not allowed, and
+ // options that don't reference `old` must stay untouched.
+ let mut rewritten = Vec::new();
+ for (key, value) in options.iter() {
+ let Some(suffix) = FIELD_SCOPED_FIELD_LIST_SUFFIXES
+ .iter()
+ .find(|suffix| key.starts_with(FIELDS_PREFIX) &&
key.ends_with(**suffix))
+ else {
+ continue;
+ };
+ let key_fields = &key[FIELDS_PREFIX.len()..key.len() - suffix.len()];
+ let new_key_fields = rename_in_field_list(key_fields, old, new);
+ let new_value = rename_in_field_list(value, old, new);
+ if new_key_fields != key_fields || new_value != *value {
+ rewritten.push((
+ key.clone(),
+ format!("{FIELDS_PREFIX}{new_key_fields}{suffix}"),
+ new_value,
+ ));
+ }
+ }
+ for (old_key, new_key, new_value) in rewritten {
+ options.remove(&old_key);
+ options.insert(new_key, new_value);
+ }
+}
+
+/// Rename exact matches of `old` inside a comma-separated field list,
+/// preserving order and any entries that don't match.
+fn rename_in_field_list(list: &str, old: &str, new: &str) -> String {
+ list.split(',')
+ .map(|field| if field == old { new } else { field })
+ .collect::<Vec<_>>()
+ .join(",")
}
/// Remove a dropped column's field-scoped aggregation option keys so no
@@ -724,4 +783,91 @@ mod tests {
}
}
}
+
+ #[test]
+ fn test_rename_field_scoped_options_case2_rewrites_key_only() {
+ let mut options: HashMap<String, String> = [
+ ("fields.a.aggregate-function", "sum"),
+ ("fields.a.ignore-retract", "true"),
+ ("fields.a.distinct", "true"),
+ ("fields.a.list-agg-delimiter", ";"),
+ ("fields.a.map.storage-layout", "shared-shredding"),
+ ("fields.a.map.shared-shredding.max-columns", "64"),
+ (
+ "fields.a.map.shared-shredding.column-placement-policy",
+ "even",
+ ),
+ // Same leading characters but a different column: must not move.
+ ("fields.ab.aggregate-function", "max"),
+ ]
+ .into_iter()
+ .map(|(key, value)| (key.to_string(), value.to_string()))
+ .collect();
+
+ rename_field_scoped_options(&mut options, "a", "b");
+
+ for (suffix, value) in [
+ ("aggregate-function", "sum"),
+ ("ignore-retract", "true"),
+ ("distinct", "true"),
+ ("list-agg-delimiter", ";"),
+ ("map.storage-layout", "shared-shredding"),
+ ("map.shared-shredding.max-columns", "64"),
+ ("map.shared-shredding.column-placement-policy", "even"),
+ ] {
+ assert_eq!(
+ options
+ .get(&format!("fields.b.{suffix}"))
+ .map(String::as_str),
+ Some(value),
+ "expected fields.b.{suffix} to carry the original value"
+ );
+ assert!(
+ !options.contains_key(&format!("fields.a.{suffix}")),
+ "old key fields.a.{suffix} should be gone"
+ );
+ }
+ assert_eq!(
+ options
+ .get("fields.ab.aggregate-function")
+ .map(String::as_str),
+ Some("max")
+ );
+ }
+
+ #[test]
+ fn test_rename_field_scoped_options_case3_rewrites_key_and_value() {
+ let mut options: HashMap<String, String> = [
+ // Rename hits both the key's field list and the value.
+ ("fields.g1,g2.sequence-group", "g2,price"),
+ // No reference to the renamed column: must stay untouched.
+ ("fields.other.sequence-group", "price"),
+ // nested-key: the column can appear in the key and in the value.
+ ("fields.g2.nested-key", "g2"),
+ ]
+ .into_iter()
+ .map(|(key, value)| (key.to_string(), value.to_string()))
+ .collect();
+
+ rename_field_scoped_options(&mut options, "g2", "g3");
+
+ assert_eq!(
+ options
+ .get("fields.g1,g3.sequence-group")
+ .map(String::as_str),
+ Some("g3,price")
+ );
+ assert_eq!(
+ options
+ .get("fields.other.sequence-group")
+ .map(String::as_str),
+ Some("price")
+ );
+ assert_eq!(
+ options.get("fields.g3.nested-key").map(String::as_str),
+ Some("g3")
+ );
+ assert!(!options.contains_key("fields.g1,g2.sequence-group"));
+ assert!(!options.contains_key("fields.g2.nested-key"));
+ }
}
diff --git a/crates/paimon/src/spec/schema.rs b/crates/paimon/src/spec/schema.rs
index b3c48d13..3e2e024b 100644
--- a/crates/paimon/src/spec/schema.rs
+++ b/crates/paimon/src/spec/schema.rs
@@ -372,9 +372,10 @@ impl TableSchema {
name,
&new_name,
);
- // Field-scoped aggregation options encode the column in
the
- // key (`fields.<col>.aggregate-function` /
`.list-agg-delimiter`),
- // so they must be rewritten too, mirroring Java
+ // Field-scoped options encode column names in the key
+ // (`fields.<col>.aggregate-function`,
`fields.<cols>.sequence-group`,
+ // ...) and, for field-list options, in the value too, so
they
+ // must be rewritten as well, mirroring Java
// `SchemaManager.applyRenameColumnsToOptions`.
rename_field_scoped_options(&mut new_schema.options, name,
&new_name);
}
@@ -3297,6 +3298,191 @@ mod tests {
);
}
+ #[test]
+ fn test_rename_column_rewrites_sequence_group_options() {
+ // `sequence-group` is rejected by Rust's create-time merge-engine
+ // validation, so the fixture carries the option on a table without
+ // `merge-engine` — the shape in which Java-written metadata arrives.
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("version", DataType::Int(IntType::new()))
+ .column("source_order", DataType::Int(IntType::new()))
+ .column("price", DataType::Int(IntType::new()))
+ .column("quantity", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option(
+ "fields.version,source_order.sequence-group",
+ "price,quantity",
+ )
+ .build()
+ .unwrap(),
+ );
+
+ // Rename a sequence field (key side) and a protected field (value
side).
+ let new_schema = table_schema
+ .apply_changes(vec![
+ crate::spec::SchemaChange::rename_column(
+ "source_order".to_string(),
+ "order_seq".to_string(),
+ ),
+ crate::spec::SchemaChange::rename_column("price".to_string(),
"amount".to_string()),
+ ])
+ .unwrap();
+
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.version,order_seq.sequence-group")
+ .map(String::as_str),
+ Some("amount,quantity")
+ );
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.version,source_order.sequence-group"),
+ None
+ );
+ }
+
+ #[test]
+ fn test_rename_column_rewrites_nested_key_options() {
+ // Same rationale as the sequence-group test: `nested-key` cannot be
+ // created through Rust, but Java-written tables carry it.
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("profile", DataType::Int(IntType::new()))
+ .column("region", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("fields.profile.nested-key", "region")
+ .build()
+ .unwrap(),
+ );
+
+ let new_schema = table_schema
+ .apply_changes(vec![
+
crate::spec::SchemaChange::rename_column("profile".to_string(),
"info".to_string()),
+ crate::spec::SchemaChange::rename_column("region".to_string(),
"area".to_string()),
+ ])
+ .unwrap();
+
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.info.nested-key")
+ .map(String::as_str),
+ Some("area")
+ );
+ assert_eq!(new_schema.options().get("fields.profile.nested-key"),
None);
+ }
+
+ #[test]
+ fn test_rename_column_rewrites_remaining_case2_suffixes() {
+ // `ignore-retract` / `distinct` are rejected by Rust's create-time
+ // merge-engine validation; the fixture carries them (and the
+ // map-shredding options, which Rust does honor) as plain metadata,
+ // like a Java-written schema.
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("price", DataType::Int(IntType::new()))
+ .column(
+ "props",
+ DataType::Map(MapType::new(
+ DataType::VarChar(VarCharType::string_type()),
+ DataType::Int(IntType::new()),
+ )),
+ )
+ .primary_key(["id"])
+ .option("fields.price.ignore-retract", "true")
+ .option("fields.price.distinct", "true")
+ .option("fields.props.map.storage-layout", "shared-shredding")
+ .option("fields.props.map.shared-shredding.max-columns", "64")
+ .build()
+ .unwrap(),
+ );
+
+ let new_schema = table_schema
+ .apply_changes(vec![
+ crate::spec::SchemaChange::rename_column("price".to_string(),
"amount".to_string()),
+ crate::spec::SchemaChange::rename_column(
+ "props".to_string(),
+ "properties".to_string(),
+ ),
+ ])
+ .unwrap();
+
+ for (key, value) in [
+ ("fields.amount.ignore-retract", "true"),
+ ("fields.amount.distinct", "true"),
+ ("fields.properties.map.storage-layout", "shared-shredding"),
+ ("fields.properties.map.shared-shredding.max-columns", "64"),
+ ] {
+ assert_eq!(
+ new_schema.options().get(key).map(String::as_str),
+ Some(value),
+ "expected {key} to follow the rename"
+ );
+ }
+ for old_key in [
+ "fields.price.ignore-retract",
+ "fields.price.distinct",
+ "fields.props.map.storage-layout",
+ "fields.props.map.shared-shredding.max-columns",
+ ] {
+ assert_eq!(new_schema.options().get(old_key), None);
+ }
+ }
+
+ #[test]
+ fn test_rename_column_field_scoped_options_match_whole_names_only() {
+ // `price2` merely starts with `price`: its keys must not move when
+ // `price` is renamed, while a value-side exact reference to `price`
+ // still gets rewritten (Java `applyNotNestedColumnRename` semantics).
+ let table_schema = TableSchema::new(
+ 0,
+ &Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("price", DataType::Int(IntType::new()))
+ .column("price2", DataType::Int(IntType::new()))
+ .primary_key(["id"])
+ .option("fields.price2.sequence-group", "price")
+ .option("fields.price2.aggregate-function", "sum")
+ .build()
+ .unwrap(),
+ );
+
+ let new_schema = table_schema
+ .apply_changes(vec![crate::spec::SchemaChange::rename_column(
+ "price".to_string(),
+ "amount".to_string(),
+ )])
+ .unwrap();
+
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.price2.sequence-group")
+ .map(String::as_str),
+ Some("amount")
+ );
+ assert_eq!(
+ new_schema
+ .options()
+ .get("fields.price2.aggregate-function")
+ .map(String::as_str),
+ Some("sum")
+ );
+ assert_eq!(
+ new_schema.options().get("fields.amount.sequence-group"),
+ None
+ );
+ }
+
fn assert_primary_key_index_column_changes_rejected(
table_schema: &TableSchema,
column_name: &str,
