[BEAM-7274] Implement the Protobuf schema provider #8690

alexvanboxel · 2019-05-26T20:25:02Z

The default implementation for Protobuf messages. Supports both
generated as dynamic messages.

NOTE: This implementation doesn't support Unknown Fields. I like to have a discussion with the reviewers how todo this.

alexvanboxel · 2019-05-27T06:37:31Z

@reuvenlax here is the first draft of Protobuf support. I hope the unit test testCoder and testCoderOnDynamic proves that having oneof's work as nullable fields (if you look at the Java API., they are actually treated as such. I think almost all permutations are unit-tested.

Things to discuss: Unknown fields.

reuvenlax · 2019-05-29T06:27:57Z

I'm still reviewing this. A few high-level thoughts off the bat:

I'm not thrilled about creating a ProtoRow object, as I've so far attempted to prevent Row from becoming an "open" class that can by extended by users. I'd be happier if we could do this the way we did Avro classes, by generating a set of getters/setters (it would probably be more efficient as well)
There are various places where logical types would be preferable - e.g. all the places where multiple proto types map to a single Beam schema type. As written today the conversion is lossy - if you selected just that field, there would be no way to get the original proto type back; LogicalTypes provide a simple way of fixing this, though we could defer that to a future PR.

alexvanboxel · 2019-05-29T07:09:01Z

@reuvenlax I'll wait for your complete review, but keep in mind that I wanted a uniform way for handling concrete typed classes and DynamicType (DynamicType do require access via the FieldDescriptor). I had trouble getting it in the same way as the AvroUtils, but maybe this is due to my unfamiliarity with it. (I'm open to try to convert it, as most of the effort is spend in the tests anyway. The for internal use only note in the Java doc didn't seem encouraging to use though.

On LogicalTypes: Are you talking about Fixed32/Fixed64/UInt32/UInt64? I simply took the Scalar mapping described here: https://developers.google.com/protocol-buffers/docs/proto#scalar

If you are talking about Timestamp... well I like to have this configurable. Having it default to DATETIME will be a helper for making it available in Beam SQL or in output IO's like BigQuery.

alexvanboxel · 2019-05-29T11:35:53Z

Addition info, so I tried to get the logic using the FieldValueGetter with partial success, see experiment: alexvanboxel@d1a8e16

My main issue is with RowWithGetters private get(Map/List)Value values that assume they have enough information to construct the embedded complex data structures. I could get rid of my ProtoRow if I could let the handling of the List/Row/Map to it's dedicated FieldValueGetter

alexvanboxel · 2019-05-31T12:46:22Z

I got it to work with the RowWithGetters but not without modifying it. See alexvanboxel@1da5715

I keep on handling the List/Map/Row with the Getters as I need more context then the other implementations. Personally I find this better and more flexible then doing it in the RowWithGetters as you have more ways to give context (example: proto map is a repeatable message).

I looked for reference to the GenericRecord in Avro, but was disappointed that the GenericRecord gets materialised in a RowWithStorage. I'm quite sure with having the getters that handle collection this could be changed to RowWithGetters

...tensions/protobuf/src/test/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchemaTest.java

...a/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchema.java

reuvenlax · 2019-06-02T10:23:37Z

...rotobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoDescriptorSerializer.java

+ *
+ * <p>Follow https://github.com/protocolbuffers/protobuf/issues/6101 for the state of the feature
+ * request in ProtoBuf.
+ */


Could you instead convert the Descriptor to a proto (using the Descriptor.toProto method) and store that? Then I think you won't need this class at all.

So I removed this class as it was not super elegant, but .toProto() is not a solution as it only works in one way. A descriptor is always contained within a FileDescriptor.
I've introduced a Domain class that could group descriptors that are related with each other. This is an important component that a ProtoBuf schema registry should provision to make all Descriptors crosswired. See the Javadoc in the ProtoDomain class.

...java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoRow.java

...a/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchema.java

alexvanboxel · 2019-06-06T17:50:49Z

ok, I was wondering if I should take action on this. Get it in master and then rebase the other section. _/ _/ Alex Van Boxel

