feat(kotlin): Add unsigned array support and tests for arrays and str…

…ings (#1891)

<!--
**Thanks for contributing to Fury.**

**If this is your first time opening a PR on fury, you can refer to
[CONTRIBUTING.md](https://github.com/apache/fury/blob/main/CONTRIBUTING.md).**

Contribution Checklist

- The **Apache Fury (incubating)** community has restrictions on the
naming of pr titles. You can also find instructions in
[CONTRIBUTING.md](https://github.com/apache/fury/blob/main/CONTRIBUTING.md).

- Fury has a strong focus on performance. If the PR you submit will have
an impact on performance, please benchmark it first and provide the
benchmark result here.
-->

## What does this PR do?

This PR adds tests for serializing/deserializing:
- Strings (same as Java)
- Primitive arrays in Kotlin (same as Java)
- Array<T> in kotlin (same as Java)
- Unsigned arrays in kotlin - `UByteArray`, `UIntArray`, `UShortArray`,
`ULongArray`

Unsigned arrays in kotlin are currently marked experimental, and are
subject to API changes (hence the annotations needed to suppress those
warnings).

These types are implemented as a view over the signed arrays e.g.
UByteArray is a view over ByteArray with contents reinterpreted as
UByte, so serializers. The current implementation delegate to existing
serializers for corresponding signed types.

The xlang type id is set to LIST for unsigned types.

## Related issues

#683

<!--
Is there any related issue? Please attach here.

- #xxxx0
- #xxxx1
- #xxxx2
-->

## Does this PR introduce any user-facing change?

Yes. Unsigned primitives no longer need to be registered for
`fury-kotlin`.

- [ ] Does this PR introduce any public API change?
- [ ] Does this PR introduce any binary protocol compatibility change?

## Benchmark
N/A

kotlin/README.md

@@ -8,14 +8,19 @@ Fury Kotlin provides additional tests and implementation support for Kotlin type
 
 Fury Kotlin is tested and works with the following types:
 
-- primitives: `Byte`, `Boolean`, `Int`, `Short`, `Long`, `Char`, `Float`, `Double`, `UByte`, `UShort`, `UInt`, `UShort`, `UInt`.
+- primitives: `Byte`, `Boolean`, `Int`, `Short`, `Long`, `Char`, `Float`, `Double`, `UByte`, `UShort`, `UInt`, `ULong`.
 - `Byte`, `Boolean`, `Int`, `Short`, `Long`, `Char`, `Float`, `Double` works out of the box with the default fury java implementation.
 - stdlib `collection`: `ArrayDeque`, `ArrayList`, `HashMap`,`HashSet`, `LinkedHashSet`, `LinkedHashMap`.
 - `ArrayList`, `HashMap`,`HashSet`, `LinkedHashSet`, `LinkedHashMap` works out of the box with the default fury java implementation.
+- `String` works out of the box with the default fury java implementation.
+- arrays: `Array`, `BooleanArray`, `ByteArray`, `CharArray`, `DoubleArray`, `FloatArray`, `IntArray`, `LongArray`, `ShortArray`
+- all standard array types work out of the box with the default fury java implementation.
+- unsigned arrays: `UByteArray`, `UShortArray`, `UIntArray`, `ULongArray`
 
 Additional support is added for the following classes in kotlin:
 
 - Unsigned primitives: `UByte`, `UShort`, `UInt`, `ULong`
+- Unsigned array types: `UByteArray`, `UShortArray`, `UIntArray`, `ULongArray`
 - Empty collections: `emptyList`, `emptyMap`, `emptySet`
 - Collections: `ArrayDeque`
 

kotlin/src/main/java/org/apache/fury/serializer/kotlin/KotlinSerializers.java

@@ -19,6 +19,10 @@
 
 package org.apache.fury.serializer.kotlin;
 
+import kotlin.UByteArray;
+import kotlin.UIntArray;
+import kotlin.ULongArray;
+import kotlin.UShortArray;
 import org.apache.fury.Fury;
 import org.apache.fury.resolver.ClassResolver;
 import org.apache.fury.serializer.collection.CollectionSerializers;
@@ -72,5 +76,15 @@ public static void registerSerializers(Fury fury) {
         Class arrayDequeClass = KotlinToJavaClass.INSTANCE.getArrayDequeClass();
         resolver.register(arrayDequeClass);
         resolver.registerSerializer(arrayDequeClass, new KotlinArrayDequeSerializer(fury, arrayDequeClass));
+
+        // Unsigned array classes: UByteArray, UShortArray, UIntArray, ULongArray.
+        resolver.register(UByteArray.class);
+        resolver.registerSerializer(UByteArray.class, new UByteArraySerializer(fury));
+        resolver.register(UShortArray.class);
+        resolver.registerSerializer(UShortArray.class, new UShortArraySerializer(fury));
+        resolver.register(UIntArray.class);
+        resolver.registerSerializer(UIntArray.class, new UIntArraySerializer(fury));
+        resolver.register(ULongArray.class);
+        resolver.registerSerializer(ULongArray.class, new ULongArraySerializer(fury));
     }
 }

kotlin/src/main/kotlin/org/apache/fury/serializer/kotlin/UnsignedArraySerializers.kt

@@ -0,0 +1,112 @@
+/*
+ * 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.
+ */
+
+@file:OptIn(ExperimentalUnsignedTypes::class)
+
+package org.apache.fury.serializer.kotlin
+
+import org.apache.fury.Fury
+import org.apache.fury.memory.MemoryBuffer
+import org.apache.fury.serializer.Serializer
+import org.apache.fury.type.Type
+
+abstract class AbstractDelegatingArraySerializer<T, T_Delegate>(
+    fury: Fury,
+    cls: Class<T>,
+    private val delegateClass: Class<T_Delegate>
+) : Serializer<T> (fury, cls) {
+
+    // Lazily initialize the delegatingSerializer here to avoid lookup cost.
+    private val delegatingSerializer by lazy {
+        fury.classResolver.getSerializer(delegateClass)
+    }
+
+    abstract fun toDelegateClass(value: T): T_Delegate
+
+    abstract fun fromDelegateClass(value: T_Delegate): T
+
+    override fun getXtypeId(): Short {
+        return (-Type.LIST.id).toShort()
+    }
+
+    override fun xwrite(buffer: MemoryBuffer, value: T) {
+        write(buffer, value)
+    }
+
+    override fun xread(buffer: MemoryBuffer): T {
+        return read(buffer)
+    }
+
+    override fun write(buffer: MemoryBuffer, value: T) {
+        delegatingSerializer.write(buffer, toDelegateClass(value))
+    }
+
+    override fun read(buffer: MemoryBuffer): T {
+        val delegatedValue = delegatingSerializer.read(buffer)
+        return fromDelegateClass(delegatedValue)
+    }
+}
+
+class UByteArraySerializer(
+    fury: Fury,
+)  : AbstractDelegatingArraySerializer<UByteArray, ByteArray>(
+    fury,
+    UByteArray::class.java,
+    ByteArray::class.java
+) {
+    override fun toDelegateClass(value: UByteArray) = value.toByteArray()
+    override fun fromDelegateClass(value: ByteArray) = value.toUByteArray()
+    override fun copy(value: UByteArray): UByteArray = value.copyOf()
+}
+
+class UShortArraySerializer(
+    fury: Fury,
+)  : AbstractDelegatingArraySerializer<UShortArray, ShortArray>(
+    fury,
+    UShortArray::class.java,
+    ShortArray::class.java
+) {
+    override fun toDelegateClass(value: UShortArray) = value.toShortArray()
+    override fun fromDelegateClass(value: ShortArray) = value.toUShortArray()
+    override fun copy(value: UShortArray) = value.copyOf()
+}
+
+class UIntArraySerializer(
+    fury: Fury,
+)  : AbstractDelegatingArraySerializer<UIntArray, IntArray>(
+    fury,
+    UIntArray::class.java,
+    IntArray::class.java
+) {
+    override fun toDelegateClass(value: UIntArray) = value.toIntArray()
+    override fun fromDelegateClass(value: IntArray) = value.toUIntArray()
+    override fun copy(value: UIntArray) = value.copyOf()
+}
+
+class ULongArraySerializer(
+    fury: Fury,
+)  : AbstractDelegatingArraySerializer<ULongArray, LongArray>(
+    fury,
+    ULongArray::class.java,
+    LongArray::class.java
+) {
+    override fun toDelegateClass(value: ULongArray) = value.toLongArray()
+    override fun fromDelegateClass(value: LongArray) = value.toULongArray()
+    override fun copy(value: ULongArray) = value.copyOf()
+}

kotlin/src/test/kotlin/org/apache/fury/serializer/kotlin/ArraySerializerTest.kt

@@ -0,0 +1,220 @@
+/*
+ * 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.kotlin
+
+import org.apache.fury.Fury
+import org.apache.fury.config.Language
+import org.testng.Assert.assertEquals
+import org.testng.annotations.Test
+
+@OptIn(ExperimentalUnsignedTypes::class)
+class ArraySerializerTest {
+    @Test
+    fun testSimpleArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = arrayOf("Apple", "Banana", "Orange", "Pineapple")
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testMultidimensional() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = Array(2) {Array<Int>(2){0} }
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testAnyArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+
+        val array = arrayOf<Any?>("Apple", 1, null, 3.141, 1.2f)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testEmptyArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = emptyArray<Any?>()
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testBooleanArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = booleanArrayOf(true, false)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testByteArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = byteArrayOf(0xFF.toByte(), 0xCA.toByte(), 0xFF.toByte())
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testCharArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = charArrayOf('a', 'b', 'c')
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testDoubleArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = doubleArrayOf(1.0, 2.0, 3.0)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testFloatArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = floatArrayOf(1.0f, 2.0f)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testIntArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = intArrayOf(1, 2, 3)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testLongArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = longArrayOf(1L, 2L, 3L)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testShortArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = shortArrayOf(1, 2, 3)
+        assertEquals(array, fury.deserialize(fury.serialize(array)))
+    }
+
+    @Test
+    fun testUByteArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = ubyteArrayOf(0xFFu, 0xEFu, 0x00u)
+        assert(array.contentEquals(fury.deserialize(fury.serialize(array)) as UByteArray))
+    }
+
+    @Test
+    fun testUShortArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = ushortArrayOf(1u, 2u)
+        assert(array.contentEquals(fury.deserialize(fury.serialize(array)) as UShortArray))
+    }
+
+    @Test
+    fun testUIntArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = uintArrayOf(1u, 2u)
+        assert(array.contentEquals(fury.deserialize(fury.serialize(array)) as UIntArray))
+    }
+    @Test
+    fun testULongArray() {
+        val fury: Fury = Fury.builder()
+            .withLanguage(Language.JAVA)
+            .requireClassRegistration(true)
+            .build()
+        KotlinSerializers.registerSerializers(fury)
+
+        val array = ulongArrayOf(1u, 2u, 3u)
+        assert(array.contentEquals(fury.deserialize(fury.serialize(array)) as ULongArray))
+    }
+}