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 3cfaa512cb [core] Mark table 'type' option as immutable (#8564)
3cfaa512cb is described below
commit 3cfaa512cb2c3411cc7022470896a221ef9bbd8c
Author: Jiajia Li <[email protected]>
AuthorDate: Mon Jul 13 10:19:44 2026 +0800
[core] Mark table 'type' option as immutable (#8564)
---
.../main/java/org/apache/paimon/CoreOptions.java | 1 +
.../org/apache/paimon/schema/SchemaManager.java | 42 ++++++++++--
.../apache/paimon/schema/SchemaManagerTest.java | 74 ++++++++++++++++++++++
.../paimon/flink/AbstractFlinkTableFactory.java | 30 +++++++--
4 files changed, 136 insertions(+), 11 deletions(-)
diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
index 0dd64d8d25..9b7798948f 100644
--- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
+++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
@@ -106,6 +106,7 @@ public class CoreOptions implements Serializable {
public static final String BLOB_DESCRIPTOR_PREFIX = "blob-descriptor.";
+ @Immutable
public static final ConfigOption<TableType> TYPE =
key("type")
.enumType(TableType.class)
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 06f6a1e077..6db4136f1b 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
@@ -327,6 +327,11 @@ public class SchemaManager implements Serializable {
Objects.equals(oldValue, newValue)
|| isUnchangedNormalizedKey(
setOption.key(), oldValue, newValue,
oldTableSchema);
+ // reject 'type' even without snapshots: format tables hold
data but
+ // create no snapshots, so the snapshot check would not catch
them
+ if (!unchanged &&
CoreOptions.TYPE.key().equals(setOption.key())) {
+ throw new UnsupportedOperationException("Change 'type' is
not supported yet.");
+ }
if (hasSnapshots.get() && !unchanged) {
checkAlterTableOption(oldOptions, setOption.key(),
oldValue, newValue);
}
@@ -335,6 +340,9 @@ public class SchemaManager implements Serializable {
}
} else if (change instanceof RemoveOption) {
RemoveOption removeOption = (RemoveOption) change;
+ if (CoreOptions.TYPE.key().equals(removeOption.key())) {
+ throw new UnsupportedOperationException("Change 'type' is
not supported yet.");
+ }
if (hasSnapshots.get()) {
checkResetTableOption(oldOptions, removeOption.key());
}
@@ -1273,23 +1281,45 @@ 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.
+ * Whether an option change is a semantic no-op: 'primary-key'/'partition'
are normalized into
+ * dedicated schema fields (old value null), and 'type' compares
case-insensitively, defaulting
+ * when unset.
*/
public static boolean isUnchangedNormalizedKey(
String key,
@Nullable String oldValue,
@Nullable String newValue,
TableSchema tableSchema) {
- if (oldValue != null || newValue == null) {
+ return isUnchangedNormalizedKey(
+ key, oldValue, newValue, tableSchema.primaryKeys(),
tableSchema.partitionKeys());
+ }
+
+ /**
+ * Overload for callers holding the primary/partition keys rather than a
{@link TableSchema}.
+ */
+ public static boolean isUnchangedNormalizedKey(
+ String key,
+ @Nullable String oldValue,
+ @Nullable String newValue,
+ List<String> primaryKeys,
+ List<String> partitionKeys) {
+ if (newValue == null) {
+ return false;
+ }
+ if (CoreOptions.TYPE.key().equals(key)) {
+ // 'type' compares case-insensitively (like convertToEnum); an
unset type is the default
+ String effectiveOld =
+ oldValue == null ?
CoreOptions.TYPE.defaultValue().toString() : oldValue;
+ return newValue.equalsIgnoreCase(effectiveOld);
+ }
+ if (oldValue != null) {
return false;
}
if (CoreOptions.PRIMARY_KEY.key().equals(key)) {
- return
normalizeKeyList(newValue).equals(tableSchema.primaryKeys());
+ return normalizeKeyList(newValue).equals(primaryKeys);
}
if (CoreOptions.PARTITION.key().equals(key)) {
- return
normalizeKeyList(newValue).equals(tableSchema.partitionKeys());
+ return normalizeKeyList(newValue).equals(partitionKeys);
}
return false;
}
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 b982ff1df1..2c065176d1 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
@@ -481,6 +481,15 @@ public class SchemaManagerTest {
SchemaManager manager = new SchemaManager(LocalFileIO.create(),
tableRoot);
manager.createTable(schema);
+ // 'type' is rejected even without snapshots (format tables hold data
but create none)
+ assertThatThrownBy(
+ () ->
manager.commitChanges(SchemaChange.setOption("type", "format-table")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Change 'type' is not supported yet.");
+ assertThatThrownBy(() ->
manager.commitChanges(SchemaChange.removeOption("type")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Change 'type' is not supported yet.");
+
// set immutable options and set primary keys
manager.commitChanges(
SchemaChange.setOption("primary-key", "f0, f1"),
@@ -542,6 +551,18 @@ public class SchemaManagerTest {
SchemaChange.setOption("merge-engine",
"deduplicate")))
.isInstanceOf(UnsupportedOperationException.class)
.hasMessage("Change 'merge-engine' is not supported yet.");
+
+ // flipping the type in place would build a different table kind over
the same data
+ assertThatThrownBy(
+ () ->
manager.commitChanges(SchemaChange.setOption("type", "format-table")))
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Change 'type' is not supported yet.");
+
+ // setting the default type explicitly is not a change
+ assertThatCode(() ->
manager.commitChanges(SchemaChange.setOption("type", "table")))
+ .doesNotThrowAnyException();
+ assertThatCode(() -> table.copy(Collections.singletonMap("type",
"table")))
+ .doesNotThrowAnyException();
}
@Test
@@ -573,6 +594,59 @@ public class SchemaManagerTest {
.isFalse();
}
+ @Test
+ public void testIsUnchangedNormalizedKeyWithKeyLists() {
+ List<String> primaryKeys = Arrays.asList("f0", "f1");
+ List<String> partitionKeys = Collections.singletonList("f0");
+ // an explicit type equal to the default is not a change
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type",
+ null,
+ CoreOptions.TYPE.defaultValue().toString(),
+ primaryKeys,
+ partitionKeys))
+ .isTrue();
+ // default type matched case-insensitively
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type", null, "TABLE", primaryKeys,
partitionKeys))
+ .isTrue();
+ // a different type is a real change
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type", null, "format-table", primaryKeys,
partitionKeys))
+ .isFalse();
+ // primary-key / partition restated with the same normalized value are
no-ops
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "primary-key", null, "f0, f1", primaryKeys,
partitionKeys))
+ .isTrue();
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "partition", null, "f0", primaryKeys,
partitionKeys))
+ .isTrue();
+ // an explicitly stored type restated with different case is still a
no-op
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type", "table", "TABLE", primaryKeys,
partitionKeys))
+ .isTrue();
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type", "format-table", "FORMAT-TABLE",
primaryKeys, partitionKeys))
+ .isTrue();
+ // a genuinely different explicit type is a real change
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "type", "table", "format-table", primaryKeys,
partitionKeys))
+ .isFalse();
+ // non-type keys with a non-null old value are treated as changes
+ assertThat(
+ SchemaManager.isUnchangedNormalizedKey(
+ "primary-key", "f0", "f0,f1", primaryKeys,
partitionKeys))
+ .isFalse();
+ }
+
@Test
public void testAlterUnchangedNormalizedOptionsOnNonEmptyTable() throws
Exception {
Map<String, String> tableOptions = new HashMap<>();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
index c6d8ccdf50..c13a1bb158 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
@@ -18,6 +18,8 @@
package org.apache.paimon.flink;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.TableType;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
@@ -139,14 +141,34 @@ public abstract class AbstractFlinkTableFactory
Table buildPaimonTable(DynamicTableFactory.Context context) {
CatalogTable origin = context.getCatalogTable().getOrigin();
+ // matches Flink's schema (asserted below); also lets the preflight
treat a
+ // dynamic option that restates a normalized/default value as a no-op
+ Schema schema =
FlinkCatalog.fromCatalogTable(context.getCatalogTable());
+
+ // FormatCatalogTable.getOptions() drops 'type'; restore the format
table's
+ // effective type so a no-op 'type'='format-table' isn't seen as a
change
+ Map<String, String> originOptions = origin.getOptions();
+ if (origin instanceof FormatCatalogTable) {
+ originOptions = new HashMap<>(originOptions);
+ originOptions.putIfAbsent(CoreOptions.TYPE.key(),
TableType.FORMAT_TABLE.toString());
+ }
+ Map<String, String> preflightOptions = originOptions;
Map<String, String> dynamicOptions = getDynamicConfigOptions(context);
dynamicOptions.forEach(
(key, newValue) -> {
- String oldValue = origin.getOptions().get(key);
- if (!Objects.equals(oldValue, newValue)) {
+ String oldValue = preflightOptions.get(key);
+ boolean unchanged =
+ Objects.equals(oldValue, newValue)
+ || SchemaManager.isUnchangedNormalizedKey(
+ key,
+ oldValue,
+ newValue,
+ schema.primaryKeys(),
+ schema.partitionKeys());
+ if (!unchanged) {
SchemaManager.checkAlterTableOption(
- origin.getOptions(), key, oldValue, newValue);
+ preflightOptions, key, oldValue, newValue);
}
});
Map<String, String> newOptions = new HashMap<>();
@@ -184,8 +206,6 @@ public abstract class AbstractFlinkTableFactory
table.fileIO().setRuntimeContext(runtimeContext);
}
// notice that the Paimon table schema must be the same with the
Flink's
- Schema schema =
FlinkCatalog.fromCatalogTable(context.getCatalogTable());
-
RowType rowType = toLogicalType(schema.rowType());
List<String> partitionKeys = schema.partitionKeys();
List<String> primaryKeys = schema.primaryKeys();
