feat(bigquery-jdbc): add Graph metadata support in getTables and getTableTypes methods (#14868)

b/570033146
b/570032972

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Author: Neenu1995

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryConnection.java

@@ -164,7 +164,9 @@ public class BigQueryConnection extends BigQueryNoOpsConnection {
               BigQueryJdbcUrlUtility.OAUTH_SA_IMPERSONATION_TOKEN_LIFETIME_PROPERTY_NAME,
               BigQueryJdbcUrlUtility.OAUTH_SA_EMAIL_PROPERTY_NAME,
               BigQueryJdbcUrlUtility.OAUTH_PVT_KEY_PATH_PROPERTY_NAME,
-              BigQueryJdbcUrlUtility.OAUTH_CLIENT_ID_PROPERTY_NAME)
+              BigQueryJdbcUrlUtility.OAUTH_CLIENT_ID_PROPERTY_NAME,
+              BigQueryJdbcUrlUtility.ENABLE_DIAGNOSTIC_TELEMETRY_PROPERTY_NAME,
+              BigQueryJdbcUrlUtility.ENABLE_GRAPH_METADATA_PROPERTY_NAME)
           .build();
   private HeaderProvider headerProvider;
   BigQueryReadClient bigQueryReadClient = null;

java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java

@@ -70,6 +70,7 @@
 import java.util.Comparator;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.Callable;
