Skip to content
This repository was archived by the owner on Feb 24, 2026. It is now read-only.

Commit b1c827f

Browse files
yoshi-automationyoshi-automation
authored
feat: add a Flush API to enable finer grained data commit needs for dataflow. (#272)
fix: add resource definition for Table/ReadStream/WriteStream message fix: add proper resource_reference for messages chore: update copyright committer: @xiaozhenliugg PiperOrigin-RevId: 311188524 Source-Author: Google APIs <noreply@google.com> Source-Date: Tue May 12 13:14:37 2020 -0700 Source-Repo: googleapis/googleapis Source-Sha: bf17ae5fd93929beb44ac4c6b04f5088c3ee4a02 Source-Link: googleapis/googleapis@bf17ae5
1 parent 9c9471a commit b1c827f

File tree

11 files changed

+1911
-126
lines changed

11 files changed

+1911
-126
lines changed

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/BigQueryWriteClient.java‎

Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
import com.google.cloud.bigquery.storage.v1alpha2.Storage.CreateWriteStreamRequest;
2727
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamRequest;
2828
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamResponse;
29+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsRequest;
30+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsResponse;
2931
import com.google.cloud.bigquery.storage.v1alpha2.Storage.GetWriteStreamRequest;
3032
import com.google.cloud.bigquery.storage.v1alpha2.Stream.WriteStream;
3133
import com.google.cloud.bigquery.storage.v1alpha2.stub.BigQueryWriteStub;
@@ -591,6 +593,108 @@ public final BatchCommitWriteStreamsResponse batchCommitWriteStreams(
591593
return stub.batchCommitWriteStreamsCallable();
592594
}
593595

596+
// AUTO-GENERATED DOCUMENTATION AND METHOD
597+
/**
598+
* Flushes rows to a BUFFERED stream. If users are appending rows to BUFFERED stream, flush
599+
* operation is required in order for the rows to become available for reading. A Flush operation
600+
* flushes up to any previously flushed offset in a BUFFERED stream, to the offset specified in
601+
* the request.
602+
*
603+
* <p>Sample code:
604+
*
605+
* <pre><code>
606+
* try (BigQueryWriteClient bigQueryWriteClient = BigQueryWriteClient.create()) {
607+
* WriteStreamName writeStream = WriteStreamName.of("[PROJECT]", "[DATASET]", "[TABLE]", "[STREAM]");
608+
* FlushRowsResponse response = bigQueryWriteClient.flushRows(writeStream);
609+
* }
610+
* </code></pre>
611+
*
612+
* @param writeStream Required. The stream that is the target of the flush operation.
613+
* @throws com.google.api.gax.rpc.ApiException if the remote call fails
614+
*/
615+
public final FlushRowsResponse flushRows(WriteStreamName writeStream) {
616+
FlushRowsRequest request =
617+
FlushRowsRequest.newBuilder()
618+
.setWriteStream(writeStream == null ? null : writeStream.toString())
619+
.build();
620+
return flushRows(request);
621+
}
622+
623+
// AUTO-GENERATED DOCUMENTATION AND METHOD
624+
/**
625+
* Flushes rows to a BUFFERED stream. If users are appending rows to BUFFERED stream, flush
626+
* operation is required in order for the rows to become available for reading. A Flush operation
627+
* flushes up to any previously flushed offset in a BUFFERED stream, to the offset specified in
628+
* the request.
629+
*
630+
* <p>Sample code:
631+
*
632+
* <pre><code>
633+
* try (BigQueryWriteClient bigQueryWriteClient = BigQueryWriteClient.create()) {
634+
* WriteStreamName writeStream = WriteStreamName.of("[PROJECT]", "[DATASET]", "[TABLE]", "[STREAM]");
635+
* FlushRowsResponse response = bigQueryWriteClient.flushRows(writeStream.toString());
636+
* }
637+
* </code></pre>
638+
*
639+
* @param writeStream Required. The stream that is the target of the flush operation.
640+
* @throws com.google.api.gax.rpc.ApiException if the remote call fails
641+
*/
642+
public final FlushRowsResponse flushRows(String writeStream) {
643+
FlushRowsRequest request = FlushRowsRequest.newBuilder().setWriteStream(writeStream).build();
644+
return flushRows(request);
645+
}
646+
647+
// AUTO-GENERATED DOCUMENTATION AND METHOD
648+
/**
649+
* Flushes rows to a BUFFERED stream. If users are appending rows to BUFFERED stream, flush
650+
* operation is required in order for the rows to become available for reading. A Flush operation
651+
* flushes up to any previously flushed offset in a BUFFERED stream, to the offset specified in
652+
* the request.
653+
*
654+
* <p>Sample code:
655+
*
656+
* <pre><code>
657+
* try (BigQueryWriteClient bigQueryWriteClient = BigQueryWriteClient.create()) {
658+
* WriteStreamName writeStream = WriteStreamName.of("[PROJECT]", "[DATASET]", "[TABLE]", "[STREAM]");
659+
* FlushRowsRequest request = FlushRowsRequest.newBuilder()
660+
* .setWriteStream(writeStream.toString())
661+
* .build();
662+
* FlushRowsResponse response = bigQueryWriteClient.flushRows(request);
663+
* }
664+
* </code></pre>
665+
*
666+
* @param request The request object containing all of the parameters for the API call.
667+
* @throws com.google.api.gax.rpc.ApiException if the remote call fails
668+
*/
669+
public final FlushRowsResponse flushRows(FlushRowsRequest request) {
670+
return flushRowsCallable().call(request);
671+
}
672+
673+
// AUTO-GENERATED DOCUMENTATION AND METHOD
674+
/**
675+
* Flushes rows to a BUFFERED stream. If users are appending rows to BUFFERED stream, flush
676+
* operation is required in order for the rows to become available for reading. A Flush operation
677+
* flushes up to any previously flushed offset in a BUFFERED stream, to the offset specified in
678+
* the request.
679+
*
680+
* <p>Sample code:
681+
*
682+
* <pre><code>
683+
* try (BigQueryWriteClient bigQueryWriteClient = BigQueryWriteClient.create()) {
684+
* WriteStreamName writeStream = WriteStreamName.of("[PROJECT]", "[DATASET]", "[TABLE]", "[STREAM]");
685+
* FlushRowsRequest request = FlushRowsRequest.newBuilder()
686+
* .setWriteStream(writeStream.toString())
687+
* .build();
688+
* ApiFuture&lt;FlushRowsResponse&gt; future = bigQueryWriteClient.flushRowsCallable().futureCall(request);
689+
* // Do something
690+
* FlushRowsResponse response = future.get();
691+
* }
692+
* </code></pre>
693+
*/
694+
public final UnaryCallable<FlushRowsRequest, FlushRowsResponse> flushRowsCallable() {
695+
return stub.flushRowsCallable();
696+
}
697+
594698
@Override
595699
public final void close() {
596700
stub.close();

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/BigQueryWriteSettings.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,8 @@
3333
import com.google.cloud.bigquery.storage.v1alpha2.Storage.CreateWriteStreamRequest;
3434
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamRequest;
3535
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamResponse;
36+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsRequest;
37+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsResponse;
3638
import com.google.cloud.bigquery.storage.v1alpha2.Storage.GetWriteStreamRequest;
3739
import com.google.cloud.bigquery.storage.v1alpha2.Stream.WriteStream;
3840
import com.google.cloud.bigquery.storage.v1alpha2.stub.BigQueryWriteStubSettings;
@@ -102,6 +104,11 @@ public UnaryCallSettings<GetWriteStreamRequest, WriteStream> getWriteStreamSetti
102104
return ((BigQueryWriteStubSettings) getStubSettings()).batchCommitWriteStreamsSettings();
103105
}
104106

107+
/** Returns the object with the settings used for calls to flushRows. */
108+
public UnaryCallSettings<FlushRowsRequest, FlushRowsResponse> flushRowsSettings() {
109+
return ((BigQueryWriteStubSettings) getStubSettings()).flushRowsSettings();
110+
}
111+
105112
public static final BigQueryWriteSettings create(BigQueryWriteStubSettings stub)
106113
throws IOException {
107114
return new BigQueryWriteSettings.Builder(stub.toBuilder()).build();
@@ -229,6 +236,11 @@ public UnaryCallSettings.Builder<GetWriteStreamRequest, WriteStream> getWriteStr
229236
return getStubSettingsBuilder().batchCommitWriteStreamsSettings();
230237
}
231238

239+
/** Returns the builder for the settings used for calls to flushRows. */
240+
public UnaryCallSettings.Builder<FlushRowsRequest, FlushRowsResponse> flushRowsSettings() {
241+
return getStubSettingsBuilder().flushRowsSettings();
242+
}
243+
232244
@Override
233245
public BigQueryWriteSettings build() throws IOException {
234246
return new BigQueryWriteSettings(this);

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/stub/BigQueryWriteStub.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
import com.google.cloud.bigquery.storage.v1alpha2.Storage.CreateWriteStreamRequest;
2727
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamRequest;
2828
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamResponse;
29+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsRequest;
30+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsResponse;
2931
import com.google.cloud.bigquery.storage.v1alpha2.Storage.GetWriteStreamRequest;
3032
import com.google.cloud.bigquery.storage.v1alpha2.Stream.WriteStream;
3133
import javax.annotation.Generated;
@@ -62,6 +64,10 @@ public UnaryCallable<GetWriteStreamRequest, WriteStream> getWriteStreamCallable(
6264
throw new UnsupportedOperationException("Not implemented: batchCommitWriteStreamsCallable()");
6365
}
6466

67+
public UnaryCallable<FlushRowsRequest, FlushRowsResponse> flushRowsCallable() {
68+
throw new UnsupportedOperationException("Not implemented: flushRowsCallable()");
69+
}
70+
6571
@Override
6672
public abstract void close();
6773
}

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/stub/BigQueryWriteStubSettings.java‎

Lines changed: 27 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@
3838
import com.google.cloud.bigquery.storage.v1alpha2.Storage.CreateWriteStreamRequest;
3939
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamRequest;
4040
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamResponse;
41+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsRequest;
42+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsResponse;
4143
import com.google.cloud.bigquery.storage.v1alpha2.Storage.GetWriteStreamRequest;
4244
import com.google.cloud.bigquery.storage.v1alpha2.Stream.WriteStream;
4345
import com.google.common.collect.ImmutableList;
@@ -99,6 +101,7 @@ public class BigQueryWriteStubSettings extends StubSettings<BigQueryWriteStubSet
99101
finalizeWriteStreamSettings;
100102
private final UnaryCallSettings<BatchCommitWriteStreamsRequest, BatchCommitWriteStreamsResponse>
101103
batchCommitWriteStreamsSettings;
104+
private final UnaryCallSettings<FlushRowsRequest, FlushRowsResponse> flushRowsSettings;
102105

103106
/** Returns the object with the settings used for calls to createWriteStream. */
104107
public UnaryCallSettings<CreateWriteStreamRequest, WriteStream> createWriteStreamSettings() {
@@ -127,6 +130,11 @@ public UnaryCallSettings<GetWriteStreamRequest, WriteStream> getWriteStreamSetti
127130
return batchCommitWriteStreamsSettings;
128131
}
129132

133+
/** Returns the object with the settings used for calls to flushRows. */
134+
public UnaryCallSettings<FlushRowsRequest, FlushRowsResponse> flushRowsSettings() {
135+
return flushRowsSettings;
136+
}
137+
130138
@BetaApi("A restructuring of stub classes is planned, so this may break in the future")
131139
public BigQueryWriteStub createStub() throws IOException {
132140
if (getTransportChannelProvider()
@@ -201,6 +209,7 @@ protected BigQueryWriteStubSettings(Builder settingsBuilder) throws IOException
201209
getWriteStreamSettings = settingsBuilder.getWriteStreamSettings().build();
202210
finalizeWriteStreamSettings = settingsBuilder.finalizeWriteStreamSettings().build();
203211
batchCommitWriteStreamsSettings = settingsBuilder.batchCommitWriteStreamsSettings().build();
212+
flushRowsSettings = settingsBuilder.flushRowsSettings().build();
204213
}
205214

206215
/** Builder for BigQueryWriteStubSettings. */
@@ -218,6 +227,7 @@ public static class Builder extends StubSettings.Builder<BigQueryWriteStubSettin
218227
private final UnaryCallSettings.Builder<
219228
BatchCommitWriteStreamsRequest, BatchCommitWriteStreamsResponse>
220229
batchCommitWriteStreamsSettings;
230+
private final UnaryCallSettings.Builder<FlushRowsRequest, FlushRowsResponse> flushRowsSettings;
221231

222232
private static final ImmutableMap<String, ImmutableSet<StatusCode.Code>>
223233
RETRYABLE_CODE_DEFINITIONS;
@@ -270,12 +280,15 @@ protected Builder(ClientContext clientContext) {
270280

271281
batchCommitWriteStreamsSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
272282

283+
flushRowsSettings = UnaryCallSettings.newUnaryCallSettingsBuilder();
284+
273285
unaryMethodSettingsBuilders =
274286
ImmutableList.<UnaryCallSettings.Builder<?, ?>>of(
275287
createWriteStreamSettings,
276288
getWriteStreamSettings,
277289
finalizeWriteStreamSettings,
278-
batchCommitWriteStreamsSettings);
290+
batchCommitWriteStreamsSettings,
291+
flushRowsSettings);
279292

280293
initDefaults(this);
281294
}
@@ -311,6 +324,11 @@ private static Builder initDefaults(Builder builder) {
311324
.setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("idempotent"))
312325
.setRetrySettings(RETRY_PARAM_DEFINITIONS.get("default"));
313326

327+
builder
328+
.flushRowsSettings()
329+
.setRetryableCodes(RETRYABLE_CODE_DEFINITIONS.get("non_idempotent"))
330+
.setRetrySettings(RETRY_PARAM_DEFINITIONS.get("default"));
331+
314332
return builder;
315333
}
316334

