This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1bb9dacae7 [core] Fix false-positive immutability check for normalized
options. (#8170)
1bb9dacae7 is described below
commit 1bb9dacae7acea9d8e9378ec9d0e5a4fd1c95bf2
Author: Wenchao Wu <[email protected]>
AuthorDate: Wed Jun 24 15:02:35 2026 +0800
[core] Fix false-positive immutability check for normalized options. (#8170)
Schema.normalizePrimaryKeys() / normalizePartitionKeys() strip these
keys from the options map during table creation and store them in
dedicated schema fields instead. When the table is later read via Spark
4.x (where the DataSource V2 framework merges Table.properties() into
scan options), the same primary-key / partition value reappears in
dynamicOptions, but the stored options map no longer contains it.
checkImmutability() compares null (old) against the actual value (new),
treats it as a change, and throws:
"Change 'primary-key' is not supported yet."
Fix: skip the immutability check when oldValue is null for
primary-key/partition and the new value matches the schema's actual
primary/partition keys — the value hasn't changed, it was just
normalized out of the options map.
---
.../org/apache/paimon/schema/SchemaManager.java | 47 ++++++++++++---
.../paimon/table/AbstractFileStoreTable.java | 9 ++-
.../apache/paimon/schema/SchemaManagerTest.java | 66 ++++++++++++++++++++++
3 files changed, 112 insertions(+), 10 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
index bfe3b7de84..0791c2539e 100644
--- a/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
+++ b/paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
@@ -321,14 +321,18 @@ public class SchemaManager implements Serializable {
for (SchemaChange change : changes) {
if (change instanceof SetOption) {
SetOption setOption = (SetOption) change;
- if (hasSnapshots.get()) {
- checkAlterTableOption(
- oldOptions,
- setOption.key(),
- oldOptions.get(setOption.key()),
- setOption.value());
+ String oldValue = oldOptions.get(setOption.key());
+ String newValue = setOption.value();
+ boolean unchanged =
+ Objects.equals(oldValue, newValue)
+ || isUnchangedNormalizedKey(
+ setOption.key(), oldValue, newValue,
oldTableSchema);
+ if (hasSnapshots.get() && !unchanged) {
+ checkAlterTableOption(oldOptions, setOption.key(),
oldValue, newValue);
+ }
+ if (!unchanged) {
+ newOptions.put(setOption.key(), setOption.value());
}
- newOptions.put(setOption.key(), setOption.value());
} else if (change instanceof RemoveOption) {
RemoveOption removeOption = (RemoveOption) change;
if (hasSnapshots.get()) {
@@ -1254,6 +1258,35 @@ public class SchemaManager implements Serializable {
}
}
+ /**
+ * Checks whether a key whose old value is null actually hasn't changed.
This handles keys like
+ * 'primary-key' and 'partition' that are stripped from options during
schema normalization and
+ * stored in dedicated schema fields instead.
+ */
+ public static boolean isUnchangedNormalizedKey(
+ String key,
+ @Nullable String oldValue,
+ @Nullable String newValue,
+ TableSchema tableSchema) {
+ if (oldValue != null || newValue == null) {
+ return false;
+ }
+ if (CoreOptions.PRIMARY_KEY.key().equals(key)) {
+ return
normalizeKeyList(newValue).equals(tableSchema.primaryKeys());
+ }
+ if (CoreOptions.PARTITION.key().equals(key)) {
+ return
normalizeKeyList(newValue).equals(tableSchema.partitionKeys());
+ }
+ return false;
+ }
+
+ private static List<String> normalizeKeyList(String value) {
+ return Arrays.stream(value.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ }
+
public static void checkAlterTableOption(
Map<String, String> options, String key, @Nullable String
oldValue, String newValue) {
if (CoreOptions.IMMUTABLE_OPTIONS.contains(key)) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 0249a8e6b3..a35167d884 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -337,7 +337,9 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
dynamicOptions.forEach(
(k, newValue) -> {
String oldValue = oldOptions.get(k);
- if (!Objects.equals(oldValue, newValue)) {
+ if (!Objects.equals(oldValue, newValue)
+ && !SchemaManager.isUnchangedNormalizedKey(
+ k, oldValue, newValue, tableSchema)) {
SchemaManager.checkAlterTableOption(oldOptions, k,
oldValue, newValue);
}
});
@@ -347,12 +349,13 @@ abstract class AbstractFileStoreTable implements
FileStoreTable {
Map<String, String> dynamicOptions, boolean tryTimeTravel) {
Map<String, String> options = new HashMap<>(tableSchema.options());
- // merge non-null dynamic options into schema.options
+ // merge dynamic options into schema.options
dynamicOptions.forEach(
(k, v) -> {
if (v == null) {
options.remove(k);
- } else {
+ } else if (!SchemaManager.isUnchangedNormalizedKey(
+ k, tableSchema.options().get(k), v, tableSchema)) {
options.put(k, v);
}
});
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
index 4b7e6f3b31..59a4f90f61 100644
--- a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
@@ -515,6 +515,72 @@ public class SchemaManagerTest {
.hasMessage("Change 'merge-engine' is not supported yet.");
}
+ @Test
+ public void testCopyWithPrimaryKeyInOptions() throws Exception {
+ // Table created with primary-key in options — normalizePrimaryKeys
strips it from the
+ // options map and stores it in the dedicated primaryKeys field. When
the same value
+ // reappears in dynamicOptions (e.g. Spark 4.x merging
Table.properties() into scan
+ // options), the immutability check should recognize it hasn't
actually changed.
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("primary-key", "f0,f1");
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ tableOptions,
+ "");
+ Path tableRoot = new Path(tempDir.toString(), "table");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(),
tableRoot);
+ manager.createTable(schema);
+
+ FileStoreTable table =
FileStoreTableFactory.create(LocalFileIO.create(), tableRoot);
+ FileStoreTable copied =
table.copy(Collections.singletonMap("primary-key", "f0,f1"));
+ assertThatCode(() ->
copied.schema().toSchema()).doesNotThrowAnyException();
+ assertThat(copied.schema().options()).doesNotContainKey("primary-key");
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "primary-key", null, null, copied.schema()))
+ .isFalse();
+ }
+
+ @Test
+ public void testAlterUnchangedNormalizedOptionsOnNonEmptyTable() throws
Exception {
+ Map<String, String> tableOptions = new HashMap<>();
+ tableOptions.put("primary-key", "f0,f1");
+ tableOptions.put("partition", "f0");
+ tableOptions.put("bucket", "1");
+ Schema schema =
+ new Schema(
+ rowType.getFields(),
+ Collections.emptyList(),
+ Collections.emptyList(),
+ tableOptions,
+ "");
+ Path tableRoot = new Path(tempDir.toString(), "table");
+ SchemaManager manager = new SchemaManager(LocalFileIO.create(),
tableRoot);
+ manager.createTable(schema);
+
+ FileStoreTable table =
FileStoreTableFactory.create(LocalFileIO.create(), tableRoot);
+ String commitUser = UUID.randomUUID().toString();
+ TableWriteImpl<?> write =
+
table.newWrite(commitUser).withIOManager(IOManager.create(tempDir + "/io"));
+ TableCommitImpl commit = table.newCommit(commitUser);
+ write.write(GenericRow.of(1, 10L, BinaryString.fromString("apple")));
+ commit.commit(1, write.prepareCommit(false, 1));
+ write.close();
+ commit.close();
+
+ TableSchema latest =
+ manager.commitChanges(
+ SchemaChange.setOption("primary-key", "f0,f1"),
+ SchemaChange.setOption("partition", "f0"));
+ assertThat(latest.primaryKeys()).containsExactly("f0", "f1");
+ assertThat(latest.partitionKeys()).containsExactly("f0");
+ assertThat(latest.options()).doesNotContainKeys("primary-key",
"partition");
+ assertThatCode(latest::toSchema).doesNotThrowAnyException();
+ }
+
@Test
public void testDropPrimaryKeyOnEmptyTable() throws Exception {
Path tableRoot = new Path(tempDir.toString(), "table");
