Add support for BigQuery resumable uploads via a write channel

- Move BlobWriteChannel and BlobReadChannel to core module
- Rename BlobWriteChannel and BlobReadChannel to WriteChannel and ReadChannel
- Add abstract class BaseWriteChannel implementing entity-agnostic channel functionality
- Add BlobWriteChannel and BlobReadChannel implementation to gcloud-java-storage
- Add LoadConfiguration and modify LoadJobInfo to take configuration as a parameter
- Add BigQuery.writer method to return a writer given LoadConfiguration
- Add BigQueryRpc.open and .write methods to implement write channel
- Add TableDataWriteChannel class to support bigquery resumable streaming inserts
- Add unit and integration tests
- Update bigquery example with load-data action

Author: mziccard

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

@@ -662,4 +662,12 @@ Page<List<FieldValue>> listTableData(TableId tableId, TableDataListOption... opt
    * @throws BigQueryException upon failure
    */
   QueryResponse getQueryResults(JobId job, QueryResultsOption... options) throws BigQueryException;
+
+  /**
+   * Returns a channel to write data to be inserted into a BigQuery table. Data format and other
+   * options can be configured using the {@link LoadConfiguration} parameter.
+   *
+   * @throws BigQueryException upon failure
+   */
+  TableDataWriteChannel writer(LoadConfiguration loadConfiguration);
 }

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

@@ -619,6 +619,10 @@ private static QueryResult.Builder transformQueryResults(JobId jobId, List<Table
         .results(transformTableData(rowsPb));
   }
 
+  public TableDataWriteChannel writer(LoadConfiguration loadConfiguration) {
+    return new TableDataWriteChannel(options(), setProjectId(loadConfiguration));
+  }
+
   private Map<BigQueryRpc.Option, ?> optionMap(Option... options) {
     Map<BigQueryRpc.Option, Object> optionMap = Maps.newEnumMap(BigQueryRpc.Option.class);
     for (Option option : options) {
@@ -698,8 +702,7 @@ public TableId apply(TableId tableId) {
     if (job instanceof LoadJobInfo) {
       LoadJobInfo loadJob = (LoadJobInfo) job;
       LoadJobInfo.Builder loadBuilder = loadJob.toBuilder();
-      loadBuilder.destinationTable(setProjectId(loadJob.destinationTable()));
-      return loadBuilder.build();
+      return loadBuilder.configuration(setProjectId(loadJob.configuration())).build();
     }
     return job;
   }
@@ -711,4 +714,10 @@ private QueryRequest setProjectId(QueryRequest request) {
     }
     return builder.build();
   }
+
+  private LoadConfiguration setProjectId(LoadConfiguration configuration) {
+    LoadConfiguration.Builder builder = configuration.toBuilder();
+    builder.destinationTable(setProjectId(configuration.destinationTable()));
+    return builder.build();
+  }
 }

gcloud-java-bigquery/src/main/java/com/google/gcloud/bigquery/CopyJobInfo.java

@@ -95,8 +95,8 @@ public Builder destinationTable(TableId destinationTable) {
     /**
      * Sets whether the job is allowed to create new tables.
      *
-     * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.link">
-     *     Jobs: Link Configuration</a>
+     * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy.createDisposition">
+     *     Create Disposition</a>
      */
     public Builder createDisposition(CreateDisposition createDisposition) {
       this.createDisposition = createDisposition;
@@ -106,8 +106,8 @@ public Builder createDisposition(CreateDisposition createDisposition) {
     /**
      * Sets the action that should occur if the destination table already exists.
      *
-     * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.link">
-     *     Jobs: Link Configuration</a>
+     * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy.writeDisposition">
+     *     Write Disposition</a>
      */
     public Builder writeDisposition(WriteDisposition writeDisposition) {
       this.writeDisposition = writeDisposition;
@@ -145,8 +145,8 @@ public TableId destinationTable() {
   /**
    * Returns whether the job is allowed to create new tables.
    *
-   * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy">
-   *     Jobs: Copy Configuration</a>
+   * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy.createDisposition">
+   *     Create Disposition</a>
    */
   public CreateDisposition createDisposition() {
     return this.createDisposition;
@@ -155,8 +155,8 @@ public CreateDisposition createDisposition() {
   /**
    * Returns the action that should occur if the destination table already exists.
    *
-   * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy">
-   *     Jobs: Copy Configuration</a>
+   * @see <a href="https://cloud.google.com/bigquery/docs/reference/v2/jobs#configuration.copy.writeDisposition">
+   *     Write Disposition</a>
    */
   public WriteDisposition writeDisposition() {
     return writeDisposition;