feat(scala): support native copy for Scala collections

scala/fory-scala/src/main/scala/org/apache/fory/serializer/scala/CollectionSerializer.scala

@@ -19,6 +19,7 @@
 
 package org.apache.fory.serializer.scala
 
+import org.apache.fory.context.CopyContext
 import org.apache.fory.context.ReadContext
 import org.apache.fory.context.WriteContext
 import org.apache.fory.resolver.TypeResolver
@@ -82,6 +83,91 @@ abstract class AbstractScalaCollectionSerializer[A, T <: Iterable[A]](
   override def onCollectionRead(collection: util.Collection[_]): T = {
     collection.asInstanceOf[JavaCollectionBuilder[A, T]].builder.result()
   }
+
+  override def copy(copyContext: CopyContext, value: T): T = {
+    if (isImmutable) {
+      value
+    } else if (
+        value.isInstanceOf[mutable.IndexedSeq[_]] &&
+        !value.isInstanceOf[mutable.Growable[_]]) {
+      copyIndexedSeq(copyContext, value, value.asInstanceOf[mutable.IndexedSeq[A]])
+    } else if (value.isInstanceOf[mutable.Iterable[_]]) {
+      newMutableCopy(value, value.size) match {
+        case result: mutable.Iterable[_] with mutable.Growable[_] =>
+          val growable = result.asInstanceOf[mutable.Iterable[A] with mutable.Growable[A]]
+          copyContext.reference(value, growable.asInstanceOf[T])
+          copyElements(copyContext, value, growable)
+          growable.asInstanceOf[T]
+        case _ =>
+          copyWithBuilder(copyContext, value, newCopyBuilder(value))
+      }
+    } else {
+      copyWithBuilder(copyContext, value, newCopyBuilder(value))
+    }
+  }
+
+  protected def newCopyBuilder(value: T): mutable.Builder[A, T] =
+    value.iterableFactory.newBuilder[A].asInstanceOf[mutable.Builder[A, T]]
+
+  private def newMutableCopy(value: T, numElements: Int): scala.collection.Iterable[A] = {
+    val builder = newCopyBuilder(value)
+    builder.sizeHint(numElements)
+    builder.result()
+  }
+
+  private def copyElements(
+      copyContext: CopyContext,
+      value: T,
+      result: mutable.Growable[A]): Unit = {
+    val iterator = value.iterator
+    while (iterator.hasNext) {
+      result.addOne(copyContext.copyObject(iterator.next()).asInstanceOf[A])
+    }
+  }
+
+  private def copyWithBuilder(
+      copyContext: CopyContext,
+      value: T,
+      builder: mutable.Builder[A, T]): T = {
+    val iterator = value.iterator
+    while (iterator.hasNext) {
+      builder.addOne(copyContext.copyObject(iterator.next()).asInstanceOf[A])
+    }
+    // A recursive copy through a mutable child may already have constructed and registered the
+    // canonical immutable copy of this collection. Reuse it instead of returning a second builder
+    // result that would split the cycle's identity.
+    val recursiveCopy = copyContext.getCopyObject(value)
+    if (recursiveCopy != null) {
+      recursiveCopy
+    } else {
+      val result = builder.result()
+      copyContext.reference(value, result)
+      result
+    }
+  }
+
+  private def copyIndexedSeq(
+      copyContext: CopyContext,
+      value: T,
+      indexed: mutable.IndexedSeq[A]): T = {
+    val result = indexed match {
+      case arraySeq: mutable.ArraySeq[_] =>
+        val sourceArray = arraySeq.array.asInstanceOf[AnyRef]
+        val array =
+          java.lang.reflect.Array.newInstance(sourceArray.getClass.getComponentType, indexed.size)
+        mutable.ArraySeq.make(array.asInstanceOf[Array[_]]).asInstanceOf[mutable.IndexedSeq[A]]
+      case _ =>
+        mutable.ArraySeq.make(new Array[Any](indexed.size)).asInstanceOf[mutable.IndexedSeq[A]]
+    }
+    val copied = result.asInstanceOf[T]
+    copyContext.reference(value, copied)
+    var i = 0
+    while (i < indexed.size) {
+      result.update(i, copyContext.copyObject(indexed(i)).asInstanceOf[A])
+      i += 1
+    }
+    copied
+  }
 }
 
 /**
@@ -171,6 +257,12 @@ class ScalaSortedSetSerializer[A, T <: scala.collection.SortedSet[A]](
     typeResolver: TypeResolver,
     cls: Class[T])
   extends AbstractScalaCollectionSerializer[A, T](typeResolver, cls) {
+  override protected def newCopyBuilder(value: T): mutable.Builder[A, T] = {
+    val factory = value.sortedIterableFactory.evidenceIterableFactory[Any](
+      value.ordering.asInstanceOf[Ordering[Any]])
+    factory.newBuilder.asInstanceOf[mutable.Builder[A, T]]
+  }
+
   override def onCollectionWrite(
       writeContext: WriteContext,
       value: T): util.Collection[_] = {

scala/fory-scala/src/main/scala/org/apache/fory/serializer/scala/MapSerializer.scala

@@ -20,6 +20,7 @@
 package org.apache.fory.serializer.scala
 
 import org.apache.fory.collection.MapEntry
+import org.apache.fory.context.CopyContext
 import org.apache.fory.context.ReadContext
 import org.apache.fory.context.WriteContext
 import org.apache.fory.resolver.TypeResolver
@@ -45,7 +46,9 @@ import scala.collection.{Factory, mutable}
  * invoke [[java.util.Map#put]] to add it into builder.</li>
  * <li>`onMapRead`: create scala map from builder.</li>
  */
