Skip to content

[core] Mark table 'type' option as immutable #8564

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
plusplusjiajia wants to merge 1 commit into
base: master
Choose a base branch
from
+136 −11
Open

[core] Mark table 'type' option as immutable #8564

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions paimon-api/src/main/java/org/apache/paimon/CoreOptions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
42 changes: 36 additions & 6 deletions paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand All @@ -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());
}
Expand Down Expand Up @@ -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<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;
}
Expand Down
74 changes: 74 additions & 0 deletions paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -573,6 +594,59 @@ public void testCopyWithPrimaryKeyInOptions() throws Exception {
.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<>();
Expand Down
30 changes: 25 additions & 5 deletions .../paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<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<>();
Expand Down Expand Up @@ -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<String> partitionKeys = schema.partitionKeys();
List<String> primaryKeys = schema.primaryKeys();
Expand Down
Loading