@@ -322,13 +340,15 @@ protected Builder(BigQueryWriteStubSettings settings) {
322340
getWriteStreamSettings = settings.getWriteStreamSettings.toBuilder();
323341
finalizeWriteStreamSettings = settings.finalizeWriteStreamSettings.toBuilder();
324342
batchCommitWriteStreamsSettings = settings.batchCommitWriteStreamsSettings.toBuilder();
343+
flushRowsSettings = settings.flushRowsSettings.toBuilder();
325344

326345
unaryMethodSettingsBuilders =
327346
ImmutableList.<UnaryCallSettings.Builder<?, ?>>of(
328347
createWriteStreamSettings,
329348
getWriteStreamSettings,
330349
finalizeWriteStreamSettings,
331-
batchCommitWriteStreamsSettings);
350+
batchCommitWriteStreamsSettings,
351+
flushRowsSettings);
332352
}
333353

334354
// NEXT_MAJOR_VER: remove 'throws Exception'
@@ -377,6 +397,11 @@ public UnaryCallSettings.Builder<GetWriteStreamRequest, WriteStream> getWriteStr
377397
return batchCommitWriteStreamsSettings;
378398
}
379399

400+
/** Returns the builder for the settings used for calls to flushRows. */
401+
public UnaryCallSettings.Builder<FlushRowsRequest, FlushRowsResponse> flushRowsSettings() {
402+
return flushRowsSettings;
403+
}
404+
380405
@Override
381406
public BigQueryWriteStubSettings build() throws IOException {
382407
return new BigQueryWriteStubSettings(this);

‎google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1alpha2/stub/GrpcBigQueryWriteStub.java‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,8 @@
3131
import com.google.cloud.bigquery.storage.v1alpha2.Storage.CreateWriteStreamRequest;
3232
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamRequest;
3333
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FinalizeWriteStreamResponse;
34+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsRequest;
35+
import com.google.cloud.bigquery.storage.v1alpha2.Storage.FlushRowsResponse;
3436
import com.google.cloud.bigquery.storage.v1alpha2.Storage.GetWriteStreamRequest;
3537
import com.google.cloud.bigquery.storage.v1alpha2.Stream.WriteStream;
3638
import com.google.common.collect.ImmutableMap;
@@ -103,6 +105,14 @@ public class GrpcBigQueryWriteStub extends BigQueryWriteStub {
103105
.setResponseMarshaller(
104106
ProtoUtils.marshaller(BatchCommitWriteStreamsResponse.getDefaultInstance()))
105107
.build();
108+
private static final MethodDescriptor<FlushRowsRequest, FlushRowsResponse>
109+
flushRowsMethodDescriptor =
110+
MethodDescriptor.<FlushRowsRequest, FlushRowsResponse>newBuilder()
111+
.setType(MethodDescriptor.MethodType.UNARY)
112+
.setFullMethodName("google.cloud.bigquery.storage.v1alpha2.BigQueryWrite/FlushRows")
113+
.setRequestMarshaller(ProtoUtils.marshaller(FlushRowsRequest.getDefaultInstance()))
114+
.setResponseMarshaller(ProtoUtils.marshaller(FlushRowsResponse.getDefaultInstance()))
115+
.build();
106116

107117
private final BackgroundResource backgroundResources;
108118

@@ -113,6 +123,7 @@ public class GrpcBigQueryWriteStub extends BigQueryWriteStub {
113123
finalizeWriteStreamCallable;
114124
private final UnaryCallable<BatchCommitWriteStreamsRequest, BatchCommitWriteStreamsResponse>
115125
batchCommitWriteStreamsCallable;
126+
private final UnaryCallable<FlushRowsRequest, FlushRowsResponse> flushRowsCallable;
116127

117128
private final GrpcStubCallableFactory callableFactory;
118129

@@ -212,6 +223,19 @@ public Map<String, String> extract(BatchCommitWriteStreamsRequest request) {
212223
}
213224
})
214225
.build();
226+
GrpcCallSettings<FlushRowsRequest, FlushRowsResponse> flushRowsTransportSettings =
227+
GrpcCallSettings.<FlushRowsRequest, FlushRowsResponse>newBuilder()
228+
.setMethodDescriptor(flushRowsMethodDescriptor)
229+
.setParamsExtractor(
230+
new RequestParamsExtractor<FlushRowsRequest>() {
231+
@Override
232+
public Map<String, String> extract(FlushRowsRequest request) {
233+
ImmutableMap.Builder<String, String> params = ImmutableMap.builder();
234+
params.put("write_stream", String.valueOf(request.getWriteStream()));
235+
return params.build();
236+
}
237+
})
238+
.build();
215239

216240
this.createWriteStreamCallable =
217241
callableFactory.createUnaryCallable(
@@ -234,6 +258,9 @@ public Map<String, String> extract(BatchCommitWriteStreamsRequest request) {
234258
batchCommitWriteStreamsTransportSettings,
235259
settings.batchCommitWriteStreamsSettings(),
236260
clientContext);
261+
this.flushRowsCallable =
262+
callableFactory.createUnaryCallable(
263+
flushRowsTransportSettings, settings.flushRowsSettings(), clientContext);
237264

238265
backgroundResources = new BackgroundResourceAggregation(clientContext.getBackgroundResources());
239266
}
@@ -260,6 +287,10 @@ public UnaryCallable<GetWriteStreamRequest, WriteStream> getWriteStreamCallable(
260287
return batchCommitWriteStreamsCallable;
261288
}
262289

290+
public UnaryCallable<FlushRowsRequest, FlushRowsResponse> flushRowsCallable() {
291+
return flushRowsCallable;
292+
}
293+
263294
@Override
264295
public final void close() {
265296
shutdown();

0 commit comments

Comments
 (0)