[core] Skip schema update when no changes detected

paimon-api/src/main/java/org/apache/paimon/schema/TableSchema.java

@@ -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)
@@ -353,7 +354,24 @@ public boolean equals(Object o) {
     @Override
     public int hashCode() {
         return Objects.hash(
-                version, fields, partitionKeys, primaryKeys, options, comment, timeMillis);
+                version,
+                fields,
+                highestFieldId,
+                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) {

paimon-core/src/main/java/org/apache/paimon/schema/SchemaManager.java

@@ -59,6 +59,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;
 
@@ -108,6 +111,7 @@
 @ThreadSafe
 public class SchemaManager implements Serializable {
 
+    private static final Logger LOG = LoggerFactory.getLogger(SchemaManager.class);
     private static final String SCHEMA_PREFIX = "schema-";
 
     private final FileIO fileIO;
@@ -266,6 +270,12 @@ public TableSchema commitChanges(List<SchemaChange> changes)
                     new LazyField<>(() -> identifierFromPath(tableRoot.toString(), true, branch));
             TableSchema newTableSchema =
                     generateTableSchema(oldTableSchema, changes, hasSnapshots, lazyIdentifier);
+            if (oldTableSchema.sameContent(newTableSchema)) {
+                LOG.info(
+                        "No schema change detected for table {}. Skipping schema update.",
+                        lazyIdentifier.get());
+                return oldTableSchema;
+            }
             try {
                 boolean success = commit(newTableSchema);
                 if (success) {

paimon-core/src/test/java/org/apache/paimon/schema/SchemaManagerTest.java

@@ -894,4 +894,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);
+    }
 }