Skip to content

Commit

Permalink
Add support for BigQuery resumable uploads via a write channel
Browse files Browse the repository at this point in the history
- 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
  • Loading branch information
mziccard committed Jan 15, 2016
1 parent e737516 commit 8978335
Show file tree
Hide file tree
Showing 36 changed files with 2,140 additions and 1,132 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -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;
}
Expand All @@ -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();
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand Down
Loading

0 comments on commit 8978335

Please sign in to comment.