Using Jobs.insert instead of Jobs.query for queries

google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQuery.java

@@ -935,7 +935,7 @@ Page<List<FieldValue>> listTableData(String datasetId, String tableId,
   boolean cancel(JobId jobId);
 
   /**
-   * Runs the query associated with the request.
+   * Runs the query associated with the request, using an internally-generated random JobId.
    *
    * <p>Example of running a query.
    * <pre> {@code
@@ -986,6 +986,19 @@ Page<List<FieldValue>> listTableData(String datasetId, String tableId,
    */
   QueryResponse query(QueryRequest request);
 
+  /**
+   * Runs the query associated with the request, using the given job id.
+   *
+   * <p>See {@link #query(QueryRequest)} for examples on populating a {@link QueryRequest}. The
+   * recommended way to populate a JobId is the following:
+   *
+   * <p>
+   * <pre> {@code
+   *  JobId jobId = JobId.of(UUID.randomUUID().toString());
+   * }</pre>
+   */
+  QueryResponse query(QueryRequest request, JobId jobId);
+
   /**
    * Returns results of the query associated with the provided job.
    *

google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/BigQueryImpl.java

@@ -37,8 +37,11 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.Callable;
 
 final class BigQueryImpl extends BaseService<BigQueryOptions> implements BigQuery {
@@ -560,39 +563,47 @@ public Boolean call() {
 
   @Override
   public QueryResponse query(final QueryRequest request) {
+    JobId jobId = JobId.of(UUID.randomUUID().toString());
+    return query(request, jobId);
+  }
+
+  @Override
+  public QueryResponse query(final QueryRequest request, final JobId jobId) {
+    final JobId completeJobId = jobId.setProjectId(getOptions().getProjectId());
     try {
-      com.google.api.services.bigquery.model.QueryResponse results =
-          runWithRetries(new Callable<com.google.api.services.bigquery.model.QueryResponse>() {
+      QueryJobConfiguration queryJobConfiguration =
+          QueryJobConfiguration.newBuilder(request.getQuery())
+          .setPositionalParameters(request.getPositionalParameters())
+          .setNamedParameters(request.getNamedParameters())
+          .setDefaultDataset(request.getDefaultDataset())
+          .setDryRun(request.dryRun())
+          .setUseQueryCache(request.useQueryCache())
+          .setUseLegacySql(request.useLegacySql())
+          .build();
+
+      final JobInfo jobInfo =
+          JobInfo.newBuilder(queryJobConfiguration).setJobId(completeJobId).build();
+
+      com.google.api.services.bigquery.model.Job jobPb =
+          runWithRetries(new Callable<com.google.api.services.bigquery.model.Job>() {
             @Override
-            public com.google.api.services.bigquery.model.QueryResponse call() {
-              return bigQueryRpc.query(request.setProjectId(getOptions().getProjectId()).toPb());
+            public com.google.api.services.bigquery.model.Job call() {
+              return bigQueryRpc.create(jobInfo.toPb(),
+                  Collections.<BigQueryRpc.Option, Object>emptyMap());
             }
           }, getOptions().getRetrySettings(), EXCEPTION_HANDLER, getOptions().getClock());
-      QueryResponse.Builder builder = QueryResponse.newBuilder();
-      JobId completeJobId = JobId.fromPb(results.getJobReference());
-      builder.setJobId(completeJobId);
-      builder.setNumDmlAffectedRows(results.getNumDmlAffectedRows());
-      builder.setJobCompleted(results.getJobComplete());
-      List<TableRow> rowsPb = results.getRows();
-      if (results.getJobComplete()) {
-        builder.setJobCompleted(true);
-        QueryResult.Builder resultBuilder = transformQueryResults(completeJobId, rowsPb,
-            results.getPageToken(), getOptions(), ImmutableMap.<BigQueryRpc.Option, Object>of());
-        resultBuilder.setTotalBytesProcessed(results.getTotalBytesProcessed());
-        resultBuilder.setCacheHit(results.getCacheHit());
-        if (results.getSchema() != null) {
-          resultBuilder.setSchema(Schema.fromPb(results.getSchema()));
-        }
-        if (results.getTotalRows() != null) {
-          resultBuilder.setTotalRows(results.getTotalRows().longValue());
-        }
-        builder.setResult(resultBuilder.build());
+      Job job = Job.fromPb(this, jobPb);
+
+      List<QueryResultsOption> options = new ArrayList<>();
+      if (request.getMaxWaitTime() != null) {
+        options.add(QueryResultsOption.maxWaitTime(request.getMaxWaitTime()));
       }
-      if (results.getErrors() != null) {
-        builder.setExecutionErrors(
-            Lists.transform(results.getErrors(), BigQueryError.FROM_PB_FUNCTION));
+      if (request.getPageSize() != null) {
+        options.add(QueryResultsOption.pageSize(request.getPageSize()));
       }
-      return builder.build();
+      return getQueryResults(job.getJobId(),
+          options.toArray(new QueryResultsOption[options.size()]));
+
     } catch (RetryHelper.RetryHelperException e) {
       throw BigQueryException.translateAndThrow(e);
     }

google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/QueryJobConfiguration.java

@@ -16,15 +16,19 @@
 
 package com.google.cloud.bigquery;
 
+import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkNotNull;
 
 import com.google.api.services.bigquery.model.JobConfigurationQuery;
+import com.google.api.services.bigquery.model.QueryParameter;
 import com.google.cloud.bigquery.JobInfo.CreateDisposition;
 import com.google.cloud.bigquery.JobInfo.WriteDisposition;
 import com.google.cloud.bigquery.JobInfo.SchemaUpdateOption;
+import com.google.cloud.bigquery.QueryRequest.Builder;
 import com.google.common.base.MoreObjects.ToStringHelper;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
@@ -41,6 +45,8 @@ public final class QueryJobConfiguration extends JobConfiguration {
   private static final long serialVersionUID = -1108948249081804890L;
 
   private final String query;
+  private final ImmutableList<QueryParameterValue> positionalParameters;
+  private final ImmutableMap<String, QueryParameterValue> namedParameters;
   private final TableId destinationTable;
   private final Map<String, ExternalTableDefinition> tableDefinitions;
   private final List<UserDefinedFunction> userDefinedFunctions;
@@ -80,6 +86,8 @@ public static final class Builder
       extends JobConfiguration.Builder<QueryJobConfiguration, Builder> {
 
     private String query;
+    private List<QueryParameterValue> positionalParameters = Lists.newArrayList();
+    private Map<String, QueryParameterValue> namedParameters = Maps.newHashMap();
     private TableId destinationTable;
     private Map<String, ExternalTableDefinition> tableDefinitions;
     private List<UserDefinedFunction> userDefinedFunctions;
@@ -102,6 +110,8 @@ private Builder() {
     private Builder(QueryJobConfiguration jobConfiguration) {
       this();
       this.query = jobConfiguration.query;
+      this.namedParameters = jobConfiguration.namedParameters;
+      this.positionalParameters = jobConfiguration.positionalParameters;
       this.destinationTable = jobConfiguration.destinationTable;
       this.tableDefinitions = jobConfiguration.tableDefinitions;
       this.userDefinedFunctions = jobConfiguration.userDefinedFunctions;
@@ -122,6 +132,21 @@ private Builder(com.google.api.services.bigquery.model.JobConfiguration configur
       this();
       JobConfigurationQuery queryConfigurationPb = configurationPb.getQuery();
       this.query = queryConfigurationPb.getQuery();
+      if (queryConfigurationPb.getQueryParameters() != null && !queryConfigurationPb.getQueryParameters().isEmpty()) {
+        if (queryConfigurationPb.getQueryParameters().get(0).getName() == null) {
+          setPositionalParameters(
+              Lists.transform(queryConfigurationPb.getQueryParameters(), QueryRequest.POSITIONAL_PARAMETER_FROM_PB_FUNCTION));
+        } else {
+          Map<String, QueryParameterValue> values = Maps.newHashMap();
+          for (QueryParameter queryParameterPb : queryConfigurationPb.getQueryParameters()) {
+            checkNotNull(queryParameterPb.getName());
+            QueryParameterValue value = QueryParameterValue.fromPb(
+                queryParameterPb.getParameterValue(), queryParameterPb.getParameterType());
+            values.put(queryParameterPb.getName(), value);
+          }
+          setNamedParameters(values);
+        }
+      }
       allowLargeResults = queryConfigurationPb.getAllowLargeResults();
       useQueryCache = queryConfigurationPb.getUseQueryCache();
       flattenResults = queryConfigurationPb.getFlattenResults();
@@ -175,6 +200,88 @@ public Builder setQuery(String query) {
     }
 
 
+    /**
+     * Adds a positional query parameter to the list of query parameters. See
+     * {@link #setPositionalParameters(Iterable)} for more details on the input requirements.
+     *
+     * <p>A positional parameter cannot be added after named parameters have been added.
+     */
+    public Builder addPositionalParameter(QueryParameterValue value) {
+      checkNotNull(value);
+      if (!namedParameters.isEmpty()) {
+        throw new IllegalStateException(
+            "Positional parameters can't be combined with named parameters");
+      }
+      positionalParameters.add(value);
+      return this;
+    }
+
+    /**
+     * Sets the query parameters to a list of positional query parameters to use in the query.
+     *
+     * <p>The set of query parameters must either be all positional or all named parameters.
+     * Positional parameters are denoted in the query with a question mark (?).
+     *
+     * <p>Additionally, useLegacySql must be set to false; query parameters cannot be used with
+     * legacy SQL.
+     *
+     * <p>The values parameter can be set to null to clear out the positional
+     * parameters so that named parameters can be used instead.
+     */
+    public Builder setPositionalParameters(Iterable<QueryParameterValue> values) {
+      if (values == null || Iterables.isEmpty(values)) {
+        positionalParameters = Lists.newArrayList();
+      } else {
+        if (!this.namedParameters.isEmpty()) {
+          throw new IllegalStateException(
+              "Positional parameters can't be combined with named parameters");
+        }
+        this.positionalParameters = Lists.newArrayList(values);
+      }
+      return this;
+    }
+
+    /**
+     * Adds a named query parameter to the set of query parameters. See
+     * {@link #setNamedParameters(Map)} for more details on the input requirements.
+     *
+     * <p>A named parameter cannot be added after positional parameters have been added.
+     */
+    public Builder addNamedParameter(String name, QueryParameterValue value) {
+      checkNotNull(value);
+      if (!this.positionalParameters.isEmpty()) {
+        throw new IllegalStateException(
+            "Named parameters can't be combined with positional parameters");
+      }
+      namedParameters.put(name, value);
+      return this;
+    }
+
+    /**
+     * Sets the query parameters to a set of named query parameters to use in the query.
+     *
+     * <p>The set of query parameters must either be all positional or all named parameters. Named
+     * parameters are denoted using an @ prefix, e.g. @myParam for a parameter named "myParam".
+     *
+     * <p>Additionally, useLegacySql must be set to false; query parameters cannot be used with
+     * legacy SQL.
+     *
+     * <p>The values parameter can be set to null to clear out the named parameters so that
+     * positional parameters can be used instead.
+     */
+    public Builder setNamedParameters(Map<String, QueryParameterValue> values) {
+      if (values == null || values.isEmpty()) {
+        namedParameters = Maps.newHashMap();
+      } else {
+        if (!this.positionalParameters.isEmpty()) {
+          throw new IllegalStateException(
+              "Named parameters can't be combined with positional parameters");
+        }
+        this.namedParameters = Maps.newHashMap(values);
+      }
+      return this;
+    }
+
     /**
      * Sets the table where to put query results. If not provided a new table is created. This value
      * is required if {@link Builder#setAllowLargeResults(Boolean)} is set to {@code true}.
@@ -373,6 +480,16 @@ public QueryJobConfiguration build() {
   private QueryJobConfiguration(Builder builder) {
     super(builder);
     this.query = checkNotNull(builder.query);
+    checkNotNull(builder.positionalParameters);
+    checkNotNull(builder.namedParameters);
+    if (!builder.positionalParameters.isEmpty()) {
+      checkArgument(builder.namedParameters.isEmpty());
+    }
+    if (!builder.namedParameters.isEmpty()) {
+      checkArgument(builder.positionalParameters.isEmpty());
+    }
+    positionalParameters = ImmutableList.copyOf(builder.positionalParameters);
+    namedParameters = ImmutableMap.copyOf(builder.namedParameters);
     this.allowLargeResults = builder.allowLargeResults;
     this.createDisposition = builder.createDisposition;
     this.defaultDataset = builder.defaultDataset;
@@ -458,6 +575,21 @@ public String getQuery() {
   }
 
 
+  /**
+   * Returns the positional query parameters to use for the query.
+   */
+  public List<QueryParameterValue> getPositionalParameters() {
+    return positionalParameters;
+  }
+
+  /**
+   * Returns the named query parameters to use for the query.
+   */
+  public Map<String, QueryParameterValue> getNamedParameters() {
+    return namedParameters;
+  }
+
+
   /**
    * Returns the external tables definitions. If querying external data sources outside of BigQuery,
    * this value describes the data format, location and other properties of the data
@@ -547,6 +679,8 @@ public Builder toBuilder() {
   ToStringHelper toStringHelper() {
     return super.toStringHelper()
         .add("query", query)
+        .add("positionalParameters", positionalParameters)
+        .add("namedParameters", namedParameters)
         .add("destinationTable", destinationTable)
         .add("defaultDataset", defaultDataset)
         .add("allowLargeResults", allowLargeResults)
@@ -573,7 +707,8 @@ public boolean equals(Object obj) {
   @Override
   public int hashCode() {
     return Objects.hash(baseHashCode(), allowLargeResults, createDisposition, destinationTable,
-        defaultDataset, flattenResults, priority, query, tableDefinitions, useQueryCache,
+        defaultDataset, flattenResults, priority, query, positionalParameters,
+        namedParameters, tableDefinitions, useQueryCache,
         userDefinedFunctions, writeDisposition, dryRun, useLegacySql, maximumBillingTier,
         schemaUpdateOptions);
   }
@@ -596,6 +731,17 @@ com.google.api.services.bigquery.model.JobConfiguration toPb() {
         new com.google.api.services.bigquery.model.JobConfiguration();
     JobConfigurationQuery queryConfigurationPb = new JobConfigurationQuery();
     queryConfigurationPb.setQuery(query);
+    if (!positionalParameters.isEmpty()) {
+      List<QueryParameter> queryParametersPb
+          = Lists.transform(positionalParameters,
+              QueryRequest.POSITIONAL_PARAMETER_TO_PB_FUNCTION);
+      queryConfigurationPb.setQueryParameters(queryParametersPb);
+    } else if (!namedParameters.isEmpty()) {
+      List<QueryParameter> queryParametersPb
+          = Lists.transform(namedParameters.entrySet().asList(),
+              QueryRequest.NAMED_PARAMETER_TO_PB_FUNCTION);
+      queryConfigurationPb.setQueryParameters(queryParametersPb);
+    }
     configurationPb.setDryRun(dryRun());
     if (allowLargeResults != null) {
       queryConfigurationPb.setAllowLargeResults(allowLargeResults);

google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/QueryParameterValue.java

@@ -431,6 +431,8 @@ static QueryParameterValue fromPb(
           valuePb.getArrayValues()) {
         arrayValues.add(fromPb(elementValuePb, typePb.getArrayType()));
       }
+      valueBuilder.setArrayValues(arrayValues);
+      valueBuilder.setArrayType(StandardSQLTypeName.valueOf(typePb.getArrayType().getType()));
     }
     valueBuilder.setType(StandardSQLTypeName.valueOf(typePb.getType()));
 

google-cloud-bigquery/src/main/java/com/google/cloud/bigquery/spi/v2/BigQueryRpc.java

@@ -16,6 +16,7 @@
 
 package com.google.cloud.bigquery.spi.v2;
 
+import com.google.api.core.BetaApi;
 import com.google.api.services.bigquery.model.Dataset;
 import com.google.api.services.bigquery.model.GetQueryResultsResponse;
 import com.google.api.services.bigquery.model.Job;
@@ -209,6 +210,7 @@ Tuple<String, Iterable<TableRow>> listTableData(String projectId, String dataset
    *
    * @throws BigQueryException upon failure
    */
+  @BetaApi
   QueryResponse query(QueryRequest request);
 
   /**