You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/04/20 20:27:45 UTC
[GitHub] [beam] reuvenlax opened a new pull request #14591: Dataflow update support for schemas
reuvenlax opened a new pull request #14591:
URL: https://github.com/apache/beam/pull/14591
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #14591:
URL: https://github.com/apache/beam/pull/14591#issuecomment-826926354
Thanks @reuvenlax!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #14591:
URL: https://github.com/apache/beam/pull/14591#issuecomment-826034586
Run XVR_Direct PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax merged pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
reuvenlax merged pull request #14591:
URL: https://github.com/apache/beam/pull/14591
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #14591:
URL: https://github.com/apache/beam/pull/14591#issuecomment-824155120
@reuvenlax it looks like this breaks `CommonCoderTest` (which runs the test cases in `standard_coders_test.yaml`). Maybe there needs to be some fallback for when encoding_position isn't specified?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax merged pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
reuvenlax merged pull request #14591:
URL: https://github.com/apache/beam/pull/14591
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on a change in pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on a change in pull request #14591:
URL: https://github.com/apache/beam/pull/14591#discussion_r617974791
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
##########
@@ -336,27 +387,30 @@ public InstrumentedType prepare(InstrumentedType instrumentedType) {
// The decode method of the generated Coder delegates to this method to evaluate all of the
// per-field Coders.
- static Row decodeDelegate(Schema schema, Coder[] coders, InputStream inputStream)
+ static Row decodeDelegate(
+ Schema schema, Coder[] coders, int[] encodingPosToIndex, InputStream inputStream)
throws IOException {
int fieldCount = VAR_INT_CODER.decode(inputStream);
BitSet nullFields = NULL_LIST_CODER.decode(inputStream);
- List<Object> fieldValues = Lists.newArrayListWithCapacity(coders.length);
- for (int i = 0; i < fieldCount; ++i) {
+ Object[] fieldValues = new Object[coders.length];
+ for (int encodingPos = 0; encodingPos < fieldCount; ++encodingPos) {
+ int rowIndex = encodingPosToIndex[encodingPos];
// In the case of a schema change going backwards, fieldCount might be > coders.length,
// in which case we drop the extra fields.
Review comment:
Added test. This is mostly so that if people emergency rollback updates things don't start crashing on them. (ideally you would roll back to a previous state snapshot, but in emergency situations it's nice that something works)
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/SchemaCoderCloudObjectTranslator.java
##########
@@ -61,11 +62,15 @@ public CloudObject toCloudObject(SchemaCoder target, SdkComponents sdkComponents
FROM_ROW_FUNCTION,
StringUtils.byteArrayToJsonString(
SerializableUtils.serializeToByteArray(target.getFromRowFunction())));
- Structs.addString(
- base,
- SCHEMA,
- StringUtils.byteArrayToJsonString(
- SchemaTranslation.schemaToProto(target.getSchema(), true).toByteArray()));
+
+ try {
+ Structs.addString(
+ base,
+ SCHEMA,
+ JsonFormat.printer().print(SchemaTranslation.schemaToProto(target.getSchema(), true)));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
Review comment:
yes
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on pull request #14591:
URL: https://github.com/apache/beam/pull/14591#issuecomment-825267794
Run XVR_Direct PostCommit
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a change in pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #14591:
URL: https://github.com/apache/beam/pull/14591#discussion_r618817939
##########
File path: runners/core-construction-java/src/test/java/org/apache/beam/runners/core/construction/CommonCoderTest.java
##########
@@ -332,6 +332,8 @@ private static Object convertValue(Object value, CommonCoder coderSpec, Coder co
} else if (s.equals(getUrn(StandardCoders.Enum.ROW))) {
Schema schema;
try {
+ // SchemaApi.Schema.Builder schemaBuilder = SchemaApi.Schema.newBuilder();
+ // JsonFormat.parser().merge(coderSpec.getPayload(), schemaBuilder);
Review comment:
nit: this looks like it was accidentally added
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java
##########
@@ -215,7 +215,19 @@ public static Schema schemaFromProto(SchemaApi.Schema protoSchema) {
}
builder.setOptions(optionsFromProto(protoSchema.getOptionsList()));
Schema schema = builder.build();
- schema.setEncodingPositions(encodingLocationMap);
+
+ Preconditions.checkState(encodingLocationMap.size() == schema.getFieldCount());
+ long dinstictEncodingPositions = encodingLocationMap.values().stream().distinct().count();
+ Preconditions.checkState(dinstictEncodingPositions <= schema.getFieldCount());
+ if (dinstictEncodingPositions < schema.getFieldCount() && schema.getFieldCount() > 0) {
+ // This means that encoding positions were not specified in the proto. Generally, we don't
+ // expect this to happen,
+ // but if it does happen, we expect none to be specified - in which case the should all be
+ // zero.
Review comment:
nit
```suggestion
// but if it does happen, we expect none to be specified - in which case they should all be
// zero.
```
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaTranslation.java
##########
@@ -215,7 +215,19 @@ public static Schema schemaFromProto(SchemaApi.Schema protoSchema) {
}
builder.setOptions(optionsFromProto(protoSchema.getOptionsList()));
Schema schema = builder.build();
- schema.setEncodingPositions(encodingLocationMap);
+
+ Preconditions.checkState(encodingLocationMap.size() == schema.getFieldCount());
+ long dinstictEncodingPositions = encodingLocationMap.values().stream().distinct().count();
+ Preconditions.checkState(dinstictEncodingPositions <= schema.getFieldCount());
+ if (dinstictEncodingPositions < schema.getFieldCount() && schema.getFieldCount() > 0) {
+ // This means that encoding positions were not specified in the proto. Generally, we don't
+ // expect this to happen,
Review comment:
Right now Python's row coder never sets encoding position, so Python pipelines that use SqlTransform will trigger this path
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #14591:
URL: https://github.com/apache/beam/pull/14591#issuecomment-824444101
Also fixed CommonCoderTest
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] TheNeuralBit commented on a change in pull request #14591: [BEAM-10277][BEAM-12198] Dataflow update support for schemas
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #14591:
URL: https://github.com/apache/beam/pull/14591#discussion_r617676763
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/coders/RowCoderGenerator.java
##########
@@ -336,27 +387,30 @@ public InstrumentedType prepare(InstrumentedType instrumentedType) {
// The decode method of the generated Coder delegates to this method to evaluate all of the
// per-field Coders.
- static Row decodeDelegate(Schema schema, Coder[] coders, InputStream inputStream)
+ static Row decodeDelegate(
+ Schema schema, Coder[] coders, int[] encodingPosToIndex, InputStream inputStream)
throws IOException {
int fieldCount = VAR_INT_CODER.decode(inputStream);
BitSet nullFields = NULL_LIST_CODER.decode(inputStream);
- List<Object> fieldValues = Lists.newArrayListWithCapacity(coders.length);
- for (int i = 0; i < fieldCount; ++i) {
+ Object[] fieldValues = new Object[coders.length];
+ for (int encodingPos = 0; encodingPos < fieldCount; ++encodingPos) {
+ int rowIndex = encodingPosToIndex[encodingPos];
// In the case of a schema change going backwards, fieldCount might be > coders.length,
// in which case we drop the extra fields.
Review comment:
Is there anything in `RowCoderTest` exercising the `i >= coders.length` case? I've always been a little confused about what would trigger it. I wouldn't think it would be possible for the old coder to read data from the new coder. Maybe we should exercise removing a field from the end of the field list?
##########
File path: runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/util/SchemaCoderCloudObjectTranslator.java
##########
@@ -61,11 +62,15 @@ public CloudObject toCloudObject(SchemaCoder target, SdkComponents sdkComponents
FROM_ROW_FUNCTION,
StringUtils.byteArrayToJsonString(
SerializableUtils.serializeToByteArray(target.getFromRowFunction())));
- Structs.addString(
- base,
- SCHEMA,
- StringUtils.byteArrayToJsonString(
- SchemaTranslation.schemaToProto(target.getSchema(), true).toByteArray()));
+
+ try {
+ Structs.addString(
+ base,
+ SCHEMA,
+ JsonFormat.printer().print(SchemaTranslation.schemaToProto(target.getSchema(), true)));
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
Review comment:
`CloudObjectsTest` should be exercising this right?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org