Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import com.google.cloud.bigquery.storage.v1.TableName;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonNull;
import com.google.gson.JsonObject;
import com.google.protobuf.Descriptors.DescriptorValidationException;
import java.io.IOException;
Expand Down Expand Up @@ -319,10 +320,9 @@ public int[] executeBatch() throws SQLException {
if (useWriteAPI()) {
try (BigQueryWriteClient writeClient = this.connection.getBigQueryWriteClient()) {
LOG.info("Using Write API for bulk INSERT operation.");
ArrayList<BigQueryJdbcParameter> currentParameterList = this.batchParameters.peek();
if (this.insertSchema == null && this.insertTableName == null) {
QueryStatistics insertJobQueryStatistics =
getQueryStatistics(getWriteBatchJobConfiguration(currentParameterList));
getQueryStatistics(getJobConfig(this.currentQuery).build());
setInsertMetadata(insertJobQueryStatistics);
}
Comment on lines 323 to 327

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.

critical

Bypassing parameter binding for the dry run query by using getJobConfig(this.currentQuery).build() will cause executeBatch() to fail in production for any parameterized query (e.g., INSERT INTO table (col) VALUES (?)).

BigQuery requires all parameters to be bound or declared even during dry runs; otherwise, the backend will throw a BigQueryException indicating that a query parameter or placeholder was not found.

The unit test testExecuteBatchNullWithWriteAPI did not catch this because it mocks bigQuery.create(any(JobInfo.class)) to return a mocked job and statistics directly, bypassing the actual BigQuery dry run validation.

To fix this, we should retain the parameter binding for the dry run. If there was an issue with handling null values in BigQueryParameterHandler, we should address that specific issue within BigQueryParameterHandler or getWriteBatchJobConfiguration rather than removing parameter binding entirely.

Suggested change
if (this.insertSchema == null && this.insertTableName == null) {
QueryStatistics insertJobQueryStatistics =
getQueryStatistics(getWriteBatchJobConfiguration(currentParameterList));
getQueryStatistics(getJobConfig(this.currentQuery).build());
setInsertMetadata(insertJobQueryStatistics);
}
ArrayList<BigQueryJdbcParameter> currentParameterList = this.batchParameters.peek();
if (this.insertSchema == null && this.insertTableName == null) {
QueryStatistics insertJobQueryStatistics =
getQueryStatistics(getWriteBatchJobConfiguration(currentParameterList));
setInsertMetadata(insertJobQueryStatistics);
}


Expand Down Expand Up @@ -392,7 +392,9 @@ private long bulkInsertWithWriteAPI(BigQueryWriteClient bigQueryWriteClient)
JsonObject rowObject = new JsonObject();
for (int j = 0; j < parameterList.size(); j++) {
BigQueryJdbcParameter parameter = parameterList.get(j);
if (parameter.getSqlType() == StandardSQLTypeName.STRING) {
if (parameter.getValue() == null) {
rowObject.add(fieldLists.get(j).getName(), JsonNull.INSTANCE);
} else if (parameter.getSqlType() == StandardSQLTypeName.STRING) {
rowObject.addProperty(fieldLists.get(j).getName(), parameter.getValue().toString());
} else {
rowObject.addProperty(fieldLists.get(j).getName(), gson.toJson(parameter.getValue()));
Expand Down Expand Up @@ -449,17 +451,6 @@ private void setInsertMetadata(QueryStatistics statistics) throws SQLException {
this.insertTableName, this.insertSchema.toString());
}

QueryJobConfiguration getWriteBatchJobConfiguration(
ArrayList<BigQueryJdbcParameter> currentParameterList) throws SQLException {
LOG.finer("++enter++");
BigQueryParameterHandler batchHandler =
new BigQueryParameterHandler(this.parameterCount, currentParameterList);
QueryJobConfiguration.Builder jobConfiguration = getJobConfig(this.currentQuery);
jobConfiguration.setParameterMode("POSITIONAL");
jobConfiguration = batchHandler.configureParameters(jobConfiguration);
return jobConfiguration.build();
}

QueryJobConfiguration getStandardBatchJobConfiguration(String query) throws SQLException {
Comment on lines 453 to 454

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.

critical

Please restore the getWriteBatchJobConfiguration helper method to support parameter binding during the dry run for the Write API path.

  QueryJobConfiguration getWriteBatchJobConfiguration(
      ArrayList<BigQueryJdbcParameter> currentParameterList) throws SQLException {
    LOG.finer("++enter++");
    BigQueryParameterHandler batchHandler =
        new BigQueryParameterHandler(this.parameterCount, currentParameterList);
    QueryJobConfiguration.Builder jobConfiguration = getJobConfig(this.currentQuery);
    jobConfiguration.setParameterMode("POSITIONAL");
    jobConfiguration = batchHandler.configureParameters(jobConfiguration);
    return jobConfiguration.build();
  }

  QueryJobConfiguration getStandardBatchJobConfiguration(String query) throws SQLException {

LOG.finer("++enter++");
QueryJobConfiguration.Builder jobConfiguration = getJobConfig(query);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,11 +21,23 @@
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;

import com.google.cloud.bigquery.BigQuery;
import com.google.cloud.bigquery.Field;
import com.google.cloud.bigquery.Job;
import com.google.cloud.bigquery.JobInfo;
import com.google.cloud.bigquery.JobStatistics.QueryStatistics;
import com.google.cloud.bigquery.Schema;
import com.google.cloud.bigquery.StandardSQLTypeName;
import com.google.cloud.bigquery.TableId;
import com.google.cloud.bigquery.storage.v1.BigQueryWriteClient;
import com.google.cloud.bigquery.storage.v1.BigQueryWriteSettings;
import com.google.cloud.bigquery.storage.v1.TableFieldSchema;
import com.google.cloud.bigquery.storage.v1.TableSchema;
import com.google.cloud.bigquery.storage.v1.WriteStream;
import java.sql.Array;
import java.sql.Date;
import java.sql.ParameterMetaData;
Expand All @@ -39,6 +51,7 @@
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.Calendar;
import java.util.Collections;
import java.util.TimeZone;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
Expand Down Expand Up @@ -269,4 +282,60 @@ public void testSetObjectWithJavaTime() throws Exception {
preparedStatement.setObject(4, instant);
assertEquals(Timestamp.class, preparedStatement.parameterHandler.getType(4));
}

@Test
public void testExecuteBatchNullWithWriteAPI() throws Exception {
BigQuery bigQuery = mock(BigQuery.class);
doReturn("SQL").when(connection).getQueryDialect();
doReturn(bigQuery).when(connection).getBigQuery();
doReturn(true).when(connection).isEnableWriteAPI();
doReturn(1).when(connection).getWriteAPIActivationRowCount();
doReturn(100).when(connection).getWriteAPIAppendRowCount();

BigQueryPreparedStatement insertStmt =
new BigQueryPreparedStatement(
connection, "INSERT INTO dataset.table (WJXBFS1) VALUES (?)");

// setObject(1, null) defaults to STRING sql type
insertStmt.setObject(1, null);
insertStmt.addBatch();

// Mock parameter-free dry run job statistics
Job dryRunJob = mock(Job.class);
QueryStatistics statistics = mock(QueryStatistics.class);
Schema schema = Schema.of(Field.of("WJXBFS1", StandardSQLTypeName.FLOAT64));
doReturn(QueryStatistics.StatementType.INSERT).when(statistics).getStatementType();
doReturn(schema).when(statistics).getSchema();
doReturn(Collections.singletonList(TableId.of("proj", "ds", "tbl")))
.when(statistics)
.getReferencedTables();
doReturn(statistics).when(dryRunJob).getStatistics();
doReturn(dryRunJob).when(bigQuery).create(any(JobInfo.class));

BigQueryWriteClient writeClient = mock(BigQueryWriteClient.class);
BigQueryWriteSettings writeSettings = mock(BigQueryWriteSettings.class);
doReturn(BigQueryWriteSettings.newBuilder()).when(writeSettings).toBuilder();
doReturn(writeSettings).when(writeClient).getSettings();
WriteStream writeStream = mock(WriteStream.class);
doReturn("projects/p/datasets/d/tables/t/streams/s").when(writeStream).getName();
TableFieldSchema fieldSchema =
TableFieldSchema.newBuilder()
.setName("WJXBFS1")
.setType(TableFieldSchema.Type.DOUBLE)
.setMode(TableFieldSchema.Mode.NULLABLE)
.build();
TableSchema tableSchema = TableSchema.newBuilder().addFields(fieldSchema).build();
doReturn(tableSchema).when(writeStream).getTableSchema();
doReturn(writeStream).when(writeClient).createWriteStream(any());
doReturn(writeClient).when(connection).getBigQueryWriteClient();

// Dry run succeeds without parameter mismatch exception
try {
insertStmt.executeBatch();
} catch (Exception ignored) {
// Ignore gRPC connection errors in unit test environment
}
assertNotNull(insertStmt.insertSchema);
assertEquals("WJXBFS1", insertStmt.insertSchema.getFields().get(0).getName());
}
}
Loading