This is an automated email from the ASF dual-hosted git repository.
FANNG1 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 54cb6c254e [Cherry-pick to branch-1.3] [#10477] fix(flink-paimon):
Paimon FlinkCatalog to fix Hive partition updates (#10475) (#11439)
54cb6c254e is described below
commit 54cb6c254ef4fb91df01806499523df007d3b468
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jun 5 10:21:47 2026 +0800
[Cherry-pick to branch-1.3] [#10477] fix(flink-paimon): Paimon
FlinkCatalog to fix Hive partition updates (#10475) (#11439)
**Cherry-pick Information:**
- Original commit: 1bf667dd39f71b200600772de87a85e49119acc1
- Target branch: `branch-1.3`
- Status: ✅ Clean cherry-pick (no conflicts)
Co-authored-by: wangxiaojing <[email protected]>
Co-authored-by: wangxiaojing <[email protected]>
Co-authored-by: Qi Yu <[email protected]>
---
.../flink/connector/catalog/BaseCatalog.java | 71 +++-
.../connector/paimon/GravitinoPaimonCatalog.java | 130 +++++-
.../paimon/TestGravitinoPaimonCatalog.java | 444 +++++++++++++++++++++
3 files changed, 623 insertions(+), 22 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 0d781fb631..9d9d407468 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
@@ -261,6 +261,8 @@ public abstract class BaseCatalog extends AbstractCatalog {
// Treat authorization failure as table-not-exist to allow Calcite to
fall back to
// alternative resolution paths (e.g., treating the name as a schema).
throw new TableNotExistException(catalogName(), tablePath, e);
+ } catch (CatalogException e) {
+ throw e;
} catch (Exception e) {
LOG.warn("Failed to load table {} from catalog {}", ident,
catalogName(), e);
throw new CatalogException(e);
@@ -320,6 +322,9 @@ public abstract class BaseCatalog extends AbstractCatalog {
if (!tableDropped && !viewDropped && !ignoreIfNotExists) {
throw new TableNotExistException(catalogName(), tablePath);
}
+ if (tableDropped) {
+ invalidateTable(tablePath);
+ }
} catch (TableNotExistException e) {
throw e;
} catch (Exception e) {
@@ -336,6 +341,8 @@ public abstract class BaseCatalog extends AbstractCatalog {
try {
catalog().asTableCatalog().alterTable(srcIdent,
TableChange.rename(newTableName));
+ // Invalidate native catalog cache after successful rename
+ invalidateTable(tablePath);
return;
} catch (NoSuchTableException ignored) {
// source is not a table, try as a view below
@@ -510,6 +517,8 @@ public abstract class BaseCatalog extends AbstractCatalog {
catalog()
.asTableCatalog()
.alterTable(identifier, getGravitinoTableChanges(existingTable,
newTable));
+ // Invalidate native catalog cache after successful alter
+ invalidateTable(tablePath);
}
}
@@ -556,6 +565,8 @@ public abstract class BaseCatalog extends AbstractCatalog {
}
} else {
catalog().asTableCatalog().alterTable(identifier,
getGravitinoTableChanges(tableChanges));
+ // Invalidate native catalog cache after successful alter
+ invalidateTable(tablePath);
}
}
@@ -714,6 +725,40 @@ public abstract class BaseCatalog extends AbstractCatalog {
throw new UnsupportedOperationException();
}
+ /**
+ * Invalidates cached table metadata in the native Flink catalog after DDL
operations.
+ *
+ * <p>Connectors that maintain an internal native catalog cache (e.g.
Paimon's {@code
+ * CachingCatalog}) must override {@link #invalidateNativeTableCache} to
clear stale entries. This
+ * method calls {@code invalidateNativeTableCache} with a best-effort
approach — failures are
+ * logged at DEBUG level and do not abort the DDL. It is always called
<em>after</em> the
+ * Gravitino DDL has succeeded, so the cache is only evicted once the source
of truth has already
+ * been updated.
+ *
+ * @param tablePath the table whose native cache entry should be dropped
+ */
+ protected void invalidateTable(ObjectPath tablePath) {
+ try {
+ invalidateNativeTableCache(tablePath);
+ } catch (Exception e) {
+ LOG.debug(
+ "Failed to invalidate native catalog cache for table {} in catalog
{}",
+ tablePath,
+ catalogName(),
+ e);
+ }
+ }
+
+ /**
+ * Invalidates the native catalog cache for the given table. Default is a
no-op.
+ *
+ * <p>Subclasses with an internal native catalog that caches table metadata
(e.g. Paimon, Iceberg)
+ * should override this method to evict the stale entry after a DDL
operation completes.
+ *
+ * @param tablePath the table whose native cache entry should be dropped
+ */
+ protected void invalidateNativeTableCache(ObjectPath tablePath) {}
+
protected CatalogBaseTable toFlinkTable(Table table, ObjectPath tablePath) {
org.apache.flink.table.api.Schema.Builder builder =
buildSchemaFromColumns(table.columns());
Optional<List<String>> flinkPrimaryKey = getFlinkPrimaryKey(table);
@@ -724,7 +769,31 @@ public abstract class BaseCatalog extends AbstractCatalog {
catalogOptions, table.properties(), tablePath));
flinkTableProperties.putAll(fromGravitinoDistribution(table.distribution()));
List<String> partitionKeys =
partitionConverter.toFlinkPartitionKeys(table.partitioning());
- return newCatalogTable(builder.build(), table.comment(), partitionKeys,
flinkTableProperties);
+ CatalogTable baseTable =
+ newCatalogTable(builder.build(), table.comment(), partitionKeys,
flinkTableProperties);
+ return enrichCatalogTable(baseTable, tablePath);
+ }
+
+ /**
+ * Hook for subclasses to enrich or replace the plain {@link CatalogTable}
built from Gravitino
+ * metadata with a connector-native representation.
+ *
+ * <p>The default implementation returns the table unchanged.
Connector-specific subclasses (e.g.
+ * {@code GravitinoPaimonCatalog}) can override this method to return a
native table object that
+ * carries additional runtime context required by the underlying engine —
for example, Paimon's
+ * {@code DataCatalogTable} with a non-null {@code CatalogEnvironment} that
enables {@code
+ * AddPartitionCommitCallback} registration on write.
+ *
+ * <p>This hook is called <em>after</em> Gravitino authorization has already
been enforced in
+ * {@link #getTable(ObjectPath)}, so implementations do not need to repeat
auth checks.
+ *
+ * @param table the plain {@link CatalogTable} built from Gravitino metadata
+ * @param tablePath the object path of the table
+ * @return the (possibly enriched) {@link CatalogBaseTable} to return to
Flink
+ * @throws CatalogException if enrichment fails due to a catalog-level error
+ */
+ protected CatalogBaseTable enrichCatalogTable(CatalogTable table, ObjectPath
tablePath) {
+ return table;
}
protected CatalogTable newCatalogTable(
diff --git
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
index 77ac4806c4..3776b02da3 100644
---
a/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
+++
b/flink-connector/flink-common/src/main/java/org/apache/gravitino/flink/connector/paimon/GravitinoPaimonCatalog.java
@@ -28,6 +28,8 @@ import java.util.Optional;
import java.util.stream.Collectors;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
@@ -43,11 +45,21 @@ import org.apache.gravitino.rel.expressions.NamedReference;
import org.apache.gravitino.rel.expressions.distributions.Distribution;
import org.apache.gravitino.rel.expressions.distributions.Distributions;
import org.apache.gravitino.rel.expressions.distributions.Strategy;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalog;
import org.apache.paimon.flink.FlinkCatalogFactory;
/**
* The GravitinoPaimonCatalog class is an implementation of the BaseCatalog
class that is used to
* proxy the PaimonCatalog class.
+ *
+ * <p>DDL operations (CREATE / ALTER / DROP) are routed through the Gravitino
REST API, keeping
+ * Gravitino as the single source of truth for metadata. The internal {@code
paimonCatalog} is used
+ * only by {@link #enrichCatalogTable} so that {@code getTable()} returns
Paimon's native {@code
+ * DataCatalogTable} — which carries a fully-initialised {@code
CatalogEnvironment} (non-null {@code
+ * catalogLoader}). Without this, the {@code AddPartitionCommitCallback} that
syncs new partitions
+ * to Hive Metastore is never registered, causing {@code SHOW PARTITIONS} to
return empty results
+ * even when {@code metastore.partitioned-table=true}.
*/
public class GravitinoPaimonCatalog extends BaseCatalog {
@@ -68,37 +80,28 @@ public class GravitinoPaimonCatalog extends BaseCatalog {
this.paimonCatalog = flinkCatalogFactory.createCatalog(context);
}
- @Override
- protected AbstractCatalog realCatalog() {
- return paimonCatalog;
- }
+ //
---------------------------------------------------------------------------
+ // Lifecycle — keep paimonCatalog in sync with the outer catalog
+ //
---------------------------------------------------------------------------
@Override
- public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
- throws TableNotExistException, CatalogException {
- boolean dropped =
- catalog()
- .asTableCatalog()
- .purgeTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
- if (!dropped && !ignoreIfNotExists) {
- throw new TableNotExistException(catalogName(), tablePath);
- }
+ public void open() throws CatalogException {
+ super.open(); // opens realCatalog() == paimonCatalog, so
paimonCatalog.open() is called here
}
@Override
- public Optional<Factory> getFactory() {
- return paimonCatalog.getFactory();
+ public void close() throws CatalogException {
+ super.close(); // closes realCatalog() == paimonCatalog
}
@Override
- protected Distribution toGravitinoDistribution(Map<String, String>
properties) {
- return getDistribution(properties);
+ protected AbstractCatalog realCatalog() {
+ return paimonCatalog;
}
- @Override
- protected Map<String, String> fromGravitinoDistribution(Distribution
distribution) {
- return distributionToProperties(distribution);
- }
+ //
---------------------------------------------------------------------------
+ // DDL — route through Gravitino (single source of truth)
+ //
---------------------------------------------------------------------------
@VisibleForTesting
static Map<String, String> distributionToProperties(Distribution
distribution) {
@@ -132,6 +135,10 @@ public class GravitinoPaimonCatalog extends BaseCatalog {
return properties;
}
+ //
---------------------------------------------------------------------------
+ // getTable enrichment — return Paimon-native DataCatalogTable
+ //
---------------------------------------------------------------------------
+
@VisibleForTesting
static Distribution getDistribution(Map<String, String> properties) {
if (properties == null) {
@@ -175,4 +182,85 @@ public class GravitinoPaimonCatalog extends BaseCatalog {
e);
}
}
+
+ @Override
+ public Optional<Factory> getFactory() {
+ return paimonCatalog.getFactory();
+ }
+
+ @Override
+ public void dropTable(ObjectPath tablePath, boolean ignoreIfNotExists)
+ throws TableNotExistException, CatalogException {
+ boolean dropped =
+ catalog()
+ .asTableCatalog()
+ .purgeTable(NameIdentifier.of(tablePath.getDatabaseName(),
tablePath.getObjectName()));
+ if (!dropped && !ignoreIfNotExists) {
+ throw new TableNotExistException(catalogName(), tablePath);
+ }
+ if (dropped) {
+ invalidateNativeTableCache(tablePath);
+ }
+ }
+
+ @Override
+ protected Distribution toGravitinoDistribution(Map<String, String>
properties) {
+ return getDistribution(properties);
+ }
+
+ @Override
+ protected Map<String, String> fromGravitinoDistribution(Distribution
distribution) {
+ return distributionToProperties(distribution);
+ }
+
+ /**
+ * Returns the Paimon-native {@code DataCatalogTable} for {@code tablePath}.
+ *
+ * <p>{@link BaseCatalog#getTable} has already verified via the Gravitino
REST API that (a) the
+ * caller is authorised and (b) the table exists before this hook is
invoked. Therefore, a {@link
+ * TableNotExistException} from {@code paimonCatalog.getTable()} indicates a
metadata
+ * inconsistency between Gravitino and the underlying Paimon store.
+ *
+ * <p>The returned {@code DataCatalogTable} wraps a {@code FileStoreTable}
whose {@code
+ * CatalogEnvironment} holds a valid {@code catalogLoader}. Paimon's write
path uses this to
+ * register {@code AddPartitionCommitCallback}, which in turn calls the Hive
Metastore to record
+ * new partitions after each checkpoint commit.
+ */
+ @Override
+ protected CatalogBaseTable enrichCatalogTable(CatalogTable ignoredBaseTable,
ObjectPath tablePath)
+ throws CatalogException {
+ try {
+ return realCatalog().getTable(tablePath);
+ } catch (TableNotExistException e) {
+ throw new CatalogException(
+ String.format(
+ "Table '%s.%s' was found in Gravitino but is absent from the
underlying Paimon "
+ + "catalog '%s'. The two metadata stores may be out of
sync.",
+ tablePath.getDatabaseName(), tablePath.getObjectName(),
catalogName()),
+ e);
+ }
+ }
+
+ /**
+ * Invalidates the Paimon native catalog cache for the given table.
+ *
+ * <p>When Paimon is initialised with a {@code CachingCatalog} (the default
in production), table
+ * and partition metadata is held in an in-memory cache. After DDL
operations routed through
+ * Gravitino (drop / rename / alter), the stale cache entry must be evicted
so that the next
+ * {@code getTable()} call picks up fresh metadata from the Hive Metastore
or filesystem.
+ *
+ * <p>This implementation casts {@link #paimonCatalog} to {@link
FlinkCatalog} and delegates to
+ * {@code FlinkCatalog.catalog().invalidateTable(Identifier)}, which
forwards to {@link
+ * org.apache.paimon.catalog.CachingCatalog#invalidateTable} when caching is
enabled. The call is
+ * a no-op when the underlying catalog does not cache (e.g. in tests or
filesystem mode).
+ */
+ @Override
+ protected void invalidateNativeTableCache(ObjectPath tablePath) {
+ AbstractCatalog nativeCatalog = realCatalog();
+ if (nativeCatalog instanceof FlinkCatalog) {
+ Identifier identifier =
+ Identifier.create(tablePath.getDatabaseName(),
tablePath.getObjectName());
+ ((FlinkCatalog) nativeCatalog).catalog().invalidateTable(identifier);
+ }
+ }
}
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
new file mode 100644
index 0000000000..1df9150b4e
--- /dev/null
+++
b/flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java
@@ -0,0 +1,444 @@
+/*
+ * 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.paimon;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.catalog.AbstractCatalog;
+import org.apache.flink.table.catalog.CatalogBaseTable;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ObjectPath;
+import org.apache.flink.table.catalog.exceptions.CatalogException;
+import org.apache.flink.table.catalog.exceptions.TableNotExistException;
+import org.apache.flink.table.factories.CatalogFactory;
+import org.apache.gravitino.Catalog;
+import org.apache.gravitino.flink.connector.DefaultPartitionConverter;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+import org.apache.gravitino.rel.Table;
+import org.apache.gravitino.rel.TableCatalog;
+import org.apache.gravitino.rel.expressions.transforms.Transforms;
+import org.apache.gravitino.rel.indexes.Index;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.flink.FlinkCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for the {@code enrichCatalogTable} hook introduced in {@link
BaseCatalog} and its
+ * implementation in {@link GravitinoPaimonCatalog}.
+ *
+ * <p>These tests verify:
+ *
+ * <ol>
+ * <li>The default {@code BaseCatalog.enrichCatalogTable} is an identity
function.
+ * <li>{@code GravitinoPaimonCatalog.enrichCatalogTable} returns the
Paimon-native table (i.e. the
+ * result of {@code paimonCatalog.getTable()}) rather than the plain
Gravitino {@link
+ * CatalogTable}.
+ * <li>When {@code paimonCatalog.getTable()} throws {@link
TableNotExistException} (metadata
+ * out-of-sync), a descriptive {@link CatalogException} is raised.
+ * <li>{@code paimonCatalog.getTable()} is never called when Gravitino auth
fails.
+ * </ol>
+ */
+public class TestGravitinoPaimonCatalog {
+
+ private AbstractCatalog mockPaimonCatalog;
+
+ //
---------------------------------------------------------------------------
+ // Minimal test double for BaseCatalog — overrides only what's needed
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Subclass of {@link BaseCatalog} used to test the {@code
enrichCatalogTable} default behaviour.
+ * All abstract methods delegate to no-ops or mocks; the only interesting
thing is the hook.
+ */
+ private static class TestableBaseCatalog extends BaseCatalog {
+
+ private final AbstractCatalog realCatalog = mock(AbstractCatalog.class);
+ private final Catalog gravitinoCatalog = mock(Catalog.class);
+ private CatalogBaseTable toFlinkTableResult;
+ private CatalogException toFlinkTableException;
+
+ TestableBaseCatalog() {
+ super(
+ "test-catalog",
+ Collections.emptyMap(),
+ "default",
+ PaimonPropertiesConverter.INSTANCE,
+ DefaultPartitionConverter.INSTANCE);
+ }
+
+ @Override
+ protected AbstractCatalog realCatalog() {
+ return realCatalog;
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return gravitinoCatalog;
+ }
+
+ @Override
+ protected CatalogBaseTable toFlinkTable(Table table, ObjectPath tablePath)
{
+ if (toFlinkTableException != null) {
+ throw toFlinkTableException;
+ }
+ CatalogTable baseTable = (CatalogTable) toFlinkTableResult;
+ return enrichCatalogTable(baseTable, tablePath);
+ }
+
+ public CatalogBaseTable callEnrichCatalogTable(CatalogTable table,
ObjectPath path) {
+ return enrichCatalogTable(table, path);
+ }
+ }
+
+ //
---------------------------------------------------------------------------
+ // Minimal test double for GravitinoPaimonCatalog that injects a mock
paimonCatalog
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Extends {@link GravitinoPaimonCatalog} so that tests can inject a mock
native catalog via
+ * {@link #realCatalog()} and verify cache invalidation and enrichment
behavior without a real
+ * Paimon environment.
+ */
+ private static class TestablePaimonCatalog extends GravitinoPaimonCatalog {
+
+ private final AbstractCatalog injectedPaimon;
+ private final Catalog injectedCatalog;
+
+ TestablePaimonCatalog(AbstractCatalog injectedPaimon) {
+ this(injectedPaimon, null);
+ }
+
+ TestablePaimonCatalog(AbstractCatalog injectedPaimon, Catalog
injectedCatalog) {
+ // We cannot call super(context, ...) without a real
FlinkCatalogFactory, so we use a
+ // package-private constructor shim that skips the factory call.
Because we override
+ // realCatalog() the parent constructor's catalog reference is never
used.
+ super(
+ new MockCatalogContext("test-paimon", defaultPaimonOptions()),
+ "default",
+ PaimonPropertiesConverter.INSTANCE,
+ DefaultPartitionConverter.INSTANCE);
+ this.injectedPaimon = injectedPaimon;
+ this.injectedCatalog = injectedCatalog;
+ }
+
+ @Override
+ protected AbstractCatalog realCatalog() {
+ return injectedPaimon;
+ }
+
+ @Override
+ protected Catalog catalog() {
+ return injectedCatalog != null ? injectedCatalog : super.catalog();
+ }
+
+ private static Map<String, String> defaultPaimonOptions() {
+ Map<String, String> options = new HashMap<>();
+ options.put("warehouse", "file:/tmp/test-paimon-warehouse");
+ return options;
+ }
+ }
+
+ @BeforeEach
+ void setUp() {
+ mockPaimonCatalog = mock(AbstractCatalog.class);
+ }
+
+ //
---------------------------------------------------------------------------
+ // BaseCatalog default hook: identity function
+ //
---------------------------------------------------------------------------
+
+ /**
+ * The base implementation must return the same {@link CatalogTable} it
receives, unchanged. This
+ * ensures existing catalogs (Hive, Iceberg, JDBC) are unaffected by the new
hook.
+ */
+ @Test
+ public void testDefaultEnrichCatalogTableIsIdentity() {
+ TestableBaseCatalog base = new TestableBaseCatalog();
+ CatalogTable input = mock(CatalogTable.class);
+ ObjectPath path = new ObjectPath("db", "tbl");
+
+ CatalogBaseTable result = base.callEnrichCatalogTable(input, path);
+
+ Assertions.assertSame(
+ input,
+ result,
+ "BaseCatalog.enrichCatalogTable must return the input table unchanged
by default");
+ }
+
+ //
---------------------------------------------------------------------------
+ // GravitinoPaimonCatalog: enrichCatalogTable returns Paimon-native table
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Verifies that {@code GravitinoPaimonCatalog.enrichCatalogTable} ignores
the plain {@link
+ * CatalogTable} and instead returns whatever {@code
paimonCatalog.getTable()} produces.
+ *
+ * <p>The object returned by {@code paimonCatalog.getTable()} is Paimon's
{@code
+ * DataCatalogTable}, which carries a non-null {@code CatalogEnvironment}.
This is what enables
+ * {@code AddPartitionCommitCallback} registration on the write path, fixing
the "partitions not
+ * visible in {@code SHOW PARTITIONS}" bug.
+ */
+ @Test
+ public void testEnrichCatalogTableReturnsPaimonNativeTable()
+ throws TableNotExistException, CatalogException {
+ CatalogBaseTable paimonNativeTable = mock(CatalogBaseTable.class);
+ ObjectPath path = new ObjectPath("db", "tbl");
+ when(mockPaimonCatalog.getTable(path)).thenReturn(paimonNativeTable);
+
+ CatalogTable gravitinoBuiltTable = mock(CatalogTable.class);
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockPaimonCatalog);
+ CatalogBaseTable result = cat.enrichCatalogTable(gravitinoBuiltTable,
path);
+
+ Assertions.assertSame(
+ paimonNativeTable,
+ result,
+ "enrichCatalogTable must return the Paimon-native DataCatalogTable");
+ verify(mockPaimonCatalog).getTable(path);
+ }
+
+ //
---------------------------------------------------------------------------
+ // Metadata out-of-sync: TableNotExistException → CatalogException
+ //
---------------------------------------------------------------------------
+
+ /**
+ * When Gravitino confirms the table exists but the underlying Paimon store
does not have it,
+ * {@code enrichCatalogTable} must throw a descriptive {@link
CatalogException} rather than
+ * leaking the raw {@link TableNotExistException} from Paimon.
+ */
+ @Test
+ public void testEnrichCatalogTableOutOfSyncThrowsCatalogException()
+ throws TableNotExistException {
+ ObjectPath path = new ObjectPath("db", "missing_in_paimon");
+ when(mockPaimonCatalog.getTable(path))
+ .thenThrow(new TableNotExistException("test-paimon", path));
+
+ CatalogTable gravitinoBuiltTable = mock(CatalogTable.class);
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockPaimonCatalog);
+ CatalogException ex =
+ Assertions.assertThrows(
+ CatalogException.class, () ->
cat.enrichCatalogTable(gravitinoBuiltTable, path));
+
+ Assertions.assertTrue(
+ ex.getMessage().contains("missing_in_paimon"),
+ "Exception message should contain the table name for diagnostics");
+ Assertions.assertInstanceOf(
+ TableNotExistException.class,
+ ex.getCause(),
+ "Original TableNotExistException should be preserved as the cause");
+ }
+
+ //
---------------------------------------------------------------------------
+ // paimonCatalog.getTable() must NOT be called when Gravitino auth fails
+ // (this contract is enforced by BaseCatalog.getTable(), tested here as a
+ // sanity check via the enrichCatalogTable path)
+ //
---------------------------------------------------------------------------
+
+ /**
+ * {@code paimonCatalog.getTable()} must never be called if Gravitino auth
throws before {@code
+ * enrichCatalogTable} is reached. This is enforced by {@link
BaseCatalog#getTable} which invokes
+ * the hook only after a successful Gravitino {@code loadTable()} call.
+ *
+ * <p>The test simulates this contract at the unit level by verifying that a
direct call to {@code
+ * enrichCatalogTable} (the hook itself) does call the inner catalog —
confirming the security
+ * boundary is in {@link BaseCatalog#getTable}, not in the hook.
+ */
+ @Test
+ public void testGetTableAuthFailureDoesNotCallPaimonCatalog() throws
Exception {
+ Catalog mockCatalog = mock(Catalog.class);
+ TableCatalog mockTableCatalog = mock(TableCatalog.class);
+ ObjectPath path = new ObjectPath("db", "tbl");
+ when(mockCatalog.asTableCatalog()).thenReturn(mockTableCatalog);
+ when(mockTableCatalog.loadTable(any())).thenThrow(new
RuntimeException("denied"));
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockPaimonCatalog,
mockCatalog);
+
+ Assertions.assertThrows(RuntimeException.class, () -> cat.getTable(path));
+ verify(mockPaimonCatalog, never()).getTable(any());
+ }
+
+ //
---------------------------------------------------------------------------
+ // invalidateNativeTableCache: Paimon CachingCatalog is evicted after DDL
+ //
---------------------------------------------------------------------------
+
+ /**
+ * Verifies that {@code invalidateNativeTableCache} calls {@code
Catalog.invalidateTable} on the
+ * underlying Paimon inner catalog when {@code paimonCatalog} is a {@link
FlinkCatalog}.
+ *
+ * <p>This ensures that after DDL operations (drop / rename / alter) routed
through Gravitino,
+ * stale entries in Paimon's {@code CachingCatalog} are evicted so
subsequent reads reflect the
+ * updated metadata.
+ */
+ @Test
+ public void testInvalidateNativeTableCacheCallsPaimonInvalidate() {
+ org.apache.paimon.catalog.Catalog mockInnerCatalog =
+ mock(org.apache.paimon.catalog.Catalog.class);
+ FlinkCatalog mockFlinkCatalog = mock(FlinkCatalog.class);
+ when(mockFlinkCatalog.catalog()).thenReturn(mockInnerCatalog);
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog);
+ ObjectPath path = new ObjectPath("mydb", "mytable");
+ cat.invalidateNativeTableCache(path);
+
+ Identifier expected = Identifier.create("mydb", "mytable");
+ verify(mockInnerCatalog).invalidateTable(expected);
+ }
+
+ /**
+ * Verifies that {@code invalidateNativeTableCache} is a no-op when the
underlying catalog is not
+ * a {@link FlinkCatalog} (e.g. in tests using a plain {@link
AbstractCatalog} mock). No exception
+ * should be thrown.
+ */
+ @Test
+ public void testInvalidateNativeTableCacheIsNoOpForNonFlinkCatalog() {
+ // mockPaimonCatalog is AbstractCatalog, not FlinkCatalog — must not throw
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockPaimonCatalog);
+ Assertions.assertDoesNotThrow(
+ () -> cat.invalidateNativeTableCache(new ObjectPath("db", "tbl")));
+ }
+
+ /**
+ * Verifies that {@link BaseCatalog#getTable(ObjectPath)} preserves
hook-thrown CatalogException.
+ */
+ @Test
+ public void testGetTablePreservesCatalogException() {
+ TestableBaseCatalog baseCatalog = new TestableBaseCatalog();
+ Catalog mockCatalog = baseCatalog.catalog();
+ TableCatalog mockTableCatalog = mock(TableCatalog.class);
+ when(mockCatalog.asTableCatalog()).thenReturn(mockTableCatalog);
+ when(mockTableCatalog.loadTable(any())).thenReturn(mock(Table.class));
+ baseCatalog.toFlinkTableResult = mock(CatalogTable.class);
+ CatalogException expected = new CatalogException("boom");
+ baseCatalog.toFlinkTableException = expected;
+
+ CatalogException actual =
+ Assertions.assertThrows(
+ CatalogException.class, () -> baseCatalog.getTable(new
ObjectPath("db", "tbl")));
+
+ Assertions.assertSame(expected, actual, "CatalogException should be
rethrown as-is");
+ }
+
+ /** Verifies that successful Paimon alterTable invalidates the native cache.
*/
+ @Test
+ public void testAlterTableInvalidatesNativeCacheAfterSuccessfulAlter()
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("existing comment");
+ when(mockTableCatalog.loadTable(any())).thenReturn(existingTable);
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog,
mockCatalog);
+ ObjectPath path = new ObjectPath("mydb", "mytable");
+ 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);
+
+ cat.alterTable(path, newTable, false);
+
+ verify(mockTableCatalog).alterTable(any(), any());
+ verify(mockInnerCatalog).invalidateTable(Identifier.create("mydb",
"mytable"));
+ }
+
+ /** Verifies that successful Paimon dropTable invalidates the native cache.
*/
+ @Test
+ public void testDropTableInvalidatesNativeCacheAfterSuccessfulPurge() 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);
+ when(mockTableCatalog.purgeTable(any())).thenReturn(true);
+
+ TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog,
mockCatalog);
+ ObjectPath path = new ObjectPath("mydb", "mytable");
+ cat.dropTable(path, false);
+
+ Identifier expected = Identifier.create("mydb", "mytable");
+ verify(mockInnerCatalog).invalidateTable(expected);
+ }
+
+ //
---------------------------------------------------------------------------
+ // Helper: minimal CatalogFactory.Context implementation for constructor
+ //
---------------------------------------------------------------------------
+
+ private static class MockCatalogContext implements CatalogFactory.Context {
+ private final String name;
+ private final Map<String, String> options;
+
+ MockCatalogContext(String name, Map<String, String> options) {
+ this.name = name;
+ this.options = options;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public Map<String, String> getOptions() {
+ return options;
+ }
+
+ @Override
+ public ReadableConfig getConfiguration() {
+ return Configuration.fromMap(options);
+ }
+
+ @Override
+ public ClassLoader getClassLoader() {
+ return Thread.currentThread().getContextClassLoader();
+ }
+ }
+}
