[flink] branch release-1.14 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 release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git


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

commit 0822c539e96b19d71df3936d319a4fa694479008
Author: Yi Tang <ssnailt...@gmail.com>
AuthorDate: Wed Apr 6 15:41:18 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 0f04ccbd654..10a15be6241 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 d29e75a5ad8..baed23706dc 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.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}. */
@@ -85,6 +86,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"),
         };
     }
 
@@ -105,4 +107,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;
+    }
 }