This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch branch-1.3
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/branch-1.3 by this push:
new 17cb189454 [Cherry-pick to branch-1.3] [#12734] fix(flink-connector):
skip no-op table alter to avoid updates must not be empty (#12771)
17cb189454 is described below
commit 17cb189454cd9bac70e49de2650004963dbfe9b6
Author: geyanggang <[email protected]>
AuthorDate: Tue Sep 1 14:53:22 2026 +0800
[Cherry-pick to branch-1.3] [#12734] fix(flink-connector): skip no-op table
alter to avoid updates must not be empty (#12771)
### What changes were proposed in this pull request?
This is the branch-1.3 backport of #12735 (cherry-pick of commit
4b529ce). It resolves the cherry-pick conflicts introduced because
branch-1.3 differs from main:
- `TestBaseCatalog.java`: added the imports that exist on main but are
missing on branch-1.3 (`ObjectPath`, `TableNotExistException`,
`TableCatalog`) and dropped the conflicting `CatalogException` import
that is unused here.
- `TestGravitinoHiveCatalog.java`: removed
`testGetTableThrowsCatalogExceptionWhenForbidden` (and its now-unused
imports). That test is not part of this fix and depends on the
`ForbiddenException` -> `CatalogException` conversion in `getTable`,
which does not exist on branch-1.3.
The core fix is unchanged from #12735: skip forwarding an empty
`TableChange` list to Gravitino when a table alter results in no actual
change.
### Why are the changes needed?
A no-op `ALTER TABLE` produced an empty update list, which the server
rejects with `IllegalArgumentException: updates must not be empty`,
failing the Flink job. This backports the fix to branch-1.3.
Fix: #12734
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
`./gradlew :flink-connector:flink-common:check -PskipITs` passes on
branch-1.3.
---
.../flink/connector/catalog/BaseCatalog.java | 40 ++++--
.../flink/connector/hive/GravitinoHiveCatalog.java | 12 +-
.../flink/connector/catalog/TestBaseCatalog.java | 91 ++++++++++++
.../connector/hive/TestGravitinoHiveCatalog.java | 155 +++++++++++++++++++++
.../paimon/TestGravitinoPaimonCatalog.java | 58 +++++++-
5 files changed, 341 insertions(+), 15 deletions(-)
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
index 59874aff4a..ae806dcdbd 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/catalog/BaseCatalog.java
@@ -528,11 +528,11 @@ public abstract class BaseCatalog extends AbstractCatalog
{
throw new CatalogException(e);
}
} else {
- catalog()
- .asTableCatalog()
- .alterTable(identifier, getGravitinoTableChanges(existingTable,
newTable));
- // Invalidate native catalog cache after successful alter
- invalidateTable(tablePath);
+ TableChange[] changes = getGravitinoTableChanges(existingTable,
newTable);
+ if (alterGravitinoTable(identifier, changes)) {
+ // Invalidate native catalog cache after successful alter
+ invalidateTable(tablePath);
+ }
}
}
@@ -582,9 +582,11 @@ public abstract class BaseCatalog extends AbstractCatalog {
throw new CatalogException(e);
}
} else {
- catalog().asTableCatalog().alterTable(identifier,
getGravitinoTableChanges(tableChanges));
- // Invalidate native catalog cache after successful alter
- invalidateTable(tablePath);
+ TableChange[] changes = getGravitinoTableChanges(tableChanges);
+ if (alterGravitinoTable(identifier, changes)) {
+ // Invalidate native catalog cache after successful alter
+ invalidateTable(tablePath);
+ }
}
}
@@ -924,6 +926,28 @@ public abstract class BaseCatalog extends AbstractCatalog {
}
}
+ /**
+ * Applies the given table changes to the underlying Gravitino table,
skipping the call when there
+ * is nothing to change.
+ *
+ * <p>When {@code changes} is empty the resolved table already matches the
existing one (for
+ * example, re-applying the same options or a comment-only alter with an
unchanged comment).
+ * Gravitino's {@code TableUpdatesRequest.validate} rejects an empty update
list with "updates
+ * must not be empty", so a no-op alter must be skipped rather than
forwarded.
+ *
+ * @param identifier the identifier of the table to alter
+ * @param changes the Gravitino table changes to apply
+ * @return {@code true} if the alter was forwarded to Gravitino, {@code
false} if it was skipped
+ * because there was nothing to change
+ */
+ private boolean alterGravitinoTable(NameIdentifier identifier, TableChange[]
changes) {
+ if (changes.length == 0) {
+ return false;
+ }
+ catalog().asTableCatalog().alterTable(identifier, changes);
+ return true;
+ }
+
@VisibleForTesting
static TableChange[] getGravitinoTableChanges(
CatalogBaseTable existingTable, CatalogBaseTable newTable) {
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
index feb6362fae..551a4a51d9 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/hive/GravitinoHiveCatalog.java
@@ -18,6 +18,7 @@
*/
package org.apache.gravitino.flink.connector.hive;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import java.util.ArrayList;
import java.util.Collections;
@@ -281,7 +282,8 @@ public class GravitinoHiveCatalog extends BaseCatalog {
}
}
- private void applyGenericTableAlter(
+ @VisibleForTesting
+ void applyGenericTableAlter(
ObjectPath tablePath, Table existingTable, ResolvedCatalogTable newTable)
throws TableNotExistException, CatalogException {
NameIdentifier identifier =
@@ -311,6 +313,14 @@ public class GravitinoHiveCatalog extends BaseCatalog {
}
});
+ // When the resolved table is identical to the existing one (for example,
re-applying the same
+ // options), no TableChange is produced. Skip the alter call in that case:
Gravitino's
+ // TableUpdatesRequest.validate rejects an empty update list with "updates
must not be empty",
+ // and a no-op alter should succeed rather than fail.
+ if (changes.isEmpty()) {
+ return;
+ }
+
try {
catalog().asTableCatalog().alterTable(identifier, changes.toArray(new
TableChange[0]));
} catch (NoSuchTableException e) {
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
index e24dc13c6c..3ba0edff0a 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/catalog/TestBaseCatalog.java
@@ -33,9 +33,11 @@ import org.apache.flink.table.catalog.CatalogDatabase;
import org.apache.flink.table.catalog.CatalogDatabaseImpl;
import org.apache.flink.table.catalog.CatalogView;
import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.ResolvedCatalogView;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.TableChange;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.gravitino.Catalog;
import org.apache.gravitino.NameIdentifier;
import org.apache.gravitino.Namespace;
@@ -47,6 +49,7 @@ import
org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
import org.apache.gravitino.rel.Dialects;
import org.apache.gravitino.rel.Representation;
import org.apache.gravitino.rel.SQLRepresentation;
+import org.apache.gravitino.rel.TableCatalog;
import org.apache.gravitino.rel.ViewCatalog;
import org.apache.gravitino.rel.ViewChange;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
@@ -372,12 +375,86 @@ public class TestBaseCatalog {
}
}
+ @Test
+ public void testAlterTableWithEmptyTableChangesSkipsAlterCall() throws
Exception {
+ // Flink may invoke alterTable with an empty change list (a no-op alter).
The connector must
+ // not forward an empty update list to Gravitino, which would fail
server-side with
+ // "updates must not be empty".
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ TableCatalog tableCatalog = Mockito.mock(TableCatalog.class);
+ Mockito.when(gravitinoCatalog.asTableCatalog()).thenReturn(tableCatalog);
+
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ CatalogBaseTable table =
+ DefaultCatalogCompat.INSTANCE.createCatalogTable(
+ schema, "comment", ImmutableList.of(), ImmutableMap.of("key",
"value"));
+
+ TestableBaseCatalog catalog =
+ new TestableBaseCatalog(Mockito.mock(AbstractCatalog.class),
gravitinoCatalog, table);
+
+ catalog.alterTable(new ObjectPath("db", "tbl"), table,
Collections.emptyList(), false);
+
+ Mockito.verify(tableCatalog, Mockito.never()).alterTable(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ public void testAlterTableWithCommentOnlyAndUnchangedCommentSkipsAlterCall()
throws Exception {
+ // The two-argument alterTable diffs only the comment. When the comment is
unchanged no
+ // TableChange is produced, so the connector must skip the alter call
instead of forwarding an
+ // empty update list.
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ TableCatalog tableCatalog = Mockito.mock(TableCatalog.class);
+ Mockito.when(gravitinoCatalog.asTableCatalog()).thenReturn(tableCatalog);
+
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ CatalogBaseTable table =
+ DefaultCatalogCompat.INSTANCE.createCatalogTable(
+ schema, "comment", ImmutableList.of(), ImmutableMap.of("key",
"value"));
+
+ TestableBaseCatalog catalog =
+ new TestableBaseCatalog(Mockito.mock(AbstractCatalog.class),
gravitinoCatalog, table);
+
+ catalog.alterTable(new ObjectPath("db", "tbl"), table, false);
+
+ Mockito.verify(tableCatalog, Mockito.never()).alterTable(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ public void testAlterTableWithCommentChangeForwardsAlterCall() throws
Exception {
+ // A comment change produces a TableChange, so the alter must be forwarded
to Gravitino.
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ TableCatalog tableCatalog = Mockito.mock(TableCatalog.class);
+ Mockito.when(gravitinoCatalog.asTableCatalog()).thenReturn(tableCatalog);
+
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ CatalogBaseTable existingTable =
+ DefaultCatalogCompat.INSTANCE.createCatalogTable(
+ schema, "old comment", ImmutableList.of(), ImmutableMap.of("key",
"value"));
+ CatalogBaseTable newTable =
+ DefaultCatalogCompat.INSTANCE.createCatalogTable(
+ schema, "new comment", ImmutableList.of(), ImmutableMap.of("key",
"value"));
+
+ TestableBaseCatalog catalog =
+ new TestableBaseCatalog(
+ Mockito.mock(AbstractCatalog.class), gravitinoCatalog,
existingTable);
+
+ catalog.alterTable(new ObjectPath("db", "tbl"), newTable, false);
+
+ Mockito.verify(tableCatalog, Mockito.times(1)).alterTable(Mockito.any(),
Mockito.any());
+ }
+
private static class TestableBaseCatalog extends BaseCatalog {
private final AbstractCatalog delegate;
private final Catalog gravitinoCatalog;
+ private final CatalogBaseTable existingTable;
TestableBaseCatalog(AbstractCatalog delegate, Catalog gravitinoCatalog) {
+ this(delegate, gravitinoCatalog, null);
+ }
+
+ TestableBaseCatalog(
+ AbstractCatalog delegate, Catalog gravitinoCatalog, CatalogBaseTable
existingTable) {
super(
"test",
Collections.emptyMap(),
@@ -386,6 +463,7 @@ public class TestBaseCatalog {
Mockito.mock(PartitionConverter.class));
this.delegate = delegate;
this.gravitinoCatalog = gravitinoCatalog;
+ this.existingTable = existingTable;
}
@Override
@@ -397,5 +475,18 @@ public class TestBaseCatalog {
protected Catalog catalog() {
return gravitinoCatalog;
}
+
+ @Override
+ public CatalogBaseTable getTable(ObjectPath tablePath) throws
TableNotExistException {
+ if (existingTable != null) {
+ return existingTable;
+ }
+ return super.getTable(tablePath);
+ }
+
+ @Override
+ protected void invalidateTable(ObjectPath tablePath) {
+ // No-op: the native cache is not exercised in these unit tests.
+ }
}
}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalog.java
new file mode 100644
index 0000000000..f68d1c125b
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/hive/TestGravitinoHiveCatalog.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.gravitino.flink.connector.hive;
+
+import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.Schema;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.Column;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.flink.connector.PartitionConverter;
+import org.apache.gravitino.flink.connector.SchemaAndTablePropertiesConverter;
+import org.apache.gravitino.flink.connector.utils.DefaultCatalogCompat;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestGravitinoHiveCatalog {
+
+ @Test
+ public void testGenericTableAlterSkipsCallWhenNoChanges() throws Exception {
+ // Existing table and the resolved new table describe the same state (same
properties, same
+ // comment), so no TableChange is produced. The connector must not forward
an empty update
+ // list to Gravitino, which would fail server-side with "updates must not
be empty".
+ Map<String, String> sameProperties =
+ ImmutableMap.of("flink.connector", "kafka", "is_generic", "true");
+
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ TableCatalog tableCatalog = Mockito.mock(TableCatalog.class);
+ Mockito.when(gravitinoCatalog.asTableCatalog()).thenReturn(tableCatalog);
+
+ Table existingTable = Mockito.mock(Table.class);
+ Mockito.when(existingTable.properties()).thenReturn(sameProperties);
+ Mockito.when(existingTable.comment()).thenReturn("same comment");
+
+ ResolvedCatalogTable newTable = resolvedTable("same comment");
+
+ TestableGravitinoHiveCatalog catalog =
+ new TestableGravitinoHiveCatalog(gravitinoCatalog, sameProperties);
+
+ catalog.applyGenericTableAlter(new ObjectPath("db", "tbl"), existingTable,
newTable);
+
+ // The alter call is skipped entirely because there is nothing to change.
+ Mockito.verify(tableCatalog, Mockito.never()).alterTable(Mockito.any(),
Mockito.any());
+ }
+
+ @Test
+ public void testGenericTableAlterCallsAlterWhenPropertiesChange() throws
Exception {
+ // The resolved new table changes a property, so the connector must
forward the update.
+ Map<String, String> currentProperties =
+ ImmutableMap.of("flink.connector", "kafka", "is_generic", "true");
+ Map<String, String> updatedProperties =
+ ImmutableMap.of(
+ "flink.connector", "kafka", "flink.topic", "new-topic",
"is_generic", "true");
+
+ Catalog gravitinoCatalog = Mockito.mock(Catalog.class);
+ TableCatalog tableCatalog = Mockito.mock(TableCatalog.class);
+ Mockito.when(gravitinoCatalog.asTableCatalog()).thenReturn(tableCatalog);
+
+ Table existingTable = Mockito.mock(Table.class);
+ Mockito.when(existingTable.properties()).thenReturn(currentProperties);
+ Mockito.when(existingTable.comment()).thenReturn("comment");
+
+ ResolvedCatalogTable newTable = resolvedTable("comment");
+
+ TestableGravitinoHiveCatalog catalog =
+ new TestableGravitinoHiveCatalog(gravitinoCatalog, updatedProperties);
+
+ catalog.applyGenericTableAlter(new ObjectPath("db", "tbl"), existingTable,
newTable);
+
+ // A real change was present, so the alter call is forwarded to Gravitino.
+ Mockito.verify(tableCatalog, Mockito.times(1)).alterTable(Mockito.any(),
Mockito.any());
+ }
+
+ private static ResolvedCatalogTable resolvedTable(String comment) {
+ Schema schema = Schema.newBuilder().column("id", DataTypes.INT()).build();
+ CatalogTable table =
+ DefaultCatalogCompat.INSTANCE.createCatalogTable(
+ schema, comment, Collections.emptyList(), Collections.emptyMap());
+ ResolvedSchema resolvedSchema =
+ new ResolvedSchema(
+ Collections.singletonList(Column.physical("id", DataTypes.INT())),
+ Collections.emptyList(),
+ null);
+ return new ResolvedCatalogTable(table, resolvedSchema);
+ }
+
+ private static class TestableGravitinoHiveCatalog extends
GravitinoHiveCatalog {
+ private final Catalog gravitinoCatalog;
+ private final Map<String, String> genericTableProperties;
+
+ TestableGravitinoHiveCatalog(Catalog gravitinoCatalog) {
+ this(gravitinoCatalog, Collections.emptyMap());
+ }
+
+ TestableGravitinoHiveCatalog(
+ Catalog gravitinoCatalog, Map<String, String> genericTableProperties) {
+ super(
+ "test",
+ "default",
+ Collections.emptyMap(),
+ Mockito.mock(SchemaAndTablePropertiesConverter.class),
+ Mockito.mock(PartitionConverter.class),
+ hiveConf(),
+ null);
+ this.gravitinoCatalog = gravitinoCatalog;
+ this.genericTableProperties = genericTableProperties;
+ }
+
+ @Override
+ protected AbstractCatalog realCatalog() {
+ return Mockito.mock(AbstractCatalog.class);
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return gravitinoCatalog;
+ }
+
+ @Override
+ protected Map<String, String>
toGravitinoGenericTableProperties(ResolvedCatalogTable table) {
+ return genericTableProperties;
+ }
+
+ private static HiveConf hiveConf() {
+ HiveConf hiveConf = new HiveConf();
+ hiveConf.set("hive.metastore.uris", "thrift://localhost:9083");
+ return hiveConf;
+ }
+ }
+}
diff --git
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
index 3e29e30e2e..3d6f9e6d98 100644
---
a/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
@@ -373,13 +373,15 @@ public class TestGravitinoPaimonCatalog {
TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog,
mockCatalog);
ObjectPath path = new ObjectPath("mydb", "mytable");
+ org.apache.flink.table.api.Schema schema =
+ org.apache.flink.table.api.Schema.newBuilder().column("id",
DataTypes.INT()).build();
+ // The existing table carries an old comment so the alter produces a real
TableChange
+ // (comment update) rather than a no-op.
+ CatalogTable existingFlinkTable =
+ CatalogTable.of(schema, "old comment", Collections.emptyList(),
Collections.emptyMap());
CatalogTable newTable =
- CatalogTable.of(
- org.apache.flink.table.api.Schema.newBuilder().column("id",
DataTypes.INT()).build(),
- "new comment",
- Collections.emptyList(),
- Collections.emptyMap());
- when(mockFlinkCatalog.getTable(path)).thenReturn(newTable);
+ CatalogTable.of(schema, "new comment", Collections.emptyList(),
Collections.emptyMap());
+ when(mockFlinkCatalog.getTable(path)).thenReturn(existingFlinkTable);
cat.alterTable(path, newTable, false);
@@ -387,6 +389,50 @@ public class TestGravitinoPaimonCatalog {
verify(mockInnerCatalog).invalidateTable(Identifier.create("mydb",
"mytable"));
}
+ /**
+ * Verifies that a no-op Paimon alterTable (existing and new tables
identical) neither forwards
+ * the alter to Gravitino nor invalidates the native cache.
+ */
+ @Test
+ public void testAlterTableNoOpDoesNotInvalidateNativeCache() throws
Exception {
+ org.apache.paimon.catalog.Catalog mockInnerCatalog =
+ mock(org.apache.paimon.catalog.Catalog.class);
+ FlinkCatalog mockFlinkCatalog = mock(FlinkCatalog.class);
+ when(mockFlinkCatalog.catalog()).thenReturn(mockInnerCatalog);
+
+ Catalog mockCatalog = mock(Catalog.class);
+ TableCatalog mockTableCatalog = mock(TableCatalog.class);
+ when(mockCatalog.asTableCatalog()).thenReturn(mockTableCatalog);
+
+ Table existingTable = mock(Table.class);
+ org.apache.gravitino.rel.Column existingColumn =
+ org.apache.gravitino.rel.Column.of(
+ "id", org.apache.gravitino.rel.types.Types.IntegerType.get());
+ when(existingTable.columns())
+ .thenReturn(new org.apache.gravitino.rel.Column[] {existingColumn});
+ when(existingTable.index()).thenReturn(new Index[0]);
+ when(existingTable.properties()).thenReturn(Collections.emptyMap());
+ when(existingTable.distribution()).thenReturn(null);
+ when(existingTable.partitioning()).thenReturn(Transforms.EMPTY_TRANSFORM);
+ when(existingTable.comment()).thenReturn("same comment");
+ when(mockTableCatalog.loadTable(any())).thenReturn(existingTable);
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog,
mockCatalog);
+ ObjectPath path = new ObjectPath("mydb", "mytable");
+ org.apache.flink.table.api.Schema schema =
+ org.apache.flink.table.api.Schema.newBuilder().column("id",
DataTypes.INT()).build();
+ // The existing table and the alter target share the same comment, so no
TableChange is
+ // produced and the alter must be skipped.
+ CatalogTable sameTable =
+ CatalogTable.of(schema, "same comment", Collections.emptyList(),
Collections.emptyMap());
+ when(mockFlinkCatalog.getTable(path)).thenReturn(sameTable);
+
+ cat.alterTable(path, sameTable, false);
+
+ verify(mockTableCatalog, never()).alterTable(any(), any());
+ verify(mockInnerCatalog, never()).invalidateTable(any());
+ }
+
/** Verifies that successful Paimon dropTable invalidates the native cache.
*/
@Test
public void testDropTableInvalidatesNativeCacheAfterSuccessfulPurge() throws
Exception {
