Skip to content

[core] Skip schema update when no changes detected #7545

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
zhongyujiang wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[core] Skip schema update when no changes detected #7545

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1764428
[core] Skip schema update when no changes detected.
zhongyujiang Apr 10, 2026
a627e75
review comments.
zhongyujiang Apr 10, 2026
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
11 changes: 11 additions & 0 deletions paimon-api/src/main/java/org/apache/paimon/schema/TableSchema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,6 +342,7 @@ public boolean equals(Object o) {
}
TableSchema tableSchema = (TableSchema) o;
return version == tableSchema.version
&& highestFieldId == tableSchema.highestFieldId
&& Objects.equals(fields, tableSchema.fields)
&& Objects.equals(partitionKeys, tableSchema.partitionKeys)
&& Objects.equals(primaryKeys, tableSchema.primaryKeys)
Expand All @@ -356,6 +357,16 @@ public int hashCode() {
version, fields, partitionKeys, primaryKeys, options, comment, timeMillis);
}

/** Checks if two schemas have the same content, ignoring version and timeMillis. */
public boolean sameContent(TableSchema other) {
return Objects.equals(fields, other.fields)
&& highestFieldId == other.highestFieldId
&& Objects.equals(partitionKeys, other.partitionKeys)
&& Objects.equals(primaryKeys, other.primaryKeys)
&& Objects.equals(options, other.options)
&& Objects.equals(comment, other.comment);
}

public static List<DataField> newFields(RowType rowType) {
return rowType.getFields();
}
Expand Down
186 changes: 173 additions & 13 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 @@ -60,6 +60,9 @@
import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
import org.apache.paimon.shade.guava30.com.google.common.collect.Streams;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;

Expand Down Expand Up @@ -110,7 +113,8 @@
@ThreadSafe
public class SchemaManager implements Serializable {

public static final String SCHEMA_PREFIX = "schema-";
private static final Logger LOG = LoggerFactory.getLogger(SchemaManager.class);
private static final String SCHEMA_PREFIX = "schema-";

private final FileIO fileIO;
private final Path tableRoot;
Expand Down Expand Up @@ -266,8 +270,15 @@ public TableSchema commitChanges(List<SchemaChange> changes)
tableRoot.toString(), true, branch)));
LazyField<Identifier> lazyIdentifier =
new LazyField<>(() -> identifierFromPath(tableRoot.toString(), true, branch));
TableSchema newTableSchema =
Optional<TableSchema> newTableSchemaOpt =
generateTableSchema(oldTableSchema, changes, hasSnapshots, lazyIdentifier);
if (!newTableSchemaOpt.isPresent()) {
LOG.info(
"No schema change detected for table {}. Skipping schema update.",
lazyIdentifier.get());
return oldTableSchema;
}
TableSchema newTableSchema = newTableSchemaOpt.get();
try {
boolean success = commit(newTableSchema);
if (success) {
Expand All @@ -279,7 +290,7 @@ public TableSchema commitChanges(List<SchemaChange> changes)
}
}

