[flink] branch release-1.10 updated: [FLINK-16242][table-runtime-blink] Duplicate field serializers in BaseRowSerializer (#11212)

[lz...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch release-1.10
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-1.10 by this push:
     new d09c806  [FLINK-16242][table-runtime-blink] Duplicate field serializers in BaseRowSerializer (#11212)
d09c806 is described below

commit d09c8068ed747d564defabb735e7e10de6c77992
Author: Jiayi Liao <bu...@163.com>
AuthorDate: Wed Feb 26 10:10:55 2020 +0800

    [FLINK-16242][table-runtime-blink] Duplicate field serializers in BaseRowSerializer (#11212)
---
 .../table/runtime/typeutils/BaseRowSerializer.java |  6 +++-
 .../runtime/typeutils/BaseRowSerializerTest.java   | 34 +++++++++++++++++++++-
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializer.java b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializer.java
index d05630d..a4debaf 100644
--- a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializer.java
+++ b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializer.java
@@ -77,7 +77,11 @@ public class BaseRowSerializer extends AbstractRowSerializer<BaseRow> {
 
 	@Override
 	public TypeSerializer<BaseRow> duplicate() {
-		return new BaseRowSerializer(types, fieldSerializers);
+		TypeSerializer<?>[] duplicateFieldSerializers = new TypeSerializer[fieldSerializers.length];
+		for (int i = 0; i < fieldSerializers.length; i++) {
+			duplicateFieldSerializers[i] = fieldSerializers[i].duplicate();
+		}
+		return new BaseRowSerializer(types, duplicateFieldSerializers);
 	}
 
 	@Override
diff --git a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializerTest.java b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializerTest.java
index 50df9c5..07e9e13 100644
--- a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializerTest.java
+++ b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/runtime/typeutils/BaseRowSerializerTest.java
@@ -20,16 +20,21 @@ package org.apache.flink.table.runtime.typeutils;
 
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeutils.SerializerTestInstance;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
 import org.apache.flink.table.dataformat.BaseRow;
 import org.apache.flink.table.dataformat.BinaryArray;
 import org.apache.flink.table.dataformat.BinaryArrayWriter;
+import org.apache.flink.table.dataformat.BinaryGeneric;
 import org.apache.flink.table.dataformat.BinaryMap;
 import org.apache.flink.table.dataformat.BinaryRow;
 import org.apache.flink.table.dataformat.GenericRow;
 import org.apache.flink.table.types.logical.ArrayType;
 import org.apache.flink.table.types.logical.DoubleType;
 import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
 import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.RawType;
 import org.apache.flink.table.types.logical.VarCharType;
 import org.apache.flink.testutils.DeeplyEqualsChecker;
 
@@ -80,7 +85,8 @@ public class BaseRowSerializerTest extends SerializerTestInstance<BaseRow> {
 		return Arrays.asList(
 				testBaseRowSerializer(),
 				testLargeBaseRowSerializer(),
-				testBaseRowSerializerWithComplexTypes());
+				testBaseRowSerializerWithComplexTypes(),
+				testBaseRowSerializerWithKryo());
 	}
 
 	private static Object[] testBaseRowSerializer() {
@@ -159,6 +165,19 @@ public class BaseRowSerializerTest extends SerializerTestInstance<BaseRow> {
 		return new Object[] {serializer, data};
 	}
 
+	private static Object[] testBaseRowSerializerWithKryo() {
+		BinaryGenericSerializer<WrappedString> binaryGenericSerializer = new BinaryGenericSerializer<>(
+				new KryoSerializer<>(WrappedString.class, new ExecutionConfig()));
+		BaseRowSerializer serializer = new BaseRowSerializer(new LogicalType[]{
+				new RawType(BinaryGeneric.class, binaryGenericSerializer)},
+				new TypeSerializer[]{binaryGenericSerializer});
+
+		GenericRow row = new GenericRow(1);
+		row.setField(0, new BinaryGeneric<>(new WrappedString("a")));
+
+		return new Object[] {serializer, new GenericRow[]{row}};
+	}
+
 	// ----------------------------------------------------------------------------------------------
 
 	private static BinaryArray createArray(int... ints) {
@@ -240,4 +259,17 @@ public class BaseRowSerializerTest extends SerializerTestInstance<BaseRow> {
 			checkDeepEquals(row, serializer.copy(row, new GenericRow(row.getArity() + 1)));
 		}
 	}
+
+	/**
+	 * Class used for concurrent testing with KryoSerializer.
+	 */
+	private static class WrappedString {
+
+		private final String content;
+
+		WrappedString(String content) {
+			this.content = content;
+		}
+	}
+
 }