Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/6218#discussion_r199780422
--- Diff:
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowDeserializationSchema.java
---
@@ -201,71 +202,69 @@ private Object convert(Schema schema,
TypeInformation<?> info, Object object) {
switch (schema.getType()) {
case RECORD:
if (object instanceof IndexedRecord) {
- return convertRecord(schema,
(RowTypeInfo) info, (IndexedRecord) object);
+ return convertAvroRecordToRow(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);
+ final TypeInformation<?> elementInfo =
((BasicArrayTypeInfo<?, ?>) info).getComponentInfo();
+ return
convertToObjectArray(schema.getElementType(), elementInfo, object);
} else {
- final ObjectArrayTypeInfo<?, ?> oati =
(ObjectArrayTypeInfo<?, ?>) info;
- final TypeInformation<?> elementInfo =
oati.getComponentInfo();
- return
convertObjectArray(schema.getElementType(), elementInfo, object);
+ final TypeInformation<?> elementInfo =
((ObjectArrayTypeInfo<?, ?>) info).getComponentInfo();
+ return
convertToObjectArray(schema.getElementType(), elementInfo, object);
}
case MAP:
- final MapTypeInfo<?, ?> mti = (MapTypeInfo<?,
?>) info;
+ final MapTypeInfo<?, ?> mapTypeInfo =
(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()));
+
convertAvroType(schema.getValueType(), mapTypeInfo.getValueTypeInfo(),
entry.getValue()));
}
return convertedMap;
case UNION:
final List<Schema> types = schema.getTypes();
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);
+ return convertAvroType(types.get(1),
info, object);
} else if (size == 2 && types.get(1).getType()
== Schema.Type.NULL) {
- return convert(types.get(0), info,
object);
+ return convertAvroType(types.get(0),
info, object);
} else if (size == 1) {
- return convert(types.get(0), info,
object);
+ return convertAvroType(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);
+ return convertToDecimal(schema,
fixedBytes);
}
return fixedBytes;
case BYTES:
- final ByteBuffer bb = (ByteBuffer) object;
- bb.position(0);
- final byte[] bytes = new byte[bb.remaining()];
- bb.get(bytes);
+ final ByteBuffer byteBuffer = (ByteBuffer)
object;
+ byteBuffer.position(0);
--- End diff --
I think this is still wrong (unless it's specified somewhere in the avro
documentation) - check my comment in previous thread
---