@@ -103,6 +104,8 @@ class BigQueryDatabaseMetaData implements DatabaseMetaData {
   private static final int DEFAULT_QUEUE_CAPACITY = 5000;
   private static final String GET_EXPORTED_KEYS_SQL = "DatabaseMetaData_GetExportedKeys.sql";
   private static String exportedKeysSqlContent;
+  private static final String GET_GRAPH_TABLES_SQL = "DatabaseMetaData_GetGraphTables.sql";
+  private static String graphTablesSqlContent;
   // Declared package-private for testing.
   static final String GOOGLE_SQL_QUOTED_IDENTIFIER = "`";
   // Does not include SQL:2003 Keywords as per JDBC spec.
@@ -138,6 +141,24 @@ class BigQueryDatabaseMetaData implements DatabaseMetaData {
   static final String GOOGLE_SQL_CATALOG_SEPARATOR = ".";
   static final int GOOGLE_SQL_MAX_COL_NAME_LEN = 300;
   static final int GOOGLE_SQL_MAX_COLS_PER_TABLE = 10000;
+  private static final String[] DEFAULT_TABLE_TYPES = {
+    "EXTERNAL", "MATERIALIZED VIEW", "SNAPSHOT", "TABLE", "VIEW"
+  };
+  private static final String[] TABLE_TYPES_WITH_GRAPH = {
+    "EXTERNAL", "GRAPH TABLE", "MATERIALIZED VIEW", "SNAPSHOT", "TABLE", "VIEW"
+  };
+  static final String GRAPH_TABLE_TYPE = "GRAPH TABLE";
+  private static final String DEFAULT_GRAPH_REGION = "us";
+  // Google Cloud project IDs: letters, digits, hyphens, and optional domain prefix (e.g.
+  // domain.com:proj-1)
+  private static final Pattern VALID_CATALOG_PATTERN =
+      Pattern.compile("^[A-Za-z0-9][A-Za-z0-9:._-]*$");
+  // BigQuery dataset IDs: letters, digits, and underscores only
+  private static final Pattern VALID_SCHEMA_PATTERN = Pattern.compile("^[A-Za-z0-9_]+$");
+  // BigQuery region qualifier for regional INFORMATION_SCHEMA queries (e.g. region-us,
+  // region-us-central1)
+  private static final Pattern VALID_REGION_SCOPE_PATTERN =
+      Pattern.compile("^region-[A-Za-z0-9-]+$");
 
   String URL;
   BigQueryConnection connection;
@@ -1308,19 +1329,37 @@ private ResultSet getTablesImpl(
             () -> {
               final List<FieldValueList> collectedResults =
                   Collections.synchronizedList(new ArrayList<>());
-              processTargetTablesConcurrently(
-                  targetDatasets,
+              boolean onlyGraphTablesRequested =
+                  requestedTypes != null
+                      && requestedTypes.size() == 1
+                      && requestedTypes.contains(GRAPH_TABLE_TYPE);
+              boolean isExactGraphTableName =
+                  tableNamePattern != null
+                      && !tableNamePattern.contains("%")
+                      && tableNamePattern.contains(".");
+              if (!isExactGraphTableName && !onlyGraphTablesRequested) {
+                processTargetTablesConcurrently(
+                    targetDatasets,
+                    tableNamePattern,
+                    true,
+                    false,
+                    false,
+                    collectedResults,
+                    resultSchemaFields,
+                    (bqTable, results, fields) -> {
+                      processTableInfo(bqTable, requestedTypes, results, fields);
+                    },
+                    TableOption.fields(
+                        TableField.TABLE_REFERENCE, TableField.TYPE, TableField.DESCRIPTION));
+              }
+
+              appendGraphTableRows(
+                  effectiveCatalog,
+                  effectiveSchemaPattern,
                   tableNamePattern,
-                  true,
-                  false,
-                  false,
+                  requestedTypes,
                   collectedResults,
-                  resultSchemaFields,
-                  (bqTable, results, fields) -> {
-                    processTableInfo(bqTable, requestedTypes, results, fields);
-                  },
-                  TableOption.fields(
-                      TableField.TABLE_REFERENCE, TableField.TYPE, TableField.DESCRIPTION));
+                  resultSchemaFields);
 
               Comparator<FieldValueList> comparator = defineGetTablesComparator(resultSchemaFields);
               sortResults(collectedResults, comparator, "getTables", LOG);
@@ -1441,6 +1480,123 @@ Comparator<FieldValueList> defineGetTablesComparator(FieldList resultSchemaField
             Comparator.nullsFirst(String::compareTo));
   }
 
+  static void validateCatalogIdentifier(String catalog) throws SQLException {
+    if (catalog == null || !VALID_CATALOG_PATTERN.matcher(catalog).matches()) {
+      throw new BigQueryJdbcException(
+          String.format("Invalid catalog (project ID) identifier: '%s'", catalog));
+    }
+  }
+
+  static void validateSchemaIdentifier(String schema) throws SQLException {
+    if (schema == null || !VALID_SCHEMA_PATTERN.matcher(schema).matches()) {
+      throw new BigQueryJdbcException(
+          String.format("Invalid schema (dataset ID) identifier: '%s'", schema));
+    }
+  }
+
+  static void validateRegionScopeIdentifier(String regionScope) throws SQLException {
+    if (regionScope == null || !VALID_REGION_SCOPE_PATTERN.matcher(regionScope).matches()) {
+      throw new BigQueryJdbcException(
+          String.format("Invalid region scope identifier: '%s'", regionScope));
+    }
+  }
+
+  boolean shouldSkipGraphTables(String tableNamePattern, Set<String> requestedTypes) {
+    if (this.connection == null || !this.connection.isEnableGraphMetadata()) {
+      return true;
+    }
+    if (requestedTypes != null && !requestedTypes.contains(GRAPH_TABLE_TYPE)) {
+      return true;
+    }
+    if (tableNamePattern != null
+        && !tableNamePattern.contains("%")
+        // If table name has a dot it is expected to be a graph table name.
+        && !tableNamePattern.contains(".")) {
+      return true;
+    }
+    return false;
+  }
+
+  String resolveGraphRegion() throws SQLException {
+    String location = this.connection != null ? this.connection.getLocation() : null;
+    if (location == null || location.trim().isEmpty()) {
+      location = DEFAULT_GRAPH_REGION;
+      LOG.info(String.format("Deafulting to Graph Location %s", DEFAULT_GRAPH_REGION));
+    }
+    String normalized = location.trim().toLowerCase(Locale.US);
+    return normalized.startsWith("region-") ? normalized : "region-" + normalized;
+  }
+
+  void appendGraphTableRows(
+      String effectiveCatalog,
+      String effectiveSchemaPattern,
+      String tableNamePattern,
+      Set<String> requestedTypes,
+      List<FieldValueList> collectedResults,
+      FieldList resultSchemaResults)
+      throws SQLException {
+    if (shouldSkipGraphTables(tableNamePattern, requestedTypes)) {
+      return;
+    }
+
+    LOG.info("Appending Graph table rows");
+
+    List<String> targetCatalogs =
+        effectiveCatalog != null
+            ? Collections.singletonList(effectiveCatalog)
+            : getAccessibleCatalogNames();
+    boolean isExactDataset =
+        effectiveSchemaPattern != null && !effectiveSchemaPattern.contains("%");
+    String effectiveTablePattern = (tableNamePattern == null) ? "%" : tableNamePattern;
+    String scope = isExactDataset ? effectiveSchemaPattern : resolveGraphRegion();
+    if (isExactDataset) {
+      validateSchemaIdentifier(scope);
+    } else {
+      validateRegionScopeIdentifier(scope);
+    }
+    for (String project : targetCatalogs) {
+      validateCatalogIdentifier(project);
+      String formattedSql = replaceSqlParameters(getGraphTablesSqlContent(), project, scope);
+
+      try (PreparedStatement statement = this.connection.prepareStatement(formattedSql)) {
+        statement.setString(1, effectiveSchemaPattern == null ? "%" : effectiveSchemaPattern);
+        statement.setString(2, effectiveTablePattern);
+
+        try (ResultSet rs = statement.executeQuery()) {
+          while (rs.next()) {
+            String tableCat = rs.getString("TABLE_CAT");
+            String tableSchem = rs.getString("TABLE_SCHEM");
+            String tableName = rs.getString("TABLE_NAME");
+            String sourceTable = rs.getString("SOURCE_TABLE");
+            String remarks = sourceTable != null ? "Source: " + sourceTable : null;
+
+            List<FieldValue> values = new ArrayList<>(resultSchemaResults.size());
+            values.add(createStringFieldValue(tableCat)); // 1. TABLE_CAT
+            values.add(createStringFieldValue(tableSchem)); // 2. TABLE_SCHEM
+            values.add(createStringFieldValue(tableName)); // 3. TABLE_NAME
+            values.add(createStringFieldValue(GRAPH_TABLE_TYPE)); // 4. TABLE_TYPE
+            values.add(createStringFieldValue(remarks)); // 5. REMARKS
+            values.add(createNullFieldValue()); // 6. TYPE_CAT (always null)
+            values.add(createNullFieldValue()); // 7. TYPE_SCHEM (always null)
+            values.add(createNullFieldValue()); // 8. TYPE_NAME (always null)
+            values.add(createNullFieldValue()); // 9. SELF_REFERENCING_COL_NAME (always null)
+            values.add(createNullFieldValue()); // 10. REF_GENERATION (always null)
+            collectedResults.add(FieldValueList.of(values, resultSchemaResults));
+          }
+        }
+      } catch (SQLException e) {
+        if (e.getMessage() != null && e.getMessage().contains("Not found")) {
+          LOG.info(
+              "Dataset or project not found while querying PROPERTY_GRAPHS in project '%s'."
+                  + " Skipping.",
+              project);
+        } else {
+          throw e;
+        }
+      }
+    }
+  }
+
   @Override
   public ResultSet getSchemas() throws SQLException {
     LOG.info("getSchemas() called");
@@ -1488,7 +1644,8 @@ public ResultSet getTableTypes() throws SQLException {
     LOG.info("getTableTypes() called");
 
     final Schema tableTypesSchema = defineGetTableTypesSchema();
-    final List<FieldValueList> tableTypeRows = prepareGetTableTypesRows(tableTypesSchema);
+    final List<FieldValueList> tableTypeRows =
+        prepareGetTableTypesRows(tableTypesSchema, this.connection.isEnableGraphMetadata());
 
     BlockingQueue<BigQueryFieldValueListWrapper> queue =
         fetchAndPopulateQueue(() -> tableTypeRows, tableTypesSchema.getFields());
@@ -1504,9 +1661,13 @@ static Schema defineGetTableTypesSchema() {
   }
 
   static List<FieldValueList> prepareGetTableTypesRows(Schema schema) {
-    final String[] tableTypes = {"EXTERNAL", "MATERIALIZED VIEW", "SNAPSHOT", "TABLE", "VIEW"};
-    List<FieldValueList> rows = new ArrayList<>(tableTypes.length);
+    return prepareGetTableTypesRows(schema, false);
+  }
+
+  static List<FieldValueList> prepareGetTableTypesRows(Schema schema, boolean includeGraphTables) {
+    final String[] tableTypes = includeGraphTables ? TABLE_TYPES_WITH_GRAPH : DEFAULT_TABLE_TYPES;
     FieldList schemaFields = schema.getFields();
+    List<FieldValueList> rows = new ArrayList<>(tableTypes.length);
 
     for (String typeName : tableTypes) {
       FieldValue fieldValue = FieldValue.of(FieldValue.Attribute.PRIMITIVE, typeName);
@@ -2195,6 +2356,8 @@ public ResultSet getExportedKeys(String catalog, String schema, String table)
       return BigQueryJsonResultSet.of(resultSchema, -1, queue, null);
     }
 
+    validateCatalogIdentifier(catalog);
+    validateSchemaIdentifier(schema);
     String sql = getExportedKeysSqlContent();
     String formattedSql = replaceSqlParameters(sql, catalog, schema, table);
     PreparedStatement stmt = this.connection.prepareStatement(formattedSql);
@@ -4853,6 +5016,13 @@ private static synchronized String getExportedKeysSqlContent() {
     return exportedKeysSqlContent;
   }
 
+  static synchronized String getGraphTablesSqlContent() {
+    if (graphTablesSqlContent == null) {
+      graphTablesSqlContent = readSqlFromFile(GET_GRAPH_TABLES_SQL);
+    }
+    return graphTablesSqlContent;
+  }
+
   static String readSqlFromFile(String filename) {
     try (InputStream in = BigQueryDatabaseMetaData.class.getResourceAsStream(filename)) {
       if (in == null) {

java-bigquery-jdbc/src/main/resources/com/google/cloud/bigquery/jdbc/DatabaseMetaData_GetGraphTables.sql

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed 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
+ *
+ * https://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.
+ */
+
+SELECT
+  property_graph_catalog AS TABLE_CAT,
+  property_graph_schema AS TABLE_SCHEM,
+  CONCAT(property_graph_name, '.', JSON_VALUE(t, '$.name')) AS TABLE_NAME,
+  CONCAT(
+    JSON_VALUE(t, '$.dataSourceTable.projectId'), '.',
+    JSON_VALUE(t, '$.dataSourceTable.datasetId'), '.',
+    JSON_VALUE(t, '$.dataSourceTable.tableId')) AS SOURCE_TABLE
+FROM `%1$s.%2$s.INFORMATION_SCHEMA.PROPERTY_GRAPHS`,
+UNNEST(ARRAY_CONCAT(
+  IFNULL(JSON_QUERY_ARRAY(property_graph_metadata_json, '$.nodeTables'), []),
+  IFNULL(JSON_QUERY_ARRAY(property_graph_metadata_json, '$.edgeTables'), []))) AS t
+WHERE LOWER(property_graph_schema) LIKE LOWER(?)
+  AND LOWER(CONCAT(property_graph_name, '.', JSON_VALUE(t, '$.name'))) LIKE LOWER(?)