…

On Thu, Jun 6, 2019 at 7:49 PM reuvenlax ***@***.***> wrote: ***@***.**** commented on this pull request. ------------------------------ In sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoRow.java <#8690 (comment)>: > +import javax.annotation.Nullable; +import org.apache.beam.sdk.annotations.Experimental; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.schemas.SchemaCoder; +import org.apache.beam.sdk.transforms.SerializableFunction; +import org.apache.beam.sdk.values.Row; +import org.joda.time.Instant; + +/** + * ProtoRow extends the Row and does late materialisation. It hold a reference to the original proto + * message and has an overlay on each field of the proto message. It doesn't have it's own Coder as + * it relies on the SchemaCoder to (de)serialize the message over the wire. + * + * <p>Each row has a FieldOverlay that handles specific field conversions, as well has special + * overlays for Well Know Types, Repeatables, Maps and Nullables. + */ Yes. I want to review this first though. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#8690?email_source=notifications&email_token=AAE4EM5RTN33LJTSVV6H6ZLPZFE2XA5CNFSM4HPXM2D2YY3PNVWWK3TUL52HS4DFWFIHK3DMKJSXC5LFON2FEZLWNFSXPKTDN5WW2ZLOORPWSZGOB22KVRQ#discussion_r291299236>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAE4EM4BC2IAUU2VEUOP2CDPZFE2XANCNFSM4HPXM2DQ> .

reuvenlax · 2019-06-06T17:52:40Z

Yes. I'd prefer to keep Row as a somewhat sealed class and not introduce extra subtypes. However there was enough other things to review in this PR that I figured I could review that first.

alexvanboxel · 2019-06-06T17:56:18Z

OK, I'll fix the issues you mentioned, although I doubt that I can do the .toProto as it probably lacks contextual information to reconstruct the descriptors, as the Descriptor has a lot of possibilities to reference other components like Enums and other Messages. But I'll try... _/ _/ Alex Van Boxel

…

On Thu, Jun 6, 2019 at 7:52 PM reuvenlax ***@***.***> wrote: Yes. I'd prefer to keep Row as a somewhat sealed class and not introduce extra subtypes. However there was enough other things to review in this PR that I figured I could review that first. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#8690?email_source=notifications&email_token=AAE4EM4Y265L3KH2IBSAYBDPZFFHLA5CNFSM4HPXM2D2YY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGODXDUVAY#issuecomment-499600003>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAE4EM63XJIOB4GXYV3MEFTPZFFHLANCNFSM4HPXM2DQ> .

alexvanboxel · 2019-06-10T18:22:20Z

Yes. I'd prefer to keep Row as a somewhat sealed class and not introduce extra subtypes. However there was enough other things to review in this PR that I figured I could review that first.

Answered (and fixed comments), I also rebased the experiment to remove the ProtoRow as a subtype: alexvanboxel@1629557

reuvenlax · 2019-07-08T22:53:41Z

sdks/java/core/src/main/java/org/apache/beam/sdk/values/Row.java

+    public Builder withFieldValueGettersHandleCollections(boolean collectionHandledByGetter) {
+      this.collectionHandledByGetter = collectionHandledByGetter;
+      return this;
+    }


I'm having trouble understanding what this is for. Can you explain?

