[flink] branch master updated: [FLINK-25238][table-runtime] Fix ArrayDataSerializer#copy for customized types

[gyfora]

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

gyfora pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 99c0438253e [FLINK-25238][table-runtime] Fix ArrayDataSerializer#copy 
for customized types
99c0438253e is described below

commit 99c0438253efaf0a9626a807ce2b882402a367d1
Author: Yi Tang <ssnailt...@gmail.com>
AuthorDate: Fri Apr 1 11:48:03 2022 +0800

    [FLINK-25238][table-runtime] Fix ArrayDataSerializer#copy for customized 
types
---
 .../flink/table/runtime/typeutils/ArrayDataSerializer.java   |  4 +++-
 .../table/runtime/typeutils/ArrayDataSerializerTest.java     | 12 ++++++++++++
 2 files changed, 15 insertions(+), 1 deletion(-)

diff --git 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializer.java
 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializer.java
index 078f01a46d3..26a6b0bdea1 100644
--- 
a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializer.java
+++ 
b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializer.java
@@ -84,10 +84,12 @@ public class ArrayDataSerializer extends 
TypeSerializer<ArrayData> {
     public ArrayData copy(ArrayData from) {
         if (from instanceof GenericArrayData) {
             return copyGenericArray((GenericArrayData) from);
+        } else if (from instanceof ColumnarArrayData) {
+            return copyColumnarArray((ColumnarArrayData) from);
         } else if (from instanceof BinaryArrayData) {
             return ((BinaryArrayData) from).copy();
         } else {
-            return copyColumnarArray((ColumnarArrayData) from);
+            return toBinaryArray(from);
         }
     }
 
diff --git 
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializerTest.java
 
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializerTest.java
index 558bbbb0ca2..29fcbda37f2 100644
--- 
a/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializerTest.java
+++ 
b/flink-table/flink-table-runtime/src/test/java/org/apache/flink/table/runtime/typeutils/ArrayDataSerializerTest.java
@@ -29,6 +29,7 @@ import 
org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
 import org.apache.flink.table.data.writer.BinaryArrayWriter;
 import org.apache.flink.testutils.DeeplyEqualsChecker;
 
+import java.lang.reflect.Proxy;
 import java.nio.charset.StandardCharsets;
 
 /** A test for the {@link ArrayDataSerializer}. */
@@ -88,6 +89,7 @@ public class ArrayDataSerializerTest extends 
SerializerTestBase<ArrayData> {
             createArray("11", "haa", "ke"),
             createArray("11", "lele", "haa", "ke"),
             createColumnarArray("11", "lele", "haa", "ke"),
+            createCustomTypeArray("11", "lele", "haa", "ke"),
         };
     }
 
@@ -108,4 +110,14 @@ public class ArrayDataSerializerTest extends 
SerializerTestBase<ArrayData> {
         }
         return new ColumnarArrayData(vector, 0, vs.length);
     }
+
+    static ArrayData createCustomTypeArray(String... vs) {
+        BinaryArrayData binaryArrayData = createArray(vs);
+        Object customArrayData =
+                Proxy.newProxyInstance(
+                        ArrayDataSerializerTest.class.getClassLoader(),
+                        new Class[] {ArrayData.class},
+                        (proxy, method, args) -> 
method.invoke(binaryArrayData, args));
+        return (ArrayData) customArrayData;
+    }
 }