Skip to content

perf(bigquery-jdbc): optimize metadata introspection performance and parallel project scanning #13667

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
keshavdandeva merged 9 commits into from
Jul 7, 2026
Merged

perf(bigquery-jdbc): optimize metadata introspection performance and parallel project scanning #13667

Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b2cdc3d
perf(bigquery-jdbc): optimize metadata introspection performance and …
keshavdandeva Jul 6, 2026
7a5836f
Merge branch 'main' into jdbc/optimize-metadata-methods
keshavdandeva Jul 6, 2026
7139d25
fix: resource leak
keshavdandeva Jul 6, 2026
51295fb
Merge branch 'main' into jdbc/optimize-metadata-methods
keshavdandeva Jul 6, 2026
60af5b1
chore: address pr feedback
keshavdandeva Jul 6, 2026
fd49c62
revert missing single catalog path in getSchemas
keshavdandeva Jul 6, 2026
d1f4863
revert naming to avoid confusion
keshavdandeva Jul 6, 2026
b64c66b
revert comment
keshavdandeva Jul 6, 2026
a9d4073
Merge branch 'main' into jdbc/optimize-metadata-methods
keshavdandeva Jul 6, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
155 changes: 87 additions & 68 deletions ...-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,12 @@
import com.google.cloud.Tuple;
import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.BigQuery.DatasetListOption;
import com.google.cloud.bigquery.BigQuery.RoutineField;
import com.google.cloud.bigquery.BigQuery.RoutineListOption;
import com.google.cloud.bigquery.BigQuery.RoutineOption;
import com.google.cloud.bigquery.BigQuery.TableField;
import com.google.cloud.bigquery.BigQuery.TableListOption;
import com.google.cloud.bigquery.BigQuery.TableOption;
import com.google.cloud.bigquery.BigQueryException;
import com.google.cloud.bigquery.Dataset;
import com.google.cloud.bigquery.DatasetId;
Expand Down Expand Up @@ -1056,7 +1060,14 @@ public ResultSet getProcedureColumns(

getRoutineDetailsExecutor = connection.getMetadataExecutor();
List<Routine> fullRoutines =
fetchFullRoutineDetailsForIds(procedureIdsToGet, getRoutineDetailsExecutor, LOG);
fetchFullRoutineDetailsForIds(
procedureIdsToGet,
getRoutineDetailsExecutor,
LOG,
RoutineOption.fields(
RoutineField.ROUTINE_REFERENCE,
RoutineField.ARGUMENTS,
RoutineField.ROUTINE_TYPE));
getRoutineDetailsExecutor = null;

if (fullRoutines.isEmpty() || Thread.currentThread().isInterrupted()) {
Expand Down Expand Up @@ -1174,7 +1185,8 @@ List<RoutineId> listMatchingProcedureIdsFromDatasets(
List<Routine> fetchFullRoutineDetailsForIds(
List<RoutineId> procedureIdsToGet,
ExecutorService getRoutineDetailsExecutor,
BigQueryJdbcCustomLogger logger)
BigQueryJdbcCustomLogger logger,
RoutineOption... options)
throws InterruptedException, ExecutionException {
logger.fine("Fetching full details for %d procedure IDs.", procedureIdsToGet.size());
final List<Future<Routine>> getRoutineFutures = new ArrayList<>();
Expand All @@ -1191,7 +1203,7 @@ List<Routine> fetchFullRoutineDetailsForIds(
Callable<Routine> getCallable =
() -> {
try {
return bigquery.getRoutine(currentProcId);
return bigquery.getRoutine(currentProcId, options);
} catch (Exception e) {
logger.warning(
"Failed to get full details for routine "
Expand Down Expand Up @@ -1616,7 +1628,11 @@ public ResultSet getTables(
TableId.of(
currentDatasetId.getProject(),
currentDatasetId.getDataset(),
name)),
name),
TableOption.fields(
TableField.TABLE_REFERENCE,
TableField.TYPE,
TableField.DESCRIPTION)),
(tbl) -> tbl.getTableId().getTable(),
tableNamePattern,
tableNameRegex,
Expand Down Expand Up @@ -1930,7 +1946,9 @@ public ResultSet getColumns(
datasetId, TableListOption.pageSize(DEFAULT_PAGE_SIZE)),
(name) ->
bigquery.getTable(
TableId.of(datasetId.getProject(), datasetId.getDataset(), name)),
TableId.of(datasetId.getProject(), datasetId.getDataset(), name),
TableOption.fields(
TableField.TABLE_REFERENCE, TableField.TYPE, TableField.SCHEMA)),
(tbl) -> tbl.getTableId().getTable(),
tableNamePattern,
tableNameRegex,
Expand Down Expand Up @@ -1999,7 +2017,10 @@ private void processTableColumns(
"Schema not included in table object for "
+ tableId
+ ", fetching full table details...");
Comment thread
keshavdandeva marked this conversation as resolved.
Table fullTable = bigquery.getTable(tableId);
Table fullTable =
bigquery.getTable(
tableId,
TableOption.fields(TableField.TABLE_REFERENCE, TableField.TYPE, TableField.SCHEMA));
if (fullTable != null) {
definition = fullTable.getDefinition();
tableSchema = (definition != null) ? definition.getSchema() : null;
Expand Down Expand Up @@ -3423,49 +3444,11 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce
Runnable multiSchemaFetcher =
() -> {
final FieldList localResultSchemaFields = resultSchemaFields;
final List<Future<List<Dataset>>> apiFutures = new ArrayList<>();
final List<Dataset> collectedDatasets = new ArrayList<>();
final List<FieldValueList> collectedResults = new ArrayList<>();

try {
List<String> projectsToScanList = getAccessibleCatalogNames();

if (projectsToScanList.isEmpty()) {
LOG.info(
"No valid projects to scan (primary, specified, or additional). Returning empty"
+ " resultset.");
return;
}

ExecutorService apiExecutor = connection.getMetadataExecutor();

LOG.fine("Submitting parallel fetchMatchingDatasets tasks...");
for (String currentProjectToScan : projectsToScanList) {
if (Thread.currentThread().isInterrupted()) {
LOG.warning("Fetcher interrupted during project iteration submission.");
break;
}

Callable<List<Dataset>> apiCallable =
() -> fetchMatchingDatasets(currentProjectToScan, schemaPattern, schemaRegex);
Future<List<Dataset>> apiFuture = apiExecutor.submit(apiCallable);
apiFutures.add(apiFuture);
}
LOG.fine("Finished submitting " + apiFutures.size() + " fetchMatchingDatasets tasks.");

LOG.fine("Processing results from fetchMatchingDatasets tasks...");
for (Future<List<Dataset>> apiFuture : apiFutures) {
if (Thread.currentThread().isInterrupted()) {
LOG.warning("Fetcher interrupted while processing API futures.");
break;
}
List<Dataset> datasetsResult = apiFuture.get();
if (datasetsResult != null) {
collectedDatasets.addAll(datasetsResult);
}
}

for (Dataset dataset : collectedDatasets) {
List<Dataset> datasets = fetchMatchingDatasets(catalog, schemaPattern, schemaRegex);
for (Dataset dataset : datasets) {
if (Thread.currentThread().isInterrupted()) {
break;
}
Expand All @@ -3480,7 +3463,6 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce
} catch (Throwable t) {
handleFetcherException(t, queue, "getSchemas");
} finally {
apiFutures.forEach(f -> f.cancel(true));
finalizeFetcher(queue, localResultSchemaFields, "Multi-schema fetcher");
}
};
Expand Down Expand Up @@ -4834,35 +4816,72 @@ private void signalEndOfData(
}
}

private List<Dataset> fetchDatasetsForProject(
String project, String schemaPattern, Pattern schemaRegex) throws SQLException {
try {
List<Dataset> datasets =
findMatchingBigQueryObjects(
"Dataset",
() -> bigquery.listDatasets(project, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)),
(name) -> bigquery.getDataset(DatasetId.of(project, name)),
(ds) -> ds.getDatasetId().getDataset(),
schemaPattern,
schemaRegex,
LOG);
return datasets != null ? datasets : Collections.emptyList();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SQLException(
"Interrupted while fetching matching datasets for project " + project, e);
} catch (Exception e) {
throw new SQLException("Failed to fetch matching datasets for project " + project, e);
}
}

private List<Dataset> fetchMatchingDatasets(
String catalog, String schemaPattern, Pattern schemaRegex) throws SQLException {
List<String> projects =
(catalog != null) ? Collections.singletonList(catalog) : getAccessibleCatalogNames();
List<Dataset> allDatasets = new ArrayList<>();
for (String project : projects) {

if (projects.isEmpty()) {
return Collections.emptyList();
}

// Single project path
if (projects.size() == 1) {
return fetchDatasetsForProject(projects.get(0), schemaPattern, schemaRegex);
}

// Multi-project path
final List<Dataset> allDatasets = Collections.synchronizedList(new ArrayList<>());
final List<Future<?>> taskFutures = new ArrayList<>();
ExecutorService executor = connection.getMetadataExecutor();

try {
for (String project : projects) {
Callable<Void> task =
() -> {
List<Dataset> datasets = fetchDatasetsForProject(project, schemaPattern, schemaRegex);
allDatasets.addAll(datasets);
return null;
};
taskFutures.add(executor.submit(task));
}

waitForTasksCompletion(taskFutures);
if (Thread.currentThread().isInterrupted()) {
throw new SQLException("Interrupted while fetching matching datasets");
throw new SQLException("Interrupted while parallel-fetching matching datasets");
}
try {
List<Dataset> datasets =
findMatchingBigQueryObjects(
"Dataset",
() -> bigquery.listDatasets(project, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)),
(name) -> bigquery.getDataset(DatasetId.of(project, name)),
(ds) -> ds.getDatasetId().getDataset(),
schemaPattern,
schemaRegex,
LOG);
if (datasets != null) {
allDatasets.addAll(datasets);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new SQLException("Interrupted while fetching matching datasets", e);
} catch (Exception e) {
throw new SQLException("Failed to fetch matching datasets for project " + project, e);
} catch (ExecutionException e) {
Throwable cause = e.getCause();
if (cause instanceof SQLException) {
throw (SQLException) cause;
}
throw new SQLException("Error parallel-fetching matching datasets", e);
} finally {
taskFutures.forEach(f -> f.cancel(true));
}

return allDatasets;
}

Expand Down
Loading