OK, maybe the name could be better, but it means that the the FieldValueGetters also handle collections like ARRAY, ROW, MAP. If you go to the original implementation ( https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/values/RowWithGetters.java#L90 ) you see that RowWithGetters does naive handling of ARRAY, ROW and MAP, for protobuf you need more context (the descriptor) to handle them. That's why I need to disable the naive mapping and let the FieldValueGetters handle ARRAY, ROW and MAP.

Feel free to suggest a better name though.

@reuvenlax is this the last remaining concern you have for this PR?

#8690 (comment) is a good reference for the motivation of this as well.

I'm not crazy about the way this is implemented since it's adding state to RowWithGetter that will get checked every time a collection field is accessed. I can't think of a better way to do it without some non-trivial refactoring though. Some ideas:

Rather than storing collectionHandledByGetter in RowWithGetters and changing behavior based on it, have two alternate RowWithGetters implementations, one with the special handling for collections and one without. I think this is still an improvement over the separate ProtoRow class that was rejected since it's not explicitly tied to a particular SchemaProvider, in fact it sounds like it could be re-used for Avro GenericRecord instances.

Move the special logic for collection getters into those SchemaProvider/FieldValueGetter implementations that need it, and make RowWithGetters always behave as if collectionHandledByGetter is true. I think this could be a cleaner approach? But it's challenging because that special logic stores cached values that wouldn't be appropriate to move into the FieldValueGetter implementations. Maybe a SchemaProvider could have some way to indicate that a certain row is expensive to access and should be cached in RowWithGetters?

I've refactored from your input: I've created a RowWithGettersCachedCollection that inherits for RowWithGetters. This cached is the default.

Can you help me understand this a bit more? Why does it not work to cache lists for protocol buffers? We saw repeated array conversion to be a big problem (which is why we cache them). I'm wondering if we could instead cache a lazy array like we do with iterables.

I'll take a closer look at this code to figure it out.

Would we make a ticket out of this or is this blocking?

...a/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchema.java

...ions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchemaProvider.java

alexvanboxel · 2019-08-31T11:49:11Z

@reuvenlax I think I tackled all your concerns you posted. The build is failing, but this is due to a warning in Jenkins that has nothing todo with this PR (I investigated). The PR is also rebased.

Reverted changes to Row so it's back to the original implementation, except for the need of collectionHandledByGetter see remark above. (better name always welcome)
ProtoDescriptorSerializer is gone. I defined a ProtoDomain class that the user needs to use themself if they want to use DynamicMessages. having one ProtoDomain for all messages that belong together will make sure object equality works.
ProtoSchemaProvider only handles compiled Proto classes, it will give an error on DynamicMessages.
ProtoSchema is the workhorse, it's a bit like AvroSchemaUtils.

It would be great having this in the next release. From then on we can do incremental changes like:

Documentation
Option handling
Logical types

I also backported the ProtoDomain on the DynamicMessage PR to align it with this.

Sorry it took so long, but I needed to prioritize something different, but related with Proto (a proto schema registry).

alexvanboxel · 2019-09-03T11:29:05Z

Run Java PreCommit

alexvanboxel · 2019-09-09T21:49:33Z

Requesting @pabloem as a reviewer.

TheNeuralBit

Thanks @alexvanboxel I'm excited to add one more schema provider to Beam :)

TheNeuralBit · 2019-10-01T22:20:02Z

...tensions/protobuf/src/test/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchemaTest.java

+
+    Message messageCoded = schemaCoderCoded.getFromRowFunction().apply(rowCoded);
+    assertEquals(message, messageCoded);
+  }


I think CoderProperties.coderDecodeEncodeEqual actually implements a lot of the logic here. It even serializes/deserializes the coder before encoding and decoding the elements. What do you think about using that here?

There's also CoderProperties.coderSerializable which just clones the coder and checks that they're equal.

TheNeuralBit · 2019-10-02T00:11:17Z

...ions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchemaProvider.java

