feat(java): implement sublist serializers

java/fury-core/src/main/java/org/apache/fury/resolver/ClassResolver.java

@@ -139,6 +139,7 @@
 import org.apache.fury.serializer.collection.ImmutableCollectionSerializers;
 import org.apache.fury.serializer.collection.MapSerializer;
 import org.apache.fury.serializer.collection.MapSerializers;
+import org.apache.fury.serializer.collection.SubListSerializers;
 import org.apache.fury.serializer.collection.SynchronizedSerializers;
 import org.apache.fury.serializer.collection.UnmodifiableSerializers;
 import org.apache.fury.serializer.scala.SingletonCollectionSerializer;
@@ -337,6 +338,7 @@ private void addDefaultSerializers() {
     SynchronizedSerializers.registerSerializers(fury);
     UnmodifiableSerializers.registerSerializers(fury);
     ImmutableCollectionSerializers.registerSerializers(fury);
+    SubListSerializers.registerSerializers(fury, true);
     if (fury.getConfig().registerGuavaTypes()) {
       GuavaCollectionSerializers.registerDefaultSerializers(fury);
     }

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

@@ -19,7 +19,156 @@
 
 package org.apache.fury.serializer.collection;
 
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.fury.Fury;
+import org.apache.fury.config.Language;
+import org.apache.fury.logging.Logger;
+import org.apache.fury.logging.LoggerFactory;
+import org.apache.fury.memory.MemoryBuffer;
+import org.apache.fury.reflect.ReflectionUtils;
+import org.apache.fury.serializer.ObjectSerializer;
+
+@SuppressWarnings({"rawtypes", "unchecked"})
 public class SubListSerializers {
-  // ArrayListSubListSerializer
-  // JavaUtilSubListSerializer
+  private static final Logger LOG = LoggerFactory.getLogger(SubListSerializers.class);
+
+  private static final Class<?> SubListClass;
+  private static final Class<?> RandomAccessSubListClass;
+  private static final Class<?> ArrayListSubListClass;
+  private static final Class<?> ImmutableSubListClass;
+
+  private interface Stub {}
+
+  static {
+    Class<?> sublistClass;
+    try {
+      sublistClass = Class.forName("java.util.SubList");
+    } catch (ClassNotFoundException e) {
+      sublistClass = ReflectionUtils.loadClass("java.util.AbstractList$SubList");
+    }
+    SubListClass = sublistClass;
+    Class<?> randomAccessSubListClass;
+    try {
+      randomAccessSubListClass = Class.forName("java.util.RandomAccessSubList");
+    } catch (ClassNotFoundException e) {
+      randomAccessSubListClass =
+          ReflectionUtils.loadClass("java.util.AbstractList$RandomAccessSubList");
+    }
+    RandomAccessSubListClass = randomAccessSubListClass;
+    try {
+      ArrayListSubListClass = Class.forName("java.util.ArrayList$SubList");
+    } catch (ClassNotFoundException e) {
+      throw new RuntimeException(e);
+    }
+    Class<?> cls;
+    try {
+      cls = Class.forName("java.util.ImmutableCollections$SubList");
+    } catch (ClassNotFoundException e) {
+      class ImmutableSubListStub implements Stub {}
+
+      cls = ImmutableSubListStub.class;
+    }
+    ImmutableSubListClass = cls;
+  }
+
+  public static void registerSerializers(Fury fury, boolean preserveView) {
+    for (Class<?> cls :
+        new Class[] {
+          SubListClass, RandomAccessSubListClass, ArrayListSubListClass, ImmutableSubListClass
+        }) {
+      if (preserveView && fury.getConfig().getLanguage() == Language.JAVA) {
+        fury.registerSerializer(cls, new SubListViewSerializer(fury, (Class<List>) cls));
+      } else {
+        fury.registerSerializer(cls, new SubListSerializer(fury, (Class<List>) cls));
+      }
+    }
+  }
+
+  public static final class SubListViewSerializer extends CollectionSerializer<List> {
+    private ObjectSerializer dataSerializer;
+    private boolean serializedBefore;
+
+    public SubListViewSerializer(Fury fury, Class cls) {
+      super(fury, Stub.class.isAssignableFrom(cls) ? (Class<List>) ArrayListSubListClass : cls);
+      assert fury.getLanguage() == Language.JAVA;
+    }
+
+    @Override
+    public Collection onCollectionWrite(MemoryBuffer buffer, List value) {
+      throw new IllegalStateException();
+    }
+
+    @Override
+    public List onCollectionRead(Collection collection) {
+      throw new IllegalStateException();
+    }
+
+    @Override
+    public void write(MemoryBuffer buffer, List value) {
+      checkSerialization(value);
+      (getObjectSerializer()).write(buffer, value);
+    }
+
+    @Override
+    public List read(MemoryBuffer buffer) {
+      List value = (List) (getObjectSerializer()).read(buffer);
+      checkSerialization(value);
+      return value;
+    }
+
+    private ObjectSerializer getObjectSerializer() {
+      ObjectSerializer dataSerializer = this.dataSerializer;
+      if (dataSerializer == null) {
+        dataSerializer = this.dataSerializer = new ObjectSerializer(fury, type);
+      }
+      return dataSerializer;
+    }
+
+    @Override
+    public List copy(List value) {
+      throw new UnsupportedOperationException(
+          "parent list didn't copy modCount, but sublist does copy it");
+    }
+
+    private void checkSerialization(Object value) {
+      if (!serializedBefore) {
+        serializedBefore = true;
+        LOG.warn(
+            "List view of type {} is being serialized/deserialized, this is not recommended, please don't "
+                + "serialize such types, it's not allowed for serialization by JDK too. "
+                + "To ensure consistency between view and the raw List, we must serialize raw List together even if "
+                + "the raw List is not referenced other places. This may cause extra data be serialized if the original List "
+                + "is not referenced in the object graph being serialized, but this is necessary. "
+                + "Otherwise, serializing multiple view of same original list will bring data duplication, "
+                + "and if you update the view, the original list won't be updated too. "
+                + "If you want to serialize SubList view as a standard List, you can register a serializer by "
+                + "`fury.registerSerializer(cls, new SubListSerializer(fury, (Class<List>) cls))`, object type of deserialized "
+                + "value will be {}",
+            value.getClass(),
+            ArrayList.class);
+      }
+    }
+  }
+
+  public static final class SubListSerializer extends CollectionSerializer {
+
+    public SubListSerializer(Fury fury, Class<List> type) {
+      super(fury, type, true);
+      fury.getClassResolver().setSerializer(type, this);
+    }
+
+    @Override
+    public Collection newCollection(MemoryBuffer buffer) {
+      int numElements = buffer.readVarUint32Small7();
+      setNumElements(numElements);
+      return new ArrayList(numElements);
+    }
+
+    @Override
+    public Collection newCollection(Collection collection) {
+      return new ArrayList(collection.size());
+    }
+  }
 }

java/fury-core/src/main/resources/META-INF/native-image/org.apache.fury/fury-core/native-image.properties

@@ -114,6 +114,12 @@ Args=--initialize-at-build-time=org.apache.fury.memory.MemoryBuffer,\
     java.time.ZonedDateTime,\
     java.util.ArrayDeque,\
     java.util.ArrayList,\
+    java.util.SubList,\
+    java.util.AbstractList$SubList,\
+    java.util.RandomAccessSubList,\
+    java.util.AbstractList$RandomAccessSubList,\
+    java.util.ArrayList$SubList,\
+    java.util.ImmutableCollections$SubList,\
     java.util.Arrays$ArrayList,\
     java.util.BitSet,\
     java.util.Collections$EmptyList,\
@@ -278,6 +284,9 @@ Args=--initialize-at-build-time=org.apache.fury.memory.MemoryBuffer,\
     org.apache.fury.serializer.ArraySerializers$ShortArraySerializer,\
     org.apache.fury.serializer.ArraySerializers$StringArraySerializer,\
     org.apache.fury.serializer.ArraySerializers,\
+    org.apache.fury.serializer.collection.SubListSerializers,\
+    org.apache.fury.serializer.collection.SubListSerializers$SubListViewSerializer,\
+    org.apache.fury.serializer.collection.SubListSerializers$SubListSerializer,\
     org.apache.fury.serializer.JavaSerializer$1,\
     org.apache.fury.serializer.JavaSerializer$2,\
     org.apache.fury.serializer.JavaSerializer$3,\

java/fury-core/src/test/java/org/apache/fury/serializer/collection/SubListSerializersTest.java

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.fury.serializer.collection;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+import org.apache.fury.Fury;
+import org.apache.fury.FuryTestBase;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+public class SubListSerializersTest extends FuryTestBase {
+  @Test
+  public void testSubListViewSerialization() {
+    Fury fury = builder().build();
+    List<Integer> data = new ArrayList<>();
+    Collections.addAll(data, 1, 2, 3, 4, 5, 6, 7);
+    int length = fury.serialize(data).length;
+
+    List<Integer> list1 = new ArrayList<>(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list1).length > length);
+    // TODO parent list didn't copy modCount, but sublist does copy it.
+    // Assert.assertEquals(fury.copy(list1), list1);
+
+    List<Integer> list2 = new LinkedList<>(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list2).length > length);
+    // TODO parent list didn't copy modCount, but sublist does copy it.
+    // Assert.assertEquals(fury.copy(list2), list2);
+
+    List<Integer> list3 = Collections.unmodifiableList(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list3).length > length);
+    // TODO parent list didn't copy modCount, but sublist does copy it.
+    // Assert.assertEquals(fury.copy(list3), list3);
+
+    List<Integer> list4 = Collections.synchronizedList(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list4).length > length);
+    // TODO parent list didn't copy modCount, but sublist does copy it.
+    // Assert.assertEquals(fury.copy(list4), list4);
+
+    List<Object> list5 = Arrays.asList(data.toArray()).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list5).length > length);
+    // TODO parent list didn't copy modCount, but sublist does copy it.
+    // Assert.assertEquals(fury.copy(list5), list5);
+  }
+
+  @Test
+  public void testSubListNoViewSerialization() {
+    Fury fury = builder().build();
+    SubListSerializers.registerSerializers(fury, false);
+    List<Integer> data = new ArrayList<>();
+    Collections.addAll(data, 1, 2, 3, 4, 5, 6, 7);
+    int length = fury.serialize(data).length;
+
+    List<Integer> list1 = new ArrayList<>(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list1).length < length);
+    Assert.assertEquals(fury.copy(list1), list1);
+
+    List<Integer> list2 = new LinkedList<>(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list2).length < length);
+    Assert.assertEquals(fury.copy(list2), list2);
+
+    List<Integer> list3 = Collections.unmodifiableList(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list3).length < length);
+    Assert.assertEquals(fury.copy(list3), list3);
+
+    List<Integer> list4 = Collections.synchronizedList(data).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list4).length < length);
+    Assert.assertEquals(fury.copy(list4), list4);
+
+    List<Object> list5 = Arrays.asList(data.toArray()).subList(1, 3);
+    Assert.assertTrue(fury.serialize(list5).length < length);
+    Assert.assertEquals(fury.copy(list5), list5);
+  }
+}