From 039a3a60976182e3ecf2eb1daa64b805cce2931c Mon Sep 17 00:00:00 2001 From: Keshav Dandeva Date: Wed, 8 Jul 2026 21:38:55 +0000 Subject: [PATCH 1/5] feat(bigquery-jdbc): migrate `getExportedKeys` to BQ API --- .../jdbc/BigQueryDatabaseMetaData.java | 149 +++++++++++------- .../jdbc/DatabaseMetaData_GetExportedKeys.sql | 71 --------- .../jdbc/it/ITDatabaseMetadataTest.java | 49 +++++- 3 files changed, 136 insertions(+), 133 deletions(-) delete mode 100644 java-bigquery-jdbc/src/main/resources/com/google/cloud/bigquery/jdbc/DatabaseMetaData_GetExportedKeys.sql diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java index 75414771600c..32b35ff6aa29 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java @@ -52,15 +52,11 @@ import com.google.cloud.bigquery.exception.BigQueryJdbcException; import com.google.cloud.bigquery.jdbc.BigQueryJdbcTypeMappings.ColumnTypeInfo; import com.google.cloud.bigquery.jdbc.utils.BigQueryJdbcVersionUtility; -import java.io.BufferedReader; -import java.io.InputStream; -import java.io.InputStreamReader; import java.sql.Connection; import java.sql.DatabaseMetaData; import java.sql.ResultSet; import java.sql.RowIdLifetime; import java.sql.SQLException; -import java.sql.Statement; import java.sql.Types; import java.util.ArrayList; import java.util.Arrays; @@ -68,7 +64,6 @@ import java.util.Comparator; import java.util.HashSet; import java.util.List; -import java.util.Scanner; import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.Callable; @@ -89,17 +84,14 @@ * * @see BigQueryStatement */ -// TODO(neenu): test and verify after post MVP implementation. class BigQueryDatabaseMetaData implements DatabaseMetaData { final BigQueryJdbcCustomLogger LOG = new BigQueryJdbcCustomLogger(this.toString()); private static final String DATABASE_PRODUCT_NAME = "Google BigQuery"; private static final String DATABASE_PRODUCT_VERSION = "2.0"; private static final String DRIVER_NAME = "GoogleJDBCDriverForGoogleBigQuery"; - private static final String SCHEMA_TERM = "Dataset"; private static final String CATALOG_TERM = "Project"; private static final String PROCEDURE_TERM = "Procedure"; - private static final String GET_EXPORTED_KEYS_SQL = "DatabaseMetaData_GetExportedKeys.sql"; private static final int DEFAULT_PAGE_SIZE = 500; private static final int DEFAULT_QUEUE_CAPACITY = 5000; // Declared package-private for testing. @@ -2413,17 +2405,6 @@ Schema defineGetVersionColumnsSchema() { return Schema.of(fields); } - private void closeStatementIgnoreException(Statement statement) { - if (statement == null) { - return; - } - try { - statement.close(); - } catch (SQLException e) { - // pass - } - } - @Override public ResultSet getPrimaryKeys(String catalog, String schema, String table) throws SQLException { if ((catalog != null && catalog.isEmpty()) @@ -2569,16 +2550,51 @@ public ResultSet getImportedKeys(String catalog, String schema, String table) @Override public ResultSet getExportedKeys(String catalog, String schema, String table) throws SQLException { - String sql = readSqlFromFile(GET_EXPORTED_KEYS_SQL); - Statement stmt = this.connection.createStatement(); - try { - stmt.closeOnCompletion(); - String formattedSql = replaceSqlParameters(sql, catalog, schema, table); - return stmt.executeQuery(formattedSql); - } catch (SQLException e) { - closeStatementIgnoreException(stmt); - throw new BigQueryJdbcException("Error executing getExportedKeys", e); + if ((catalog != null && catalog.isEmpty()) + || (schema != null && schema.isEmpty()) + || table == null + || table.isEmpty()) { + LOG.warning( + "Returning empty ResultSet as required parameters are null/empty, or catalog/schema parameters are empty."); + return new BigQueryJsonResultSet(); } + + final Schema resultSchema = defineForeignKeyResultSetSchema(); + final FieldList resultSchemaFields = resultSchema.getFields(); + + final List collectedResults = Collections.synchronizedList(new ArrayList<>()); + List targetDatasets = getTargetDatasets(catalog, schema); + + boolean ignoreAccessErrors = (catalog == null); + processTargetTablesConcurrently( + targetDatasets, + null, + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + (bqTable, results, fields) -> { + TableConstraints constraints = bqTable.getTableConstraints(); + if (constraints != null && constraints.getForeignKeys() != null) { + for (ForeignKey fk : constraints.getForeignKeys()) { + TableId pkTableId = fk.getReferencedTable(); + if (pkTableId != null + && equalsOrNullMatchesAll(catalog, pkTableId.getProject()) + && equalsOrNullMatchesAll(schema, pkTableId.getDataset()) + && table.equals(pkTableId.getTable())) { + processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields); + } + } + } + }); + + Comparator comparator = defineFkTableSortComparator(resultSchemaFields); + sortResults(collectedResults, comparator, "getExportedKeys", LOG); + + final BlockingQueue queue = + new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); + populateQueue(collectedResults, queue, resultSchemaFields); + signalEndOfData(queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @Override @@ -5063,24 +5079,6 @@ private boolean equalsOrNullMatchesAll(String expected, String actual) { return expected == null || expected.equals(actual); } - static String readSqlFromFile(String filename) { - InputStream in; - in = BigQueryDatabaseMetaData.class.getResourceAsStream(filename); - BufferedReader reader = new BufferedReader(new InputStreamReader(in)); - StringBuilder builder = new StringBuilder(); - try (Scanner scanner = new Scanner(reader)) { - while (scanner.hasNextLine()) { - String line = scanner.nextLine(); - builder.append(line).append("\n"); - } - } - return builder.toString(); - } - - String replaceSqlParameters(String sql, String... params) throws SQLException { - return String.format(sql, (Object[]) params); - } - private void writeErrorToQueue(BlockingQueue queue, Throwable t) { Exception ex = (t instanceof Exception) ? (Exception) t : new Exception(t); BigQueryFieldValueListWrapper element = BigQueryFieldValueListWrapper.ofError(ex); @@ -5168,7 +5166,7 @@ private void processTargetTablesConcurrently( boolean ignoreAccessErrors, TableProcessor processor) throws SQLException { - if (targetDatasets.size() == 1) { + if (targetDatasets.size() == 1 && tableName != null) { processSingleTable( targetDatasets.get(0), tableName, @@ -5184,18 +5182,51 @@ private void processTargetTablesConcurrently( try { for (DatasetId datasetId : targetDatasets) { - taskFutures.add( - executor.submit( - () -> { - processSingleTable( - datasetId, - tableName, - collectedResults, - resultSchemaFields, - ignoreAccessErrors, - processor); - return null; - })); + if (tableName != null) { + taskFutures.add( + executor.submit( + () -> { + processSingleTable( + datasetId, + tableName, + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + })); + } else { + Page tablesPage; + try { + tablesPage = + bigquery.listTables(datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)); + } catch (BigQueryException e) { + if (ignoreAccessErrors && (e.getCode() == 404 || e.getCode() == 403)) { + LOG.info( + String.format( + "Dataset '%s' not found/accessible in project '%s' (API error %d). Skipping.", + datasetId.getDataset(), datasetId.getProject(), e.getCode())); + continue; + } + throw new SQLException("Error while listing tables: " + e.getMessage(), e); + } + if (tablesPage != null) { + for (Table table : tablesPage.iterateAll()) { + taskFutures.add( + executor.submit( + () -> { + processSingleTable( + datasetId, + table.getTableId().getTable(), + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + })); + } + } + } } waitForTasksCompletion(taskFutures); if (Thread.currentThread().isInterrupted()) { diff --git a/java-bigquery-jdbc/src/main/resources/com/google/cloud/bigquery/jdbc/DatabaseMetaData_GetExportedKeys.sql b/java-bigquery-jdbc/src/main/resources/com/google/cloud/bigquery/jdbc/DatabaseMetaData_GetExportedKeys.sql deleted file mode 100644 index 4058f6bff60a..000000000000 --- a/java-bigquery-jdbc/src/main/resources/com/google/cloud/bigquery/jdbc/DatabaseMetaData_GetExportedKeys.sql +++ /dev/null @@ -1,71 +0,0 @@ -/* - * Copyright 2024 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 PKTABLE_CAT, - PKTABLE_SCHEM, - PKTABLE_NAME, - PRIMARY.column_name AS PKCOLUMN_NAME, - FOREIGN.constraint_catalog AS FKTABLE_CAT, - FOREIGN.constraint_schema AS FKTABLE_SCHEM, - FOREIGN.table_name AS FKTABLE_NAME, - FOREIGN.column_name AS FKCOLUMN_NAME, - FOREIGN.ordinal_position AS KEY_SEQ, - NULL AS UPDATE_RULE, - NULL AS DELETE_RULE, - FOREIGN.constraint_name AS FK_NAME, - PRIMARY.constraint_name AS PK_NAME, - NULL AS DEFERRABILITY -FROM (SELECT DISTINCT CCU.table_catalog AS PKTABLE_CAT, - CCU.table_schema AS PKTABLE_SCHEM, - CCU.table_name AS PKTABLE_NAME, - TC.constraint_catalog, - TC.constraint_schema, - TC.constraint_name, - TC.table_catalog, - TC.table_schema, - TC.table_name, - TC.constraint_type, - KCU.column_name, - KCU.ordinal_position, - KCU.position_in_unique_constraint - FROM `%1$s.%2$s.INFORMATION_SCHEMA.TABLE_CONSTRAINTS` TC - INNER JOIN - `%1$s.%2$s.INFORMATION_SCHEMA.KEY_COLUMN_USAGE` KCU - USING - (constraint_catalog, - constraint_schema, - constraint_name, - table_catalog, - table_schema, - table_name) - INNER JOIN - `%1$s.%2$s.INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE` CCU - USING - (constraint_catalog, - constraint_schema, - constraint_name) - WHERE constraint_type = 'FOREIGN KEY') FOREIGN - INNER JOIN (SELECT * - FROM `%1$s.%2$s.INFORMATION_SCHEMA.KEY_COLUMN_USAGE` - WHERE position_in_unique_constraint IS NULL - AND RTRIM(table_name) = '%3$s') PRIMARY -ON - FOREIGN.PKTABLE_CAT = PRIMARY.table_catalog - AND FOREIGN.PKTABLE_SCHEM = PRIMARY.table_schema - AND FOREIGN.PKTABLE_NAME = PRIMARY.table_name - AND FOREIGN.position_in_unique_constraint = - PRIMARY.ordinal_position -ORDER BY FKTABLE_CAT, FKTABLE_SCHEM, FKTABLE_NAME, KEY_SEQ \ No newline at end of file diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java index e40ba8bccd8c..61ef3727d398 100644 --- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java @@ -24,7 +24,6 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; -import com.google.cloud.ServiceOptions; import java.sql.Connection; import java.sql.DatabaseMetaData; import java.sql.DriverManager; @@ -48,7 +47,7 @@ import org.junit.jupiter.api.Test; public class ITDatabaseMetadataTest extends ITBase { - static final String PROJECT_ID = ServiceOptions.getDefaultProjectId(); + static final String PROJECT_ID = "bigquery-devtools-drivers"; private static final Random random = new Random(); private static final int randomNumber = random.nextInt(9999); private static String DATASET; @@ -59,7 +58,7 @@ public class ITDatabaseMetadataTest extends ITBase { private static final String CONSTRAINTS_TABLE_NAME3 = "JDBC_CONSTRAINTS_TEST_TABLE3"; private static final Pattern VERSION_PATTERN = Pattern.compile("^(\\d+)\\.(\\d+)(?:\\.\\d+)+\\s*.*"); - private static final String DEFAULT_CATALOG = ServiceOptions.getDefaultProjectId(); + private static final String DEFAULT_CATALOG = "bigquery-devtools-drivers"; private static final String TABLE_NAME = "JDBC_DBMETADATA_TEST_TABLE" + randomNumber; @BeforeAll @@ -329,6 +328,50 @@ public void testTableConstraints() throws SQLException { connection.close(); } + @Test + public void testGetExportedKeys() throws SQLException { + try (Connection connection = DriverManager.getConnection(ITBase.connectionUrl)) { + DatabaseMetaData metaData = connection.getMetaData(); + + // Table 2 exports keys to Table + ResultSet exportedKeys2 = + metaData.getExportedKeys(PROJECT_ID, CONSTRAINTS_DATASET, CONSTRAINTS_TABLE_NAME2); + Assertions.assertTrue(exportedKeys2.next()); + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME2, exportedKeys2.getString(3)); // PKTABLE_NAME + Assertions.assertEquals("first_name", exportedKeys2.getString(4)); // PKCOLUMN_NAME + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME, exportedKeys2.getString(7)); // FKTABLE_NAME + Assertions.assertEquals("name", exportedKeys2.getString(8)); // FKCOLUMN_NAME + Assertions.assertEquals(1, exportedKeys2.getInt(9)); // KEY_SEQ + Assertions.assertEquals("my_fk", exportedKeys2.getString(12)); // FK_NAME + + Assertions.assertTrue(exportedKeys2.next()); + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME2, exportedKeys2.getString(3)); + Assertions.assertEquals("last_name", exportedKeys2.getString(4)); + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME, exportedKeys2.getString(7)); + Assertions.assertEquals("second_name", exportedKeys2.getString(8)); + Assertions.assertEquals(2, exportedKeys2.getInt(9)); + Assertions.assertEquals("my_fk", exportedKeys2.getString(12)); + Assertions.assertFalse(exportedKeys2.next()); + + // Table 3 exports keys to Table + ResultSet exportedKeys3 = + metaData.getExportedKeys(PROJECT_ID, CONSTRAINTS_DATASET, CONSTRAINTS_TABLE_NAME3); + Assertions.assertTrue(exportedKeys3.next()); + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME3, exportedKeys3.getString(3)); + Assertions.assertEquals("address", exportedKeys3.getString(4)); + Assertions.assertEquals(CONSTRAINTS_TABLE_NAME, exportedKeys3.getString(7)); + Assertions.assertEquals("address", exportedKeys3.getString(8)); + Assertions.assertEquals(1, exportedKeys3.getInt(9)); + Assertions.assertEquals("my_fk2", exportedKeys3.getString(12)); + Assertions.assertFalse(exportedKeys3.next()); + + // Table does not export keys to anything (it only imports them) + ResultSet exportedKeys1 = + metaData.getExportedKeys(PROJECT_ID, CONSTRAINTS_DATASET, CONSTRAINTS_TABLE_NAME); + Assertions.assertFalse(exportedKeys1.next()); + } + } + @Test public void testMetadataResultSetsDoNotInterfere() throws SQLException { try (Connection connection = DriverManager.getConnection(ITBase.connectionUrl)) { From 7c889f7061043964bf1c9fc9c31fd83efa192b0f Mon Sep 17 00:00:00 2001 From: Keshav Dandeva Date: Wed, 8 Jul 2026 22:31:30 +0000 Subject: [PATCH 2/5] resolve deadlock risk via async queue population --- .../jdbc/BigQueryDatabaseMetaData.java | 82 ++++++++++--------- .../jdbc/it/ITDatabaseMetadataTest.java | 5 +- 2 files changed, 47 insertions(+), 40 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java index 32b35ff6aa29..4488fc811559 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java @@ -1812,8 +1812,7 @@ public ResultSet getCatalogs() throws SQLException { final BlockingQueue queue = new LinkedBlockingQueue<>(catalogRows.isEmpty() ? 1 : catalogRows.size() + 1); - populateQueue(catalogRows, queue, schemaFields); - signalEndOfData(queue, schemaFields); + populateQueueAsync(catalogRows, queue, schemaFields); return BigQueryJsonResultSet.of( catalogsSchema, catalogRows.size(), queue, null, new Future[0]); @@ -1844,8 +1843,7 @@ public ResultSet getTableTypes() { BlockingQueue queue = new LinkedBlockingQueue<>(tableTypeRows.size() + 1); - populateQueue(tableTypeRows, queue, tableTypesSchema.getFields()); - signalEndOfData(queue, tableTypesSchema.getFields()); + populateQueueAsync(tableTypeRows, queue, tableTypesSchema.getFields()); return BigQueryJsonResultSet.of( tableTypesSchema, tableTypeRows.size(), queue, null, new Future[0]); @@ -2439,8 +2437,7 @@ public ResultSet getPrimaryKeys(String catalog, String schema, String table) thr final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueue(collectedResults, queue, resultSchemaFields); - signalEndOfData(queue, resultSchemaFields); + populateQueueAsync(collectedResults, queue, resultSchemaFields); return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @@ -2542,8 +2539,7 @@ public ResultSet getImportedKeys(String catalog, String schema, String table) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueue(collectedResults, queue, resultSchemaFields); - signalEndOfData(queue, resultSchemaFields); + populateQueueAsync(collectedResults, queue, resultSchemaFields); return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @@ -2563,7 +2559,7 @@ public ResultSet getExportedKeys(String catalog, String schema, String table) final FieldList resultSchemaFields = resultSchema.getFields(); final List collectedResults = Collections.synchronizedList(new ArrayList<>()); - List targetDatasets = getTargetDatasets(catalog, schema); + List targetDatasets = getTargetDatasets(catalog, null); boolean ignoreAccessErrors = (catalog == null); processTargetTablesConcurrently( @@ -2592,8 +2588,7 @@ && equalsOrNullMatchesAll(schema, pkTableId.getDataset()) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueue(collectedResults, queue, resultSchemaFields); - signalEndOfData(queue, resultSchemaFields); + populateQueueAsync(collectedResults, queue, resultSchemaFields); return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @@ -2651,8 +2646,7 @@ && equalsOrNullMatchesAll(parentSchema, pkTableId.getDataset()) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueue(collectedResults, queue, resultSchemaFields); - signalEndOfData(queue, resultSchemaFields); + populateQueueAsync(collectedResults, queue, resultSchemaFields); return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @@ -2669,8 +2663,7 @@ public ResultSet getTypeInfo() { final BlockingQueue queue = new LinkedBlockingQueue<>(typeInfoRows.size() + 1); - populateQueue(typeInfoRows, queue, schemaFields); - signalEndOfData(queue, schemaFields); + populateQueueAsync(typeInfoRows, queue, schemaFields); return BigQueryJsonResultSet.of( typeInfoSchema, typeInfoRows.size(), queue, null, new Future[0]); } @@ -3593,8 +3586,7 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce } Comparator comparator = defineGetSchemasComparator(resultSchemaFields); sortResults(collectedResults, comparator, "getSchemas", LOG); - populateQueue(collectedResults, queue, resultSchemaFields); - signalEndOfData(queue, resultSchemaFields); + populateQueueAsync(collectedResults, queue, resultSchemaFields); return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); } @@ -4907,6 +4899,19 @@ private void waitForTasksCompletion(List> taskFutures) throws Executio LOG.info("Finished waiting for tasks."); } + private void populateQueueAsync( + List collectedResults, + BlockingQueue queue, + FieldList resultSchemaFields) { + connection + .getMetadataExecutor() + .submit( + () -> { + populateQueue(collectedResults, queue, resultSchemaFields); + signalEndOfData(queue, resultSchemaFields); + }); + } + private void populateQueue( List collectedResults, BlockingQueue queue, @@ -5196,36 +5201,37 @@ private void processTargetTablesConcurrently( return null; })); } else { - Page
tablesPage; try { - tablesPage = + Page
tablesPage = bigquery.listTables(datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)); + if (tablesPage != null) { + for (Table table : tablesPage.iterateAll()) { + if (table.getDefinition() != null + && table.getDefinition().getType() == TableDefinition.Type.TABLE) { + taskFutures.add( + executor.submit( + () -> { + processSingleTable( + datasetId, + table.getTableId().getTable(), + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + })); + } + } + } } catch (BigQueryException e) { if (ignoreAccessErrors && (e.getCode() == 404 || e.getCode() == 403)) { LOG.info( - String.format( - "Dataset '%s' not found/accessible in project '%s' (API error %d). Skipping.", - datasetId.getDataset(), datasetId.getProject(), e.getCode())); + "Dataset '%s' not found/accessible in project '%s' (API error %d). Skipping.", + datasetId.getDataset(), datasetId.getProject(), e.getCode()); continue; } throw new SQLException("Error while listing tables: " + e.getMessage(), e); } - if (tablesPage != null) { - for (Table table : tablesPage.iterateAll()) { - taskFutures.add( - executor.submit( - () -> { - processSingleTable( - datasetId, - table.getTableId().getTable(), - collectedResults, - resultSchemaFields, - ignoreAccessErrors, - processor); - return null; - })); - } - } } } waitForTasksCompletion(taskFutures); diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java index 61ef3727d398..149858a91b4f 100644 --- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java @@ -24,6 +24,7 @@ import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import com.google.cloud.ServiceOptions; import java.sql.Connection; import java.sql.DatabaseMetaData; import java.sql.DriverManager; @@ -47,7 +48,7 @@ import org.junit.jupiter.api.Test; public class ITDatabaseMetadataTest extends ITBase { - static final String PROJECT_ID = "bigquery-devtools-drivers"; + static final String PROJECT_ID = ServiceOptions.getDefaultProjectId(); private static final Random random = new Random(); private static final int randomNumber = random.nextInt(9999); private static String DATASET; @@ -58,7 +59,7 @@ public class ITDatabaseMetadataTest extends ITBase { private static final String CONSTRAINTS_TABLE_NAME3 = "JDBC_CONSTRAINTS_TEST_TABLE3"; private static final Pattern VERSION_PATTERN = Pattern.compile("^(\\d+)\\.(\\d+)(?:\\.\\d+)+\\s*.*"); - private static final String DEFAULT_CATALOG = "bigquery-devtools-drivers"; + private static final String DEFAULT_CATALOG = ServiceOptions.getDefaultProjectId(); private static final String TABLE_NAME = "JDBC_DBMETADATA_TEST_TABLE" + randomNumber; @BeforeAll From 0c39a62746226a5905512ab3892616df005d4249 Mon Sep 17 00:00:00 2001 From: Keshav Dandeva Date: Wed, 8 Jul 2026 22:47:41 +0000 Subject: [PATCH 3/5] address pr feedback --- .../jdbc/BigQueryDatabaseMetaData.java | 39 ++++++++++--------- 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java index 4488fc811559..5264ca827da5 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java @@ -1812,10 +1812,9 @@ public ResultSet getCatalogs() throws SQLException { final BlockingQueue queue = new LinkedBlockingQueue<>(catalogRows.isEmpty() ? 1 : catalogRows.size() + 1); - populateQueueAsync(catalogRows, queue, schemaFields); + Future fetcherFuture = populateQueueAsync(catalogRows, queue, schemaFields); - return BigQueryJsonResultSet.of( - catalogsSchema, catalogRows.size(), queue, null, new Future[0]); + return BigQueryJsonResultSet.of(catalogsSchema, catalogRows.size(), queue, null, fetcherFuture); } Schema defineGetCatalogsSchema() { @@ -1843,10 +1842,11 @@ public ResultSet getTableTypes() { BlockingQueue queue = new LinkedBlockingQueue<>(tableTypeRows.size() + 1); - populateQueueAsync(tableTypeRows, queue, tableTypesSchema.getFields()); + Future fetcherFuture = + populateQueueAsync(tableTypeRows, queue, tableTypesSchema.getFields()); return BigQueryJsonResultSet.of( - tableTypesSchema, tableTypeRows.size(), queue, null, new Future[0]); + tableTypesSchema, tableTypeRows.size(), queue, null, fetcherFuture); } static Schema defineGetTableTypesSchema() { @@ -2437,8 +2437,8 @@ public ResultSet getPrimaryKeys(String catalog, String schema, String table) thr final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueueAsync(collectedResults, queue, resultSchemaFields); - return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + Future fetcherFuture = populateQueueAsync(collectedResults, queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); } private Schema defineGetPrimaryKeysSchema() { @@ -2539,8 +2539,8 @@ public ResultSet getImportedKeys(String catalog, String schema, String table) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueueAsync(collectedResults, queue, resultSchemaFields); - return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + Future fetcherFuture = populateQueueAsync(collectedResults, queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); } @Override @@ -2588,8 +2588,8 @@ && equalsOrNullMatchesAll(schema, pkTableId.getDataset()) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueueAsync(collectedResults, queue, resultSchemaFields); - return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + Future fetcherFuture = populateQueueAsync(collectedResults, queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); } @Override @@ -2646,8 +2646,8 @@ && equalsOrNullMatchesAll(parentSchema, pkTableId.getDataset()) final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - populateQueueAsync(collectedResults, queue, resultSchemaFields); - return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + Future fetcherFuture = populateQueueAsync(collectedResults, queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); } @Override @@ -2663,9 +2663,9 @@ public ResultSet getTypeInfo() { final BlockingQueue queue = new LinkedBlockingQueue<>(typeInfoRows.size() + 1); - populateQueueAsync(typeInfoRows, queue, schemaFields); + Future fetcherFuture = populateQueueAsync(typeInfoRows, queue, schemaFields); return BigQueryJsonResultSet.of( - typeInfoSchema, typeInfoRows.size(), queue, null, new Future[0]); + typeInfoSchema, typeInfoRows.size(), queue, null, fetcherFuture); } Schema defineGetTypeInfoSchema() { @@ -3586,8 +3586,8 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce } Comparator comparator = defineGetSchemasComparator(resultSchemaFields); sortResults(collectedResults, comparator, "getSchemas", LOG); - populateQueueAsync(collectedResults, queue, resultSchemaFields); - return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + Future fetcherFuture = populateQueueAsync(collectedResults, queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); } // Multi-Catalog Path: fan out using connection-scoped metadataExecutor @@ -4899,11 +4899,11 @@ private void waitForTasksCompletion(List> taskFutures) throws Executio LOG.info("Finished waiting for tasks."); } - private void populateQueueAsync( + private Future populateQueueAsync( List collectedResults, BlockingQueue queue, FieldList resultSchemaFields) { - connection + return connection .getMetadataExecutor() .submit( () -> { @@ -5230,6 +5230,7 @@ private void processTargetTablesConcurrently( datasetId.getDataset(), datasetId.getProject(), e.getCode()); continue; } + taskFutures.forEach(future -> future.cancel(true)); throw new SQLException("Error while listing tables: " + e.getMessage(), e); } } From ccef49e57faf03ec289c7d9c4e3be95772e0f72c Mon Sep 17 00:00:00 2001 From: Keshav Dandeva Date: Wed, 8 Jul 2026 22:56:14 +0000 Subject: [PATCH 4/5] nit: early returns --- .../jdbc/BigQueryDatabaseMetaData.java | 82 ++++++++++--------- 1 file changed, 43 insertions(+), 39 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java index 5264ca827da5..5c7094efcf2d 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java @@ -2570,16 +2570,18 @@ public ResultSet getExportedKeys(String catalog, String schema, String table) ignoreAccessErrors, (bqTable, results, fields) -> { TableConstraints constraints = bqTable.getTableConstraints(); - if (constraints != null && constraints.getForeignKeys() != null) { - for (ForeignKey fk : constraints.getForeignKeys()) { - TableId pkTableId = fk.getReferencedTable(); - if (pkTableId != null - && equalsOrNullMatchesAll(catalog, pkTableId.getProject()) - && equalsOrNullMatchesAll(schema, pkTableId.getDataset()) - && table.equals(pkTableId.getTable())) { - processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields); - } + if (constraints == null || constraints.getForeignKeys() == null) { + return; + } + for (ForeignKey fk : constraints.getForeignKeys()) { + TableId pkTableId = fk.getReferencedTable(); + if (pkTableId == null + || !equalsOrNullMatchesAll(catalog, pkTableId.getProject()) + || !equalsOrNullMatchesAll(schema, pkTableId.getDataset()) + || !table.equals(pkTableId.getTable())) { + continue; } + processForeignKey(fk, pkTableId, bqTable.getTableId(), results, fields); } }); @@ -5200,39 +5202,41 @@ private void processTargetTablesConcurrently( processor); return null; })); - } else { - try { - Page
tablesPage = - bigquery.listTables(datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)); - if (tablesPage != null) { - for (Table table : tablesPage.iterateAll()) { - if (table.getDefinition() != null - && table.getDefinition().getType() == TableDefinition.Type.TABLE) { - taskFutures.add( - executor.submit( - () -> { - processSingleTable( - datasetId, - table.getTableId().getTable(), - collectedResults, - resultSchemaFields, - ignoreAccessErrors, - processor); - return null; - })); - } + continue; + } + + try { + Page
tablesPage = + bigquery.listTables(datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)); + if (tablesPage != null) { + for (Table table : tablesPage.iterateAll()) { + if (table.getDefinition() == null + || table.getDefinition().getType() != TableDefinition.Type.TABLE) { + continue; } + taskFutures.add( + executor.submit( + () -> { + processSingleTable( + datasetId, + table.getTableId().getTable(), + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + })); } - } catch (BigQueryException e) { - if (ignoreAccessErrors && (e.getCode() == 404 || e.getCode() == 403)) { - LOG.info( - "Dataset '%s' not found/accessible in project '%s' (API error %d). Skipping.", - datasetId.getDataset(), datasetId.getProject(), e.getCode()); - continue; - } - taskFutures.forEach(future -> future.cancel(true)); - throw new SQLException("Error while listing tables: " + e.getMessage(), e); } + } catch (BigQueryException e) { + if (ignoreAccessErrors && (e.getCode() == 404 || e.getCode() == 403)) { + LOG.info( + "Dataset '%s' not found/accessible in project '%s' (API error %d). Skipping.", + datasetId.getDataset(), datasetId.getProject(), e.getCode()); + continue; + } + taskFutures.forEach(future -> future.cancel(true)); + throw new SQLException("Error while listing tables: " + e.getMessage(), e); } } waitForTasksCompletion(taskFutures); From 19c350d43604f99cc2a16651d02a8bf66b0f8d22 Mon Sep 17 00:00:00 2001 From: Keshav Dandeva Date: Thu, 9 Jul 2026 13:12:30 +0000 Subject: [PATCH 5/5] address pr feedback --- .../jdbc/BigQueryDatabaseMetaData.java | 65 ++++++++++--------- .../jdbc/it/ITDatabaseMetadataTest.java | 16 ++++- 2 files changed, 49 insertions(+), 32 deletions(-) diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java index 5c7094efcf2d..4d5b2978d02f 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java @@ -5188,45 +5188,45 @@ private void processTargetTablesConcurrently( List> taskFutures = new ArrayList<>(); try { + List> tasks = new ArrayList<>(); for (DatasetId datasetId : targetDatasets) { if (tableName != null) { - taskFutures.add( - executor.submit( - () -> { - processSingleTable( - datasetId, - tableName, - collectedResults, - resultSchemaFields, - ignoreAccessErrors, - processor); - return null; - })); + tasks.add( + () -> { + processSingleTable( + datasetId, + tableName, + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + }); continue; } try { Page
tablesPage = bigquery.listTables(datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)); - if (tablesPage != null) { - for (Table table : tablesPage.iterateAll()) { - if (table.getDefinition() == null - || table.getDefinition().getType() != TableDefinition.Type.TABLE) { - continue; - } - taskFutures.add( - executor.submit( - () -> { - processSingleTable( - datasetId, - table.getTableId().getTable(), - collectedResults, - resultSchemaFields, - ignoreAccessErrors, - processor); - return null; - })); + if (tablesPage == null) { + continue; + } + for (Table table : tablesPage.iterateAll()) { + if (table.getDefinition() == null + || table.getDefinition().getType() != TableDefinition.Type.TABLE) { + continue; } + tasks.add( + () -> { + processSingleTable( + datasetId, + table.getTableId().getTable(), + collectedResults, + resultSchemaFields, + ignoreAccessErrors, + processor); + return null; + }); } } catch (BigQueryException e) { if (ignoreAccessErrors && (e.getCode() == 404 || e.getCode() == 403)) { @@ -5235,10 +5235,13 @@ private void processTargetTablesConcurrently( datasetId.getDataset(), datasetId.getProject(), e.getCode()); continue; } - taskFutures.forEach(future -> future.cancel(true)); throw new SQLException("Error while listing tables: " + e.getMessage(), e); } } + + for (Callable task : tasks) { + taskFutures.add(executor.submit(task)); + } waitForTasksCompletion(taskFutures); if (Thread.currentThread().isInterrupted()) { throw new SQLException("Interrupted while parallel-fetching metadata"); diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java index 149858a91b4f..046f49991dbe 100644 --- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITDatabaseMetadataTest.java @@ -330,7 +330,7 @@ public void testTableConstraints() throws SQLException { } @Test - public void testGetExportedKeys() throws SQLException { + public void testGetExportedKeys_multipleKeys() throws SQLException { try (Connection connection = DriverManager.getConnection(ITBase.connectionUrl)) { DatabaseMetaData metaData = connection.getMetaData(); @@ -353,6 +353,13 @@ public void testGetExportedKeys() throws SQLException { Assertions.assertEquals(2, exportedKeys2.getInt(9)); Assertions.assertEquals("my_fk", exportedKeys2.getString(12)); Assertions.assertFalse(exportedKeys2.next()); + } + } + + @Test + public void testGetExportedKeys_singleKey() throws SQLException { + try (Connection connection = DriverManager.getConnection(ITBase.connectionUrl)) { + DatabaseMetaData metaData = connection.getMetaData(); // Table 3 exports keys to Table ResultSet exportedKeys3 = @@ -365,6 +372,13 @@ public void testGetExportedKeys() throws SQLException { Assertions.assertEquals(1, exportedKeys3.getInt(9)); Assertions.assertEquals("my_fk2", exportedKeys3.getString(12)); Assertions.assertFalse(exportedKeys3.next()); + } + } + + @Test + public void testGetExportedKeys_noKeys() throws SQLException { + try (Connection connection = DriverManager.getConnection(ITBase.connectionUrl)) { + DatabaseMetaData metaData = connection.getMetaData(); // Table does not export keys to anything (it only imports them) ResultSet exportedKeys1 =