Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

TANGO 2130 #1903

Merged
merged 6 commits into from
Oct 4, 2024
Merged

TANGO 2130 #1903

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c6ef9d1
TANGO-2130 : Tests for primitive types
adamallegro Sep 5, 2024
d908207
TANGO-2130 : Array and Map primitives - tests
adamallegro Sep 6, 2024
e615e51
TANGO-2130 : Checkpoint after primitives
adamallegro Sep 11, 2024
659bba4
TANGO-2130 : Unit tests for records and for simple type mappings
adamallegro Sep 12, 2024
03c5d90
TANGO-2130 : Updated and refactored tests
adamallegro Sep 24, 2024
cb8147e
TANGO-2130 : Fixes after review
adamallegro Sep 27, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .vscode/settings.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
{
"java.compile.nullAnalysis.mode": "automatic"
}
95 changes: 67 additions & 28 deletions ...mes/consumers/consumer/sender/googlebigquery/avro/GoogleBigQueryAvroToProtoConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,7 @@

import com.google.cloud.bigquery.storage.v1.TableSchema;
import com.google.cloud.bigquery.storage.v1.ToProtoConverter;
import com.google.protobuf.ByteString;
import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.google.protobuf.*;
import org.apache.avro.generic.GenericRecord;

import java.math.BigInteger;
Expand All @@ -14,65 +12,106 @@
public class GoogleBigQueryAvroToProtoConverter implements ToProtoConverter<GenericRecord> {
@Override
public DynamicMessage convertToProtoMessage(Descriptors.Descriptor protoSchema, TableSchema tableSchema, GenericRecord inputObject, boolean ignoreUnknownFields) {
return convertToProtoMessage(protoSchema, inputObject);
}

public DynamicMessage convertToProtoMessage(Descriptors.Descriptor protoSchema, GenericRecord inputObject) {
return createMessage(protoSchema, inputObject);
}

private DynamicMessage createMessage(Descriptors.Descriptor protoSchema, GenericRecord inputObject) {
DynamicMessage.Builder messageBuilder = DynamicMessage.newBuilder(protoSchema);

protoSchema.getFields().forEach(field -> {
String fieldName = field.getName();
Object fieldValue = inputObject.get(fieldName);
convertSimpleField(inputObject, field, messageBuilder);
});

return messageBuilder.build();
}

private void convertSimpleField(GenericRecord inputObject, Descriptors.FieldDescriptor field, Message.Builder messageBuilder) {
String fieldName = field.getName();
Object fieldValue = inputObject.get(fieldName);

if (fieldValue != null) {
if (field.isRepeated()) {
if (!(fieldValue instanceof Map<?,?>)) {
for (Object el : (Iterable<?>) fieldValue) {
messageBuilder.addRepeatedField(field, toProtobufValue(field, el));
}
} else {
for (Map.Entry<?, ?> el: ((Map<?, ?>) fieldValue).entrySet()) {
DynamicMessage.Builder entryBuilder = DynamicMessage.newBuilder(field.getMessageType());
entryBuilder.setField(field.getMessageType().findFieldByName("key"), el.getKey().toString());
entryBuilder.setField(field.getMessageType().findFieldByName("value"), el.getValue().toString());
messageBuilder.addRepeatedField(field, entryBuilder.build());
}
if (fieldValue != null) {
if (field.isRepeated()) {
if (!(fieldValue instanceof Map<?, ?>)) {
for (Object el : (Iterable<?>) fieldValue) {
messageBuilder.addRepeatedField(field, toProtobufValue(field, el));
}
} else {
messageBuilder.setField(field, toProtobufValue(field, fieldValue));
for (Map.Entry<?, ?> el : ((Map<?, ?>) fieldValue).entrySet()) {
DynamicMessage.Builder entryBuilder = DynamicMessage.newBuilder(field.getMessageType());
Descriptors.FieldDescriptor valueField = field.getMessageType().findFieldByName("value");
entryBuilder.setField(field.getMessageType().findFieldByName("key"), el.getKey().toString());
entryBuilder.setField(valueField, toProtobufValue(valueField, el.getValue()));
messageBuilder.addRepeatedField(field, entryBuilder.build());
}
}
} else {
messageBuilder.setField(field, toProtobufValue(field, fieldValue));
}
});
}
}

return messageBuilder.build();

private <T extends Number> T toNumber(Object object, Class<T> clazz) {
if (object instanceof Number) {
return numberToSpecific((Number) object, clazz);
} else if (object instanceof String) {
return numberToSpecific(Double.parseDouble((String) object), clazz);
} else if (object instanceof Boolean) {
return numberToSpecific((Boolean) object ? 1 : 0, clazz);
} else return clazz.cast(object);

}

private <T extends Number> T numberToSpecific(Number object, Class<T> clazz) {
if (clazz == Integer.class) {
return clazz.cast((object).intValue());
} else if (clazz == Long.class) {
return clazz.cast((object).longValue());
} else if (clazz == Float.class) {
return clazz.cast((object).floatValue());
} else if (clazz == Double.class) {
return clazz.cast((object).doubleValue());
} else return clazz.cast(object);
}

private Object toProtobufValue(Descriptors.FieldDescriptor fieldDescriptor, Object value) {
switch (fieldDescriptor.getType()) {
case UINT32:
case INT32:
case SINT32:
case SFIXED32:
return toNumber(value, Integer.class);
case INT64:
case SINT64:
case FIXED64:
case UINT64:
case SFIXED64:
case BOOL:
return toNumber(value, Long.class);
case FLOAT:
return toNumber(value, Float.class);
case DOUBLE:
return value;
return toNumber(value, Double.class);
case STRING:
return value.toString();
case BOOL:
if (value instanceof Boolean) {
return value;
} else if (value instanceof Number) {
return ((Number) value).intValue() != 0;
} else
return Boolean.parseBoolean(value.toString());

case ENUM:
Descriptors.EnumDescriptor enumDescriptor = fieldDescriptor.getEnumType();
return enumDescriptor.findValueByName((String) value);
case FIXED64:
case UINT64:
BigInteger bigInt = (BigInteger) value;
return bigInt.longValue();
case BYTES:
ByteBuffer byteBuffer = (ByteBuffer) value;
return ByteString.copyFrom(byteBuffer);
case FIXED32:
case UINT32:
return ((Long) value).intValue();
case MESSAGE:
Descriptors.Descriptor messageDescriptor = fieldDescriptor.getMessageType();
Expand Down
14 changes: 14 additions & 0 deletions ...ovy/pl/allegro/tech/hermes/consumers/consumer/sender/googlebigquery/avro/AvroTrait.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,14 @@
package pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro

import org.apache.avro.AvroRuntimeException
import org.apache.avro.Schema
import org.apache.avro.generic.GenericDatumReader
import org.apache.avro.generic.GenericRecord
import org.apache.avro.io.DatumReader
import org.apache.avro.io.DecoderFactory

trait AvroTrait {
Schema getSchemaFromResources(String fileName) {
return new Schema.Parser().parse(AvroTrait.class.getClassLoader().getResourceAsStream("schemas/${fileName}.avsc"))
}
}
94 changes: 94 additions & 0 deletions ...umers/consumer/sender/googlebigquery/avro/conversions/ConvertAndCastPrimitivesTest.groovy
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
package pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.conversions

import com.google.protobuf.DynamicMessage
import org.apache.avro.Schema
import org.apache.avro.generic.GenericRecord
import org.apache.avro.generic.GenericRecordBuilder
import org.junit.jupiter.api.Test
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.AvroTrait
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.GoogleBigQueryAvroToProtoConverter
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.BoolProto
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.DoubleProto
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.FloatProto
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.Int32Proto
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.Int64Proto
import pl.allegro.tech.hermes.consumers.consumer.sender.googlebigquery.avro.descriptor.StringProto
import spock.lang.Specification

class ConvertAndCastPrimitivesTest extends Specification implements AvroTrait {

@Test

void 'should convert and cast primitive types'() {

given:
Schema schema = getSchemaFromResources("${suite}/${avroType}")
String fieldName = "field"
GenericRecord record = new GenericRecordBuilder(schema).set(fieldName, avroValue).build()
GoogleBigQueryAvroToProtoConverter converter = new GoogleBigQueryAvroToProtoConverter()

when:
DynamicMessage message = converter.convertToProtoMessage(protoDescriptor, record)


then:
message != null
println(message)
def field = message.getDescriptorForType().findFieldByName(fieldName)
def converted_value = message.getField(field)
if (field.getDefaultValue() != avroValue) {
record.hasField(fieldName) == message.hasField(field)
}
converted_value == expectedProtoValue


where:
suite | avroType | protoDescriptor | avroValue | expectedProtoValue
"primitives" | "string" | StringProto.PrimitivesString.getDescriptor() | "value" | "value"
"primitives" | "string" | StringProto.PrimitivesString.getDescriptor() | 12 | "12"
"primitives" | "string" | StringProto.PrimitivesString.getDescriptor() | 12l | "12"
"primitives" | "string" | StringProto.PrimitivesString.getDescriptor() | 12f | "12.0"
"primitives" | "string" | StringProto.PrimitivesString.getDescriptor() | 12d | "12.0"

"primitives" | "int" | Int32Proto.PrimitivesInt32.getDescriptor() | 12 | 12
"primitives" | "long" | Int32Proto.PrimitivesInt32.getDescriptor() | 12l | 12
"primitives" | "float" | Int32Proto.PrimitivesInt32.getDescriptor() | 12f | 12
"primitives" | "double" | Int32Proto.PrimitivesInt32.getDescriptor() | 12d | 12
"primitives" | "string" | Int32Proto.PrimitivesInt32.getDescriptor() | "12" | 12
"primitives" | "boolean" | Int32Proto.PrimitivesInt32.getDescriptor() | true | 1
"primitives" | "boolean" | Int32Proto.PrimitivesInt32.getDescriptor() | false | 0

"primitives" | "int" | Int64Proto.PrimitivesInt64.getDescriptor() | 12 | 12l
"primitives" | "long" | Int64Proto.PrimitivesInt64.getDescriptor() | 12l | 12l
"primitives" | "float" | Int64Proto.PrimitivesInt64.getDescriptor() | 12f | 12l
"primitives" | "double" | Int64Proto.PrimitivesInt64.getDescriptor() | 12d | 12l
"primitives" | "string" | Int64Proto.PrimitivesInt64.getDescriptor() | "12.0" | 12l
"primitives" | "boolean" | Int64Proto.PrimitivesInt64.getDescriptor() | true | 1l
"primitives" | "boolean" | Int64Proto.PrimitivesInt64.getDescriptor() | false | 0l

"primitives" | "int" | FloatProto.PrimitivesFloat.getDescriptor() | 12 | 12.0f
"primitives" | "long" | FloatProto.PrimitivesFloat.getDescriptor() | 12l | 12.0f
"primitives" | "float" | FloatProto.PrimitivesFloat.getDescriptor() | 12.234f | 12.234f
"primitives" | "double" | FloatProto.PrimitivesFloat.getDescriptor() | 12.234d | 12.234f
"primitives" | "string" | FloatProto.PrimitivesFloat.getDescriptor() | "12.234" | 12.234f
"primitives" | "boolean" | FloatProto.PrimitivesFloat.getDescriptor() | true | 1f
"primitives" | "boolean" | FloatProto.PrimitivesFloat.getDescriptor() | false | 0f

"primitives" | "int" | DoubleProto.PrimitivesDouble.getDescriptor() | 12 | 12.0d
"primitives" | "long" | DoubleProto.PrimitivesDouble.getDescriptor() | 12l | 12.0d
"primitives" | "float" | DoubleProto.PrimitivesDouble.getDescriptor() | 12.5f | 12.5d
"primitives" | "double" | DoubleProto.PrimitivesDouble.getDescriptor() | 12.234d | 12.234d
"primitives" | "string" | DoubleProto.PrimitivesDouble.getDescriptor() | "12.234" | 12.234d
"primitives" | "boolean" | DoubleProto.PrimitivesDouble.getDescriptor() | true | 1d
"primitives" | "boolean" | DoubleProto.PrimitivesDouble.getDescriptor() | false | 0d

"primitives" | "boolean" | BoolProto.PrimitivesBool.getDescriptor() | true | true
"primitives" | "boolean" | BoolProto.PrimitivesBool.getDescriptor() | false | false
"primitives" | "string" | BoolProto.PrimitivesBool.getDescriptor() | "true" | true
"primitives" | "string" | BoolProto.PrimitivesBool.getDescriptor() | "false" | false
"primitives" | "string" | BoolProto.PrimitivesBool.getDescriptor() | "True" | true
"primitives" | "string" | BoolProto.PrimitivesBool.getDescriptor() | "False" | false
"primitives" | "boolean" | BoolProto.PrimitivesBool.getDescriptor() | 1 | true
"primitives" | "boolean" | BoolProto.PrimitivesBool.getDescriptor() | 0 | false
}
}
Loading