[
https://issues.apache.org/jira/browse/BEAM-7274?focusedWorklogId=252929&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-252929
]
ASF GitHub Bot logged work on BEAM-7274:
----------------------------------------
Author: ASF GitHub Bot
Created on: 03/Jun/19 06:20
Start Date: 03/Jun/19 06:20
Worklog Time Spent: 10m
Work Description: alexvanboxel commented on pull request #8690:
[BEAM-7274] Implement the Protobuf schema provider
URL: https://github.com/apache/beam/pull/8690#discussion_r289699837
##########
File path:
sdks/java/extensions/protobuf/src/main/java/org/apache/beam/sdk/extensions/protobuf/ProtoRow.java
##########
@@ -0,0 +1,556 @@
+/*
+ * 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.extensions.protobuf;
+
+import com.google.protobuf.ByteString;
+import com.google.protobuf.Descriptors;
+import com.google.protobuf.DynamicMessage;
+import com.google.protobuf.Message;
+import com.google.protobuf.Timestamp;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+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.
+ */
+@Experimental(Experimental.Kind.SCHEMAS)
+class ProtoRow extends Row {
+
+ private Message containingMessage;
+ private ArrayList<FieldOverlay> fieldOverlay;
+
+ ProtoRow(Schema schema, ArrayList<FieldOverlay> fieldOverlay, Message
message) {
+ super(schema);
+ this.fieldOverlay = fieldOverlay;
+ this.containingMessage = message;
+ }
+
+ @Nullable
+ @Override
+ @SuppressWarnings({"TypeParameterUnusedInFormals", "unchecked"})
+ public <T> T getValue(int fieldIdx) {
+ return (T) fieldOverlay.get(fieldIdx).get(this.containingMessage);
+ }
+
+ @Override
+ public int getFieldCount() {
+ return fieldOverlay.size();
+ }
+
+ @Override
+ public List<Object> getValues() {
+ List values = new ArrayList();
+ for (FieldOverlay get : fieldOverlay) {
+ values.add(get.get(containingMessage));
+ }
+ return values;
+ }
+
+ /**
+ * Protobuf FieldOverlay is the interface that each implementation needs to
implement to handle a
+ * specific field types.
+ */
+ public interface FieldOverlay<ValueT> {
+
+ /** Convert the overlayed field in the Message and convert it to the Row
field. */
+ ValueT get(Message object);
+
+ ValueT convertGetObject(Object object);
+
+ /** Convert the Row field and set it on the overlayed field of the
message. */
+ void set(Message.Builder object, ValueT value);
+
+ Object convertSetObject(Object value);
+
+ /** Return the Beam Schema Field of this overlayed field. */
+ Schema.Field getSchemaField();
+ }
+
+ /** Overlay for Protobuf primitive types. Primitive values are just passed
through. */
+ static class PrimitiveOverlay implements FieldOverlay<Object> {
+ protected Descriptors.FieldDescriptor fieldDescriptor;
+ private Schema.Field field;
+
+ PrimitiveOverlay(Descriptors.FieldDescriptor fieldDescriptor) {
+ this.fieldDescriptor = fieldDescriptor;
+ this.field =
+ Schema.Field.of(
+ fieldDescriptor.getName(),
ProtoSchema.convertType(fieldDescriptor.getType()));
+ }
+
+ @Override
+ public Object get(Message message) {
+ return convertGetObject(message.getField(fieldDescriptor));
+ }
+
+ @Override
+ public Object convertGetObject(Object object) {
+ return object;
+ }
+
+ @Override
+ public void set(Message.Builder message, Object value) {
+ message.setField(fieldDescriptor, value);
+ }
+
+ @Override
+ public Object convertSetObject(Object value) {
+ return value;
+ }
+
+ @Override
+ public Schema.Field getSchemaField() {
+ return field;
+ }
+ }
+
+ /**
+ * Overlay for Bytes. Protobuf Bytes are natively represented as ByteStrings
that requires special
+ * handling for byte[] of size 0.
+ */
+ static class BytesOverlay extends PrimitiveOverlay {
+ BytesOverlay(Descriptors.FieldDescriptor fieldDescriptor) {
+ super(fieldDescriptor);
+ }
+
+ @Override
+ public Object convertGetObject(Object object) {
+ // return object;
+ return ((ByteString) object).toByteArray();
+ }
+
+ @Override
+ public void set(Message.Builder message, Object value) {
+ if (value != null && ((byte[]) value).length > 0) {
+ // Protobuf messages BYTES doesn't like emptry bytes?!
+ message.setField(fieldDescriptor, convertSetObject(value));
+ }
+ }
+
+ @Override
+ public Object convertSetObject(Object value) {
+ if (value != null) {
+ return ByteString.copyFrom((byte[]) value);
+ }
+ return null;
+ }
+ }
+
+ /**
+ * Overlay handler for the Well Known Type "Wrapper". These wrappers make it
possible to have
+ * nullable primitives.
+ */
+ static class WrapperOverlay<ValueT> implements FieldOverlay<ValueT> {
+ private Descriptors.FieldDescriptor fieldDescriptor;
+ private Descriptors.FieldDescriptor valueDescriptor;
+ private FieldOverlay value;
+
+ WrapperOverlay(ProtoSchema protoSchema, Descriptors.FieldDescriptor
fieldDescriptor) {
+ this.fieldDescriptor = fieldDescriptor;
+ this.valueDescriptor =
fieldDescriptor.getMessageType().findFieldByName("value");
+ this.value = protoSchema.createFieldLayer(valueDescriptor, false);
+ }
+
+ @Override
+ public ValueT get(Message message) {
+ if (message.hasField(fieldDescriptor)) {
+
+ Message wrapper = (Message) message.getField(fieldDescriptor);
+ return (ValueT) value.get(wrapper);
+ }
+ return null;
+ }
+
+ @Override
+ public ValueT convertGetObject(Object object) {
+ return (ValueT) object;
+ }
+
+ @Override
+ public void set(Message.Builder message, ValueT value) {
+ if (value != null) {
+ DynamicMessage.Builder builder =
+ DynamicMessage.newBuilder(fieldDescriptor.getMessageType());
+ this.value.set(builder, value);
+ message.setField(fieldDescriptor, builder.build());
+ }
+ }
+
+ @Override
+ public Object convertSetObject(Object value) {
+ return value;
+ }
+
+ @Override
+ public Schema.Field getSchemaField() {
+ return Schema.Field.of(
+ fieldDescriptor.getName(),
+
ProtoSchema.convertType(valueDescriptor.getType()).withNullable(true));
+ }
+ }
+
+ /**
+ * Overlay handler for the Well Known Type "Timestamp". This wrappers
converts from a single Row
+ * DATETIME and a protobuf "Timestamp" messsage.
+ */
+ static class TimestampOverlay implements FieldOverlay<Instant> {
+ protected Schema.Field field;
+ private Descriptors.FieldDescriptor fieldDescriptor;
+ private Descriptors.FieldDescriptor secondsDescriptor;
+ private Descriptors.FieldDescriptor nanosDescriptor;
+
+ TimestampOverlay(Descriptors.FieldDescriptor fieldDescriptor) {
+ this.fieldDescriptor = fieldDescriptor;
+ this.secondsDescriptor =
fieldDescriptor.getMessageType().findFieldByName("seconds");
+ this.nanosDescriptor =
fieldDescriptor.getMessageType().findFieldByName("nanos");
+ this.field =
+ Schema.Field.of(fieldDescriptor.getName(),
Schema.FieldType.DATETIME).withNullable(true);
+ }
+
+ @Override
+ public Instant get(Message message) {
+ if (message.hasField(fieldDescriptor)) {
+ Message wrapper = (Message) message.getField(fieldDescriptor);
+ return convertGetObject(wrapper);
+ }
+ return null;
+ }
+
+ @Override
+ public Instant convertGetObject(Object object) {
Review comment:
Like to make this switchable via options, Instant will be enough for most
Beam use-cases and will make it work nicely with Beam SQL.
----------------------------------------------------------------
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:
[email protected]
Issue Time Tracking
-------------------
Worklog Id: (was: 252929)
Time Spent: 2.5h (was: 2h 20m)
> Protobuf Beam Schema support
> ----------------------------
>
> Key: BEAM-7274
> URL: https://issues.apache.org/jira/browse/BEAM-7274
> Project: Beam
> Issue Type: Improvement
> Components: sdk-java-core
> Reporter: Alex Van Boxel
> Assignee: Alex Van Boxel
> Priority: Minor
> Time Spent: 2.5h
> Remaining Estimate: 0h
>
> Add support for the new Beam Schema to the Protobuf extension.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
