[Java] support immutable collection/map jit and generics optimization

java/fury-core/src/main/java/io/fury/builder/BaseObjectCodecBuilder.java

@@ -903,6 +903,7 @@ protected Expression deserializeForCollection(
     // if add branch by `ArrayList`, generated code will be > 325 bytes.
     // and List#add is more likely be inlined if there is only one subclass.
     Expression hookRead = readCollectionCodegen(buffer, collection, size, elementType);
+    hookRead = new Invoke(serializer, "onCollectionRead", COLLECTION_TYPE, hookRead);
     Expression action =
         new If(
             supportHook,
@@ -988,6 +989,7 @@ protected Expression deserializeForMap(
             });
     // first newMap to create map, last newMap as expr value
     Expression hookRead = new ListExpression(size, newMap, readKeyValues, newMap);
+    hookRead = new Invoke(serializer, "onMapRead", MAP_TYPE, hookRead);
     Expression action =
         new If(supportHook, hookRead, new Invoke(serializer, "read", MAP_TYPE, buffer), false);
     if (generateNewMethod) {

java/fury-core/src/main/java/io/fury/serializer/ChildContainerSerializers.java

@@ -134,8 +134,8 @@ public void writeHeader(MemoryBuffer buffer, T value) {
       }
     }
 
-    public T newCollection(MemoryBuffer buffer, int numElements) {
-      T collection = super.newCollection(buffer, numElements);
+    public Collection newCollection(MemoryBuffer buffer, int numElements) {
+      Collection collection = super.newCollection(buffer, numElements);
       readAndSetFields(buffer, collection, slotsSerializers);
       return collection;
     }
@@ -149,7 +149,7 @@ public ChildArrayListSerializer(Fury fury, Class<T> cls) {
 
     @Override
     public T newCollection(MemoryBuffer buffer, int numElements) {
-      T collection = super.newCollection(buffer, numElements);
+      T collection = (T) super.newCollection(buffer, numElements);
       collection.ensureCapacity(numElements);
       return collection;
     }
@@ -180,8 +180,8 @@ public void writeHeader(MemoryBuffer buffer, T value) {
     }
 
     @Override
-    public T newMap(MemoryBuffer buffer, int numElements) {
-      T map = super.newMap(buffer, numElements);
+    public Map newMap(MemoryBuffer buffer, int numElements) {
+      Map map = super.newMap(buffer, numElements);
       readAndSetFields(buffer, map, slotsSerializers);
       return map;
     }

java/fury-core/src/main/java/io/fury/serializer/CollectionSerializers.java

@@ -228,13 +228,13 @@ public static void xwriteElements(Fury fury, MemoryBuffer buffer, Collection val
     @Override
     public T read(MemoryBuffer buffer) {
       int numElements = buffer.readPositiveVarInt();
-      T collection = newCollection(buffer, numElements);
+      Collection collection = newCollection(buffer, numElements);
       readElements(fury, buffer, collection, numElements);
-      return collection;
+      return onCollectionRead(collection);
     }
 
     protected final void readElements(
-        Fury fury, MemoryBuffer buffer, T collection, int numElements) {
+        Fury fury, MemoryBuffer buffer, Collection collection, int numElements) {
       Serializer elemSerializer = this.elemSerializer;
       // clear the elemSerializer to avoid conflict if the nested
       // serialization has collection field.
@@ -267,7 +267,7 @@ protected final void readElements(
     private void javaReadWithGenerics(
         Fury fury,
         MemoryBuffer buffer,
-        T collection,
+        Collection collection,
         int numElements,
         GenericType elemGenericType) {
       Serializer elemSerializer;
@@ -306,9 +306,9 @@ private void javaReadWithGenerics(
     @Override
     public T xread(MemoryBuffer buffer) {
       int numElements = buffer.readPositiveVarInt();
-      T collection = newCollection(buffer, numElements);
+      Collection collection = newCollection(buffer, numElements);
       xreadElements(fury, buffer, collection, numElements);
-      return collection;
+      return onCollectionRead(collection);
     }
 
     public static void xreadElements(
@@ -382,7 +382,7 @@ public void writeHeader(MemoryBuffer buffer, T value) {}
      *   <li>read elements
      * </ol>
      */
-    public T newCollection(MemoryBuffer buffer, int numElements) {
+    public Collection newCollection(MemoryBuffer buffer, int numElements) {
       if (constructor == null) {
         constructor = ReflectionUtils.newAccessibleNoArgConstructor(type);
       }
@@ -395,6 +395,10 @@ public T newCollection(MemoryBuffer buffer, int numElements) {
             "Please provide public no arguments constructor for class " + type, e);
       }
     }
+
+    public T onCollectionRead(Collection collection) {
+      return (T) collection;
+    }
   }
 
   public static final class ArrayListSerializer extends CollectionSerializer<ArrayList> {

java/fury-core/src/main/java/io/fury/serializer/Container.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2023 The Fury Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.fury.serializer;
+
+import java.util.AbstractCollection;
+import java.util.AbstractMap;
+import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Set;
+
+class Container {}
+
+/**
+ * A collection container to hold collection elements by array.
+ *
+ * @author chaokunyang
+ */
+class CollectionContainer<T> extends AbstractCollection<T> {
+  final Object[] elements;
+  int size;
+
+  public CollectionContainer(int capacity) {
+    elements = new Object[capacity];
+  }
+
+  @Override
+  public Iterator<T> iterator() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public int size() {
+    return size;
+  }
+
+  @Override
+  public boolean add(Object o) {
+    elements[size++] = o;
+    return true;
+  }
+}
+
+/**
+ * A sorted collection container to hold collection elements and comparator.
+ *
+ * @author chaokunyang
+ */
+class SortedCollectionContainer<T> extends CollectionContainer<T> {
+  Comparator<T> comparator;
+
+  public SortedCollectionContainer(Comparator<T> comparator, int capacity) {
+    super(capacity);
+    this.comparator = comparator;
+  }
+}
+
+/**
+ * A map container to hold map key and value elements by arrays.
+ *
+ * @author chaokunyang
+ */
+class MapContainer<K, V> extends AbstractMap<K, V> {
+  final Object[] keyArray;
+  final Object[] valueArray;
+  int size;
+
+  public MapContainer(int capacity) {
+    keyArray = new Object[capacity];
+    valueArray = new Object[capacity];
+  }
+
+  @Override
+  public Set<Entry<K, V>> entrySet() {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public V put(K key, V value) {
+    keyArray[size] = key;
+    valueArray[size++] = value;
+    return null;
+  }
+}
+
+/**
+ * A sorted map container to hold map data and comparator.
+ *
+ * @author chaokunyang
+ */
+class SortedMapContainer<K, V> extends MapContainer<K, V> {
+
+  final Comparator<K> comparator;
+
+  public SortedMapContainer(Comparator<K> comparator, int capacity) {
+    super(capacity);
+    this.comparator = comparator;
+  }
+}