Skip to content

feat(bigquery-jdbc): add queryProcessThreadCount connection property #13480

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

Closed
Neenu1995 wants to merge 1 commit into from
Closed

feat(bigquery-jdbc): add queryProcessThreadCount connection property #13480

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
16 changes: 11 additions & 5 deletions java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -174,6 +174,7 @@ public class BigQueryConnection extends BigQueryNoOpsConnection {
boolean useQueryCache;
String queryDialect;
int metadataFetchThreadCount;
int queryProcessThreadCount;
boolean allowLargeResults;
String destinationTable;
String destinationDataset;
Expand Down Expand Up @@ -340,19 +341,20 @@ public class BigQueryConnection extends BigQueryNoOpsConnection {
this.filterTablesOnDefaultDataset = ds.getFilterTablesOnDefaultDataset();
this.requestGoogleDriveScope = ds.getRequestGoogleDriveScope();
this.metadataFetchThreadCount = ds.getMetadataFetchThreadCount();
this.queryProcessThreadCount = ds.getQueryProcessThreadCount();
this.requestReason = ds.getRequestReason();
this.connectionPoolSize = ds.getConnectionPoolSize();
this.listenerPoolSize = ds.getListenerPoolSize();
this.partnerToken = ds.getPartnerToken();

this.headerProvider = createHeaderProvider();
this.bigQuery = getBigQueryConnection();
// Fixed thread pool limits concurrent calls to prevent BigQuery metadata API throttling
// (independent tasks).
this.metadataExecutor = BigQueryJdbcMdc.newFixedThreadPool(metadataFetchThreadCount);
// Use a bounded cached thread pool to prevent unbounded thread creation (and OOMs)
// under heavy load, while ensuring a limit (e.g., 100) high enough to prevent deadlocks
// between interdependent producer/consumer tasks (like nextPageWorker and
// populateBufferWorker).
this.queryExecutor = BigQueryJdbcMdc.newBoundedCachedThreadPool(100);
// Cached thread pool prevents deadlocks between interdependent producer/consumer query
// execution tasks.
this.queryExecutor = BigQueryJdbcMdc.newBoundedCachedThreadPool(queryProcessThreadCount);
}
}

Expand Down Expand Up @@ -708,6 +710,10 @@ int getMetadataFetchThreadCount() {
return this.metadataFetchThreadCount;
}

int getQueryProcessThreadCount() {
return this.queryProcessThreadCount;
}

