BigQuery: Add table validity check for BigQueryMetastoreCatalog

bigquery/src/main/java/org/apache/iceberg/gcp/bigquery/BigQueryMetastoreCatalog.java

@@ -280,6 +280,16 @@ public Map<String, String> loadNamespaceMetadata(Namespace namespace) {
     }
   }
 
+  @Override
+  protected boolean isValidIdentifier(TableIdentifier identifier) {
+    try {
+      validateNamespace(identifier.namespace());
+    } catch (IllegalArgumentException e) {
+      return false;
+    }
+    return true;
+  }
+
   @Override
   public String name() {
     return catalogName;

bigquery/src/test/java/org/apache/iceberg/gcp/bigquery/TestBigQueryCatalog.java

@@ -21,11 +21,19 @@
 import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE;
 import static org.apache.iceberg.CatalogUtil.ICEBERG_CATALOG_TYPE_BIGQUERY;
 import static org.apache.iceberg.gcp.bigquery.BigQueryMetastoreCatalog.PROJECT_ID;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.MetadataTableUtils;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableMetadata;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.catalog.CatalogTests;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
@@ -39,6 +47,8 @@
 import org.junit.jupiter.api.io.TempDir;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
 
 public class TestBigQueryCatalog extends CatalogTests<BigQueryMetastoreCatalog> {
   @TempDir private File tempFolder;
@@ -144,9 +154,61 @@ public void createTableTransaction(int formatVersion) {
   @Test
   public void testCreateTableWithDefaultColumnValue() {}
 
-  @Disabled("BigQuery Metastore does not support multi layer namespaces")
   @Test
-  public void testLoadMetadataTable() {}
+  public void testLoadMetadataTable() {
+
+    // Create a spy of the catalog to verify method calls
+    BigQueryMetastoreCatalog spyCatalog = spy(catalog);
+
+    // Create mock objects
+    TableOperations mockOps = Mockito.mock(TableOperations.class);
+    TableMetadata mockMetadata = Mockito.mock(TableMetadata.class);
+    Table mockMetadataTable = Mockito.mock(Table.class);
+
+    // Mock the table operations to return metadata (indicating base table exists)
+    when(mockOps.current()).thenReturn(mockMetadata);
+
+    // Create the expected base table identifier that will be extracted from the metadata table
+    // identifier
+    TableIdentifier expectedBaseTableId = TableIdentifier.of("dataset1", "table1");
+    when(spyCatalog.newTableOps(expectedBaseTableId)).thenReturn(mockOps);
+
+    // Use MockedStatic to mock the static MetadataTableUtils.createMetadataTableInstance call
+    try (MockedStatic<MetadataTableUtils> mockedUtils =
+        Mockito.mockStatic(MetadataTableUtils.class)) {
+      mockedUtils
+          .when(
+              () ->
+                  MetadataTableUtils.createMetadataTableInstance(
+                      any(TableOperations.class),
+                      any(String.class),
+                      any(TableIdentifier.class),
+                      any(TableIdentifier.class),
+                      any()))
+          .thenReturn(mockMetadataTable);
+
+      // Create a metadata table identifier
+      TableIdentifier metadataTableId =
+          TableIdentifier.of(Namespace.of("dataset1", "table1"), "partitions");
+
+      // Call loadTable which should trigger the metadata table loading path
+      Table result = spyCatalog.loadTable(metadataTableId);
+
+      // Verify that MetadataTableUtils.createMetadataTableInstance was called
+      // This confirms that loadMetadataTable path was taken
+      mockedUtils.verify(
+          () ->
+              MetadataTableUtils.createMetadataTableInstance(
+                  any(TableOperations.class),
+                  any(String.class),
+                  any(TableIdentifier.class),
+                  any(TableIdentifier.class),
+                  any()));
+
+      // Verify the result is the mocked metadata table
+      assertThat(result).isSameAs(mockMetadataTable);
+    }
+  }
 
   @Disabled("BigQuery Metastore does not support rename tables")
   @Test
@@ -171,4 +233,30 @@ public void renameTableNamespaceMissing() {
   public void testRenameTableMissingSourceTable() {
     super.testRenameTableMissingSourceTable();
   }
+
+  @Test
+  public void testIsValidIdentifierWithValidSingleLevelNamespace() {
+    TableIdentifier validIdentifier = TableIdentifier.of("dataset1", "table1");
+    assertThat(catalog.isValidIdentifier(validIdentifier)).isTrue();
+  }
+
+  @Test
+  public void testIsValidIdentifierWithInvalidMultiLevelNamespace() {
+    TableIdentifier invalidIdentifier =
+        TableIdentifier.of(Namespace.of("level1", "level2"), "table1");
+    assertThat(catalog.isValidIdentifier(invalidIdentifier)).isFalse();
+  }
+
+  @Test
+  public void testIsValidIdentifierWithThreeLevelNamespace() {
+    TableIdentifier invalidIdentifier =
+        TableIdentifier.of(Namespace.of("level1", "level2", "level3"), "table1");
+    assertThat(catalog.isValidIdentifier(invalidIdentifier)).isFalse();
+  }
+
+  @Test
+  public void testIsValidIdentifierWithEmptyNamespace() {
+    TableIdentifier invalidIdentifier = TableIdentifier.of(Namespace.empty(), "table1");
+    assertThat(catalog.isValidIdentifier(invalidIdentifier)).isFalse();
+  }
 }