You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "reuvenlax (via GitHub)" <gi...@apache.org> on 2023/06/02 05:30:35 UTC
[GitHub] [beam] reuvenlax opened a new pull request, #26975: Add upsert and delete support to BigQueryIO
reuvenlax opened a new pull request, #26975:
URL: https://github.com/apache/beam/pull/26975
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214541990
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RowUpdateInformation.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.auto.value.AutoValue;
+
+/**
+ * This class indicates how to apply a row update to BigQuery. If UPDATE or DELETE is selected as
+ * the update type, then a sequence number must also be supplied to order the updates. Incorrect
+ * sequence numbers will result in unexpected state in the BigQuery table.
+ */
+@AutoValue
+public abstract class RowUpdateInformation {
+ public enum UpdateType {
+ // Insert the row into the table, ignoring the primary key. Inserting the same row twice will
+ // result in two copies
+ // of the row in the table.
+ INSERT,
Review Comment:
Curious why INSERT is needed in the first place. Looks like you just convert INSERT records into regular INSERTS and leave out the CDC semantics. If you were to force UPSERT for everything - wouldn't it make things safer, esp. with the possibility that with AT_LEAST_ONCE required semantics these INSERTS can result in duplicates (even though the input feed is "perfect" and there is no end-user error in the pipeline design).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1581140826
@slilichenko @johnjcasey any more comments on this PR? Sorry there are so many files (though most of them have minimal changes just to forward a new parameter).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1573226680
Assigning reviewers. If you would like to opt out of this review, comment `assign to next reviewer`:
R: @robertwb for label java.
R: @manavgarg for label io.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214568162
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -3068,6 +3148,17 @@ public WriteResult expand(PCollection<T> input) {
!getAutoSchemaUpdate(),
"withAutoSchemaUpdate only supported when using storage-api writes.");
}
+ if (getRowUpdateFn() != null) {
+ if (method != Method.STORAGE_API_AT_LEAST_ONCE) {
+ checkArgument(
+ getRowUpdateFn() == null,
+ "Row updates only supported when using STORAGE_API_AT_LEAST_ONCE.");
+ }
+ checkArgument(
+ getCreateDisposition() == CreateDisposition.CREATE_NEVER,
+ "CREATE_IF_NEEDED is not supported when applying row updates. Tables must be precreated "
+ + "with a primary key specified.");
+ }
Review Comment:
"supports CDC writes" basically means "has a primary key" (today it also requires clustering on those keys, though that requirement is being dropped soon). Unfortunately primary keys are still not exposed via the CRUD API, so we can't check this right now (this is the same reason we don't support CREATE_IF_NEEDED; once there is API support, we could support this creation mode as long as the user specifies a primary key). I believe support in the CRUD api will be there soon, at which point we can do this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222122708
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
+ * .apply(BigQueryIO.applyRowMutations()
+ * .to(my_project:my_dataset.my_table)
+ * .withSchema(schema)
+ * .withMethod(Write.Method.STORAGE_API_AT_LEAST_ONCE)
+ * .withCreateDisposition(Write.CreateDisposition.CREATE_NEVER));
+ * }</pre>
+ *
+ * <p>If writing a type other than TableRow (e.g. using {@link BigQueryIO#writeGenericRecords} or
+ * writing a custom user type), then the {@link Write#withRowMutationFn} method can be used to set
+ * an update type and sequence number for each record. For example:
+ *
+ * <pre>{@code
+ * PCollection<CdcEvent> cdcEvent = ...;
+ *
+ * cdcEvent.apply(BigQueryIO.write()
+ * .to("my-project:my_dataset.my_table")
+ * .withSchema(schema)
+ * .withFormatFunction(CdcEvent::getTableRow)
+ * .withRowMutationFn(cdc -> RowMutationInformation.of(cdc.getChangeType(), cdc.getSequenceNumber()))
Review Comment:
Naming is confusing. withRowMutationFn() implies that it will return RowMutation, not RowMutationInformation. Either make it return RowMutation in a single step, or rename to withRowMutationInformationFn.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -92,8 +93,26 @@ public class FakeDatasetService implements DatasetService, Serializable {
public void close() throws Exception {}
static class Stream {
+ static class Entry {
+ enum UpdateType {
+ INSERT,
Review Comment:
Is INSERT still needed?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -586,11 +639,35 @@ public ApiFuture<AppendRowsResponse> appendRows(long offset, ProtoRows rows)
}
TableRow tableRow =
TableRowToStorageApiProto.tableRowFromMessage(
- DynamicMessage.parseFrom(protoDescriptor, bytes));
+ DynamicMessage.parseFrom(protoDescriptor, bytes), false);
if (shouldFailRow.apply(tableRow)) {
rowIndexToErrorMessage.put(i, "Failing row " + tableRow.toPrettyString());
}
- tableRows.add(tableRow);
+ String insertTypeStr = null;
+ long csn = -1;
+ Descriptors.FieldDescriptor fieldDescriptor =
+ protoDescriptor.findFieldByName("_CHANGE_TYPE");
+ if (fieldDescriptor != null) {
+ insertTypeStr = (String) msg.getField(fieldDescriptor);
+ }
+ fieldDescriptor = protoDescriptor.findFieldByName("_CHANGE_SEQUENCE_NUMBER");
+ if (fieldDescriptor != null) {
+ csn = (long) msg.getField(fieldDescriptor);
+ }
+ Stream.Entry.UpdateType insertType = Stream.Entry.UpdateType.INSERT;
Review Comment:
Remove references to INSERT
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RowMutation.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+
+/**
+ * A convenience class for applying row updates to BigQuery using {@link
+ * BigQueryIO.Write#applyRowMutations}. This class encapsulates a {@link TableRow} payload along
+ * with information how to update the row. A sequence number must also be supplied to order the
+ * updates. Incorrect sequence numbers will result in unexpected state in the BigQuery table.
+ */
+@AutoValue
+public abstract class RowMutation {
+ public abstract TableRow getTableRow();
+
+ public abstract RowMutationInformation getMutationInformation();
+
+ public static RowMutation of(TableRow tableRow, RowMutationInformation rowMutationInformation) {
+ return new AutoValue_RowMutation(tableRow, rowMutationInformation);
+ }
+
+ public static class RowMutationCoder extends AtomicCoder<RowMutation> {
+ private static final RowMutationCoder INSTANCE = new RowMutationCoder();
+
+ public static RowMutationCoder of() {
+ return INSTANCE;
+ }
+
+ @Override
+ public void encode(RowMutation value, OutputStream outStream) throws IOException {
+ TableRowJsonCoder.of().encode(value.getTableRow(), outStream);
+ VarIntCoder.of()
+ .encode(value.getMutationInformation().getMutationType().ordinal(), outStream);
+ VarLongCoder.of().encode(value.getMutationInformation().getSequenceNumber(), outStream);
+ }
+
+ @Override
+ public RowMutation decode(InputStream inStream) throws IOException {
+ return RowMutation.of(
+ TableRowJsonCoder.of().decode(inStream),
+ RowMutationInformation.of(
Review Comment:
Not a bug, but this relies on parameter ordering. If the parameters get reversed, the coder would break. Perhaps, explicitly decode each part before constructing the Info object.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -160,7 +177,20 @@ void flush(long position) {
throw new RuntimeException("");
}
for (; nextFlushPosition <= position; ++nextFlushPosition) {
- tableContainer.addRow(stream.get((int) nextFlushPosition), "");
+ applyEntry(stream.get((int) nextFlushPosition));
+ }
+ }
+
+ void applyEntry(Entry entry) {
+ switch (entry.updateType) {
+ case INSERT:
Review Comment:
Remove?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
Review Comment:
can applyRowUpdates() explicitly set the method; making it less verbose?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java:
##########
@@ -75,6 +77,11 @@
* with the Storage write API.
*/
public class TableRowToStorageApiProto {
+ private static final String CDC_CHANGE_SQN_COLUMN = "_CHANGE_SEQUENCE_NUMBER";
+ private static final String CDC_CHANGE_TYPE_COLUMN = "_CHANGE_TYPE";
+ private static final Set<String> CDC_COLUMNS =
Review Comment:
This seems to be the third time these constants are defined in different places. Perhaps consolidate.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -586,11 +639,35 @@ public ApiFuture<AppendRowsResponse> appendRows(long offset, ProtoRows rows)
}
TableRow tableRow =
TableRowToStorageApiProto.tableRowFromMessage(
- DynamicMessage.parseFrom(protoDescriptor, bytes));
+ DynamicMessage.parseFrom(protoDescriptor, bytes), false);
if (shouldFailRow.apply(tableRow)) {
rowIndexToErrorMessage.put(i, "Failing row " + tableRow.toPrettyString());
}
- tableRows.add(tableRow);
+ String insertTypeStr = null;
+ long csn = -1;
+ Descriptors.FieldDescriptor fieldDescriptor =
+ protoDescriptor.findFieldByName("_CHANGE_TYPE");
Review Comment:
Change to refer to constants?
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -160,7 +177,20 @@ void flush(long position) {
throw new RuntimeException("");
}
for (; nextFlushPosition <= position; ++nextFlushPosition) {
- tableContainer.addRow(stream.get((int) nextFlushPosition), "");
+ applyEntry(stream.get((int) nextFlushPosition));
+ }
+ }
+
+ void applyEntry(Entry entry) {
+ switch (entry.updateType) {
+ case INSERT:
+ tableContainer.addRow(entry.tableRow, "");
+ break;
+ case UPSERT:
+ tableContainer.upsertRow(entry.tableRow, entry.sqn);
+ break;
+ case DELETE:
+ tableContainer.deleteRow(entry.tableRow, entry.sqn);
}
Review Comment:
break; for consistency + default: throw RuntimeException()?
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProtoTest.java:
##########
@@ -446,9 +449,32 @@ private void assertBaseRecord(DynamicMessage msg, Map<String, Object> expectedFi
public void testMessageFromGenericRecord() throws Exception {
Descriptors.Descriptor descriptor =
TableRowToStorageApiProto.getDescriptorFromTableSchema(
- AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA), true);
+ AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA),
+ true,
+ false);
DynamicMessage msg =
- AvroGenericRecordToStorageApiProto.messageFromGenericRecord(descriptor, nestedRecord);
+ AvroGenericRecordToStorageApiProto.messageFromGenericRecord(
+ descriptor, nestedRecord, null, -1);
+
+ assertEquals(2, msg.getAllFields().size());
+
+ Map<String, Descriptors.FieldDescriptor> fieldDescriptors =
+ descriptor.getFields().stream()
+ .collect(Collectors.toMap(Descriptors.FieldDescriptor::getName, Functions.identity()));
+ DynamicMessage nestedMsg = (DynamicMessage) msg.getField(fieldDescriptors.get("nested"));
+ assertBaseRecord(nestedMsg, baseProtoExpectedFields);
+ }
+
+ @Test
+ public void testCdcFields() throws Exception {
+ Descriptors.Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(
+ AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA),
+ true,
+ false);
+ DynamicMessage msg =
+ AvroGenericRecordToStorageApiProto.messageFromGenericRecord(
+ descriptor, nestedRecord, null, -1);
Review Comment:
Doesn't look like it's actually testing CDC fields - changeType is null.
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/TableContainer.java:
##########
@@ -53,12 +113,64 @@ long addRow(TableRow row, String id) {
}
}
+ void upsertRow(TableRow row, long sequenceNumber) {
+ List<Object> primaryKey = getPrimaryKey(row);
+ if (primaryKey == null) {
+ throw new RuntimeException("Upserts only allowed when using primary keys");
+ }
+ long lastSequenceNumberForKey = lastSequenceNumber.getOrDefault(primaryKey, -1L);
Review Comment:
Makes an assumption that sequence numbers will be positive, which is not necessarily the case. Recommend changing to negative infinitiy.
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProtoTest.java:
##########
@@ -1186,4 +1189,37 @@ public void testIgnoreUnknownNestedField() throws Exception {
((TableRow) unknown.get("nestedvalue1")).getF().get(BASE_TABLE_ROW.getF().size()).getV());
assertEquals("foobar", ((TableRow) unknown.get("nestedvaluenof1")).get("unknown"));
}
+
+ @Test
+ public void testCdcFields() throws Exception {
+ TableRow tableRow =
+ new TableRow()
+ .set("nestedValue1", BASE_TABLE_ROW)
+ .set("nestedValue2", BASE_TABLE_ROW)
+ .set("nestedValueNoF1", BASE_TABLE_ROW_NO_F)
+ .set("nestedValueNoF2", BASE_TABLE_ROW_NO_F);
+
+ Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(NESTED_TABLE_SCHEMA, true, true);
+ assertNotNull(descriptor.findFieldByName("_CHANGE_TYPE"));
Review Comment:
Use constants?
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java:
##########
@@ -392,6 +393,26 @@ public void testMessageFromTableRow() throws Exception {
assertBaseRecord(nestedMsg);
}
+ @Test
+ public void testCdcFields() throws Exception {
+ Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(
+ BeamRowToStorageApiProto.protoTableSchemaFromBeamSchema(NESTED_SCHEMA), true, true);
+ assertNotNull(descriptor.findFieldByName("_CHANGE_TYPE"));
+ assertNotNull(descriptor.findFieldByName("_CHANGE_SEQUENCE_NUMBER"));
+ DynamicMessage msg =
+ BeamRowToStorageApiProto.messageFromBeamRow(descriptor, NESTED_ROW, "UPDATE", 42);
+ assertEquals(5, msg.getAllFields().size());
+
+ Map<String, FieldDescriptor> fieldDescriptors =
+ descriptor.getFields().stream()
+ .collect(Collectors.toMap(FieldDescriptor::getName, Functions.identity()));
+ DynamicMessage nestedMsg = (DynamicMessage) msg.getField(fieldDescriptors.get("nested"));
+ assertBaseRecord(nestedMsg);
+ assertEquals("UPDATE", msg.getField(descriptor.findFieldByName("_CHANGE_TYPE")));
Review Comment:
Use constants?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1574334539
R: @johnjcasey
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1589498590
Run Java_GCP_IO_Direct PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222215575
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
Review Comment:
Not sure I understand - this is standin for user code (it's not a Beam function).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222106621
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java:
##########
@@ -172,7 +178,7 @@ public static TableSchema protoTableSchemaFromAvroSchema(Schema schema) {
* @return A dynamic message representation of a Proto payload to be used for StorageWrite API
*/
public static DynamicMessage messageFromGenericRecord(
- Descriptor descriptor, GenericRecord record) {
+ Descriptor descriptor, GenericRecord record, @Nullable String changeType, long csn) {
Review Comment:
Perhaps rename 'csn' to a more descriptive name.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222219585
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
Review Comment:
Never mind - it's the function body, not signature. Please ignore.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1227296763
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
Review Comment:
Done - added to resolveMethod
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1224798744
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProtoTest.java:
##########
@@ -1186,4 +1189,37 @@ public void testIgnoreUnknownNestedField() throws Exception {
((TableRow) unknown.get("nestedvalue1")).getF().get(BASE_TABLE_ROW.getF().size()).getV());
assertEquals("foobar", ((TableRow) unknown.get("nestedvaluenof1")).get("unknown"));
}
+
+ @Test
+ public void testCdcFields() throws Exception {
+ TableRow tableRow =
+ new TableRow()
+ .set("nestedValue1", BASE_TABLE_ROW)
+ .set("nestedValue2", BASE_TABLE_ROW)
+ .set("nestedValueNoF1", BASE_TABLE_ROW_NO_F)
+ .set("nestedValueNoF2", BASE_TABLE_ROW_NO_F);
+
+ Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(NESTED_TABLE_SCHEMA, true, true);
+ assertNotNull(descriptor.findFieldByName("_CHANGE_TYPE"));
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1224803177
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java:
##########
@@ -172,7 +178,7 @@ public static TableSchema protoTableSchemaFromAvroSchema(Schema schema) {
* @return A dynamic message representation of a Proto payload to be used for StorageWrite API
*/
public static DynamicMessage messageFromGenericRecord(
- Descriptor descriptor, GenericRecord record) {
+ Descriptor descriptor, GenericRecord record, @Nullable String changeType, long csn) {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214530192
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -3068,6 +3148,17 @@ public WriteResult expand(PCollection<T> input) {
!getAutoSchemaUpdate(),
"withAutoSchemaUpdate only supported when using storage-api writes.");
}
+ if (getRowUpdateFn() != null) {
+ if (method != Method.STORAGE_API_AT_LEAST_ONCE) {
Review Comment:
Why not "checkArgument(method == STORAGE_API_AT_LEAST_ONCE)" instead of if and check?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1589562153
Run Java_GCP_IO_Direct PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222117313
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
Review Comment:
is `getUpdateType(tableRow)` a valid method signature?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214536953
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -3068,6 +3148,17 @@ public WriteResult expand(PCollection<T> input) {
!getAutoSchemaUpdate(),
"withAutoSchemaUpdate only supported when using storage-api writes.");
}
+ if (getRowUpdateFn() != null) {
+ if (method != Method.STORAGE_API_AT_LEAST_ONCE) {
+ checkArgument(
+ getRowUpdateFn() == null,
+ "Row updates only supported when using STORAGE_API_AT_LEAST_ONCE.");
+ }
+ checkArgument(
+ getCreateDisposition() == CreateDisposition.CREATE_NEVER,
+ "CREATE_IF_NEEDED is not supported when applying row updates. Tables must be precreated "
+ + "with a primary key specified.");
+ }
Review Comment:
Does it make sense for non-dynamic destination to fetch the schema and see if the table supports CDC writes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1217515278
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -3068,6 +3148,17 @@ public WriteResult expand(PCollection<T> input) {
!getAutoSchemaUpdate(),
"withAutoSchemaUpdate only supported when using storage-api writes.");
}
+ if (getRowUpdateFn() != null) {
+ if (method != Method.STORAGE_API_AT_LEAST_ONCE) {
+ checkArgument(
+ getRowUpdateFn() == null,
+ "Row updates only supported when using STORAGE_API_AT_LEAST_ONCE.");
+ }
+ checkArgument(
+ getCreateDisposition() == CreateDisposition.CREATE_NEVER,
+ "CREATE_IF_NEEDED is not supported when applying row updates. Tables must be precreated "
+ + "with a primary key specified.");
+ }
Review Comment:
The way to do that is to query INFORMATION_SCHEMA (https://cloud.google.com/bigquery/docs/information-schema-table-constraints). The downside is an extra delay, permissions and complexity. Perhaps at some point an optional "withCDCTableValidation" method to make this an explicit request?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222118721
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
Review Comment:
can applyRowUpdates() explicitly set the method; making the overall BigQueryIO usage less verbose in this case? Or, for consistency, call it a new method? I understand that the latter has pretty broad implications on the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1224798288
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
Review Comment:
Removed sentence
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
Review Comment:
ack
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1224799157
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -586,11 +639,35 @@ public ApiFuture<AppendRowsResponse> appendRows(long offset, ProtoRows rows)
}
TableRow tableRow =
TableRowToStorageApiProto.tableRowFromMessage(
- DynamicMessage.parseFrom(protoDescriptor, bytes));
+ DynamicMessage.parseFrom(protoDescriptor, bytes), false);
if (shouldFailRow.apply(tableRow)) {
rowIndexToErrorMessage.put(i, "Failing row " + tableRow.toPrettyString());
}
- tableRows.add(tableRow);
+ String insertTypeStr = null;
+ long csn = -1;
+ Descriptors.FieldDescriptor fieldDescriptor =
+ protoDescriptor.findFieldByName("_CHANGE_TYPE");
+ if (fieldDescriptor != null) {
+ insertTypeStr = (String) msg.getField(fieldDescriptor);
+ }
+ fieldDescriptor = protoDescriptor.findFieldByName("_CHANGE_SEQUENCE_NUMBER");
+ if (fieldDescriptor != null) {
+ csn = (long) msg.getField(fieldDescriptor);
+ }
+ Stream.Entry.UpdateType insertType = Stream.Entry.UpdateType.INSERT;
Review Comment:
It's used internally, and also validates that we don't accidentally use the same stream object for both CDC and inserts
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax merged pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax merged PR #26975:
URL: https://github.com/apache/beam/pull/26975
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222104714
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProto.java:
##########
@@ -172,7 +178,7 @@ public static TableSchema protoTableSchemaFromAvroSchema(Schema schema) {
* @return A dynamic message representation of a Proto payload to be used for StorageWrite API
*/
public static DynamicMessage messageFromGenericRecord(
- Descriptor descriptor, GenericRecord record) {
+ Descriptor descriptor, GenericRecord record, @Nullable String changeType, long csn) {
Review Comment:
Update Javadoc with details on the two new parameters.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222114197
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
Review Comment:
I would remove reference to "Row inserts" and only leave the "upsert" logic. This should match the BigQuery documentation where INSERTS for CDC are not discussed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1589289181
Run Java_GCP_IO_Direct PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1588186062
R: @slilichenko any more comments on this PR?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1589397452
Run Java_GCP_IO_Direct PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] slilichenko commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "slilichenko (via GitHub)" <gi...@apache.org>.
slilichenko commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1222118721
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
Review Comment:
can applyRowUpdates() explicitly set the method; making the overall BigQueryIO usage less verbose in this case? Or, for consistency, make it a new method? I understand that the latter has pretty broad implications on the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1224798569
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -477,6 +476,56 @@
* reviewers mentioned <a
* href="https://github.com/apache/beam/blob/master/sdks/java/io/google-cloud-platform/OWNERS">
* here</a>.
+ *
+ * <h3>Upserts and deletes</h3>
+ *
+ * The connector also supports streaming row updates to BigQuery, with the following qualifications:
+ * - The CREATE_IF_NEEDED CreateDisposition is not supported. Tables must be precreated with primary
+ * keys. - Only the STORAGE_WRITE_API_AT_LEAST_ONCE method is supported.
+ *
+ * <p>Two types of updates are supported. UPSERT replaces the row with the matching primary key or
+ * inserts the row if non exists. DELETE removes the row with the matching primary key. Row inserts
+ * are still allowed as normal using a separate instance of the sink, however care must be taken not
+ * to violate primary key uniqueness constraints, as those constraints are not enforced by BigQuery.
+ * If a table contains multiple rows with the same primary key, then row updates may not work as
+ * expected. In particular, these inserts should _only_ be done using the exactly-once sink
+ * (STORAGE_WRITE_API), as the at-least once sink may duplicate inserts, violating the constraint.
+ *
+ * <p>Since PCollections are unordered, in order to properly sequence updates a sequence number must
+ * be set on each update. BigQuery uses this sequence number to ensure that updates are correctly
+ * applied to the table even if they arrive out of order.
+ *
+ * <p>The simplest way to apply row updates if applying {@link TableRow} object is to use the {@link
+ * Write#applyRowMutations} method. Each {@link RowMutation} element contains a {@link TableRow}, an
+ * update type (UPSERT or DELETE), and a sequence number to order the updates.
+ *
+ * <pre>{@code
+ * PCollection<TableRow> rows = ...;
+ * row.apply(MapElements
+ * .into(new TypeDescriptor<RowMutation>(){})
+ * .via(tableRow -> RowMutation.of(tableRow, getUpdateType(tableRow), getSequenceNumber(tableRow))))
+ * .apply(BigQueryIO.applyRowMutations()
+ * .to(my_project:my_dataset.my_table)
+ * .withSchema(schema)
+ * .withMethod(Write.Method.STORAGE_API_AT_LEAST_ONCE)
+ * .withCreateDisposition(Write.CreateDisposition.CREATE_NEVER));
+ * }</pre>
+ *
+ * <p>If writing a type other than TableRow (e.g. using {@link BigQueryIO#writeGenericRecords} or
+ * writing a custom user type), then the {@link Write#withRowMutationFn} method can be used to set
+ * an update type and sequence number for each record. For example:
+ *
+ * <pre>{@code
+ * PCollection<CdcEvent> cdcEvent = ...;
+ *
+ * cdcEvent.apply(BigQueryIO.write()
+ * .to("my-project:my_dataset.my_table")
+ * .withSchema(schema)
+ * .withFormatFunction(CdcEvent::getTableRow)
+ * .withRowMutationFn(cdc -> RowMutationInformation.of(cdc.getChangeType(), cdc.getSequenceNumber()))
Review Comment:
done
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RowMutation.java:
##########
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.api.services.bigquery.model.TableRow;
+import com.google.auto.value.AutoValue;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.beam.sdk.coders.AtomicCoder;
+import org.apache.beam.sdk.coders.VarIntCoder;
+import org.apache.beam.sdk.coders.VarLongCoder;
+
+/**
+ * A convenience class for applying row updates to BigQuery using {@link
+ * BigQueryIO.Write#applyRowMutations}. This class encapsulates a {@link TableRow} payload along
+ * with information how to update the row. A sequence number must also be supplied to order the
+ * updates. Incorrect sequence numbers will result in unexpected state in the BigQuery table.
+ */
+@AutoValue
+public abstract class RowMutation {
+ public abstract TableRow getTableRow();
+
+ public abstract RowMutationInformation getMutationInformation();
+
+ public static RowMutation of(TableRow tableRow, RowMutationInformation rowMutationInformation) {
+ return new AutoValue_RowMutation(tableRow, rowMutationInformation);
+ }
+
+ public static class RowMutationCoder extends AtomicCoder<RowMutation> {
+ private static final RowMutationCoder INSTANCE = new RowMutationCoder();
+
+ public static RowMutationCoder of() {
+ return INSTANCE;
+ }
+
+ @Override
+ public void encode(RowMutation value, OutputStream outStream) throws IOException {
+ TableRowJsonCoder.of().encode(value.getTableRow(), outStream);
+ VarIntCoder.of()
+ .encode(value.getMutationInformation().getMutationType().ordinal(), outStream);
+ VarLongCoder.of().encode(value.getMutationInformation().getSequenceNumber(), outStream);
+ }
+
+ @Override
+ public RowMutation decode(InputStream inStream) throws IOException {
+ return RowMutation.of(
+ TableRowJsonCoder.of().decode(inStream),
+ RowMutationInformation.of(
Review Comment:
not worried here since the parameters have different types. If anyone reversed the parameters without updating this code, there would be a compile error
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/TableRowToStorageApiProto.java:
##########
@@ -75,6 +77,11 @@
* with the Storage write API.
*/
public class TableRowToStorageApiProto {
+ private static final String CDC_CHANGE_SQN_COLUMN = "_CHANGE_SEQUENCE_NUMBER";
+ private static final String CDC_CHANGE_TYPE_COLUMN = "_CHANGE_TYPE";
+ private static final Set<String> CDC_COLUMNS =
Review Comment:
refactored
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java:
##########
@@ -92,8 +93,26 @@ public class FakeDatasetService implements DatasetService, Serializable {
public void close() throws Exception {}
static class Stream {
+ static class Entry {
+ enum UpdateType {
+ INSERT,
Review Comment:
yeah, because we now use it internal to the test fixture
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/TableContainer.java:
##########
@@ -53,12 +113,64 @@ long addRow(TableRow row, String id) {
}
}
+ void upsertRow(TableRow row, long sequenceNumber) {
+ List<Object> primaryKey = getPrimaryKey(row);
+ if (primaryKey == null) {
+ throw new RuntimeException("Upserts only allowed when using primary keys");
+ }
+ long lastSequenceNumberForKey = lastSequenceNumber.getOrDefault(primaryKey, -1L);
Review Comment:
done
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/AvroGenericRecordToStorageApiProtoTest.java:
##########
@@ -446,9 +449,32 @@ private void assertBaseRecord(DynamicMessage msg, Map<String, Object> expectedFi
public void testMessageFromGenericRecord() throws Exception {
Descriptors.Descriptor descriptor =
TableRowToStorageApiProto.getDescriptorFromTableSchema(
- AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA), true);
+ AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA),
+ true,
+ false);
DynamicMessage msg =
- AvroGenericRecordToStorageApiProto.messageFromGenericRecord(descriptor, nestedRecord);
+ AvroGenericRecordToStorageApiProto.messageFromGenericRecord(
+ descriptor, nestedRecord, null, -1);
+
+ assertEquals(2, msg.getAllFields().size());
+
+ Map<String, Descriptors.FieldDescriptor> fieldDescriptors =
+ descriptor.getFields().stream()
+ .collect(Collectors.toMap(Descriptors.FieldDescriptor::getName, Functions.identity()));
+ DynamicMessage nestedMsg = (DynamicMessage) msg.getField(fieldDescriptors.get("nested"));
+ assertBaseRecord(nestedMsg, baseProtoExpectedFields);
+ }
+
+ @Test
+ public void testCdcFields() throws Exception {
+ Descriptors.Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(
+ AvroGenericRecordToStorageApiProto.protoTableSchemaFromAvroSchema(NESTED_SCHEMA),
+ true,
+ false);
+ DynamicMessage msg =
+ AvroGenericRecordToStorageApiProto.messageFromGenericRecord(
+ descriptor, nestedRecord, null, -1);
Review Comment:
good catch - fixed.
##########
sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BeamRowToStorageApiProtoTest.java:
##########
@@ -392,6 +393,26 @@ public void testMessageFromTableRow() throws Exception {
assertBaseRecord(nestedMsg);
}
+ @Test
+ public void testCdcFields() throws Exception {
+ Descriptor descriptor =
+ TableRowToStorageApiProto.getDescriptorFromTableSchema(
+ BeamRowToStorageApiProto.protoTableSchemaFromBeamSchema(NESTED_SCHEMA), true, true);
+ assertNotNull(descriptor.findFieldByName("_CHANGE_TYPE"));
+ assertNotNull(descriptor.findFieldByName("_CHANGE_SEQUENCE_NUMBER"));
+ DynamicMessage msg =
+ BeamRowToStorageApiProto.messageFromBeamRow(descriptor, NESTED_ROW, "UPDATE", 42);
+ assertEquals(5, msg.getAllFields().size());
+
+ Map<String, FieldDescriptor> fieldDescriptors =
+ descriptor.getFields().stream()
+ .collect(Collectors.toMap(FieldDescriptor::getName, Functions.identity()));
+ DynamicMessage nestedMsg = (DynamicMessage) msg.getField(fieldDescriptors.get("nested"));
+ assertBaseRecord(nestedMsg);
+ assertEquals("UPDATE", msg.getField(descriptor.findFieldByName("_CHANGE_TYPE")));
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1589288919
Run Java PreCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214659977
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/RowUpdateInformation.java:
##########
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.bigquery;
+
+import com.google.auto.value.AutoValue;
+
+/**
+ * This class indicates how to apply a row update to BigQuery. If UPDATE or DELETE is selected as
+ * the update type, then a sequence number must also be supplied to order the updates. Incorrect
+ * sequence numbers will result in unexpected state in the BigQuery table.
+ */
+@AutoValue
+public abstract class RowUpdateInformation {
+ public enum UpdateType {
+ // Insert the row into the table, ignoring the primary key. Inserting the same row twice will
+ // result in two copies
+ // of the row in the table.
+ INSERT,
Review Comment:
removed INSERT support
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #26975:
URL: https://github.com/apache/beam/pull/26975#issuecomment-1574335497
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a diff in pull request #26975: Add upsert and delete support to BigQueryIO
Posted by "reuvenlax (via GitHub)" <gi...@apache.org>.
reuvenlax commented on code in PR #26975:
URL: https://github.com/apache/beam/pull/26975#discussion_r1214958826
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.java:
##########
@@ -3068,6 +3148,17 @@ public WriteResult expand(PCollection<T> input) {
!getAutoSchemaUpdate(),
"withAutoSchemaUpdate only supported when using storage-api writes.");
}
+ if (getRowUpdateFn() != null) {
+ if (method != Method.STORAGE_API_AT_LEAST_ONCE) {
Review Comment:
done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org