perf(java): optimize map copy perf

java/benchmark/src/main/java/org/apache/fury/benchmark/CopyBenchmark.java

@@ -42,10 +42,40 @@ public Object kryo_copy(KryoState.KryoUserTypeState state) {
     return state.kryo.copy(state.object);
   }
 
+  @Benchmark
+  public Object fury_copy_string_map(FuryState.DataState state) {
+    return state.fury.copy(state.data.stringMap);
+  }
+
+  @Benchmark
+  public Object fury_copy_int_map(FuryState.DataState state) {
+    return state.fury.copy(state.data.intMap);
+  }
+
+  @Benchmark
+  public Object kryo_copy_string_map(KryoState.DataState state) {
+    return state.kryo.copy(state.data.stringMap);
+  }
+
+  @Benchmark
+  public Object kryo_copy_int_map(KryoState.DataState state) {
+    return state.kryo.copy(state.data.intMap);
+  }
+
+  @Benchmark
+  public Object fury_copy_list(FuryState.DataState state) {
+    return state.fury.copy(state.data.intList);
+  }
+
+  @Benchmark
+  public Object kryo_copy_list(KryoState.DataState state) {
+    return state.kryo.copy(state.data.intList);
+  }
+
   public static void main(String[] args) throws IOException {
     if (args.length == 0) {
       String commandLine =
-          "org.apache.fury.*CopyBenchmark.* -f 1 -wi 3 -i 3 -t 1 -w 2s -r 2s -rf csv "
+          "org.apache.fury.*CopyBenchmark.*map -f 1 -wi 3 -i 3 -t 1 -w 2000s -r 2s -rf csv "
               + "-p bufferType=array -p references=false";
       System.out.println(commandLine);
       args = commandLine.split(" ");

java/benchmark/src/main/java/org/apache/fury/benchmark/data/Data.java

@@ -19,6 +19,10 @@
 
 package org.apache.fury.benchmark.data;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 import org.apache.fury.util.StringUtils;
 
 public class Data {
@@ -93,4 +97,18 @@ public static String newLongStr() {
     }
     return StringUtils.random(strLength);
   }
+
+  public List<String> stringList = new ArrayList<>();
+  public List<Integer> intList = new ArrayList<>();
+  public Map<String, String> stringMap = new HashMap<>();
+  public Map<Integer, Integer> intMap = new HashMap<>();
+
+  {
+    for (int i = 0; i < 20; i++) {
+      stringList.add("hello, " + i);
+      intList.add(i);
+      stringMap.put("key" + i, "value" + i);
+      intMap.put(i, i * 2);
+    }
+  }
 }

java/benchmark/src/main/java/org/apache/fury/benchmark/state/KryoState.java

@@ -26,6 +26,7 @@
 import com.esotericsoftware.kryo.unsafe.UnsafeByteBufferInput;
 import com.esotericsoftware.kryo.unsafe.UnsafeByteBufferOutput;
 import java.util.ArrayList;
+import java.util.HashMap;
 import org.apache.fury.benchmark.IntsSerializationSuite;
 import org.apache.fury.benchmark.LongStringSerializationSuite;
 import org.apache.fury.benchmark.LongsSerializationSuite;
@@ -85,6 +86,8 @@ public void setup() {
       kryo.setRegistrationRequired(registerClass);
       kryo.register(int[].class);
       kryo.register(long[].class);
+      kryo.register(ArrayList.class);
+      kryo.register(HashMap.class);
     }
   }
 

java/fury-core/src/main/java/org/apache/fury/serializer/collection/AbstractMapSerializer.java

@@ -29,6 +29,7 @@
 import org.apache.fury.memory.MemoryBuffer;
 import org.apache.fury.reflect.ReflectionUtils;
 import org.apache.fury.reflect.TypeRef;
+import org.apache.fury.resolver.ClassInfo;
 import org.apache.fury.resolver.ClassInfoHolder;
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.resolver.RefResolver;
@@ -417,8 +418,24 @@ public T xread(MemoryBuffer buffer) {
   }
 
   protected <K, V> void copyEntry(Map<K, V> originMap, Map<K, V> newMap) {
+    ClassResolver classResolver = fury.getClassResolver();
     for (Map.Entry<K, V> entry : originMap.entrySet()) {
-      newMap.put(fury.copyObject(entry.getKey()), fury.copyObject(entry.getValue()));
+      K key = entry.getKey();
+      if (key != null) {
+        ClassInfo classInfo = classResolver.getClassInfo(key.getClass(), keyClassInfoWriteCache);
+        if (!classInfo.getSerializer().isImmutable()) {
+          key = fury.copyObject(key, classInfo.getClassId());
+        }
+      }
+      V value = entry.getValue();
+      if (value != null) {
+        ClassInfo classInfo =
+            classResolver.getClassInfo(value.getClass(), valueClassInfoWriteCache);
+        if (!classInfo.getSerializer().isImmutable()) {
+          value = fury.copyObject(value, classInfo.getClassId());
+        }
+      }
+      newMap.put(key, value);
     }
   }