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 1b5eb0af0a4e..220a5dcf5bd1 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 @@ -69,7 +69,6 @@ import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; import java.util.concurrent.LinkedBlockingQueue; -import java.util.concurrent.TimeUnit; import java.util.function.Function; import java.util.function.Supplier; import java.util.regex.Pattern; @@ -96,7 +95,6 @@ class BigQueryDatabaseMetaData implements DatabaseMetaData { private static final String GET_IMPORTED_KEYS_SQL = "DatabaseMetaData_GetImportedKeys.sql"; private static final String GET_EXPORTED_KEYS_SQL = "DatabaseMetaData_GetExportedKeys.sql"; private static final String GET_CROSS_REFERENCE_SQL = "DatabaseMetaData_GetCrossReference.sql"; - private static final int API_EXECUTOR_POOL_SIZE = 50; private static final int DEFAULT_PAGE_SIZE = 500; private static final int DEFAULT_QUEUE_CAPACITY = 5000; // Declared package-private for testing. @@ -138,13 +136,11 @@ class BigQueryDatabaseMetaData implements DatabaseMetaData { String URL; BigQueryConnection connection; private final BigQuery bigquery; - private final int metadataFetchThreadCount; BigQueryDatabaseMetaData(BigQueryConnection connection) { this.URL = connection.getConnectionUrl(); this.connection = connection; this.bigquery = connection.getBigQuery(); - this.metadataFetchThreadCount = connection.getMetadataFetchThreadCount(); } @Override @@ -774,10 +770,10 @@ public boolean dataDefinitionIgnoredInTransactions() { @Override public ResultSet getProcedures( String catalog, String schemaPattern, String procedureNamePattern) { - if ((catalog == null || catalog.isEmpty()) + if ((catalog != null && catalog.isEmpty()) || (schemaPattern != null && schemaPattern.isEmpty()) || (procedureNamePattern != null && procedureNamePattern.isEmpty())) { - LOG.warning("Returning empty ResultSet as catalog is null/empty or a pattern is empty."); + LOG.warning("Returning empty ResultSet as catalog is empty or a pattern is empty."); return new BigQueryJsonResultSet(); } @@ -802,16 +798,7 @@ public ResultSet getProcedures( try { List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaPattern, - schemaRegex, - LOG); + fetchMatchingDatasets(catalogParam, schemaPattern, schemaRegex); if (datasetsToScan.isEmpty()) { LOG.info("Fetcher thread found no matching datasets. Finishing."); @@ -874,33 +861,21 @@ public ResultSet getProcedures( Thread.currentThread().interrupt(); LOG.warning("Fetcher thread interrupted while waiting for API future result."); break; - } catch (ExecutionException e) { - LOG.warning( - "Error executing findMatchingRoutines task: " - + e.getMessage() - + ". Cause: " - + e.getCause()); } catch (CancellationException e) { LOG.warning("A findMatchingRoutines task was cancelled."); } } - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetProceduresComparator(localResultSchemaFields); - sortResults(collectedResults, comparator, "getProcedures", LOG); - } - - if (!Thread.currentThread().isInterrupted()) { - populateQueue(collectedResults, queue, localResultSchemaFields); - } + Comparator comparator = + defineGetProceduresComparator(localResultSchemaFields); + sortResults(collectedResults, comparator, "getProcedures", LOG); + populateQueue(collectedResults, queue, localResultSchemaFields); } catch (Throwable t) { - LOG.severe("Unexpected error in procedure fetcher runnable: " + t.getMessage()); + handleFetcherException(t, queue, "getProcedures"); } finally { apiFutures.forEach(f -> f.cancel(true)); - signalEndOfData(queue, localResultSchemaFields); - LOG.info("Procedure fetcher thread finished."); + finalizeFetcher(queue, localResultSchemaFields, "Procedure fetcher"); } }; @@ -1023,8 +998,8 @@ Comparator defineGetProceduresComparator(FieldList resultSchemaF public ResultSet getProcedureColumns( String catalog, String schemaPattern, String procedureNamePattern, String columnNamePattern) { - if (catalog == null || catalog.isEmpty()) { - LOG.warning("Returning empty ResultSet because catalog (project) is null or empty."); + if (catalog != null && catalog.isEmpty()) { + LOG.warning("Returning empty ResultSet because catalog (project) is empty."); return new BigQueryJsonResultSet(); } if ((schemaPattern != null && schemaPattern.isEmpty()) @@ -1056,7 +1031,7 @@ public ResultSet getProcedureColumns( try { List datasetsToScan = - fetchMatchingDatasetsForProcedureColumns(catalogParam, schemaPattern, schemaRegex); + fetchMatchingDatasets(catalogParam, schemaPattern, schemaRegex); if (datasetsToScan.isEmpty() || Thread.currentThread().isInterrupted()) { LOG.info( "Fetcher: No matching datasets or interrupted early. Catalog: " + catalogParam); @@ -1093,29 +1068,15 @@ public ResultSet getProcedureColumns( processProcedureArgumentsSequentially( fullRoutines, columnNameRegex, collectedResults, resultSchema.getFields(), LOG); - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetProcedureColumnsComparator(resultSchema.getFields()); - sortResults(collectedResults, comparator, "getProcedureColumns", LOG); - populateQueue(collectedResults, queue, resultSchema.getFields()); - } + Comparator comparator = + defineGetProcedureColumnsComparator(resultSchema.getFields()); + sortResults(collectedResults, comparator, "getProcedureColumns", LOG); + populateQueue(collectedResults, queue, resultSchema.getFields()); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - LOG.warning( - "Fetcher: Interrupted in main try block for catalog " - + catalogParam - + ". Error: " - + e.getMessage()); } catch (Throwable t) { - LOG.severe( - "Fetcher: Unexpected error in main try block for catalog " - + catalogParam - + ". Error: " - + t.getMessage()); + handleFetcherException(t, queue, "getProcedureColumns"); } finally { - signalEndOfData(queue, resultSchema.getFields()); - LOG.info("Procedure column fetcher thread finished for catalog: " + catalogParam); + finalizeFetcher(queue, resultSchema.getFields(), "Procedure column fetcher"); } }; @@ -1127,27 +1088,6 @@ public ResultSet getProcedureColumns( return resultSet; } - private List fetchMatchingDatasetsForProcedureColumns( - String catalogParam, String schemaPattern, Pattern schemaRegex) throws InterruptedException { - LOG.fine( - "Fetching matching datasets for catalog '%s', schemaPattern '%s'", - catalogParam, schemaPattern); - List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets(catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaPattern, - schemaRegex, - LOG); - LOG.info( - "Found %d datasets to scan for procedures in catalog '%s'.", - datasetsToScan.size(), catalogParam); - return datasetsToScan; - } - List listMatchingProcedureIdsFromDatasets( List datasetsToScan, String procedureNamePattern, @@ -1155,7 +1095,7 @@ List listMatchingProcedureIdsFromDatasets( ExecutorService listRoutinesExecutor, String catalogParam, BigQueryJdbcCustomLogger logger) - throws InterruptedException { + throws InterruptedException, ExecutionException { logger.fine( "Listing matching procedure IDs from %d datasets for catalog '%s'.", @@ -1221,9 +1161,6 @@ List listMatchingProcedureIdsFromDatasets( } } } - } catch (ExecutionException e) { - logger.warning( - "Error getting routine list result for catalog " + catalogParam + ": " + e.getCause()); } catch (CancellationException e) { logger.warning("Routine list task cancelled for catalog: " + catalogParam); } @@ -1238,7 +1175,7 @@ List fetchFullRoutineDetailsForIds( List procedureIdsToGet, ExecutorService getRoutineDetailsExecutor, BigQueryJdbcCustomLogger logger) - throws InterruptedException { + throws InterruptedException, ExecutionException { logger.fine("Fetching full details for %d procedure IDs.", procedureIdsToGet.size()); final List> getRoutineFutures = new ArrayList<>(); final List fullRoutines = Collections.synchronizedList(new ArrayList<>()); @@ -1281,8 +1218,6 @@ List fetchFullRoutineDetailsForIds( if (fullRoutine != null) { fullRoutines.add(fullRoutine); } - } catch (ExecutionException e) { - logger.warning("Error processing getRoutine future result: " + e.getCause()); } catch (CancellationException e) { logger.warning("getRoutine detail task cancelled."); } @@ -1618,11 +1553,11 @@ public ResultSet getTables( String effectiveCatalog = effectiveIdentifiers.x(); String effectiveSchemaPattern = effectiveIdentifiers.y(); - if ((effectiveCatalog == null || effectiveCatalog.isEmpty()) + if ((effectiveCatalog != null && effectiveCatalog.isEmpty()) || (effectiveSchemaPattern != null && effectiveSchemaPattern.isEmpty()) || (tableNamePattern != null && tableNamePattern.isEmpty())) { LOG.warning( - "Returning empty ResultSet as one or more patterns are empty or catalog is null."); + "Returning empty ResultSet as one or more patterns are empty or catalog is empty."); return new BigQueryJsonResultSet(); } @@ -1652,16 +1587,7 @@ public ResultSet getTables( try { List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaParam, - schemaRegex, - LOG); + fetchMatchingDatasets(catalogParam, schemaParam, schemaRegex); if (datasetsToScan.isEmpty()) { LOG.info("Fetcher thread found no matching datasets. Returning empty resultset."); @@ -1721,33 +1647,21 @@ public ResultSet getTables( Thread.currentThread().interrupt(); LOG.warning("Fetcher thread interrupted while waiting for API future result."); break; - } catch (ExecutionException e) { - LOG.warning( - "Error executing findMatchingTables task: " - + e.getMessage() - + ". Cause: " - + e.getCause()); } catch (CancellationException e) { LOG.warning("A findMatchingTables task was cancelled."); } } - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetTablesComparator(localResultSchemaFields); - sortResults(collectedResults, comparator, "getTables", LOG); - } - - if (!Thread.currentThread().isInterrupted()) { - populateQueue(collectedResults, queue, localResultSchemaFields); - } + Comparator comparator = + defineGetTablesComparator(localResultSchemaFields); + sortResults(collectedResults, comparator, "getTables", LOG); + populateQueue(collectedResults, queue, localResultSchemaFields); } catch (Throwable t) { - LOG.severe("Unexpected error in table fetcher runnable: " + t.getMessage()); + handleFetcherException(t, queue, "getTables"); } finally { apiFutures.forEach(f -> f.cancel(true)); - signalEndOfData(queue, localResultSchemaFields); - LOG.info("Table fetcher thread finished."); + finalizeFetcher(queue, localResultSchemaFields, "Table fetcher"); } }; @@ -1956,12 +1870,12 @@ public ResultSet getColumns( String effectiveCatalog = effectiveIdentifiers.x(); String effectiveSchemaPattern = effectiveIdentifiers.y(); - if ((effectiveCatalog == null || effectiveCatalog.isEmpty()) + if ((effectiveCatalog != null && effectiveCatalog.isEmpty()) || (effectiveSchemaPattern != null && effectiveSchemaPattern.isEmpty()) || (tableNamePattern != null && tableNamePattern.isEmpty()) || (columnNamePattern != null && columnNamePattern.isEmpty())) { LOG.warning( - "Returning empty ResultSet as one or more patterns are empty or catalog is null."); + "Returning empty ResultSet as one or more patterns are empty or catalog is empty."); return new BigQueryJsonResultSet(); } @@ -1990,16 +1904,7 @@ public ResultSet getColumns( try { List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaParam, - schemaRegex, - LOG); + fetchMatchingDatasets(catalogParam, schemaParam, schemaRegex); if (datasetsToScan.isEmpty()) { LOG.info("Fetcher thread found no matching datasets. Returning empty resultset."); @@ -2057,22 +1962,16 @@ public ResultSet getColumns( waitForTasksCompletion(taskFutures); - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetColumnsComparator(localResultSchemaFields); - sortResults(collectedResults, comparator, "getColumns", LOG); - } - - if (!Thread.currentThread().isInterrupted()) { - populateQueue(collectedResults, queue, localResultSchemaFields); - } + Comparator comparator = + defineGetColumnsComparator(localResultSchemaFields); + sortResults(collectedResults, comparator, "getColumns", LOG); + populateQueue(collectedResults, queue, localResultSchemaFields); } catch (Throwable t) { - LOG.severe("Unexpected error in column fetcher runnable: " + t.getMessage()); + handleFetcherException(t, queue, "getColumns"); } finally { taskFutures.forEach(f -> f.cancel(true)); - signalEndOfData(queue, localResultSchemaFields); - LOG.info("Column fetcher thread finished."); + finalizeFetcher(queue, localResultSchemaFields, "Column fetcher"); } }; @@ -3503,22 +3402,33 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce final Schema resultSchema = defineGetSchemasSchema(); final FieldList resultSchemaFields = resultSchema.getFields(); + if (catalog != null) { + // Single-Catalog Path: completely synchronous on caller thread + final BlockingQueue queue = new LinkedBlockingQueue<>(); + List datasets = fetchMatchingDatasets(catalog, schemaPattern, schemaRegex); + List collectedResults = new ArrayList<>(); + for (Dataset dataset : datasets) { + processSchemaInfo(dataset, collectedResults, resultSchemaFields); + } + Comparator comparator = defineGetSchemasComparator(resultSchemaFields); + sortResults(collectedResults, comparator, "getSchemas", LOG); + populateQueue(collectedResults, queue, resultSchemaFields); + signalEndOfData(queue, resultSchemaFields); + return BigQueryJsonResultSet.of(resultSchema, -1, queue, null); + } + + // Multi-Catalog Path: fan out using connection-scoped metadataExecutor final BlockingQueue queue = new LinkedBlockingQueue<>(DEFAULT_QUEUE_CAPACITY); - final List collectedResults = Collections.synchronizedList(new ArrayList<>()); - final String catalogParam = catalog; - - Runnable schemaFetcher = + Runnable multiSchemaFetcher = () -> { final FieldList localResultSchemaFields = resultSchemaFields; - List projectsToScanList = new ArrayList<>(); + final List>> apiFutures = new ArrayList<>(); + final List collectedDatasets = new ArrayList<>(); + final List collectedResults = new ArrayList<>(); try { - if (catalogParam != null) { - projectsToScanList.add(catalogParam); - } else { - projectsToScanList.addAll(getAccessibleCatalogNames()); - } + List projectsToScanList = getAccessibleCatalogNames(); if (projectsToScanList.isEmpty()) { LOG.info( @@ -3527,76 +3437,59 @@ public ResultSet getSchemas(String catalog, String schemaPattern) throws SQLExce return; } + ExecutorService apiExecutor = connection.getMetadataExecutor(); + + LOG.fine("Submitting parallel fetchMatchingDatasets tasks..."); for (String currentProjectToScan : projectsToScanList) { if (Thread.currentThread().isInterrupted()) { - LOG.warning( - "Schema fetcher interrupted during project iteration for project: " - + currentProjectToScan); + LOG.warning("Fetcher interrupted during project iteration submission."); break; } - LOG.info("Fetching schemas for project: " + currentProjectToScan); - List datasetsInProject = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - currentProjectToScan, - BigQuery.DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(currentProjectToScan, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaPattern, - schemaRegex, - LOG); - if (datasetsInProject.isEmpty() || Thread.currentThread().isInterrupted()) { - LOG.info( - "Fetcher thread found no matching datasets in project: " - + currentProjectToScan); - continue; - } + Callable> apiCallable = + () -> fetchMatchingDatasets(currentProjectToScan, schemaPattern, schemaRegex); + Future> apiFuture = apiExecutor.submit(apiCallable); + apiFutures.add(apiFuture); + } + LOG.fine("Finished submitting " + apiFutures.size() + " fetchMatchingDatasets tasks."); - LOG.fine("Processing found datasets for project: " + currentProjectToScan); - for (Dataset dataset : datasetsInProject) { - if (Thread.currentThread().isInterrupted()) { - LOG.warning( - "Schema fetcher interrupted during dataset iteration for project: " - + currentProjectToScan); - break; - } - processSchemaInfo(dataset, collectedResults, localResultSchemaFields); + LOG.fine("Processing results from fetchMatchingDatasets tasks..."); + for (Future> apiFuture : apiFutures) { + if (Thread.currentThread().isInterrupted()) { + LOG.warning("Fetcher interrupted while processing API futures."); + break; + } + List datasetsResult = apiFuture.get(); + if (datasetsResult != null) { + collectedDatasets.addAll(datasetsResult); } } - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetSchemasComparator(localResultSchemaFields); - sortResults(collectedResults, comparator, "getSchemas", LOG); + for (Dataset dataset : collectedDatasets) { + if (Thread.currentThread().isInterrupted()) { + break; + } + processSchemaInfo(dataset, collectedResults, localResultSchemaFields); } - if (!Thread.currentThread().isInterrupted()) { - populateQueue(collectedResults, queue, localResultSchemaFields); - } + Comparator comparator = + defineGetSchemasComparator(localResultSchemaFields); + sortResults(collectedResults, comparator, "getSchemas", LOG); + populateQueue(collectedResults, queue, localResultSchemaFields); } catch (Throwable t) { - LOG.severe("Unexpected error in schema fetcher runnable: " + t.getMessage()); - Exception ex = (t instanceof Exception) ? (Exception) t : new Exception(t); - try { - queue.put(BigQueryFieldValueListWrapper.ofError(ex)); - } catch (InterruptedException ie) { - LOG.warning("Failed to put exception to queue due to interruption."); - Thread.currentThread().interrupt(); - } + handleFetcherException(t, queue, "getSchemas"); } finally { - signalEndOfData(queue, localResultSchemaFields); - LOG.info("Schema fetcher thread finished."); + apiFutures.forEach(f -> f.cancel(true)); + finalizeFetcher(queue, localResultSchemaFields, "Multi-schema fetcher"); } }; - Future fetcherFuture = connection.getExecutorService().submit(schemaFetcher); + Future fetcherFuture = connection.getExecutorService().submit(multiSchemaFetcher); BigQueryJsonResultSet resultSet = BigQueryJsonResultSet.of(resultSchema, -1, queue, null, fetcherFuture); - LOG.info("Submitted background task for getSchemas to metadata executor"); + LOG.info("Submitted background task for multi-catalog getSchemas to query executor"); return resultSet; } @@ -3738,11 +3631,11 @@ Schema defineGetClientInfoPropertiesSchema() { @Override public ResultSet getFunctions(String catalog, String schemaPattern, String functionNamePattern) { - if ((catalog == null || catalog.isEmpty()) + if ((catalog != null && catalog.isEmpty()) || (schemaPattern != null && schemaPattern.isEmpty()) || (functionNamePattern != null && functionNamePattern.isEmpty())) { LOG.warning( - "Returning empty ResultSet as catalog is null/empty or a pattern is empty for" + "Returning empty ResultSet as catalog is empty or a pattern is empty for" + " getFunctions."); return new BigQueryJsonResultSet(); } @@ -3768,16 +3661,7 @@ public ResultSet getFunctions(String catalog, String schemaPattern, String funct try { List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaPattern, - schemaRegex, - LOG); + fetchMatchingDatasets(catalogParam, schemaPattern, schemaRegex); if (datasetsToScan.isEmpty()) { LOG.info("Fetcher thread found no matching datasets. Returning empty resultset."); @@ -3852,10 +3736,9 @@ public ResultSet getFunctions(String catalog, String schemaPattern, String funct LOG.warning( "Function fetcher thread interrupted while waiting for API future result."); break; - } catch (ExecutionException | CancellationException e) { + } catch (CancellationException e) { LOG.warning( - "Error or cancellation in findMatchingRoutines (for functions) task: " - + e.getMessage()); + "Cancellation in findMatchingRoutines (for functions) task: " + e.getMessage()); } } Comparator comparator = @@ -3863,11 +3746,10 @@ public ResultSet getFunctions(String catalog, String schemaPattern, String funct sortResults(collectedResults, comparator, "getFunctions", LOG); populateQueue(collectedResults, queue, localResultSchemaFields); } catch (Throwable t) { - LOG.severe("Unexpected error in function fetcher runnable: " + t.getMessage()); + handleFetcherException(t, queue, "getFunctions"); } finally { apiFutures.forEach(f -> f.cancel(true)); - signalEndOfData(queue, localResultSchemaFields); - LOG.info("Function fetcher thread finished."); + finalizeFetcher(queue, localResultSchemaFields, "Function fetcher"); } }; @@ -3970,8 +3852,8 @@ Comparator defineGetFunctionsComparator(FieldList resultSchemaFi @Override public ResultSet getFunctionColumns( String catalog, String schemaPattern, String functionNamePattern, String columnNamePattern) { - if (catalog == null || catalog.isEmpty()) { - LOG.warning("Returning empty ResultSet catalog (project) is null or empty."); + if (catalog != null && catalog.isEmpty()) { + LOG.warning("Returning empty ResultSet catalog (project) is empty."); return new BigQueryJsonResultSet(); } if ((schemaPattern != null && schemaPattern.isEmpty()) @@ -4004,16 +3886,7 @@ public ResultSet getFunctionColumns( try { List datasetsToScan = - findMatchingBigQueryObjects( - "Dataset", - () -> - bigquery.listDatasets( - catalogParam, DatasetListOption.pageSize(DEFAULT_PAGE_SIZE)), - (name) -> bigquery.getDataset(DatasetId.of(catalogParam, name)), - (ds) -> ds.getDatasetId().getDataset(), - schemaPattern, - schemaRegex, - LOG); + fetchMatchingDatasets(catalogParam, schemaPattern, schemaRegex); if (datasetsToScan.isEmpty() || Thread.currentThread().isInterrupted()) { LOG.info( @@ -4051,29 +3924,15 @@ public ResultSet getFunctionColumns( processFunctionParametersSequentially( fullFunctions, columnNameRegex, collectedResults, resultSchemaFields, LOG); - if (!Thread.currentThread().isInterrupted()) { - Comparator comparator = - defineGetFunctionColumnsComparator(resultSchemaFields); - sortResults(collectedResults, comparator, "getFunctionColumns", LOG); - populateQueue(collectedResults, queue, resultSchemaFields); - } + Comparator comparator = + defineGetFunctionColumnsComparator(resultSchemaFields); + sortResults(collectedResults, comparator, "getFunctionColumns", LOG); + populateQueue(collectedResults, queue, resultSchemaFields); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - LOG.warning( - "Fetcher: Interrupted in main try block for catalog " - + catalogParam - + ". Error: " - + e.getMessage()); } catch (Throwable t) { - LOG.severe( - "Fetcher: Unexpected error in main try block for catalog " - + catalogParam - + ". Error: " - + t.getMessage()); + handleFetcherException(t, queue, "getFunctionColumns"); } finally { - signalEndOfData(queue, resultSchemaFields); - LOG.info("Function column fetcher thread finished for catalog: " + catalogParam); + finalizeFetcher(queue, resultSchemaFields, "Function column fetcher"); } }; @@ -4159,7 +4018,7 @@ List listMatchingFunctionIdsFromDatasets( ExecutorService listRoutinesExecutor, String catalogParam, BigQueryJdbcCustomLogger logger) - throws InterruptedException { + throws InterruptedException, ExecutionException { logger.fine( "Listing matching function IDs from %d datasets for catalog '%s'.", @@ -4224,12 +4083,6 @@ List listMatchingFunctionIdsFromDatasets( } } } - } catch (ExecutionException e) { - logger.warning( - "Error getting routine (function) list result for catalog " - + catalogParam - + ": " - + e.getCause()); } catch (CancellationException e) { logger.warning("Routine (function) list task cancelled for catalog: " + catalogParam); } @@ -4634,7 +4487,8 @@ List findMatchingBigQueryObjects( Function nameExtractor, String pattern, Pattern regex, - BigQueryJdbcCustomLogger logger) { + BigQueryJdbcCustomLogger logger) + throws InterruptedException { boolean needsList = needsListing(pattern); List resultList = new ArrayList<>(); @@ -4688,14 +4542,17 @@ List findMatchingBigQueryObjects( logger.warning( "BigQueryException finding %ss for pattern '%s': %s (Code: %d)", objectTypeName, pattern, e.getMessage(), e.getCode()); + throw e; } } catch (InterruptedException e) { Thread.currentThread().interrupt(); logger.warning("Interrupted while finding " + objectTypeName + "s."); + throw e; } catch (Exception e) { logger.severe( "Unexpected exception finding %ss for pattern '%s': %s", objectTypeName, pattern, e.getMessage()); + throw new RuntimeException(e); } return resultList; } @@ -4890,7 +4747,7 @@ private Long getLongValueOrNull(FieldValueList fvl, int index) { } } - private void waitForTasksCompletion(List> taskFutures) { + private void waitForTasksCompletion(List> taskFutures) throws ExecutionException { LOG.info("Waiting for %d submitted tasks to complete...", taskFutures.size()); for (Future future : taskFutures) { try { @@ -4899,10 +4756,6 @@ private void waitForTasksCompletion(List> taskFutures) { } } catch (CancellationException e) { LOG.warning("A table processing task was cancelled."); - } catch (ExecutionException e) { - LOG.severe( - "Error executing table processing task: %s", - (e.getCause() != null ? e.getCause().getMessage() : e.getMessage())); } catch (InterruptedException e) { Thread.currentThread().interrupt(); LOG.warning( @@ -4937,51 +4790,88 @@ private void populateQueue( } } + private void handleFetcherException( + Throwable t, BlockingQueue queue, String fetcherName) { + if (t instanceof ExecutionException && t.getCause() != null) { + t = t.getCause(); + } + if (t instanceof InterruptedException) { + Thread.currentThread().interrupt(); + LOG.warning("Fetcher interrupted in " + fetcherName + ": " + t.getMessage()); + } else { + LOG.severe("Unexpected error in " + fetcherName + ": " + t.getMessage()); + } + writeErrorToQueue(queue, t); + } + + private void finalizeFetcher( + BlockingQueue queue, FieldList schema, String fetcherName) { + if (Thread.currentThread().isInterrupted()) { + writeErrorToQueue(queue, new SQLException("Metadata fetch interrupted.")); + } + signalEndOfData(queue, schema); + LOG.info(fetcherName + " finished."); + } + private void signalEndOfData( BlockingQueue queue, FieldList resultSchemaFields) { try { LOG.info("Adding end signal to queue."); - queue.put(BigQueryFieldValueListWrapper.of(resultSchemaFields, null, true)); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - LOG.warning("Interrupted while sending end signal to queue."); + BigQueryFieldValueListWrapper element = + BigQueryFieldValueListWrapper.of(resultSchemaFields, null, true); + if (!queue.offer(element)) { + boolean wasInterrupted = Thread.interrupted(); + try { + queue.put(element); + } catch (InterruptedException e) { + LOG.warning("Interrupted while sending end signal to queue."); + wasInterrupted = true; + } finally { + if (wasInterrupted) { + Thread.currentThread().interrupt(); + } + } + } } catch (Exception e) { LOG.severe("Exception while sending end signal to queue: " + e.getMessage()); } } - private void shutdownExecutor(ExecutorService executor) { - if (executor == null || executor.isShutdown()) { - return; - } - LOG.info("Shutting down column executor service..."); - executor.shutdown(); - try { - if (!executor.awaitTermination(10, TimeUnit.SECONDS)) { - LOG.warning("Executor did not terminate gracefully after 10s, forcing shutdownNow()."); - List droppedTasks = executor.shutdownNow(); - LOG.warning( - "Executor shutdownNow() initiated. Dropped tasks count: " + droppedTasks.size()); - if (!executor.awaitTermination(10, TimeUnit.SECONDS)) { - LOG.severe("Executor did not terminate even after shutdownNow()."); + private List fetchMatchingDatasets( + String catalog, String schemaPattern, Pattern schemaRegex) throws SQLException { + List projects = + (catalog != null) ? Collections.singletonList(catalog) : getAccessibleCatalogNames(); + List allDatasets = new ArrayList<>(); + for (String project : projects) { + if (Thread.currentThread().isInterrupted()) { + throw new SQLException("Interrupted while fetching matching datasets"); + } + try { + List 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); } - LOG.info("Executor shutdown complete."); - } catch (InterruptedException ie) { - LOG.warning( - "Interrupted while waiting for executor termination. Forcing shutdownNow() again."); - executor.shutdownNow(); - Thread.currentThread().interrupt(); } - } - - private String getCurrentCatalogName() { - return this.connection.getCatalog(); + return allDatasets; } private List getAccessibleCatalogNames() throws SQLException { Set accessibleCatalogs = new HashSet<>(); - String primaryCatalog = getCurrentCatalogName(); + String primaryCatalog = this.connection.getCatalog(); if (primaryCatalog != null && !primaryCatalog.isEmpty()) { accessibleCatalogs.add(primaryCatalog); } @@ -5026,4 +4916,22 @@ static String readSqlFromFile(String filename) { 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); + if (!queue.offer(element)) { + boolean wasInterrupted = Thread.interrupted(); + try { + queue.put(element); + } catch (InterruptedException ie) { + LOG.warning("Failed to put exception to queue due to interruption."); + wasInterrupted = true; + } finally { + if (wasInterrupted) { + Thread.currentThread().interrupt(); + } + } + } + } } diff --git a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJsonResultSet.java b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJsonResultSet.java index b36a3e018823..998a189eae2b 100644 --- a/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJsonResultSet.java +++ b/java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJsonResultSet.java @@ -311,7 +311,9 @@ public void close() { this.isClosed = true; if (ownedTasks != null) { for (Future ownedTask : ownedTasks) { - ownedTask.cancel(true); + if (ownedTask != null) { + ownedTask.cancel(true); + } } } super.close(); diff --git a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java index 3a97f41924a8..2b72ed01b8a3 100644 --- a/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java +++ b/java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryDatabaseMetaDataTest.java @@ -1017,7 +1017,7 @@ public void testSortResults_Procedures_EmptyList() { } @Test - public void testFindMatchingBigQueryObjects_Routines_ListWithPattern() { + public void testFindMatchingBigQueryObjects_Routines_ListWithPattern() throws Exception { String catalog = "p-cat"; String schema = "d-sch"; String pattern = "proc_%"; @@ -1063,7 +1063,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListWithPattern() { } @Test - public void testFindMatchingBigQueryObjects_Routines_ListNoPattern() { + public void testFindMatchingBigQueryObjects_Routines_ListNoPattern() throws Exception { String catalog = "p-cat"; String schema = "d-sch"; String pattern = null; @@ -1102,7 +1102,7 @@ public void testFindMatchingBigQueryObjects_Routines_ListNoPattern() { } @Test - public void testFindMatchingBigQueryObjects_Routines_GetSpecific() { + public void testFindMatchingBigQueryObjects_Routines_GetSpecific() throws Exception { String catalog = "p-cat"; String schema = "d-sch"; String procNameExact = "exactprocname"; 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 789ccced452c..e40ba8bccd8c 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 @@ -792,8 +792,10 @@ public void testDatabaseMetadataGetSchemas() throws SQLException { Assertions.assertFalse(rsNoMatch.next()); // Test case 4: Get schemas with non-existent catalog - rsNoMatch = databaseMetaData.getSchemas("invalid-catalog", null); - Assertions.assertFalse(rsNoMatch.next()); + Assertions.assertThrows( + SQLException.class, + () -> databaseMetaData.getSchemas("invalid-catalog", null), + "Should throw SQLException for non-existent catalog"); connection.close(); } @@ -935,9 +937,9 @@ public void testDatabaseMetaDataGetFunctions() throws SQLException { rsEmptyFunction.next(), "Empty function name pattern should return no results"); rsEmptyFunction.close(); - // Test 9: Null catalog + // Test 9: Null catalog should return all functions (spec-compliant) ResultSet rsNullCatalog = databaseMetaData.getFunctions(null, testSchema, null); - Assertions.assertFalse(rsNullCatalog.next(), "Null catalog should return no results"); + Assertions.assertTrue(rsNullCatalog.next(), "Null catalog should return results"); rsNullCatalog.close(); connection.close(); }