Github user twalthr commented on a diff in the pull request:
https://github.com/apache/flink/pull/6218#discussion_r199759847
--- Diff:
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
---
@@ -17,154 +17,338 @@
package org.apache.flink.formats.avro;
+import org.apache.flink.annotation.PublicEvolving;
import
org.apache.flink.api.common.serialization.AbstractDeserializationSchema;
+import org.apache.flink.api.common.typeinfo.BasicArrayTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
-import org.apache.flink.formats.avro.typeutils.AvroRecordClassConverter;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.typeutils.MapTypeInfo;
+import org.apache.flink.api.java.typeutils.ObjectArrayTypeInfo;
+import org.apache.flink.api.java.typeutils.RowTypeInfo;
+import org.apache.flink.formats.avro.typeutils.AvroSchemaConverter;
import org.apache.flink.formats.avro.utils.MutableByteArrayInputStream;
import org.apache.flink.types.Row;
import org.apache.flink.util.Preconditions;
+import org.apache.avro.LogicalTypes;
import org.apache.avro.Schema;
-import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericFixed;
+import org.apache.avro.generic.IndexedRecord;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.Decoder;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.specific.SpecificData;
import org.apache.avro.specific.SpecificDatumReader;
import org.apache.avro.specific.SpecificRecord;
-import org.apache.avro.specific.SpecificRecordBase;
-import org.apache.avro.util.Utf8;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeFieldType;
+import org.joda.time.LocalDate;
+import org.joda.time.LocalTime;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.lang.reflect.Array;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.TimeZone;
/**
- * Deserialization schema from Avro bytes over {@link SpecificRecord} to
{@link Row}.
+ * Deserialization schema from Avro bytes to {@link Row}.
*
- * <p>Deserializes the <code>byte[]</code> messages into (nested) Flink
Rows.
+ * <p>Deserializes the <code>byte[]</code> messages into (nested) Flink
rows. It converts Avro types
+ * into types that are compatible with Flink's Table & SQL API.
*
- * {@link Utf8} is converted to regular Java Strings.
+ * <p>Projects with Avro records containing logical date/time types need
to add a JodaTime
+ * dependency.
+ *
+ * <p>Note: Changes in this class need to be kept in sync with the
corresponding runtime
+ * class {@link AvroRowSerializationSchema} and schema converter {@link
AvroSchemaConverter}.
*/
+@PublicEvolving
public class AvroRowDeserializationSchema extends
AbstractDeserializationSchema<Row> {
/**
- * Avro record class.
+ * Used for time conversions into SQL types.
+ */
+ private static final TimeZone LOCAL_TZ = TimeZone.getDefault();
+
+ /**
+ * Avro record class for deserialization. Might be null if record class
is not available.
*/
private Class<? extends SpecificRecord> recordClazz;
/**
- * Schema for deterministic field order.
+ * Schema string for deserialization.
+ */
+ private String schemaString;
+
+ /**
+ * Avro serialization schema.
*/
private transient Schema schema;
/**
- * Reader that deserializes byte array into a record.
+ * Type information describing the result type.
*/
- private transient DatumReader<SpecificRecord> datumReader;
+ private transient TypeInformation<Row> typeInfo;
/**
- * Input stream to read message from.
+ * Record to deserialize byte array.
*/
- private transient MutableByteArrayInputStream inputStream;
+ private transient IndexedRecord record;
/**
- * Avro decoder that decodes binary data.
+ * Reader that deserializes byte array into a record.
*/
- private transient Decoder decoder;
+ private transient DatumReader<IndexedRecord> datumReader;
/**
- * Record to deserialize byte array to.
+ * Input stream to read message from.
*/
- private SpecificRecord record;
+ private transient MutableByteArrayInputStream inputStream;
/**
- * Type information describing the result type.
+ * Avro decoder that decodes binary data.
*/
- private transient TypeInformation<Row> typeInfo;
+ private transient Decoder decoder;
/**
- * Creates a Avro deserialization schema for the given record.
+ * Creates a Avro deserialization schema for the given specific record
class. Having the
+ * concrete Avro record class might improve performance.
*
* @param recordClazz Avro record class used to deserialize Avro's
record to Flink's row
*/
- public AvroRowDeserializationSchema(Class<? extends SpecificRecordBase>
recordClazz) {
+ public AvroRowDeserializationSchema(Class<? extends SpecificRecord>
recordClazz) {
Preconditions.checkNotNull(recordClazz, "Avro record class must
not be null.");
this.recordClazz = recordClazz;
- this.schema = SpecificData.get().getSchema(recordClazz);
- this.datumReader = new SpecificDatumReader<>(schema);
- this.record = (SpecificRecord)
SpecificData.newInstance(recordClazz, schema);
- this.inputStream = new MutableByteArrayInputStream();
- this.decoder = DecoderFactory.get().binaryDecoder(inputStream,
null);
- this.typeInfo = AvroRecordClassConverter.convert(recordClazz);
+ schema = SpecificData.get().getSchema(recordClazz);
+ typeInfo = AvroSchemaConverter.convert(recordClazz);
+ schemaString = schema.toString();
+ record = (SpecificRecord) SpecificData.newInstance(recordClazz,
schema);
+ datumReader = new SpecificDatumReader<>(schema);
+ inputStream = new MutableByteArrayInputStream();
+ decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
+ }
+
+ /**
+ * Creates a Avro deserialization schema for the given Avro schema
string.
+ *
+ * @param avroSchemaString Avro schema string to deserialize Avro's
record to Flink's row
+ */
+ public AvroRowDeserializationSchema(String avroSchemaString) {
+ Preconditions.checkNotNull(avroSchemaString, "Avro schema must
not be null.");
+ recordClazz = null;
+ typeInfo = AvroSchemaConverter.convert(avroSchemaString);
+ schemaString = avroSchemaString;
+ schema = new Schema.Parser().parse(avroSchemaString);
+ record = new GenericData.Record(schema);
+ datumReader = new GenericDatumReader<>(schema);
+ inputStream = new MutableByteArrayInputStream();
+ decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
+ // check for a schema that describes a record
+ if (!(typeInfo instanceof RowTypeInfo)) {
+ throw new IllegalArgumentException("Row type
information expected.");
+ }
}
@Override
public Row deserialize(byte[] message) throws IOException {
- // read record
try {
inputStream.setBuffer(message);
- this.record = datumReader.read(record, decoder);
- } catch (IOException e) {
- throw new RuntimeException("Failed to deserialize
Row.", e);
+ final IndexedRecord read = datumReader.read(record,
decoder);
+ return convertRecord(schema, (RowTypeInfo) typeInfo,
read);
+ } catch (Exception e) {
+ throw new IOException("Failed to deserialize Avro
record.", e);
}
-
- // convert to row
- final Object row = convertToRow(schema, record);
- return (Row) row;
- }
-
- private void writeObject(ObjectOutputStream oos) throws IOException {
- oos.writeObject(recordClazz);
- }
-
- @SuppressWarnings("unchecked")
- private void readObject(ObjectInputStream ois) throws
ClassNotFoundException, IOException {
- this.recordClazz = (Class<? extends SpecificRecord>)
ois.readObject();
- this.schema = SpecificData.get().getSchema(recordClazz);
- this.datumReader = new SpecificDatumReader<>(schema);
- this.record = (SpecificRecord)
SpecificData.newInstance(recordClazz, schema);
- this.inputStream = new MutableByteArrayInputStream();
- this.decoder = DecoderFactory.get().binaryDecoder(inputStream,
null);
}
@Override
public TypeInformation<Row> getProducedType() {
return typeInfo;
}
- /**
- * Converts a (nested) Avro {@link SpecificRecord} into Flink's Row
type.
- * Avro's {@link Utf8} fields are converted into regular Java strings.
- */
- private static Object convertToRow(Schema schema, Object recordObj) {
- if (recordObj instanceof GenericRecord) {
- // records can be wrapped in a union
- if (schema.getType() == Schema.Type.UNION) {
+ //
--------------------------------------------------------------------------------------------
+
+ private Row convertRecord(Schema schema, RowTypeInfo typeInfo,
IndexedRecord record) {
+ final List<Schema.Field> fields = schema.getFields();
+ final TypeInformation<?>[] fieldInfo = typeInfo.getFieldTypes();
+ final int length = fields.size();
+ final Row row = new Row(length);
+ for (int i = 0; i < length; i++) {
+ final Schema.Field field = fields.get(i);
+ row.setField(i, convert(field.schema(), fieldInfo[i],
record.get(i)));
+ }
+ return row;
+ }
+
+ private Object convert(Schema schema, TypeInformation<?> info, Object
object) {
+ // we perform the conversion based on schema information but
enriched with pre-computed
+ // type information where useful (i.e., for arrays)
+
+ if (object == null) {
+ return null;
+ }
+ switch (schema.getType()) {
+ case RECORD:
+ if (object instanceof IndexedRecord) {
+ return convertRecord(schema,
(RowTypeInfo) info, (IndexedRecord) object);
+ }
+ throw new IllegalStateException("IndexedRecord
expected but was: " + object.getClass());
+ case ENUM:
+ case STRING:
+ return object.toString();
+ case ARRAY:
+ if (info instanceof BasicArrayTypeInfo) {
+ final BasicArrayTypeInfo<?, ?> bati =
(BasicArrayTypeInfo<?, ?>) info;
+ final TypeInformation<?> elementInfo =
bati.getComponentInfo();
+ return
convertObjectArray(schema.getElementType(), elementInfo, object);
+ } else {
+ final ObjectArrayTypeInfo<?, ?> oati =
(ObjectArrayTypeInfo<?, ?>) info;
+ final TypeInformation<?> elementInfo =
oati.getComponentInfo();
+ return
convertObjectArray(schema.getElementType(), elementInfo, object);
+ }
+ case MAP:
+ final MapTypeInfo<?, ?> mti = (MapTypeInfo<?,
?>) info;
+ final Map<String, Object> convertedMap = new
HashMap<>();
+ final Map<?, ?> map = (Map<?, ?>) object;
+ for (Map.Entry<?, ?> entry : map.entrySet()) {
+ convertedMap.put(
+ entry.getKey().toString(),
+ convert(schema.getValueType(),
mti.getValueTypeInfo(), entry.getValue()));
+ }
+ return convertedMap;
+ case UNION:
final List<Schema> types = schema.getTypes();
- if (types.size() == 2 && types.get(0).getType()
== Schema.Type.NULL && types.get(1).getType() == Schema.Type.RECORD) {
- schema = types.get(1);
+ final int size = types.size();
+ final Schema actualSchema;
+ if (size == 2 && types.get(0).getType() ==
Schema.Type.NULL) {
+ return convert(types.get(1), info,
object);
+ } else if (size == 2 && types.get(1).getType()
== Schema.Type.NULL) {
+ return convert(types.get(0), info,
object);
+ } else if (size == 1) {
+ return convert(types.get(0), info,
object);
+ } else {
+ // generic type
+ return object;
+ }
+ case FIXED:
+ final byte[] fixedBytes = ((GenericFixed)
object).bytes();
+ if (info == Types.BIG_DEC) {
+ return convertDecimal(schema,
fixedBytes);
}
- else {
- throw new RuntimeException("Currently
we only support schemas of the following form: UNION[null, RECORD]. Given: " +
schema);
+ return fixedBytes;
+ case BYTES:
+ final ByteBuffer bb = (ByteBuffer) object;
+ bb.position(0);
--- End diff --
We cannot be sure about that. It is safer to reset the position.
---
