[flink] branch release-1.10 updated: [FLINK-18168][table-runtime-blink] Fix array reuse for BinaryArrayData in converters

[twalthr]

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

twalthr 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 23ad7e3  [FLINK-18168][table-runtime-blink] Fix array reuse for 
BinaryArrayData in converters
23ad7e3 is described below

commit 23ad7e33117b7c02b28fda77596b12668b5117c1
Author: zoudan <zou...@bytedance.com>
AuthorDate: Tue Jun 9 16:12:11 2020 +0800

    [FLINK-18168][table-runtime-blink] Fix array reuse for BinaryArrayData in 
converters
    
    This closes #12542.
---
 .../apache/flink/table/dataformat/DataFormatConverters.java  |  2 +-
 .../flink/table/dataformat/DataFormatConvertersTest.java     | 12 +++++++++++-
 2 files changed, 12 insertions(+), 2 deletions(-)

diff --git 
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/DataFormatConverters.java
 
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/DataFormatConverters.java
index 5ed8aa4..9d9fcca 100644
--- 
a/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/DataFormatConverters.java
+++ 
b/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/dataformat/DataFormatConverters.java
@@ -1111,7 +1111,7 @@ public class DataFormatConverters {
                                }
                        }
                        reuseWriter.complete();
-                       return reuseArray;
+                       return reuseArray.copy();
                }
 
                @Override
diff --git 
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/dataformat/DataFormatConvertersTest.java
 
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/dataformat/DataFormatConvertersTest.java
index 2ca2c87..079e32e 100644
--- 
a/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/dataformat/DataFormatConvertersTest.java
+++ 
b/flink-table/flink-table-runtime-blink/src/test/java/org/apache/flink/table/dataformat/DataFormatConvertersTest.java
@@ -146,9 +146,18 @@ public class DataFormatConvertersTest {
        }
 
        private static void test(TypeInformation typeInfo, Object value) {
+               test(typeInfo, value, null);
+       }
+
+       private static void test(TypeInformation typeInfo, Object value, Object 
anotherValue) {
                DataFormatConverter converter = getConverter(typeInfo);
+               final Object innerValue = converter.toInternal(value);
+               if (anotherValue != null) {
+                       converter.toInternal(anotherValue);
+               }
+
                Assert.assertTrue(Arrays.deepEquals(
-                               new Object[] 
{converter.toExternal(converter.toInternal(value))}, new Object[] {value}));
+                       new Object[] {converter.toExternal(innerValue)}, new 
Object[]{value}));
        }
 
        private static DataFormatConverter getConverter(DataType dataType) {
@@ -191,6 +200,7 @@ public class DataFormatConvertersTest {
                test(BasicArrayTypeInfo.DOUBLE_ARRAY_TYPE_INFO, new Double[] 
{null, null});
                test(ObjectArrayTypeInfo.getInfoFor(Types.STRING), new String[] 
{null, null});
                test(ObjectArrayTypeInfo.getInfoFor(Types.STRING), new String[] 
{"haha", "hehe"});
+               test(ObjectArrayTypeInfo.getInfoFor(Types.STRING), new String[] 
{"haha", "hehe"}, new String[] {"aa", "bb"});
                test(new MapTypeInfo<>(Types.STRING, Types.INT), null);
 
                HashMap<String, Integer> map = new HashMap<>();