public static TableSchema generateTableSchema(
public static Optional<TableSchema> generateTableSchema(
TableSchema oldTableSchema,
List<SchemaChange> changes,
LazyField<Boolean> hasSnapshots,
Expand Down Expand Up @@ -314,7 +325,98 @@ public static TableSchema generateTableSchema(
AtomicInteger highestFieldId = new AtomicInteger(oldTableSchema.highestFieldId());
String newComment = oldTableSchema.comment();
List<String> newPrimaryKeys = oldTableSchema.primaryKeys();

// Filter out ineffective changes
List<SchemaChange> effectiveChanges = new ArrayList<>();
for (SchemaChange change : changes) {
if (change instanceof SetOption) {
SetOption setOption = (SetOption) change;
String oldValue = oldOptions.get(setOption.key());
if (oldValue == null || !oldValue.equals(setOption.value())) {
effectiveChanges.add(change);
}
} else if (change instanceof RemoveOption) {
RemoveOption removeOption = (RemoveOption) change;
if (oldOptions.containsKey(removeOption.key())) {
effectiveChanges.add(change);
}
} else if (change instanceof UpdateComment) {
UpdateComment updateComment = (UpdateComment) change;
if (!Objects.equals(oldTableSchema.comment(), updateComment.comment())) {
effectiveChanges.add(change);
}
} else if (change instanceof RenameColumn) {
RenameColumn rename = (RenameColumn) change;
DataField field = findField(oldTableSchema.fields(), rename.fieldNames());
if (field != null && !field.name().equals(rename.newName())) {
effectiveChanges.add(change);
}
} else if (change instanceof UpdateColumnType) {
UpdateColumnType update = (UpdateColumnType) change;
DataField field = findField(oldTableSchema.fields(), update.fieldNames());
if (field != null) {
DataType oldType = field.type();
DataType newType = update.newDataType();
if (update.keepNullability()) {
newType = newType.copy(oldType.isNullable());
}
if (!oldType.equals(newType)) {
effectiveChanges.add(change);
}
}
} else if (change instanceof UpdateColumnNullability) {
UpdateColumnNullability update = (UpdateColumnNullability) change;
DataField field = findField(oldTableSchema.fields(), update.fieldNames());
if (field != null) {
DataType oldType = field.type();
DataType sourceRootType =
getRootType(
oldType,
update.fieldNames().length - 1,
update.fieldNames().length);
if (sourceRootType.isNullable() != update.newNullability()) {
effectiveChanges.add(change);
}
}
} else if (change instanceof UpdateColumnComment) {
UpdateColumnComment update = (UpdateColumnComment) change;
DataField field = findField(oldTableSchema.fields(), update.fieldNames());
if (field != null
&& !Objects.equals(field.description(), update.newDescription())) {
effectiveChanges.add(change);
}
} else if (change instanceof UpdateColumnPosition) {
UpdateColumnPosition update = (UpdateColumnPosition) change;
SchemaChange.Move move = update.move();
String fieldName = move.fieldName();
DataField field = findFieldByName(newFields, fieldName);
if (field != null) {
int currentIndex = -1;
for (int i = 0; i < newFields.size(); i++) {
if (newFields.get(i).name().equals(fieldName)) {
currentIndex = i;
break;
}
}
int newIndex = calculateNewPosition(newFields, move);
if (currentIndex != newIndex) {
effectiveChanges.add(change);
}
}
} else if (change instanceof UpdateColumnDefaultValue) {
UpdateColumnDefaultValue update = (UpdateColumnDefaultValue) change;
DataField field = findField(oldTableSchema.fields(), update.fieldNames());
if (field != null
&& !Objects.equals(field.defaultValue(), update.newDefaultValue())) {
effectiveChanges.add(change);
}
} else {
// AddColumn and DropColumn always change the schema
effectiveChanges.add(change);
}
}

for (SchemaChange change : effectiveChanges) {
if (change instanceof SetOption) {
SetOption setOption = (SetOption) change;
if (hasSnapshots.get()) {
Expand Down Expand Up @@ -571,18 +673,76 @@ protected void updateLastColumn(
newFields,
oldTableSchema.partitionKeys(),
applyNotNestedColumnRename(
newPrimaryKeys, Iterables.filter(changes, RenameColumn.class)),
applyRenameColumnsToOptions(newOptions, changes),
newPrimaryKeys,
Iterables.filter(effectiveChanges, RenameColumn.class)),
applyRenameColumnsToOptions(newOptions, effectiveChanges),
newComment);

return new TableSchema(
oldTableSchema.id() + 1,
newSchema.fields(),
highestFieldId.get(),
newSchema.partitionKeys(),
newSchema.primaryKeys(),
newSchema.options(),
newSchema.comment());
TableSchema newTableSchema =
new TableSchema(
oldTableSchema.id() + 1,
newSchema.fields(),
highestFieldId.get(),
newSchema.partitionKeys(),
newSchema.primaryKeys(),
newSchema.options(),
newSchema.comment());

if (oldTableSchema.sameContent(newTableSchema)) {
return Optional.empty();
}
return Optional.of(newTableSchema);
}

private static DataField findField(List<DataField> fields, String[] fieldNames) {
if (fieldNames.length == 0) {
return null;
}
String firstName = fieldNames[0];
for (DataField field : fields) {
if (field.name().equals(firstName)) {
if (fieldNames.length == 1) {
return field;
}
// Handle nested fields
if (field.type() instanceof RowType) {
return findField(
((RowType) field.type()).getFields(),
Arrays.copyOfRange(fieldNames, 1, fieldNames.length));
}
}
}
return null;
}

private static DataField findFieldByName(List<DataField> fields, String fieldName) {
for (DataField field : fields) {
if (field.name().equals(fieldName)) {
return field;
}
}
return null;
}

private static int calculateNewPosition(List<DataField> fields, SchemaChange.Move move) {
if (move.type().equals(SchemaChange.Move.MoveType.FIRST)) {
return 0;
} else if (move.type().equals(SchemaChange.Move.MoveType.LAST)) {
return fields.size() - 1;
} else if (move.type().equals(SchemaChange.Move.MoveType.AFTER)) {
for (int i = 0; i < fields.size(); i++) {
if (fields.get(i).name().equals(move.referenceFieldName())) {
return i + 1;
}
}
} else if (move.type().equals(SchemaChange.Move.MoveType.BEFORE)) {
for (int i = 0; i < fields.size(); i++) {
if (fields.get(i).name().equals(move.referenceFieldName())) {
return i;
}
}
}
return -1;
}

// gets the rootType at the defined depth
Expand Down
10 changes: 6 additions & 4 deletions paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogServer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2455,10 +2455,12 @@ protected void alterTableImpl(Identifier identifier, List<SchemaChange> changes)
if (isFormatTable(schema.toSchema())) {
TableSchema newSchema =
SchemaManager.generateTableSchema(
schema,
changes,
new LazyField<>(() -> false),
new LazyField<>(() -> identifier));
schema,
changes,
new LazyField<>(() -> false),
new LazyField<>(() -> identifier))
.orElse(schema);

TableMetadata newTableMetadata =
createTableMetadata(
identifier,
Expand Down
32 changes: 32 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 @@ -939,4 +939,36 @@ public void testRollbackSchemaNotExist() throws Exception {
new ChangelogManager(LocalFileIO.create(), path, null)))
.hasMessageContaining("Schema 999 does not exist");
}

@Test
public void testNoChangeCommitDoesNotCreateNewSchema() throws Exception {
// Create table with an initial option foo=bar
Map<String, String> initialOptions = new HashMap<>();
initialOptions.put("foo", "bar");
Schema schemaWithOption =
new Schema(
rowType.getFields(),
Collections.emptyList(),
Collections.emptyList(),
initialOptions,
"");
SchemaManager manager = new SchemaManager(LocalFileIO.create(), path);
manager.createTable(schemaWithOption);

long initialSchemaId = manager.latest().get().id();
assertThat(manager.latest().get().options()).containsEntry("foo", "bar");

// Set option foo=bar again (no actual change)
manager.commitChanges(SchemaChange.setOption("foo", "bar"));

// Verify no new schema is created when value didn't change
long newSchemaId = manager.latest().get().id();
assertThat(newSchemaId).isEqualTo(initialSchemaId);
assertThat(manager.latest().get().options()).containsEntry("foo", "bar");

// Also test with UpdateComment when comment is unchanged
String initialComment = manager.latest().get().comment();
manager.commitChanges(SchemaChange.updateComment(initialComment));
assertThat(manager.latest().get().id()).isEqualTo(initialSchemaId);
}
}
Loading