-abstract class AbstractScalaMapSerializer[K, V, T](typeResolver: TypeResolver, cls: Class[T])
+abstract class AbstractScalaMapSerializer[K, V, T <: scala.collection.Map[K, V]](
+    typeResolver: TypeResolver,
+    cls: Class[T])
   extends MapLikeSerializer[T](typeResolver, cls) {
   private val ReferenceBytes = 4L
   // Lower-bound shallow owner cost for the retained Scala map wrapper itself. Key/value slots are
@@ -84,6 +87,72 @@ abstract class AbstractScalaMapSerializer[K, V, T](typeResolver: TypeResolver, c
   override def onMapRead(map: util.Map[_, _]): T = {
     map.asInstanceOf[MapBuilder[K, V, T]].builder.result()
   }
+
+  override def copy(copyContext: CopyContext, value: T): T = {
+    if (isImmutable) {
+      value
+    } else if (value.isInstanceOf[mutable.Map[_, _]]) {
+      newMutableMapCopy(value, value.size) match {
+        case result: mutable.Map[_, _] =>
+          val mutableResult = result.asInstanceOf[mutable.Map[K, V]]
+          copyContext.reference(value, mutableResult.asInstanceOf[T])
+          copyEntries(copyContext, value, mutableResult)
+          mutableResult.asInstanceOf[T]
+        case _ =>
+          copyWithBuilder(copyContext, value, newCopyBuilder(value))
+      }
+    } else {
+      copyWithBuilder(copyContext, value, newCopyBuilder(value))
+    }
+  }
+
+  protected def newCopyBuilder(value: T): mutable.Builder[(K, V), T] =
+    value.mapFactory.newBuilder[K, V].asInstanceOf[mutable.Builder[(K, V), T]]
+
+  private def newMutableMapCopy(value: T, numElements: Int): scala.collection.Map[K, V] = {
+    val builder = newCopyBuilder(value)
+    builder.sizeHint(numElements)
+    builder.result()
+  }
+
+  private def copyEntries(
+      copyContext: CopyContext,
+      value: T,
+      result: mutable.Map[K, V]): Unit = {
+    val iterator = value.iterator
+    while (iterator.hasNext) {
+      val entry = iterator.next()
+      result.addOne(
+        (
+          copyContext.copyObject(entry._1).asInstanceOf[K],
+          copyContext.copyObject(entry._2).asInstanceOf[V]))
+    }
+  }
+
+  private def copyWithBuilder(
+      copyContext: CopyContext,
+      value: T,
+      builder: mutable.Builder[(K, V), T]): T = {
+    val iterator = value.iterator
+    while (iterator.hasNext) {
+      val entry = iterator.next()
+      builder.addOne(
+        (
+          copyContext.copyObject(entry._1).asInstanceOf[K],
+          copyContext.copyObject(entry._2).asInstanceOf[V]))
+    }
+    // A recursive copy through a mutable child may already have constructed and registered the
+    // canonical immutable copy of this map. Reuse it instead of returning a second builder result
+    // that would split the cycle's identity.
+    val recursiveCopy = copyContext.getCopyObject(value)
+    if (recursiveCopy != null) {
+      recursiveCopy
+    } else {
+      val result = builder.result()
+      copyContext.reference(value, result)
+      result
+    }
+  }
 }
 
 /**
@@ -154,6 +223,12 @@ class ScalaSortedMapSerializer[K, V, T <: scala.collection.SortedMap[K, V]](
     typeResolver: TypeResolver,
     cls: Class[T])
   extends AbstractScalaMapSerializer[K, V, T](typeResolver, cls) {
+  override protected def newCopyBuilder(value: T): mutable.Builder[(K, V), T] = {
+    val factory = value.sortedMapFactory.sortedMapFactory[Any, Any](
+      value.ordering.asInstanceOf[Ordering[Any]])
+    factory.newBuilder.asInstanceOf[mutable.Builder[(K, V), T]]
+  }
+
   override def onMapWrite(
       writeContext: WriteContext,
       value: T): util.Map[_, _] = {

scala/fory-scala/src/main/scala/org/apache/fory/serializer/scala/RangeSerializer.scala

@@ -19,6 +19,7 @@
 
 package org.apache.fory.serializer.scala
 
+import org.apache.fory.context.CopyContext
 import org.apache.fory.context.ReadContext
 import org.apache.fory.context.WriteContext
 import org.apache.fory.reflect.FieldAccessor
@@ -38,6 +39,8 @@ class RangeSerializer[T <: Range](typeResolver: TypeResolver, cls: Class[T])
   private val rangeClass = cls
   private val graphMemoryBytes = GraphMemoryEstimates.shallowObjectBytes(cls)
 
+  override def copy(copyContext: CopyContext, value: T): T = value
+
   override def write(writeContext: WriteContext, value: T): Unit = {
     val buffer = writeContext.getBuffer
     buffer.writeVarInt32(value.start)
@@ -79,6 +82,9 @@ class NumericRangeSerializer[A, T <: NumericRange[A]](typeResolver: TypeResolver
   with Shareable {
   private val ctr = RangeUtils.lookupCache.get(cls)
   private val graphMemoryBytes = GraphMemoryEstimates.shallowObjectBytes(cls)
+
+  override def copy(copyContext: CopyContext, value: T): T = value
+
   private val getter =
     FieldAccessor.createAccessor(
       cls.getDeclaredFields.find(f => f.getType == classOf[Integral[?]]).get)

scala/fory-scala/src/test/scala/org/apache/fory/serializer/scala/NativeCopyTest.scala

@@ -0,0 +1,139 @@
+/*
+ * 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.fory.serializer.scala
+
+import org.apache.fory.Fory
+import org.apache.fory.scala.ForyScala
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.collection.mutable
+
+private case class CollectionHolder(values: List[Int])
+
+class NativeCopyTest extends AnyWordSpec with Matchers {
+  private def fory: Fory =
+    ForyScala.builder()
+      .withXlang(false)
+      .withRefTracking(true)
+      .withRefCopy(true)
+      .requireClassRegistration(false)
+      .suppressClassRegistrationWarnings(false)
+      .build()
+
+  "native Scala collection copy" should {
+    "copy immutable collections and ranges" in {
+      val runtime = fory
+      val list = List("a", "b")
+      val copiedList = runtime.copy(list).asInstanceOf[List[String]]
+      copiedList shouldEqual list
+      copiedList should not be theSameInstanceAs(list)
+
+      val range = Range.inclusive(1, 4)
+      runtime.copy(range) shouldBe theSameInstanceAs(range)
+
+      val numericRange = scala.collection.immutable.NumericRange.inclusive(1, 4, 1)
+      runtime.copy(numericRange) shouldBe theSameInstanceAs(numericRange)
+
+      val sortedSet = scala.collection.immutable.TreeSet(3, 1, 2)
+      val copiedSet = runtime.copy(sortedSet).asInstanceOf[scala.collection.immutable.TreeSet[Int]]
+      copiedSet.toList shouldEqual List(1, 2, 3)
+      copiedSet should not be theSameInstanceAs(sortedSet)
+
+      val sortedMap = scala.collection.immutable.TreeMap(3 -> "c", 1 -> "a", 2 -> "b")
+      val copiedMap =
+        runtime.copy(sortedMap).asInstanceOf[scala.collection.immutable.TreeMap[Int, String]]
+      copiedMap.toList shouldEqual List(1 -> "a", 2 -> "b", 3 -> "c")
+      copiedMap should not be theSameInstanceAs(sortedMap)
+
+      val holder = CollectionHolder(list.map(_.length))
+      val copiedHolder = runtime.copy(holder).asInstanceOf[CollectionHolder]
+      copiedHolder shouldEqual holder
+      copiedHolder should not be theSameInstanceAs(holder)
+      copiedHolder.values should not be theSameInstanceAs(holder.values)
+    }
+
+    "copy mutable collections and maps" in {
+      val runtime = fory
+      val list = mutable.ListBuffer(1, 2)
+      val copiedList = runtime.copy(list).asInstanceOf[mutable.ListBuffer[Int]]
+      copiedList shouldEqual list
+      copiedList should not be theSameInstanceAs(list)
+
+      val indexed = mutable.ArraySeq[AnyRef](null)
+      indexed.update(0, indexed)
+      val copiedIndexed = runtime.copy(indexed).asInstanceOf[mutable.ArraySeq[AnyRef]]
+      copiedIndexed(0) shouldBe theSameInstanceAs(copiedIndexed)
+    }
+
+    "preserve cycles in mutable collections and maps" in {
+      val runtime = fory
+      val list = mutable.ArrayBuffer.empty[AnyRef]
+      list += list
+      val copiedList = runtime.copy(list).asInstanceOf[mutable.ArrayBuffer[AnyRef]]
+      copiedList should not be theSameInstanceAs(list)
+      copiedList.head shouldBe theSameInstanceAs(copiedList)
+
+      val map = mutable.LinkedHashMap.empty[String, AnyRef]
+      map.put("self", map)
+      val copiedMap = runtime.copy(map).asInstanceOf[mutable.LinkedHashMap[String, AnyRef]]
+      copiedMap should not be theSameInstanceAs(map)
+      copiedMap("self") shouldBe theSameInstanceAs(copiedMap)
+    }
+
+    "preserve identity across immutable collection and mutable map cycles" in {
+      val runtime = fory
+      val map = mutable.LinkedHashMap.empty[String, AnyRef]
+      val list = List[AnyRef](map)
+      map.put("list", list)
+
+      val copiedList = runtime.copy(list).asInstanceOf[List[AnyRef]]
+      val copiedMap = copiedList.head.asInstanceOf[mutable.LinkedHashMap[String, AnyRef]]
+      if (copiedList.asInstanceOf[AnyRef] eq list.asInstanceOf[AnyRef]) {
+        throw new AssertionError("immutable list was not copied")
+      }
+      if (copiedMap.asInstanceOf[AnyRef] eq map.asInstanceOf[AnyRef]) {
+        throw new AssertionError("mutable map was not copied")
+      }
+      if (!(copiedMap("list").asInstanceOf[AnyRef] eq copiedList.asInstanceOf[AnyRef])) {
+        throw new AssertionError("copied mutable map must point to the outer copied immutable list")
+      }
+    }
+
+    "preserve identity across immutable map and mutable collection cycles" in {
+      val runtime = fory
+      val list = mutable.ListBuffer.empty[AnyRef]
+      val map = scala.collection.immutable.Map[String, AnyRef]("list" -> list)
+      list += map
+
+      val copiedMap = runtime.copy(map).asInstanceOf[scala.collection.immutable.Map[String, AnyRef]]
+      val copiedList = copiedMap("list").asInstanceOf[mutable.ListBuffer[AnyRef]]
+      if (copiedMap.asInstanceOf[AnyRef] eq map.asInstanceOf[AnyRef]) {
+        throw new AssertionError("immutable map was not copied")
+      }
+      if (copiedList.asInstanceOf[AnyRef] eq list.asInstanceOf[AnyRef]) {
+        throw new AssertionError("mutable list was not copied")
+      }
+      if (!(copiedList.head.asInstanceOf[AnyRef] eq copiedMap.asInstanceOf[AnyRef])) {
+        throw new AssertionError("copied mutable list must point to the outer copied immutable map")
+      }
+    }
+  }
+}