[flink] 01/03: [hotfix] Improve performance of GenericArraySerializer.copy()

[sr...@apache.org]

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

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

commit 28e0b83dd86cbf2a8d0eeda95391f206ba0d812b
Author: Stefan Richter <s....@data-artisans.com>
AuthorDate: Wed Jun 6 17:15:20 2018 +0200

    [hotfix] Improve performance of GenericArraySerializer.copy()
---
 .../typeutils/base/GenericArraySerializer.java       | 20 +++++++++++++-------
 1 file changed, 13 insertions(+), 7 deletions(-)

diff --git a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
index 6668c57..55ba8ab 100644
--- a/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
+++ b/flink-core/src/main/java/org/apache/flink/api/common/typeutils/base/GenericArraySerializer.java
@@ -25,6 +25,7 @@ import org.apache.flink.core.memory.DataOutputView;
 
 import java.io.IOException;
 import java.lang.reflect.Array;
+import java.util.Arrays;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
 
@@ -86,16 +87,21 @@ public final class GenericArraySerializer<C> extends TypeSerializer<C[]> {
 
 	@Override
 	public C[] copy(C[] from) {
-		C[] copy = create(from.length);
 
-		for (int i = 0; i < copy.length; i++) {
-			C val = from[i];
-			if (val != null) {
-				copy[i] = this.componentSerializer.copy(val);
+		final TypeSerializer<C> serializer = this.componentSerializer;
+
+		if (serializer.isImmutableType()) {
+			return Arrays.copyOf(from, from.length);
+		} else {
+			C[] copy = create(from.length);
+			for (int i = 0; i < copy.length; i++) {
+				C val = from[i];
+				if (val != null) {
+					copy[i] = serializer.copy(val);
+				}
 			}
+			return copy;
 		}
-
-		return copy;
 	}
 	
 	@Override