+ * external registry that maps those URN's with Descriptors.
+ */
+@Experimental(Experimental.Kind.SCHEMAS)
+public class ProtoSchemaProvider implements SchemaProvider {


Is the need for withFieldValueGettersHandleCollections(true) the only reason this can't be made to implement GetterBasedSchemaProvider?

No, each data type has it's own Provider (eg. AvroSchemaProvider). This class makes sure Proto can be auto mapped.

I'm not sure I understand this comment.

Let me elaborate. The ProtoSchemaProvider delegates the functionality from to/fromRow to the ProtoSchema class that is, yes, optimized for dynamic types. The ProtoSchema class managed most of the descriptors. that's it doesn't the default GetterBasedProvider.

If we would optimize for generated classes this could switch (but I prefer this to do in a separate PR later on).

TheNeuralBit · 2019-10-02T00:21:14Z

...a/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchema.java

+ *   <li>Protobuf enums are mapped to non-nullable string values.
+ *   <li>Enum map to their string representation


nit: Are these two lines saying the same thing?

TheNeuralBit · 2019-10-02T00:22:29Z

...a/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchema.java

+ *   <li>Protobuf maps are mapped to nullable maps, where empty maps are mapped to the null value.
+ *   <li>Protobuf repeatables are mapped to nullable arrays, where empty arrays are mapped to the
+ *       null value.


Why not just use non-nullable maps and arrays and produce empty ones when the proto ones are empty?

Business wise this makes more sense. As proto doesn't distinguish between empty map/array and not set. It's also more performant as if the default would be an empty map/array it would result in a bigger Row. It would be an option to make it later configurable. Kept it as is.

TheNeuralBit · 2019-10-02T00:26:33Z

...nsions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoFieldOverlay.java

+    @Override
+    public void set(Message.Builder message, Object value) {
+      //     builder.setField(fieldDescriptor,
+      // convertSetObject(row.getString(fieldDescriptor.getName())));


nit: can you remove this?

TheNeuralBit · 2019-10-02T00:34:15Z

...ns/protobuf/src/test/java/org/apache/beam/sdk/extensions/protobuf/ProtoSchemaValuesTest.java

+
+  @Test
+  public void testRowAndBack() {
+    setup();


I wonder if this could be re-structured a bit so that you could use @Before instead of calling the setup functions yourself. Maybe if you have two different Parameterized test classes that draw from the same data? One for message and one for dynamic message. You can have multiple suites in one outer class if you annotate it with @RunWith(Enclosed.class) (like I did here)

This is definitely a nitpicky thing... totally fine to leave it as is.

alexvanboxel · 2019-11-11T11:16:20Z

This PR is now rebased against master.This removed ProtoDomain (and it's tests) from this PR as it was part of the DymamicMessage PR that is now part of master. I think I've incorporated most of the comment of the review in this PR as well.

The default implementation for Protobuf messages. Supports both generated as dynamic messages. Add the initial release of handling of options, converting it to schema metadata. The implementation is not feature complete, be shows the viability of the feature.

Split RowWithGetters in an implementation with a version that caches collections, named RowWithGettersCachedCollection and one where all the types are handled by the FieldGetters named RowWithGetters.

alexvanboxel · 2019-11-13T08:02:49Z

@TheNeuralBit / @pabloem can you re-review. Thanks.

reuvenlax · 2019-11-19T06:47:56Z

Another comment: this delegates everything to Proto's reflection API, which can be quite inefficient. Compare with AvroSchema where we delegate straight to generated classes when they exist. Reflection might be necessary for the case of DynamicMessage (though in that case I think we should use RowWithStorage instead of RowWithGetters), but shouldn't be necessary when we have generated classes.

alexvanboxel · 2019-11-19T08:35:59Z

Another comment: this delegates everything to Proto's reflection API, which can be quite inefficient. Compare with AvroSchema where we delegate straight to generated classes when they exist.

Totally agree, but I wanted first a solution that works for every use-case. Optimization can happen afterward. I still have plans for iterative improvements, certainly related to protobuf options first, then optimize for performance (but I have to make performance tests for this to measure this)

Reflection might be necessary for the case of DynamicMessage (though in that case I think we should use RowWithStorage instead of RowWithGetters), but shouldn't be necessary when we have generated classes.

RowWithStorage could be a solution when the Row is materialized (with the getToRowFunction), but having the reflection API is still needed it you use the schema without materializing. Example just you the filter on a column function on a schema aware collection but keeping the original type, you don't want to materialize the a big proto just to filter on 1 column. Again, I like to measure this, but we need a base line.

reuvenlax · 2019-11-19T16:36:28Z

BTW I've checked out the PR to play with it a bit more. My worry is that in its current form "optimizing later" will not be so simple, since it doesn't currently fit in with the framework used by AvroSchema, AutoValueSchema, JavaBeanSchema, etc.. If you don't mind, I'll try and play with it a bit to see if I can get it to fit into that framework a bit better.

alexvanboxel · 2019-11-19T18:28:30Z

BTW I've checked out the PR to play with it a bit more. My worry is that in its current form "optimizing later" will not be so simple, since it doesn't currently fit in with the framework used by AvroSchema, AutoValueSchema, JavaBeanSchema, etc.. If you don't mind, I'll try and play with it a bit to see if I can get it to fit into that framework a bit better.

I don't mind, the thing that worries me is that it will miss the 2.18 as well (as it missed the 2.17 already). As we are depending in our business on this feature we are currently building our own version of Beam and that's a pain. For me the DynamicMessage is very important... I was thinking for the normal use-case going to a complete other codepath.

alexvanboxel · 2020-01-08T08:59:32Z

Closing this PR as it's supersede by several other PR that are already on master: Last remaining PR for full proto support is: #10502

Option/Metadata support has been split of and can be followed:

[BEAM-9035] Typed options for Row Schema and Field #10413 (options implementation)
[BEAM-9044] Protobuf options to Schema options #10529 (options support out of proto)

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch 2 times, most recently from 8f3473c to 799e5b3 Compare May 26, 2019 21:10

reuvenlax reviewed Jun 2, 2019

View reviewed changes

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch 3 times, most recently from 782b6b9 to cfadb9b Compare June 8, 2019 20:46

reuvenlax reviewed Jul 9, 2019

View reviewed changes

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch 9 times, most recently from 868ec11 to 37edf7b Compare August 31, 2019 07:30

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch from 37edf7b to 2988800 Compare September 3, 2019 09:52

kennknowles requested a review from reuvenlax September 5, 2019 17:41

pabloem self-requested a review September 19, 2019 17:02

TheNeuralBit reviewed Oct 2, 2019

View reviewed changes

This was referenced Oct 12, 2019

[BEAM-8146,BEAM-8204,BEAM-8205] Add equals and hashCode to SchemaCoder and RowCoder #9493

Merged

[BEAM-5967] Add handling of DynamicMessage in ProtoCoder #8496

Merged

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch 3 times, most recently from 938ab37 to 7e9c3b3 Compare November 11, 2019 11:06

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch from 7e9c3b3 to 9f6d4f2 Compare November 11, 2019 16:36

alexvanboxel added 2 commits November 12, 2019 13:05

[BEAM-7274] Split RowWithGetters in two implementations

e5d5c77

Split RowWithGetters in an implementation with a version that caches collections, named RowWithGettersCachedCollection and one where all the types are handled by the FieldGetters named RowWithGetters.

alexvanboxel force-pushed the feature/BEAM-7274-proto-schema branch from 9f6d4f2 to e5d5c77 Compare November 12, 2019 12:05

alexvanboxel closed this Jan 8, 2020

		* <li>Protobuf enums are mapped to non-nullable string values.
		* <li>Enum map to their string representation

[BEAM-7274] Implement the Protobuf schema provider #8690

[BEAM-7274] Implement the Protobuf schema provider #8690

Uh oh!

Conversation

alexvanboxel commented May 26, 2019

Uh oh!

alexvanboxel commented May 27, 2019

Uh oh!

reuvenlax commented May 29, 2019

Uh oh!

alexvanboxel commented May 29, 2019

Uh oh!

alexvanboxel commented May 29, 2019

Uh oh!

alexvanboxel commented May 31, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alexvanboxel commented Jun 6, 2019 via email

Uh oh!

reuvenlax commented Jun 6, 2019

Uh oh!

alexvanboxel commented Jun 6, 2019 via email

Uh oh!

alexvanboxel commented Jun 10, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

alexvanboxel commented Aug 31, 2019

Uh oh!

alexvanboxel commented Sep 3, 2019

Uh oh!

alexvanboxel commented Sep 9, 2019

Uh oh!

TheNeuralBit left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

alexvanboxel commented May 31, 2019 •

edited

Loading

alexvanboxel Nov 11, 2019 •

edited

Loading