Fix records with Option fields with a default that isn't None

avro4s-core/src/main/scala/com/sksamuel/avro4s/decoders/options.scala

@@ -1,29 +1,32 @@
 package com.sksamuel.avro4s.decoders
 
-import com.sksamuel.avro4s.{Decoder, Encoder}
+import com.sksamuel.avro4s.Decoder
+import com.sksamuel.avro4s.Encoder
 import org.apache.avro.Schema
 
+import scala.jdk.CollectionConverters.*
+
 class OptionDecoder[T](decoder: Decoder[T]) extends Decoder[Option[T]] {
 
   override def decode(schema: Schema): Any => Option[T] = {
-    // nullables must be encoded with a union of 2 elements, where null is the first type
+    // nullables must be encoded with a union of 2 elements, one of which is null
     require(schema.getType == Schema.Type.UNION, {
       "Options can only be encoded with a UNION schema"
     })
     require(schema.getTypes.size() >= 2, {
       "An option should be encoded with a UNION schema with at least 2 element types"
     })
-    require(schema.getTypes.get(0).getType == Schema.Type.NULL, {
-      "Options can only be encoded with a UNION schema with NULL as the first element type"
-    })
-    val schemaSize = schema.getTypes.size()
-    val elementSchema = schemaSize match
-      case 2 => schema.getTypes.get(1)
-      case _ => Schema.createUnion(schema.getTypes.subList(1, schemaSize))
+    val (isNull, isNotNull) = schema.getTypes().asScala.toList.partition(_.getType() == Schema.Type.NULL)
+    require(isNull.size == 1, s"exactly one of the schemas must be null (found ${isNull.size})")
+    val elementSchema = isNotNull match {
+      case List(single) => single
+      case more => Schema.createUnion(more.asJava)
+    }
+
     val decode = decoder.decode(elementSchema)
     { value => if (value == null) None else Some(decode(value)) }
   }
 }
 
 trait OptionDecoders:
-  given[T](using decoder: Decoder[T]): Decoder[Option[T]] = new OptionDecoder[T](decoder)
+  given[T](using decoder: Decoder[T]): Decoder[Option[T]] = new OptionDecoder[T](decoder)

avro4s-core/src/main/scala/com/sksamuel/avro4s/encoders/options.scala

@@ -1,27 +1,32 @@
 package com.sksamuel.avro4s.encoders
 
-import com.sksamuel.avro4s.{Encoder, FieldMapper}
+import com.sksamuel.avro4s.Encoder
+import com.sksamuel.avro4s.FieldMapper
 import org.apache.avro.Schema
 
+import scala.jdk.CollectionConverters.*
+
 class OptionEncoder[T](encoder: Encoder[T]) extends Encoder[Option[T]] {
 
   override def encode(schema: Schema): Option[T] => Any = {
-    // nullables must be encoded with a union of 2 elements, where null is the first type
+    // nullables must be encoded with a union of 2 elements, one of which is null
     require(schema.getType == Schema.Type.UNION, {
       "Options can only be encoded with a UNION schema"
     })
     require(schema.getTypes.size() >= 2, {
       "Options can only be encoded with a union schema with 2 or more types"
     })
-    require(schema.getTypes.get(0).getType == Schema.Type.NULL)
-    val schemaSize = schema.getTypes.size()
-    val elementSchema = schemaSize match
-      case 2 => schema.getTypes.get(1)
-      case _ => Schema.createUnion(schema.getTypes.subList(1, schemaSize))
+    val (isNull, isNotNull) = schema.getTypes().asScala.toList.partition(_.getType() == Schema.Type.NULL)
+    require(isNull.size == 1, s"exactly one of the schemas must be null (found ${isNull.size})")
+
+    val elementSchema = isNotNull match {
+      case List(single) => single
+      case more => Schema.createUnion(more.asJava)
+    }
     val elementEncoder = encoder.encode(elementSchema)
-    { option => option.fold(null)(value => elementEncoder(value)) }
+    { option => option.fold(null)(elementEncoder) }
   }
 }
 
 trait OptionEncoders:
-  given[T](using encoder: Encoder[T]): Encoder[Option[T]] = OptionEncoder[T](encoder)
+  given[T](using encoder: Encoder[T]): Encoder[Option[T]] = OptionEncoder[T](encoder)

avro4s-core/src/test/scala/com/sksamuel/avro4s/record/decoder/OptionDecoderTest.scala

@@ -1,6 +1,9 @@
 package com.sksamuel.avro4s.record.decoder
 
-import com.sksamuel.avro4s.{AvroSchema, Decoder, ImmutableRecord}
+import com.sksamuel.avro4s.AvroSchema
+import com.sksamuel.avro4s.Decoder
+import com.sksamuel.avro4s.Encoder
+import com.sksamuel.avro4s.ImmutableRecord
 import org.apache.avro.Schema
 import org.apache.avro.generic.GenericData
 import org.apache.avro.util.Utf8
@@ -169,6 +172,17 @@ class OptionDecoderTest extends AnyWordSpec with Matchers {
       Decoder[OptionDibble].decode(schema)(dabbleRecord) shouldBe OptionDibble(Some(Dabble))
       Decoder[OptionDibble].decode(schema)(noneRecord) shouldBe OptionDibble(None)
     }
+
+    "Option with non-None default" in {
+      case class Foo(
+        bar: Option[Int] = Some(42)
+      )
+      val schema = AvroSchema[Foo]
+      for (foo <- Seq(Foo(), Foo(Some(17)), Foo(None))) {
+        val encoded = Encoder[Foo].encode(schema).apply(foo)
+        val decoded = Decoder[Foo].decode(schema).apply(encoded)
+        decoded shouldBe foo
+      }
+    }
   }
 }
-