Copilot commented on code in PR #10475:
URL: https://github.com/apache/gravitino/pull/10475#discussion_r3339974373


##########
flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.Map;
+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.gravitino.flink.connector.DefaultPartitionConverter;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+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;
+  private GravitinoPaimonCatalog catalog;
+
+  // 
---------------------------------------------------------------------------
+  // 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 {
+
+    TestableBaseCatalog() {
+      super(
+          "test-catalog",
+          Collections.emptyMap(),
+          "default",
+          PaimonPropertiesConverter.INSTANCE,
+          DefaultPartitionConverter.INSTANCE);
+    }
+
+    @Override
+    protected AbstractCatalog realCatalog() {
+      return mock(AbstractCatalog.class);
+    }
+
+    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 the internal {@code 
paimonCatalog} field is
+   * replaced with a Mockito mock, enabling white-box verification without a 
real Paimon/Hive
+   * environment.
+   */
+  private static class TestablePaimonCatalog extends GravitinoPaimonCatalog {
+
+    private final AbstractCatalog injectedPaimon;
+
+    TestablePaimonCatalog(AbstractCatalog injectedPaimon) {
+      // 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", Collections.emptyMap()),
+          "default",
+          PaimonPropertiesConverter.INSTANCE,
+          DefaultPartitionConverter.INSTANCE);
+      this.injectedPaimon = injectedPaimon;
+    }
+
+    @Override
+    protected AbstractCatalog realCatalog() {
+      return injectedPaimon;
+    }
+  }
+
+  @BeforeEach
+  void setUp() {
+    mockPaimonCatalog = mock(AbstractCatalog.class);
+    catalog = new TestablePaimonCatalog(mockPaimonCatalog);
+  }
+
+  // 
---------------------------------------------------------------------------
+  // 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);
+    CatalogBaseTable result = catalog.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);
+
+    CatalogException ex =
+        Assertions.assertThrows(
+            CatalogException.class, () -> 
catalog.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 testPaimonCatalogNotCalledWhenEnrichIsNotReached() throws 
TableNotExistException {
+    // The hook is not called → paimonCatalog.getTable() is never invoked
+    verify(mockPaimonCatalog, never()).getTable(any());
+  }

Review Comment:
   This test is vacuous: it performs no action and simply verifies that 
`mockPaimonCatalog.getTable(any())` was never called. Since `mockPaimonCatalog` 
is a freshly-mocked instance with zero interactions in this method, the 
verification is tautological and will always pass regardless of the production 
code's behavior. Either remove the test or rewrite it to drive a real scenario 
(e.g., invoke `catalog.getTable(...)` after stubbing the Gravitino auth path to 
fail, and then assert the Paimon mock was not consulted).



##########
flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.Map;
+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.gravitino.flink.connector.DefaultPartitionConverter;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+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;
+  private GravitinoPaimonCatalog catalog;
+
+  // 
---------------------------------------------------------------------------
+  // 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 {
+
+    TestableBaseCatalog() {
+      super(
+          "test-catalog",
+          Collections.emptyMap(),
+          "default",
+          PaimonPropertiesConverter.INSTANCE,
+          DefaultPartitionConverter.INSTANCE);
+    }
+
+    @Override
+    protected AbstractCatalog realCatalog() {
+      return mock(AbstractCatalog.class);
+    }

Review Comment:
   `TestableBaseCatalog.realCatalog()` returns a freshly-created Mockito mock 
on every invocation. If the parent `open()`/`close()` lifecycle (or any other 
method that calls `realCatalog()` more than once) is ever exercised through 
this test double, the calls will hit different mock instances and 
verifications/stubbing become impossible. Construct the mock once in the 
constructor (or in the field declaration) and return that single instance.



##########
flink-connector/flink-common/src/test/java/org/apache/gravitino/flink/connector/paimon/TestGravitinoPaimonCatalog.java:
##########
@@ -0,0 +1,308 @@
+/*
+ * 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.Map;
+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.gravitino.flink.connector.DefaultPartitionConverter;
+import org.apache.gravitino.flink.connector.catalog.BaseCatalog;
+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;
+  private GravitinoPaimonCatalog catalog;
+
+  // 
---------------------------------------------------------------------------
+  // 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 {
+
+    TestableBaseCatalog() {
+      super(
+          "test-catalog",
+          Collections.emptyMap(),
+          "default",
+          PaimonPropertiesConverter.INSTANCE,
+          DefaultPartitionConverter.INSTANCE);
+    }
+
+    @Override
+    protected AbstractCatalog realCatalog() {
+      return mock(AbstractCatalog.class);
+    }
+
+    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 the internal {@code 
paimonCatalog} field is
+   * replaced with a Mockito mock, enabling white-box verification without a 
real Paimon/Hive
+   * environment.
+   */
+  private static class TestablePaimonCatalog extends GravitinoPaimonCatalog {
+
+    private final AbstractCatalog injectedPaimon;
+
+    TestablePaimonCatalog(AbstractCatalog injectedPaimon) {
+      // 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", Collections.emptyMap()),
+          "default",
+          PaimonPropertiesConverter.INSTANCE,
+          DefaultPartitionConverter.INSTANCE);
+      this.injectedPaimon = injectedPaimon;
+    }
+
+    @Override
+    protected AbstractCatalog realCatalog() {
+      return injectedPaimon;
+    }
+  }
+
+  @BeforeEach
+  void setUp() {
+    mockPaimonCatalog = mock(AbstractCatalog.class);
+    catalog = new TestablePaimonCatalog(mockPaimonCatalog);
+  }
+
+  // 
---------------------------------------------------------------------------
+  // 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);
+    CatalogBaseTable result = catalog.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);
+
+    CatalogException ex =
+        Assertions.assertThrows(
+            CatalogException.class, () -> 
catalog.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 testPaimonCatalogNotCalledWhenEnrichIsNotReached() throws 
TableNotExistException {
+    // The hook is not called → paimonCatalog.getTable() is never invoked
+    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
+   * org.apache.paimon.flink.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);
+    org.apache.paimon.flink.FlinkCatalog mockFlinkCatalog =
+        mock(org.apache.paimon.flink.FlinkCatalog.class);
+    when(mockFlinkCatalog.catalog()).thenReturn(mockInnerCatalog);
+
+    TestablePaimonCatalog cat = new TestablePaimonCatalog(mockFlinkCatalog);
+    ObjectPath path = new ObjectPath("mydb", "mytable");
+    cat.invalidateNativeTableCache(path);
+
+    org.apache.paimon.catalog.Identifier expected =
+        org.apache.paimon.catalog.Identifier.create("mydb", "mytable");

Review Comment:
   Per the project's import convention (AGENTS.md), fully qualified class names 
should not be used inline; add normal imports instead. Several types in this 
test file are referenced via FQN (e.g. `org.apache.paimon.catalog.Catalog`, 
`org.apache.paimon.flink.FlinkCatalog`, `org.apache.paimon.catalog.Identifier` 
at lines 247-258; `org.apache.flink.table.factories.CatalogFactory.Context` at 
line 279; `org.apache.flink.configuration.ReadableConfig` / 
`org.apache.flink.configuration.Configuration` at lines 299-300). Please 
replace these with proper `import` statements at the top of the file.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]

Reply via email to