You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by th...@apache.org on 2022/05/14 04:07:40 UTC
[flink] branch release-1.15 updated: [FLINK-27255] [flink-avro] flink-avro does not support ser/de of large avro schema (#19645)
This is an automated email from the ASF dual-hosted git repository.
thw pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new 20768b2f53f [FLINK-27255] [flink-avro] flink-avro does not support ser/de of large avro schema (#19645)
20768b2f53f is described below
commit 20768b2f53fb8c126d984bc85be2d34fdaa487c1
Author: Haizhou Zhao <zh...@gmail.com>
AuthorDate: Tue May 10 16:18:38 2022 -0700
[FLINK-27255] [flink-avro] flink-avro does not support ser/de of large avro schema (#19645)
Co-authored-by: Haizhou Zhao <ha...@apple.com>
---
.../avro/typeutils/GenericRecordAvroTypeInfo.java | 10 ++++-
.../avro/typeutils/SerializableAvroSchema.java | 11 +++--
.../typeutils/AvroGenericRecordTypeInfoTest.java | 50 ++++++++++++++++++++++
.../typeutils/AvroSerializerGenericRecordTest.java | 7 +--
...a => AvroSerializerLargeGenericRecordTest.java} | 11 ++---
.../flink/formats/avro/utils/AvroTestUtils.java | 30 +++++++++++++
6 files changed, 102 insertions(+), 17 deletions(-)
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/GenericRecordAvroTypeInfo.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/GenericRecordAvroTypeInfo.java
index 387014077d2..28332db5a46 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/GenericRecordAvroTypeInfo.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/GenericRecordAvroTypeInfo.java
@@ -28,6 +28,7 @@ import org.apache.avro.generic.GenericRecord;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
+import java.nio.charset.StandardCharsets;
import java.util.Objects;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -104,10 +105,15 @@ public class GenericRecordAvroTypeInfo extends TypeInformation<GenericRecord> {
}
private void writeObject(ObjectOutputStream oos) throws IOException {
- oos.writeUTF(schema.toString());
+ byte[] schemaStrInBytes = schema.toString(false).getBytes(StandardCharsets.UTF_8);
+ oos.writeInt(schemaStrInBytes.length);
+ oos.write(schemaStrInBytes);
}
private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException {
- this.schema = new Schema.Parser().parse(ois.readUTF());
+ int len = ois.readInt();
+ byte[] content = new byte[len];
+ ois.readFully(content);
+ this.schema = new Schema.Parser().parse(new String(content, StandardCharsets.UTF_8));
}
}
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/SerializableAvroSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/SerializableAvroSchema.java
index 32714951a61..12458752428 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/SerializableAvroSchema.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/typeutils/SerializableAvroSchema.java
@@ -28,6 +28,7 @@ import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.Serializable;
+import java.nio.charset.StandardCharsets;
/** A wrapper for Avro {@link Schema}, that is Java serializable. */
@Internal
@@ -52,14 +53,18 @@ final class SerializableAvroSchema implements Serializable {
oos.writeBoolean(false);
} else {
oos.writeBoolean(true);
- oos.writeUTF(schema.toString(false));
+ byte[] schemaStrInBytes = schema.toString(false).getBytes(StandardCharsets.UTF_8);
+ oos.writeInt(schemaStrInBytes.length);
+ oos.write(schemaStrInBytes);
}
}
private void readObject(ObjectInputStream ois) throws ClassNotFoundException, IOException {
if (ois.readBoolean()) {
- String schema = ois.readUTF();
- this.schema = new Parser().parse(schema);
+ int len = ois.readInt();
+ byte[] content = new byte[len];
+ ois.readFully(content);
+ this.schema = new Parser().parse(new String(content, StandardCharsets.UTF_8));
} else {
this.schema = null;
}
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroGenericRecordTypeInfoTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroGenericRecordTypeInfoTest.java
new file mode 100644
index 00000000000..dd2b6c98907
--- /dev/null
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroGenericRecordTypeInfoTest.java
@@ -0,0 +1,50 @@
+/*
+ * 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.flink.formats.avro.typeutils;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeInformationTestBase;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.formats.avro.utils.AvroTestUtils;
+
+import org.apache.avro.generic.GenericRecord;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link GenericRecordAvroTypeInfo}. */
+public class AvroGenericRecordTypeInfoTest
+ extends TypeInformationTestBase<GenericRecordAvroTypeInfo> {
+
+ @Override
+ protected GenericRecordAvroTypeInfo[] getTestData() {
+ return new GenericRecordAvroTypeInfo[] {
+ new GenericRecordAvroTypeInfo(AvroTestUtils.getSmallSchema()),
+ new GenericRecordAvroTypeInfo(AvroTestUtils.getLargeSchema())
+ };
+ }
+
+ @Test
+ void testAvroByDefault() {
+ final TypeSerializer<GenericRecord> serializer =
+ new GenericRecordAvroTypeInfo(AvroTestUtils.getLargeSchema())
+ .createSerializer(new ExecutionConfig());
+ assertThat(serializer).isInstanceOf(AvroSerializer.class);
+ }
+}
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java
index 750bae9df2d..26471e75d4f 100644
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.formats.avro.typeutils;
import org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.formats.avro.utils.AvroTestUtils;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
@@ -28,11 +29,7 @@ import org.apache.avro.generic.GenericRecordBuilder;
/** Test for {@link AvroSerializer} that tests GenericRecord. */
public class AvroSerializerGenericRecordTest extends SerializerTestBase<GenericRecord> {
- private static final Schema SCHEMA =
- new org.apache.avro.Schema.Parser()
- .parse(
- "{\"type\":\"record\",\"name\":\"Dummy\",\"namespace\":\"dummy\",\"fields\": "
- + "[{\"name\":\"afield\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+ private static final Schema SCHEMA = AvroTestUtils.getSmallSchema();
@Override
protected TypeSerializer<GenericRecord> createSerializer() {
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerLargeGenericRecordTest.java
similarity index 76%
copy from flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java
copy to flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerLargeGenericRecordTest.java
index 750bae9df2d..29092e810ed 100644
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerGenericRecordTest.java
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/typeutils/AvroSerializerLargeGenericRecordTest.java
@@ -20,19 +20,16 @@ package org.apache.flink.formats.avro.typeutils;
import org.apache.flink.api.common.typeutils.SerializerTestBase;
import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.formats.avro.utils.AvroTestUtils;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.GenericRecordBuilder;
/** Test for {@link AvroSerializer} that tests GenericRecord. */
-public class AvroSerializerGenericRecordTest extends SerializerTestBase<GenericRecord> {
+public class AvroSerializerLargeGenericRecordTest extends SerializerTestBase<GenericRecord> {
- private static final Schema SCHEMA =
- new org.apache.avro.Schema.Parser()
- .parse(
- "{\"type\":\"record\",\"name\":\"Dummy\",\"namespace\":\"dummy\",\"fields\": "
- + "[{\"name\":\"afield\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+ private static final Schema SCHEMA = AvroTestUtils.getLargeSchema();
@Override
protected TypeSerializer<GenericRecord> createSerializer() {
@@ -52,7 +49,7 @@ public class AvroSerializerGenericRecordTest extends SerializerTestBase<GenericR
@Override
protected GenericRecord[] getTestData() {
return new GenericRecord[] {
- new GenericRecordBuilder(SCHEMA).set("afield", "foo bar").build()
+ new GenericRecordBuilder(SCHEMA).set("field1", "foo bar").build()
};
}
}
diff --git a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java
index aedcaa66a2f..c4e49e0980a 100644
--- a/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java
+++ b/flink-formats/flink-avro/src/test/java/org/apache/flink/formats/avro/utils/AvroTestUtils.java
@@ -24,9 +24,11 @@ import org.apache.flink.formats.avro.generated.Colors;
import org.apache.flink.formats.avro.generated.Fixed16;
import org.apache.flink.formats.avro.generated.Fixed2;
import org.apache.flink.formats.avro.generated.User;
+import org.apache.flink.formats.avro.typeutils.AvroSerializerLargeGenericRecordTest;
import org.apache.flink.types.Row;
import org.apache.avro.Schema;
+import org.apache.avro.SchemaBuilder;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumWriter;
import org.apache.avro.generic.GenericRecord;
@@ -259,6 +261,34 @@ public final class AvroTestUtils {
return t;
}
+ /**
+ * Craft a large Avro Schema which contains more than 0xFFFF characters.
+ *
+ * <p>0xFFFF is the magical number that once a java string length is above it, then the
+ * serialization scheme changes
+ */
+ public static Schema getLargeSchema() {
+ SchemaBuilder.FieldAssembler<Schema> fields =
+ SchemaBuilder.record("LargeAvroSchema")
+ .namespace(AvroSerializerLargeGenericRecordTest.class.getName())
+ .fields();
+ for (int i = 0; i < 10000; ++i) {
+ fields = fields.optionalString("field" + i);
+ }
+ Schema schema = fields.endRecord();
+
+ assert schema.toString().length() > 0xFFFF;
+ return schema;
+ }
+
+ /** Craft a small Avro Schema which contains less than 0xFFFF characters. */
+ public static Schema getSmallSchema() {
+ return new org.apache.avro.Schema.Parser()
+ .parse(
+ "{\"type\":\"record\",\"name\":\"Dummy\",\"namespace\":\"dummy\",\"fields\": "
+ + "[{\"name\":\"afield\",\"type\":{\"type\":\"string\",\"avro.java.string\":\"String\"}}]}");
+ }
+
/**
* Writes given record using specified schema.
*