Ensure Kotlin can encode ByteArrays correctly

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/ArrayCodecProvider.kt

@@ -25,7 +25,9 @@ public class ArrayCodecProvider : CodecProvider {
     override fun <T : Any> get(clazz: Class<T>, registry: CodecRegistry): Codec<T>? = get(clazz, emptyList(), registry)
 
     override fun <T : Any> get(clazz: Class<T>, typeArguments: List<Type>, registry: CodecRegistry): Codec<T>? =
-        if (clazz.isArray) {
+        // ByteArrays must be encoded as compact BSON Binary by ByteArrayCodec.
+        // Returning null lets the registry fall through to ByteArrayCodec.
+        if (clazz.isArray && clazz != ByteArray::class.java) {
             ArrayCodec.create(clazz.kotlin, typeArguments, registry)
         } else null
 }

bson-kotlin/src/main/kotlin/org/bson/codecs/kotlin/DataClassCodec.kt

@@ -251,8 +251,10 @@ internal data class DataClassCodec<T : Any>(
 
         @Suppress("UNCHECKED_CAST")
         private fun CodecRegistry.getCodec(kParameter: KParameter, clazz: Class<Any>, types: List<Type>): Codec<Any> {
+            // ByteArrays use a dedicated ByteArrayCodec in the registry that encodes compact BSON
+            // Binary.
             val codec =
-                if (clazz.isArray) {
+                if (clazz.isArray && clazz != ByteArray::class.java) {
                     ArrayCodec.create(clazz.kotlin, types, this)
                 } else if (types.isEmpty()) {
                     this.get(clazz)

bson-kotlin/src/test/kotlin/org/bson/codecs/kotlin/DataClassCodecTest.kt

@@ -45,6 +45,7 @@ import org.bson.codecs.kotlin.samples.DataClassWithBsonExtraElements
 import org.bson.codecs.kotlin.samples.DataClassWithBsonId
 import org.bson.codecs.kotlin.samples.DataClassWithBsonIgnore
 import org.bson.codecs.kotlin.samples.DataClassWithBsonProperty
+import org.bson.codecs.kotlin.samples.DataClassWithByteArray
 import org.bson.codecs.kotlin.samples.DataClassWithCollections
 import org.bson.codecs.kotlin.samples.DataClassWithDataClassMapKey
 import org.bson.codecs.kotlin.samples.DataClassWithDefaults
@@ -122,6 +123,11 @@ class DataClassCodecTest {
             | "arraySimple": ["a", "b", "c", "d"],
             | "nestedArrays":  [["e", "f"], [], ["g", "h"]],
             | "arrayOfMaps":  [{"A": ["aa"], "B": ["bb"]}, {}, {"C": ["cc", "ccc"]}],
+            | "byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}},
+            | "nestedByteArrays": [
+            |     {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
+            |     {"${'$'}binary": {"base64": "AwQF", "subType": "00"}}
+            | ],
             |}"""
                 .trimMargin()
 
@@ -130,7 +136,9 @@ class DataClassCodecTest {
                 arrayOf("a", "b", "c", "d"),
                 arrayOf(arrayOf("e", "f"), emptyArray(), arrayOf("g", "h")),
                 arrayOf(
-                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))))
+                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))),
+                byteArrayOf(1, 2, 3, 4),
+                arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4, 5)))
 
         assertRoundTrips(expected, dataClass)
     }
@@ -140,7 +148,7 @@ class DataClassCodecTest {
         val expected =
             """{
             |    "booleanArray": [true, false],
-            |    "byteArray": [1, 2],
+            |    "byteArray": {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
             |    "charArray": ["a", "b"],
             |    "doubleArray": [ 1.1, 2.2, 3.3],
             |    "floatArray": [1.0, 2.0, 3.0],
@@ -168,6 +176,55 @@ class DataClassCodecTest {
         assertRoundTrips(expected, dataClass)
     }
 
+    @Test
+    fun testDataClassWithByteArrayEncodesAsBinary() {
+        // A ByteArray field must encode as compact BSON Binary (subType 00),
+        // via ByteArrayCodec, not as a BSON Array of Int32 (one element per byte).
+        val expected = """{"byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}}}"""
+        assertRoundTrips(expected, DataClassWithByteArray(byteArrayOf(1, 2, 3, 4)))
+    }
+
+    @Test
+    fun testDataClassWithByteArrayDoesNotExpandDocumentSize() {
+        // BSON Array of ints encoding expands size ~8-10x, breaking the 16MB limit.
+        // Binary encoding keeps a 1MB payload close to 1MB.
+        val oneMegabyte = ByteArray(1_000_000)
+        val codec = DataClassCodec.create(DataClassWithByteArray::class, registry())!!
+        val document = BsonDocument()
+        codec.encode(
+            BsonDocumentWriter(document), DataClassWithByteArray(oneMegabyte), EncoderContext.builder().build())
+        val encodedSize = document.toBsonDocument().getBinary("byteArray").data.size
+        assertEquals(1_000_000, encodedSize)
+    }
+
+    @Test
+    fun testDataClassWithObjectArrayEncodesAsBsonArray() {
+        // Ensure normal arrays and ByteArrays are handled correctly.
+        val expected =
+            """{
+            | "arraySimple": ["a", "b", "c", "d"],
+            | "nestedArrays": [["e", "f"], [], ["g", "h"]],
+            | "arrayOfMaps": [{"A": ["aa"], "B": ["bb"]}, {}, {"C": ["cc", "ccc"]}],
+            | "byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}},
+            | "nestedByteArrays": [
+            |     {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
+            |     {"${'$'}binary": {"base64": "AwQF", "subType": "00"}}
+            | ]
+            |}"""
+                .trimMargin()
+
+        val dataClass =
+            DataClassWithArrays(
+                arrayOf("a", "b", "c", "d"),
+                arrayOf(arrayOf("e", "f"), emptyArray(), arrayOf("g", "h")),
+                arrayOf(
+                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))),
+                byteArrayOf(1, 2, 3, 4),
+                arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4, 5)))
+
+        assertRoundTrips(expected, dataClass)
+    }
+
     @Test
     fun testDataClassWithDefaults() {
         val expectedDefault =

bson-kotlin/src/test/kotlin/org/bson/codecs/kotlin/samples/DataClasses.kt

@@ -52,7 +52,9 @@ data class DataClassWithCollections(
 data class DataClassWithArrays(
     val arraySimple: Array<String>,
     val nestedArrays: Array<Array<String>>,
-    val arrayOfMaps: Array<Map<String, Array<String>>>
+    val arrayOfMaps: Array<Map<String, Array<String>>>,
+    val byteArray: ByteArray,
+    val nestedByteArrays: Array<ByteArray>
 ) {
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
@@ -70,13 +72,18 @@ data class DataClassWithArrays(
             map.keys.forEach { key -> if (!map[key].contentEquals(otherMap[key])) return false }
         }
 
+        if (!byteArray.contentEquals(other.byteArray)) return false
+        if (!nestedByteArrays.contentDeepEquals(other.nestedByteArrays)) return false
+
         return true
     }
 
     override fun hashCode(): Int {
         var result = arraySimple.contentHashCode()
         result = 31 * result + nestedArrays.contentDeepHashCode()
         result = 31 * result + arrayOfMaps.contentHashCode()
+        result = 31 * result + byteArray.contentHashCode()
+        result = 31 * result + nestedByteArrays.contentDeepHashCode()
         return result
     }
 }
@@ -134,6 +141,17 @@ data class DataClassWithNativeArrays(
     }
 }
 
+data class DataClassWithByteArray(val byteArray: ByteArray) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+        other as DataClassWithByteArray
+        return byteArray.contentEquals(other.byteArray)
+    }
+
+    override fun hashCode(): Int = byteArray.contentHashCode()
+}
+
 data class DataClassWithDefaults(
     val boolean: Boolean = false,
     val string: String = "String",

bson-kotlinx/src/main/kotlin/org/bson/codecs/kotlinx/ByteArrayAsBsonBinary.kt

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2008-present MongoDB, Inc.
+ *
+ * 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 org.bson.codecs.kotlinx
+
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.SerializationException
+import kotlinx.serialization.descriptors.PrimitiveKind
+import kotlinx.serialization.descriptors.PrimitiveSerialDescriptor
+import kotlinx.serialization.descriptors.SerialDescriptor
+import kotlinx.serialization.encoding.Decoder
+import kotlinx.serialization.encoding.Encoder
+import kotlinx.serialization.modules.SerializersModule
+import org.bson.BsonBinary
+
+/**
+ * ByteArray KSerializer.
+ *
+ * Encodes and decodes `ByteArray` values to and from a compact `BsonBinary` (subtype
+ * [org.bson.BsonBinarySubType.BINARY]), matching the behavior of the standard `ByteArrayCodec` and the `bson-kotlin`
+ * `DataClassCodec`.
+ *
+ * This is an opt-in serializer: kotlinx.serialization's built-in `ByteArray` serializer encodes a `ByteArray` as a BSON
+ * array of int32 elements (one element per byte). To store a `ByteArray` field as `BsonBinary` instead, either annotate
+ * the field with `@Serializable(with = ByteArrayAsBsonBinary::class)`, or annotate it with `@Contextual` and register
+ * [ByteArrayAsBsonBinary.serializersModule] on the codec.
+ *
+ * @since 5.10
+ */
+@ExperimentalSerializationApi
+public object ByteArrayAsBsonBinary : KSerializer<ByteArray> {
+    override val descriptor: SerialDescriptor = PrimitiveSerialDescriptor("ByteArrayAsBsonBinary", PrimitiveKind.STRING)
+
+    override fun serialize(encoder: Encoder, value: ByteArray) {
+        when (encoder) {
+            is BsonEncoder -> encoder.encodeBsonValue(BsonBinary(value))
+            else -> throw SerializationException("ByteArray is not supported by ${encoder::class}")
+        }
+    }
+
+    override fun deserialize(decoder: Decoder): ByteArray {
+        return when (decoder) {
+            is BsonDecoder -> decoder.decodeBsonValue().asBinary().data
+            else -> throw SerializationException("ByteArray is not supported by ${decoder::class}")
+        }
+    }
+
+    public val serializersModule: SerializersModule = SerializersModule {
+        contextual(ByteArray::class, ByteArrayAsBsonBinary)
+    }
+}

bson-kotlinx/src/test/kotlin/org/bson/codecs/kotlinx/KotlinSerializerCodecTest.kt

@@ -74,6 +74,7 @@ import org.bson.codecs.kotlinx.samples.DataClassSealedB
 import org.bson.codecs.kotlinx.samples.DataClassSealedC
 import org.bson.codecs.kotlinx.samples.DataClassSelfReferential
 import org.bson.codecs.kotlinx.samples.DataClassWithAnnotations
+import org.bson.codecs.kotlinx.samples.DataClassWithArrays
 import org.bson.codecs.kotlinx.samples.DataClassWithBooleanMapKey
 import org.bson.codecs.kotlinx.samples.DataClassWithBsonConstructor
 import org.bson.codecs.kotlinx.samples.DataClassWithBsonDiscriminator
@@ -82,6 +83,7 @@ import org.bson.codecs.kotlinx.samples.DataClassWithBsonId
 import org.bson.codecs.kotlinx.samples.DataClassWithBsonIgnore
 import org.bson.codecs.kotlinx.samples.DataClassWithBsonProperty
 import org.bson.codecs.kotlinx.samples.DataClassWithBsonRepresentation
+import org.bson.codecs.kotlinx.samples.DataClassWithByteArray
 import org.bson.codecs.kotlinx.samples.DataClassWithCamelCase
 import org.bson.codecs.kotlinx.samples.DataClassWithCollections
 import org.bson.codecs.kotlinx.samples.DataClassWithContextualDateValues
@@ -299,6 +301,50 @@ class KotlinSerializerCodecTest {
         assertRoundTrips(expected, expectedDataClass)
     }
 
+    @Test
+    fun testDataClassWithByteArrayEncodesAsBsonArrayByDefault() {
+        // By default kotlinx.serialization encodes a ByteArray as a BSON array of int32 elements
+        // (one per byte).
+        // This differs from bson-kotlin's DataClassCodec, which encodes ByteArray as compact BSON
+        // Binary.
+        val expected = """{"byteArray": [1, 2, 3, 4]}"""
+
+        assertRoundTrips(expected, DataClassWithByteArray(byteArrayOf(1, 2, 3, 4)))
+    }
+
+    @Test
+    fun testDataClassWithObjectArrayEncodesAsBsonArray() {
+        // Object arrays encode as BSON arrays, while ByteArray fields opted in via
+        // @Serializable(with = ByteArrayAsBsonBinary::class) encode as compact BSON Binary (subType
+        // 00),
+        // consistent with the standard ByteArrayCodec and bson-kotlin's DataClassCodec. The
+        // per-type-argument
+        // annotation on nestedByteArrays applies the serializer to each inner ByteArray element.
+        val expected =
+            """{
+            | "arraySimple": ["a", "b", "c", "d"],
+            | "nestedArrays": [["e", "f"], [], ["g", "h"]],
+            | "arrayOfMaps": [{"A": ["aa"], "B": ["bb"]}, {}, {"C": ["cc", "ccc"]}],
+            | "byteArray": {"${'$'}binary": {"base64": "AQIDBA==", "subType": "00"}},
+            | "nestedByteArrays": [
+            |     {"${'$'}binary": {"base64": "AQI=", "subType": "00"}},
+            |     {"${'$'}binary": {"base64": "AwQF", "subType": "00"}}
+            | ]
+            |}"""
+                .trimMargin()
+
+        val dataClass =
+            DataClassWithArrays(
+                arrayOf("a", "b", "c", "d"),
+                arrayOf(arrayOf("e", "f"), emptyArray(), arrayOf("g", "h")),
+                arrayOf(
+                    mapOf("A" to arrayOf("aa"), "B" to arrayOf("bb")), emptyMap(), mapOf("C" to arrayOf("cc", "ccc"))),
+                byteArrayOf(1, 2, 3, 4),
+                arrayOf(byteArrayOf(1, 2), byteArrayOf(3, 4, 5)))
+
+        assertRoundTrips(expected, dataClass)
+    }
+
     @Test
     fun testDataClassWithComplexTypes() {
         val expected =

bson-kotlinx/src/test/kotlin/org/bson/codecs/kotlinx/samples/DataClasses.kt

@@ -47,6 +47,7 @@ import org.bson.BsonSymbol
 import org.bson.BsonTimestamp
 import org.bson.BsonType
 import org.bson.BsonUndefined
+import org.bson.codecs.kotlinx.ByteArrayAsBsonBinary
 import org.bson.codecs.pojo.annotations.BsonCreator
 import org.bson.codecs.pojo.annotations.BsonDiscriminator
 import org.bson.codecs.pojo.annotations.BsonExtraElements
@@ -376,3 +377,56 @@ data class DataClassWithJsonElementsNullable(
     val jsonElements: List<JsonElement?>?,
     val jsonNestedMap: Map<String, JsonElement?>?
 )
+
+@Serializable
+data class DataClassWithByteArray(val byteArray: ByteArray) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+        other as DataClassWithByteArray
+        return byteArray.contentEquals(other.byteArray)
+    }
+
+    override fun hashCode(): Int = byteArray.contentHashCode()
+}
+
+@OptIn(ExperimentalSerializationApi::class)
+@Serializable
+data class DataClassWithArrays(
+    val arraySimple: Array<String>,
+    val nestedArrays: Array<Array<String>>,
+    val arrayOfMaps: Array<Map<String, Array<String>>>,
+    @Serializable(with = ByteArrayAsBsonBinary::class) val byteArray: ByteArray,
+    val nestedByteArrays: Array<@Serializable(with = ByteArrayAsBsonBinary::class) ByteArray>
+) {
+    override fun equals(other: Any?): Boolean {
+        if (this === other) return true
+        if (javaClass != other?.javaClass) return false
+
+        other as DataClassWithArrays
+
+        if (!arraySimple.contentEquals(other.arraySimple)) return false
+        if (!nestedArrays.contentDeepEquals(other.nestedArrays)) return false
+
+        if (arrayOfMaps.size != other.arrayOfMaps.size) return false
+        arrayOfMaps.forEachIndexed { i, map ->
+            val otherMap = other.arrayOfMaps[i]
+            if (map.keys != otherMap.keys) return false
+            map.keys.forEach { key -> if (!map[key].contentEquals(otherMap[key])) return false }
+        }
+
+        if (!byteArray.contentEquals(other.byteArray)) return false
+        if (!nestedByteArrays.contentDeepEquals(other.nestedByteArrays)) return false
+
+        return true
+    }
+
+    override fun hashCode(): Int {
+        var result = arraySimple.contentHashCode()
+        result = 31 * result + nestedArrays.contentDeepHashCode()
+        result = 31 * result + arrayOfMaps.contentHashCode()
+        result = 31 * result + byteArray.contentHashCode()
+        result = 31 * result + nestedByteArrays.contentDeepHashCode()
+        return result
+    }
+}