From bbadcdb09e78928ac210b2171d84251b36e8ca9b Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Fri, 12 Aug 2022 13:13:27 +0530 Subject: [PATCH 01/12] Added exponential-back-off to create read session to avoid table-not-found error --- .../google/cloud/bigquery/ConnectionImpl.java | 41 ++++++++++++++++++- 1 file changed, 40 insertions(+), 1 deletion(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java index 3a9937a074..2e457e052a 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java @@ -21,6 +21,7 @@ import com.google.api.core.BetaApi; import com.google.api.core.InternalApi; +import com.google.api.gax.retrying.RetrySettings; import com.google.api.services.bigquery.model.GetQueryResultsResponse; import com.google.api.services.bigquery.model.JobConfigurationQuery; import com.google.api.services.bigquery.model.QueryParameter; @@ -73,6 +74,7 @@ import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; +import org.threeten.bp.Duration; /** Implementation for {@link Connection}, the generic BigQuery connection API (not JDBC). */ class ConnectionImpl implements Connection { @@ -93,6 +95,34 @@ class ConnectionImpl implements Connection { private BlockingQueue bufferRow; // initialized lazily iff we end up using Read API + // retry config to retry on table not found errors. Ref: b/241134681 + private static final BigQueryRetryConfig TABLE_NOT_FOUND_RETRY_CONFIG = + BigQueryRetryConfig.newBuilder() + .retryOnMessage("Not found") + .retryOnMessage("NOT_FOUND") + .retryOnRegEx(".*not.*found.*table.*") + .build(); + + // retry setting to retry on table not found errors. Settings uses a max timeout of 20 mins which + // could be useful for really long running jobs. Ref: b/241134681 + private static RetrySettings getTableNotFoundRetrySettings() { + double retryDelayMultiplier = 2.0; + int maxAttempts = 45; + long initialRetryDelay = 5000L; + long maxRetryDelay = 30000L; // 30secs + long totalTimeOut = 1200000L; // 20min + return RetrySettings.newBuilder() + .setMaxAttempts(maxAttempts) + .setMaxRetryDelay(Duration.ofMillis(maxRetryDelay)) + .setTotalTimeout(Duration.ofMillis(totalTimeOut)) + .setInitialRetryDelay(Duration.ofMillis(initialRetryDelay)) + .setRetryDelayMultiplier(retryDelayMultiplier) + .setInitialRpcTimeout(Duration.ofMillis(totalTimeOut)) + .setRpcTimeoutMultiplier(retryDelayMultiplier) + .setMaxRpcTimeout(Duration.ofMillis(totalTimeOut)) + .build(); + } + ConnectionImpl( ConnectionSettings connectionSettings, BigQueryOptions bigQueryOptions, @@ -835,8 +865,17 @@ BigQueryResult highThroughPutRead( .setMaxStreamCount(1) // Currently just one stream is allowed // DO a regex check using order by and use multiple streams ; + // Using exponential-back-off to create read session to avoid table-not-found error. Ref: + // b/241134681 . This approach is a short term approach and should later be replaced with a + // job status poll based solution to make it more deterministic. + ReadSession readSession = + BigQueryRetryHelper.runWithRetries( + () -> bqReadClient.createReadSession(builder.build()), + getTableNotFoundRetrySettings(), + BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER, + bigQueryOptions.getClock(), + TABLE_NOT_FOUND_RETRY_CONFIG); - ReadSession readSession = bqReadClient.createReadSession(builder.build()); bufferRow = new LinkedBlockingDeque<>(getBufferSize()); Map arrowNameToIndex = new HashMap<>(); // deserialize and populate the buffer async, so that the client isn't blocked From a965bc3365ffcb834a75b615e2ab6d22fdad3683 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Fri, 12 Aug 2022 15:07:43 +0530 Subject: [PATCH 02/12] Added testForTableNotFound IT --- .../bigquery/it/ITNightlyBigQueryTest.java | 61 +++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java index 73bd21a30d..bd08d0ac23 100644 --- a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java +++ b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java @@ -23,9 +23,11 @@ import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import com.google.cloud.ServiceOptions; import com.google.cloud.bigquery.BigQuery; import com.google.cloud.bigquery.BigQueryError; import com.google.cloud.bigquery.BigQueryException; +import com.google.cloud.bigquery.BigQueryOptions; import com.google.cloud.bigquery.BigQueryResult; import com.google.cloud.bigquery.BigQuerySQLException; import com.google.cloud.bigquery.Connection; @@ -60,6 +62,7 @@ import java.util.List; import java.util.Map; import java.util.TimeZone; +import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.logging.Level; import java.util.logging.Logger; @@ -484,6 +487,64 @@ public void testPositionalParams() assertEquals(MULTI_LIMIT_RECS, cnt); } + @Test + // This testcase reads 500k rows for a public table to make sure we do not get + // table-not-found exception. Ref: b/241134681 + public void testForTableNotFound() throws SQLException { + int recordCnt = 500000; // 500k + String query = + String.format( + "SELECT * FROM `bigquery-samples.wikipedia_benchmark.Wiki10B` LIMIT %s", recordCnt); + + String dataSet = RemoteBigQueryHelper.generateDatasetName(); + String table = "TAB_" + UUID.randomUUID(); + createDataset(dataSet); + TableId targetTable = + TableId.of( + ServiceOptions.getDefaultProjectId(), + dataSet, + table); // table will be created implicitly + + ConnectionSettings conSet = + ConnectionSettings.newBuilder() + .setUseReadAPI(true) // enable read api + .setDestinationTable(targetTable) + .setAllowLargeResults(true) + .build(); + + Connection connection = + BigQueryOptions.getDefaultInstance().getService().createConnection(conSet); + BigQueryResult bigQueryResultSet = connection.executeSelect(query); + assertNotNull(getResultHashWiki(bigQueryResultSet)); // this iterated through all the rows + assertTrue( + (recordCnt == bigQueryResultSet.getTotalRows()) + || (-1 + == bigQueryResultSet + .getTotalRows())); // either job should return the actual count or -1 if the job + // is still running + try { + deleteTable(dataSet, table); + deleteDataset(dataSet); + } catch (Exception e) { + logger.log( + Level.WARNING, + String.format( + "Error [ %s ] while deleting dataset: %s , table: %s", + e.getMessage(), dataSet, table)); + } + } + + // this iterated through all the rows (just reads the title column) + private Long getResultHashWiki(BigQueryResult bigQueryResultSet) throws SQLException { + ResultSet rs = bigQueryResultSet.getResultSet(); + long hash = 0L; + System.out.print("\n Running"); + while (rs.next()) { + hash += rs.getString("title") == null ? 0 : rs.getString("title").hashCode(); + } + return hash; + } + // asserts the value of each row private static void testForAllDataTypeValues(ResultSet rs, int cnt) throws SQLException { // Testing JSON type From b16e9f9606e9161a1b20ee10b5952bf829942759 Mon Sep 17 00:00:00 2001 From: Owl Bot Date: Fri, 12 Aug 2022 10:56:01 +0000 Subject: [PATCH 03/12] =?UTF-8?q?=F0=9F=A6=89=20Updates=20from=20OwlBot=20?= =?UTF-8?q?post-processor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md --- README.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index d01ea50599..e01960a1ab 100644 --- a/README.md +++ b/README.md @@ -59,13 +59,13 @@ implementation 'com.google.cloud:google-cloud-bigquery' If you are using Gradle without BOM, add this to your dependencies: ```Groovy -implementation 'com.google.cloud:google-cloud-bigquery:2.14.3' +implementation 'com.google.cloud:google-cloud-bigquery:2.14.4' ``` If you are using SBT, add this to your dependencies: ```Scala -libraryDependencies += "com.google.cloud" % "google-cloud-bigquery" % "2.14.3" +libraryDependencies += "com.google.cloud" % "google-cloud-bigquery" % "2.14.4" ``` ## Authentication From 6816db8ca914ecd14fc572dbc203233abe334a77 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Fri, 19 Aug 2022 14:09:57 +0530 Subject: [PATCH 04/12] Set recordCnt to 5Mil --- .../google/cloud/bigquery/it/ITNightlyBigQueryTest.java | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java index bd08d0ac23..006c126b66 100644 --- a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java +++ b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/it/ITNightlyBigQueryTest.java @@ -488,10 +488,11 @@ public void testPositionalParams() } @Test - // This testcase reads 500k rows for a public table to make sure we do not get - // table-not-found exception. Ref: b/241134681 + // This testcase reads rows in bulk for a public table to make sure we do not get + // table-not-found exception. Ref: b/241134681 . This exception has been seen while reading data + // in bulk public void testForTableNotFound() throws SQLException { - int recordCnt = 500000; // 500k + int recordCnt = 50000000; // 5Mil String query = String.format( "SELECT * FROM `bigquery-samples.wikipedia_benchmark.Wiki10B` LIMIT %s", recordCnt); From 2a15e2b8cb1d9bd1743105ca9e91efe118b0445e Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Fri, 19 Aug 2022 14:13:43 +0530 Subject: [PATCH 05/12] Add polling logic @ getQueryResultsFirstPage, Removed retrial logic on table_not_found --- .../google/cloud/bigquery/ConnectionImpl.java | 97 +++++++++++-------- 1 file changed, 55 insertions(+), 42 deletions(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java index 2e457e052a..dc1dc7fb0a 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java @@ -95,14 +95,6 @@ class ConnectionImpl implements Connection { private BlockingQueue bufferRow; // initialized lazily iff we end up using Read API - // retry config to retry on table not found errors. Ref: b/241134681 - private static final BigQueryRetryConfig TABLE_NOT_FOUND_RETRY_CONFIG = - BigQueryRetryConfig.newBuilder() - .retryOnMessage("Not found") - .retryOnMessage("NOT_FOUND") - .retryOnRegEx(".*not.*found.*table.*") - .build(); - // retry setting to retry on table not found errors. Settings uses a max timeout of 20 mins which // could be useful for really long running jobs. Ref: b/241134681 private static RetrySettings getTableNotFoundRetrySettings() { @@ -865,16 +857,7 @@ BigQueryResult highThroughPutRead( .setMaxStreamCount(1) // Currently just one stream is allowed // DO a regex check using order by and use multiple streams ; - // Using exponential-back-off to create read session to avoid table-not-found error. Ref: - // b/241134681 . This approach is a short term approach and should later be replaced with a - // job status poll based solution to make it more deterministic. - ReadSession readSession = - BigQueryRetryHelper.runWithRetries( - () -> bqReadClient.createReadSession(builder.build()), - getTableNotFoundRetrySettings(), - BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER, - bigQueryOptions.getClock(), - TABLE_NOT_FOUND_RETRY_CONFIG); + ReadSession readSession = bqReadClient.createReadSession(builder.build()); bufferRow = new LinkedBlockingDeque<>(getBufferSize()); Map arrowNameToIndex = new HashMap<>(); @@ -1034,33 +1017,63 @@ GetQueryResultsResponse getQueryResultsFirstPage(JobId jobId) { jobId.getLocation() == null && bigQueryOptions.getLocation() != null ? bigQueryOptions.getLocation() : jobId.getLocation()); - try { - GetQueryResultsResponse results = - BigQueryRetryHelper.runWithRetries( - () -> - bigQueryRpc.getQueryResultsWithRowLimit( - completeJobId.getProject(), - completeJobId.getJob(), - completeJobId.getLocation(), - connectionSettings.getMaxResultPerPage()), - bigQueryOptions.getRetrySettings(), - BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER, - bigQueryOptions.getClock(), - retryConfig); - if (results.getErrors() != null) { - List bigQueryErrors = - results.getErrors().stream() - .map(BigQueryError.FROM_PB_FUNCTION) - .collect(Collectors.toList()); - // Throwing BigQueryException since there may be no JobId and we want to stay consistent - // with the case where there there is a HTTP error - throw new BigQueryException(bigQueryErrors); + // Implementing logic to poll the Job's status using getQueryResults as + // we do not get rows, rows count and schema unless the job is complete + // Ref: b/241134681 + // This logic will wait for approx (poolingIntervalMs + 10 seconds which is the default timeout + // for getQueryResults) per iteration of the loop + long startTimeMs = System.currentTimeMillis(); + long totalTimeOutMs = 18 * 60 * 60 * 1000; // 18 hours, which is the max timeout for the job + long poolingIntervalMs = 60000; // 1 min + GetQueryResultsResponse results = null; + + while ((System.currentTimeMillis() - startTimeMs) <= totalTimeOutMs) { + try { + results = + BigQueryRetryHelper.runWithRetries( + () -> + bigQueryRpc.getQueryResultsWithRowLimit( + completeJobId.getProject(), + completeJobId.getJob(), + completeJobId.getLocation(), + connectionSettings.getMaxResultPerPage()), + bigQueryOptions.getRetrySettings(), + BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER, + bigQueryOptions.getClock(), + retryConfig); + + if (results.getErrors() != null) { + List bigQueryErrors = + results.getErrors().stream() + .map(BigQueryError.FROM_PB_FUNCTION) + .collect(Collectors.toList()); + // Throwing BigQueryException since there may be no JobId and we want to stay consistent + // with the case where there is a HTTP error + throw new BigQueryException(bigQueryErrors); + } + } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { + throw BigQueryException.translateAndThrow(e); + } + + if (results.getJobComplete()) { + break; // The job is complete + + } else { // wait for the defined poolingIntervalMs and the loop will retry + try { + Thread.sleep(poolingIntervalMs); + logger.log(Level.FINE, "Pooling getQueryResults"); + } catch (InterruptedException e) { + logger.log( + Level.FINE, + String.format( + "\n Interrupted while waiting @ getQueryResultsFirstPage with error %s", + e.getMessage())); + } } - return results; - } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { - throw BigQueryException.translateAndThrow(e); } + + return results; } @VisibleForTesting From 1ba5a0d4e9ff9245fbb36360e17ccd452dad6fff Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Fri, 19 Aug 2022 16:17:07 +0530 Subject: [PATCH 06/12] Removed getTableNotFoundRetrySettings --- .../google/cloud/bigquery/ConnectionImpl.java | 22 ------------------- 1 file changed, 22 deletions(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java index dc1dc7fb0a..1c26374bf0 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java @@ -21,7 +21,6 @@ import com.google.api.core.BetaApi; import com.google.api.core.InternalApi; -import com.google.api.gax.retrying.RetrySettings; import com.google.api.services.bigquery.model.GetQueryResultsResponse; import com.google.api.services.bigquery.model.JobConfigurationQuery; import com.google.api.services.bigquery.model.QueryParameter; @@ -74,7 +73,6 @@ import org.apache.arrow.vector.ipc.message.MessageSerializer; import org.apache.arrow.vector.types.pojo.Field; import org.apache.arrow.vector.util.ByteArrayReadableSeekableByteChannel; -import org.threeten.bp.Duration; /** Implementation for {@link Connection}, the generic BigQuery connection API (not JDBC). */ class ConnectionImpl implements Connection { @@ -95,26 +93,6 @@ class ConnectionImpl implements Connection { private BlockingQueue bufferRow; // initialized lazily iff we end up using Read API - // retry setting to retry on table not found errors. Settings uses a max timeout of 20 mins which - // could be useful for really long running jobs. Ref: b/241134681 - private static RetrySettings getTableNotFoundRetrySettings() { - double retryDelayMultiplier = 2.0; - int maxAttempts = 45; - long initialRetryDelay = 5000L; - long maxRetryDelay = 30000L; // 30secs - long totalTimeOut = 1200000L; // 20min - return RetrySettings.newBuilder() - .setMaxAttempts(maxAttempts) - .setMaxRetryDelay(Duration.ofMillis(maxRetryDelay)) - .setTotalTimeout(Duration.ofMillis(totalTimeOut)) - .setInitialRetryDelay(Duration.ofMillis(initialRetryDelay)) - .setRetryDelayMultiplier(retryDelayMultiplier) - .setInitialRpcTimeout(Duration.ofMillis(totalTimeOut)) - .setRpcTimeoutMultiplier(retryDelayMultiplier) - .setMaxRpcTimeout(Duration.ofMillis(totalTimeOut)) - .build(); - } - ConnectionImpl( ConnectionSettings connectionSettings, BigQueryOptions bigQueryOptions, From 9226bf4453682011d7671268cfe2d9a47f6e6d33 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 10:14:24 +0530 Subject: [PATCH 07/12] Updated getQueryResultsWithRowLimit - Added timeoutMs param --- .../main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java index 871590ca4b..eecf5d36f5 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java @@ -315,7 +315,7 @@ GetQueryResultsResponse getQueryResults( * @throws BigQueryException upon failure */ GetQueryResultsResponse getQueryResultsWithRowLimit( - String projectId, String jobId, String location, Integer preFetchedRowLimit); + String projectId, String jobId, String location, Integer preFetchedRowLimit, Long timeoutMs); /** * Runs a BigQuery SQL query synchronously and returns query results if the query completes within From c09ad6a9b656ab0546412410850d9f6d472da296 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 10:15:03 +0530 Subject: [PATCH 08/12] Updated testGetQueryResultsFirstPage --- .../google/cloud/bigquery/ConnectionImplTest.java | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ConnectionImplTest.java b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ConnectionImplTest.java index 9543ccebf7..4b1b93487b 100644 --- a/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ConnectionImplTest.java +++ b/google-cloud-bigquery/src/test/java/com/google/cloud/bigquery/ConnectionImplTest.java @@ -315,14 +315,22 @@ public void testNextPageTask() throws InterruptedException { @Test public void testGetQueryResultsFirstPage() { when(bigqueryRpcMock.getQueryResultsWithRowLimit( - any(String.class), any(String.class), any(String.class), any(Integer.class))) + any(String.class), + any(String.class), + any(String.class), + any(Integer.class), + any(Long.class))) .thenReturn(GET_QUERY_RESULTS_RESPONSE); GetQueryResultsResponse response = connection.getQueryResultsFirstPage(QUERY_JOB); assertNotNull(response); assertEquals(GET_QUERY_RESULTS_RESPONSE, response); verify(bigqueryRpcMock, times(1)) .getQueryResultsWithRowLimit( - any(String.class), any(String.class), any(String.class), any(Integer.class)); + any(String.class), + any(String.class), + any(String.class), + any(Integer.class), + any(Long.class)); } // calls executeSelect with a nonFast query and exercises createQueryJob From e3f9d1ff874b4a4d6525cb33d34dd9a67592bf14 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 10:15:31 +0530 Subject: [PATCH 09/12] Updated getQueryResultsWithRowLimit - Add timeoutMs --- .../java/com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc.java index d6b57a3daf..a9ef3a817a 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc.java @@ -697,7 +697,7 @@ public GetQueryResultsResponse getQueryResults( @Override public GetQueryResultsResponse getQueryResultsWithRowLimit( - String projectId, String jobId, String location, Integer maxResultPerPage) { + String projectId, String jobId, String location, Integer maxResultPerPage, Long timeoutMs) { try { return bigquery .jobs() @@ -705,6 +705,7 @@ public GetQueryResultsResponse getQueryResultsWithRowLimit( .setPrettyPrint(false) .setLocation(location) .setMaxResults(Long.valueOf(maxResultPerPage)) + .setTimeoutMs(timeoutMs) .execute(); } catch (IOException ex) { throw translate(ex); From ac13c9926c3bb76befc26846fcf39b9829d9e66e Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 10:31:40 +0530 Subject: [PATCH 10/12] Updated getQueryResultsFirstPage - Modified polling logic and refactor --- .../google/cloud/bigquery/ConnectionImpl.java | 40 ++++++++----------- 1 file changed, 17 insertions(+), 23 deletions(-) diff --git a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java index 1c26374bf0..da79bd6ba1 100644 --- a/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java +++ b/google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/ConnectionImpl.java @@ -999,14 +999,14 @@ GetQueryResultsResponse getQueryResultsFirstPage(JobId jobId) { // Implementing logic to poll the Job's status using getQueryResults as // we do not get rows, rows count and schema unless the job is complete // Ref: b/241134681 - // This logic will wait for approx (poolingIntervalMs + 10 seconds which is the default timeout - // for getQueryResults) per iteration of the loop - long startTimeMs = System.currentTimeMillis(); - long totalTimeOutMs = 18 * 60 * 60 * 1000; // 18 hours, which is the max timeout for the job - long poolingIntervalMs = 60000; // 1 min + // This logic relies on backend for poll and wait.BigQuery guarantees that jobs make forward + // progress (a job won't get stuck in pending forever). + boolean jobComplete = false; GetQueryResultsResponse results = null; + long timeoutMs = + 60000; // defaulting to 60seconds. TODO(prashant): It should be made user configurable - while ((System.currentTimeMillis() - startTimeMs) <= totalTimeOutMs) { + while (!jobComplete) { try { results = BigQueryRetryHelper.runWithRetries( @@ -1015,7 +1015,8 @@ GetQueryResultsResponse getQueryResultsFirstPage(JobId jobId) { completeJobId.getProject(), completeJobId.getJob(), completeJobId.getLocation(), - connectionSettings.getMaxResultPerPage()), + connectionSettings.getMaxResultPerPage(), + timeoutMs), bigQueryOptions.getRetrySettings(), BigQueryBaseService.BIGQUERY_EXCEPTION_HANDLER, bigQueryOptions.getClock(), @@ -1026,29 +1027,22 @@ GetQueryResultsResponse getQueryResultsFirstPage(JobId jobId) { results.getErrors().stream() .map(BigQueryError.FROM_PB_FUNCTION) .collect(Collectors.toList()); - // Throwing BigQueryException since there may be no JobId and we want to stay consistent + // Throwing BigQueryException since there may be no JobId, and we want to stay consistent // with the case where there is a HTTP error throw new BigQueryException(bigQueryErrors); } } catch (BigQueryRetryHelper.BigQueryRetryHelperException e) { throw BigQueryException.translateAndThrow(e); } + jobComplete = results.getJobComplete(); - if (results.getJobComplete()) { - break; // The job is complete - - } else { // wait for the defined poolingIntervalMs and the loop will retry - try { - Thread.sleep(poolingIntervalMs); - logger.log(Level.FINE, "Pooling getQueryResults"); - } catch (InterruptedException e) { - logger.log( - Level.FINE, - String.format( - "\n Interrupted while waiting @ getQueryResultsFirstPage with error %s", - e.getMessage())); - } - } + // This log msg at Level.FINE might indicate that the job is still running and not stuck for + // very long running jobs. + logger.log( + Level.FINE, + String.format( + "jobComplete: %s , Polling getQueryResults with timeoutMs: %s", + jobComplete, timeoutMs)); } return results; From 7398531c12293e434cfc208e83cba726b7dcedf0 Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 11:02:15 +0530 Subject: [PATCH 11/12] Removed prev differences. Add getQueryResultsWithRowLimit --- google-cloud-bigquery/clirr-ignored-differences.xml | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/google-cloud-bigquery/clirr-ignored-differences.xml b/google-cloud-bigquery/clirr-ignored-differences.xml index 1d3d826812..2d05606a37 100644 --- a/google-cloud-bigquery/clirr-ignored-differences.xml +++ b/google-cloud-bigquery/clirr-ignored-differences.xml @@ -2,14 +2,14 @@ - + 7012 - com/google/cloud/bigquery/LoadConfiguration - java.util.List getDecimalTargetTypes() + com/google/cloud/bigquery/spi/v2/BigQueryRpc + com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer, java.lang.Long) 7012 - com/google/cloud/bigquery/LoadConfiguration$Builder - com.google.cloud.bigquery.LoadConfiguration$Builder setDecimalTargetTypes(java.util.List) + com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc + com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer, java.lang.Long) \ No newline at end of file From fec3c59e49bbddcc9dbe4ae5691a59e7b2cb381d Mon Sep 17 00:00:00 2001 From: Prashant Mishra Date: Mon, 22 Aug 2022 12:20:29 +0530 Subject: [PATCH 12/12] Removed prev differences. Add getQueryResultsWithRowLimit --- google-cloud-bigquery/clirr-ignored-differences.xml | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/google-cloud-bigquery/clirr-ignored-differences.xml b/google-cloud-bigquery/clirr-ignored-differences.xml index 2d05606a37..2ad26f9464 100644 --- a/google-cloud-bigquery/clirr-ignored-differences.xml +++ b/google-cloud-bigquery/clirr-ignored-differences.xml @@ -3,13 +3,15 @@ - 7012 + 7004 com/google/cloud/bigquery/spi/v2/BigQueryRpc - com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer, java.lang.Long) + com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer) + getQueryResultsWithRowLimit is just used by ConnectionImpl at the moment so it should be fine to update the signature instead of writing an overloaded method - 7012 + 7004 com/google/cloud/bigquery/spi/v2/HttpBigQueryRpc - com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer, java.lang.Long) + com.google.api.services.bigquery.model.GetQueryResultsResponse getQueryResultsWithRowLimit(java.lang.String, java.lang.String, java.lang.String, java.lang.Integer) + getQueryResultsWithRowLimit is just used by ConnectionImpl at the moment so it should be fine to update the signature instead of writing an overloaded method \ No newline at end of file