kurtostfeld commented on code in PR #25896:
URL: https://github.com/apache/flink/pull/25896#discussion_r1978482869
##########
flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/utils/AvroKryoSerializerUtils.java:
##########
@@ -106,11 +105,55 @@ public void write(Kryo kryo, Output output, Schema
object) {
}
@Override
- public Schema read(Kryo kryo, Input input, Class<Schema> type) {
+ public Schema read(Kryo kryo, Input input, Class<? extends Schema>
type) {
String schemaAsString = input.readString();
// the parser seems to be stateful, to we need a new one for every
type.
Schema.Parser sParser = new Schema.Parser();
return sParser.parse(schemaAsString);
}
}
+
+ public static class AvroGenericDataArraySerializer
+ extends CollectionSerializer<GenericData.Array> implements
Serializable {
+ @Override
+ public void write(Kryo kryo, Output output, GenericData.Array
collection) {
+ String schemaAsString = collection.getSchema().toString(false);
+ output.writeString(schemaAsString);
+
+ int length = collection.size();
+ output.writeVarInt(length + 1, true);
+
+ // Efficiency could be improved for cases where all elements are
the same type.
+ for (Object element : collection) {
+ kryo.writeClassAndObject(output, element);
+ }
+ }
+
+ @Override
+ public GenericData.Array read(
+ Kryo kryo, Input input, Class<? extends GenericData.Array>
type) {
+ String schemaAsString = input.readString();
+ Schema.Parser schemaParser = new Schema.Parser();
+ Schema schema = schemaParser.parse(schemaAsString);
+
+ int lengthPlusOne = input.readVarInt(true);
+ int length = lengthPlusOne - 1;
+
+ // Efficiency could be improved for cases where all elements are
the same type.
+ ArrayList<Object> collection = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ collection.add(kryo.readClassAndObject(input));
+ }
+ return new GenericData.Array(schema, collection);
+ }
+
+ @Override
+ public GenericData.Array createCopy(Kryo kryo, GenericData.Array
original) {
+ String schemaAsString = original.getSchema().toString(false);
+ Schema.Parser schemaParser = new Schema.Parser();
+ Schema schema = schemaParser.parse(schemaAsString);
+
+ return new GenericData.Array(schema, Collections.emptyList());
Review Comment:
I rewrote this using the Avro serialization logic. Please re-review.
--
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.
To unsubscribe, e-mail: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