boolean isEnableWriteAPI() {
return enableWriteAPI;
}
Expand Down
9 changes: 7 additions & 2 deletions java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcMdc.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ static void clear() {
static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
MdcThreadPoolExecutor executor =
new MdcThreadPoolExecutor(
"Metadata Fetch Pool",
nThreads,
nThreads,
60L,
Expand All @@ -86,6 +87,7 @@ static ExecutorService newFixedThreadPool(int nThreads) {
*/
static ExecutorService newBoundedCachedThreadPool(int maxThreads, ThreadFactory threadFactory) {
return new MdcThreadPoolExecutor(
"Query Executor Pool",
0,
maxThreads,
60L,
Expand Down Expand Up @@ -125,15 +127,18 @@ public Thread newThread(Runnable r) {
}

private static class MdcThreadPoolExecutor extends ThreadPoolExecutor {
private final String poolName;

public MdcThreadPoolExecutor(
String poolName,
int corePoolSize,
int maximumPoolSize,
long keepAliveTime,
TimeUnit unit,
BlockingQueue<Runnable> workQueue,
ThreadFactory threadFactory) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
this.poolName = poolName;
}

private final AtomicBoolean warningLogged = new AtomicBoolean(false);
Expand All @@ -149,8 +154,8 @@ private void monitorQueueSaturation(int queueSize) {
if (queueSize >= warnThreshold) {
if (warningLogged.compareAndSet(false, true)) {
LOG.warning(
"Thread pool is saturating. Max pool size: %d, Active threads: %d, Queued tasks: %d. Consider increasing the thread count property.",
maxPoolSize, getActiveCount(), queueSize);
"[%s] Thread pool is saturating. Max pool size: %d, Active threads: %d, Queued tasks: %d. Consider increasing the metadataFetchThreadCount property.",
poolName, maxPoolSize, getActiveCount(), queueSize);
}
Comment on lines 154 to 159

@gemini-code-assist gemini-code-assist Bot Jun 15, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The warning message hardcodes metadataFetchThreadCount (which is also lowercase compared to the actual property name MetaDataFetchThreadCount). Since MdcThreadPoolExecutor is now shared between the metadata fetch pool and the query executor pool, this warning will be misleading if the query executor pool saturates.

Consider dynamically determining the property name based on the pool name to ensure the warning points to the correct configuration property.

      if (queueSize >= warnThreshold) {
        if (warningLogged.compareAndSet(false, true)) {
          String propertyName = "Query Executor Pool".equals(poolName)
              ? BigQueryJdbcUrlUtility.QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME
              : BigQueryJdbcUrlUtility.METADATA_FETCH_THREAD_COUNT_PROPERTY_NAME;
          LOG.warning(
              "[%s] Thread pool is saturating. Max pool size: %d, Active threads: %d, Queued tasks: %d. Consider increasing the %s property.",
              poolName, maxPoolSize, getActiveCount(), queueSize, propertyName);
        }

} else if (queueSize <= recoveryThreshold) {
if (warningLogged.get()) {
Expand Down
8 changes: 8 additions & 0 deletions java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcUrlUtility.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,8 @@ protected boolean removeEldestEntry(Map.Entry<String, Map<String, String>> eldes
Pattern.CASE_INSENSITIVE);
static final String METADATA_FETCH_THREAD_COUNT_PROPERTY_NAME = "MetaDataFetchThreadCount";
static final int DEFAULT_METADATA_FETCH_THREAD_COUNT_VALUE = 32;
static final String QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME = "QueryProcessThreadCount";
static final int DEFAULT_QUERY_PROCESS_THREAD_COUNT_VALUE = 100;
static final String RETRY_TIMEOUT_IN_SECS_PROPERTY_NAME = "Timeout";
static final long DEFAULT_RETRY_TIMEOUT_IN_SECS_VALUE = 0L;
static final String JOB_TIMEOUT_PROPERTY_NAME = "JobTimeout";
Expand Down Expand Up @@ -540,6 +542,12 @@ protected boolean removeEldestEntry(Map.Entry<String, Map<String, String>> eldes
"The number of threads used to call a DatabaseMetaData method.")
.setDefaultValue(String.valueOf(DEFAULT_METADATA_FETCH_THREAD_COUNT_VALUE))
.build(),
BigQueryConnectionProperty.newBuilder()
.setName(QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME)
.setDescription(
"The maximum number of threads used to process query results concurrently.")
.setDefaultValue(String.valueOf(DEFAULT_QUERY_PROCESS_THREAD_COUNT_VALUE))
.build(),
BigQueryConnectionProperty.newBuilder()
.setName(ENABLE_WRITE_API_PROPERTY_NAME)
.setDescription(
Expand Down
27 changes: 27 additions & 0 deletions java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/DataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,7 @@ public class DataSource implements javax.sql.DataSource {
private Boolean filterTablesOnDefaultDataset;
private Integer requestGoogleDriveScope;
private Integer metadataFetchThreadCount;
private Integer queryProcessThreadCount;
private String sslTrustStorePath;
private String sslTrustStorePassword;
private Map<String, String> labels;
Expand Down Expand Up @@ -248,6 +249,9 @@ public class DataSource implements javax.sql.DataSource {
.put(
BigQueryJdbcUrlUtility.METADATA_FETCH_THREAD_COUNT_PROPERTY_NAME,
(ds, val) -> ds.setMetadataFetchThreadCount(Integer.parseInt(val)))
.put(
BigQueryJdbcUrlUtility.QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME,
(ds, val) -> ds.setQueryProcessThreadCount(Integer.parseInt(val)))
.put(
BigQueryJdbcUrlUtility.SSL_TRUST_STORE_PROPERTY_NAME,
DataSource::setSSLTrustStorePath)
Expand Down Expand Up @@ -565,6 +569,11 @@ Properties createProperties() {
BigQueryJdbcUrlUtility.METADATA_FETCH_THREAD_COUNT_PROPERTY_NAME,
String.valueOf(this.metadataFetchThreadCount));
}
if (this.queryProcessThreadCount != null) {
connectionProperties.setProperty(
BigQueryJdbcUrlUtility.QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME,
String.valueOf(this.queryProcessThreadCount));
}
if (this.sslTrustStorePath != null) {
connectionProperties.setProperty(
BigQueryJdbcUrlUtility.SSL_TRUST_STORE_PROPERTY_NAME,
Expand Down Expand Up @@ -1085,6 +1094,24 @@ public void setMetadataFetchThreadCount(Integer metadataFetchThreadCount) {
this.metadataFetchThreadCount = metadataFetchThreadCount;
}

public Integer getQueryProcessThreadCount() {
return queryProcessThreadCount != null
? queryProcessThreadCount
: BigQueryJdbcUrlUtility.DEFAULT_QUERY_PROCESS_THREAD_COUNT_VALUE;
}

public void setQueryProcessThreadCount(Integer queryProcessThreadCount) {
if (queryProcessThreadCount != null) {
// Must be at least 4 to avoid thread starvation deadlocks between interdependent
// producer-consumer tasks (like nextPageWorker and populateBufferWorker).
validateMin(
queryProcessThreadCount,
4,
BigQueryJdbcUrlUtility.QUERY_PROCESS_THREAD_COUNT_PROPERTY_NAME);
}
this.queryProcessThreadCount = queryProcessThreadCount;
}

public String getSSLTrustStorePath() {
return sslTrustStorePath;
}
Expand Down
43 changes: 43 additions & 0 deletions java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/DataSourceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,4 +83,47 @@ public void testMetadataFetchThreadCountValidation() {
.contains(
"Invalid value for MetaDataFetchThreadCount. It must be greater than or equal to 1"));
}

@Test
public void testQueryProcessThreadCountValidation() {
DataSource dataSource = new DataSource();

// Setting values >= 4 should succeed
dataSource.setQueryProcessThreadCount(4);
assertEquals(4, dataSource.getQueryProcessThreadCount());

dataSource.setQueryProcessThreadCount(10);
assertEquals(10, dataSource.getQueryProcessThreadCount());

dataSource.setQueryProcessThreadCount(null); // Should fallback to default
assertEquals(
BigQueryJdbcUrlUtility.DEFAULT_QUERY_PROCESS_THREAD_COUNT_VALUE,
dataSource.getQueryProcessThreadCount());

// Setting value < 4 (e.g., 3, 0, -5) should throw BigQueryJdbcRuntimeException
BigQueryJdbcRuntimeException ex3 =
assertThrows(
BigQueryJdbcRuntimeException.class, () -> dataSource.setQueryProcessThreadCount(3));
assertTrue(
ex3.getMessage()
.contains(
"Invalid value for QueryProcessThreadCount. It must be greater than or equal to 4"));

BigQueryJdbcRuntimeException ex0 =
assertThrows(
BigQueryJdbcRuntimeException.class, () -> dataSource.setQueryProcessThreadCount(0));
assertTrue(
ex0.getMessage()
.contains(
"Invalid value for QueryProcessThreadCount. It must be greater than or equal to 4"));

BigQueryJdbcRuntimeException exNeg =
assertThrows(
BigQueryJdbcRuntimeException.class, () -> dataSource.setQueryProcessThreadCount(-5));
assertTrue(
exNeg
.getMessage()
.contains(
"Invalid value for QueryProcessThreadCount. It must be greater than or equal to 4"));
}
}
Loading