From 813a470dd639e6e5283de634764c437eeb3b0b60 Mon Sep 17 00:00:00 2001 From: Li Jiajia Date: Sat, 11 Jul 2026 12:48:34 -0400 Subject: [PATCH] [core] Mark table 'type' option as immutable The 'type' option decides which table kind the catalog builds (CatalogUtils reads it on every load), but it was missing from IMMUTABLE_OPTIONS, so ALTER TABLE SET ('type'='format-table') on an existing table succeeded. The next load then builds a different table kind (e.g. FormatTable) over the existing FileStore data layout: data files under bucket directories are no longer readable, snapshot history is orphaned, and -- since only FileStore tables wire up query auth -- an authorized table silently stops enforcing row filters and column masks. Changing a table's type in place has no supported semantics: there are no tests or docs exercising it, and replaceTable() already handles cross-type replacement explicitly via drop-and-create. Mark TYPE @Immutable so the checkAlterTableOption / checkResetTableOption / dynamic-option guards reject it, consistent with the other structural options (primary-key, partition, merge-engine, ...). Unlike the other immutable options, the type is also rejected without snapshots: table kinds like format tables never create Paimon snapshots but can still hold data, so the snapshot-based emptiness check does not apply to it. Since a table created without an explicit type is of the default type, isUnchangedNormalizedKey treats setting the default explicitly as unchanged, so redundant 'type'='table' dynamic options and idempotent ALTERs still pass. Type values are compared case-insensitively, matching OptionsUtils.convertToEnum, so 'type'='TABLE' -- and a case-only restatement of an explicitly stored type -- is also recognized as unchanged. The Flink dynamic-option preflight (AbstractFlinkTableFactory.buildPaimonTable) diffed the raw option maps and called checkAlterTableOption without that normalization, so once 'type' became immutable a documented paimon....type=table dynamic option on an implicit-type table would hit the new guard and throw. It now applies isUnchangedNormalizedKey as well -- via a List-keyed overload that takes the primary/partition keys directly, since only the schema (not a full TableSchema) is available there. For a FormatCatalogTable, whose getOptions() drops Paimon's 'type', the wrapped table's effective format-table type is restored before the check so a no-op 'type'='format-table' dynamic option is not treated as a change either. Real type changes are still rejected. --- .../java/org/apache/paimon/CoreOptions.java | 1 + .../apache/paimon/schema/SchemaManager.java | 42 +++++++++-- .../paimon/schema/SchemaManagerTest.java | 74 +++++++++++++++++++ .../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 0dd64d8d2543..9b7798948ff8 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 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 06f6a1e07768..6db4136f1bf6 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 static TableSchema generateTableSchema( 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 static TableSchema generateTableSchema( } } 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 void rollbackTo( } /** - * 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 primaryKeys, + List 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 b982ff1df1d3..2c065176d1c8 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 void testAlterImmutableOptionsOnEmptyTable() throws Exception { 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 void testAlterImmutableOptionsOnEmptyTable() throws Exception { 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 void testCopyWithPrimaryKeyInOptions() throws Exception { .isFalse(); } + @Test + public void testIsUnchangedNormalizedKeyWithKeyLists() { + List primaryKeys = Arrays.asList("f0", "f1"); + List 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 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 c6d8ccdf5047..c13a1bb158e4 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 @@ Table getPaimonTable(CatalogTable origin, DynamicTableFactory.Context context) { 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 originOptions = origin.getOptions(); + if (origin instanceof FormatCatalogTable) { + originOptions = new HashMap<>(originOptions); + originOptions.putIfAbsent(CoreOptions.TYPE.key(), TableType.FORMAT_TABLE.toString()); + } + Map preflightOptions = originOptions; Map 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 newOptions = new HashMap<>(); @@ -184,8 +206,6 @@ Table buildPaimonTable(DynamicTableFactory.Context context) { 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 partitionKeys = schema.partitionKeys(); List primaryKeys = schema.primaryKeys();