Repository: avro-kotlin/avro4k
Pull request #110 (2021-09-06 23:47:27)

Polymorphic open support (Take 2)
This is a replacement for https://github.com/avro-kotlin/avro4k/pull/109 - Forking this on my personal account to make it easier for @thake to make changes to it.


This PR adds support for open / unsealed / abstract classes to generate avro schemas and serialize/deserialize avro binary, configured like so:
```kotlin
val module = SerializersModule {
   polymorphic(UnsealedPolymorphicRoot::class) {
      subclass(UnsealedChildOne::class)
      subclass(UnsealedChildTwo::class)
   }
}
val schema = Avro(serializersModule = module).schema(UnsealedPolymorphicRoot.serializer())
```

For now it explicitly doesn't support polymorphic defaults - this is a feature someone could build on top if they want to.


Patch:
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/SerialDescriptor.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/SerialDescriptor.kt
index 36f41ce..50eedb1 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/SerialDescriptor.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/SerialDescriptor.kt
@@ -1,16 +1,25 @@
 package com.github.avrokotlin.avro4k
 
 import kotlinx.serialization.ExperimentalSerializationApi
-import kotlinx.serialization.descriptors.PolymorphicKind
-import kotlinx.serialization.descriptors.SerialDescriptor
-import kotlinx.serialization.descriptors.SerialKind
-import kotlinx.serialization.descriptors.elementDescriptors
+import kotlinx.serialization.descriptors.*
+import kotlinx.serialization.modules.SerializersModule
 
 @ExperimentalSerializationApi
-fun SerialDescriptor.leavesOfSealedClasses() : List<SerialDescriptor> {
-   return if (this.kind == PolymorphicKind.SEALED) {
-      elementDescriptors.filter {it.kind == SerialKind.CONTEXTUAL }.flatMap { it.elementDescriptors }.flatMap { it.leavesOfSealedClasses() }
-   } else {
-      listOf(this)
-   }
-}
\ No newline at end of file
+fun SerialDescriptor.leavesOfSealedClasses(): List<SerialDescriptor> {
+    return if (this.kind == PolymorphicKind.SEALED) {
+        elementDescriptors.filter { it.kind == SerialKind.CONTEXTUAL }.flatMap { it.elementDescriptors }
+            .flatMap { it.leavesOfSealedClasses() }
+    } else {
+        listOf(this)
+    }
+}
+
+@ExperimentalSerializationApi
+fun SerialDescriptor.possibleSerializationSubclasses(serializersModule: SerializersModule): List<SerialDescriptor> {
+    return when (this.kind) {
+        PolymorphicKind.SEALED -> this.leavesOfSealedClasses()
+        PolymorphicKind.OPEN -> serializersModule.getPolymorphicDescriptors(this).sortedBy { it.serialName }
+        else -> throw UnsupportedOperationException("Can't get possible serialization subclasses for the SerialDescriptor of kind ${this.kind}.")
+    }
+}
+
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/ListDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/ListDecoder.kt
index ab1646c..e09c407 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/ListDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/ListDecoder.kt
@@ -88,7 +88,7 @@ class ListDecoder(
          StructureKind.CLASS -> RecordDecoder(descriptor, array[index] as GenericRecord, serializersModule, configuration)
          StructureKind.LIST -> ListDecoder(schema.elementType, array[index] as GenericArray<*>, serializersModule, configuration)
          StructureKind.MAP -> MapDecoder(descriptor, schema.elementType, array[index] as Map<String, *>, serializersModule, configuration)
-         PolymorphicKind.SEALED -> SealedClassDecoder(descriptor,array[index] as GenericRecord, serializersModule, configuration)
+         PolymorphicKind.SEALED -> UnionDecoder(descriptor,array[index] as GenericRecord, serializersModule, configuration)
          else -> throw UnsupportedOperationException("Kind ${descriptor.kind} is currently not supported.")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RecordDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RecordDecoder.kt
index e9ab4e1..3fbe1d8 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RecordDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RecordDecoder.kt
@@ -65,7 +65,7 @@ class RecordDecoder(
             }
             decoder
          }
-         PolymorphicKind.SEALED -> SealedClassDecoder(descriptor,value as GenericRecord, serializersModule, configuration)
+         PolymorphicKind.SEALED, PolymorphicKind.OPEN -> UnionDecoder(descriptor,value as GenericRecord, serializersModule, configuration)
          else -> throw UnsupportedOperationException("Decoding descriptor of kind ${descriptor.kind} is currently not supported")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RootRecordDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RootRecordDecoder.kt
index 01a60f3..7bcf4d6 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RootRecordDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/RootRecordDecoder.kt
@@ -27,7 +27,7 @@ class RootRecordDecoder(
             serializersModule,
             configuration
          )
-         PolymorphicKind.SEALED -> SealedClassDecoder(descriptor, record, serializersModule, configuration)
+         PolymorphicKind.SEALED -> UnionDecoder(descriptor, record, serializersModule, configuration)
          else -> throw SerializationException("Non-class structure passed to root record decoder")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/SealedClassDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/UnionDecoder.kt
similarity index 79%
rename from src/main/kotlin/com/github/avrokotlin/avro4k/decoder/SealedClassDecoder.kt
rename to src/main/kotlin/com/github/avrokotlin/avro4k/decoder/UnionDecoder.kt
index 5a93dac..e5c3d89 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/SealedClassDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/decoder/UnionDecoder.kt
@@ -2,7 +2,7 @@ package com.github.avrokotlin.avro4k.decoder
 
 import com.github.avrokotlin.avro4k.AvroConfiguration
 import com.github.avrokotlin.avro4k.RecordNaming
-import com.github.avrokotlin.avro4k.leavesOfSealedClasses
+import com.github.avrokotlin.avro4k.possibleSerializationSubclasses
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.SerializationException
@@ -14,10 +14,10 @@ import org.apache.avro.Schema
 import org.apache.avro.generic.GenericRecord
 
 @ExperimentalSerializationApi
-class SealedClassDecoder (descriptor: SerialDescriptor,
-                          private val value: GenericRecord,
-                          override val serializersModule: SerializersModule,
-                          private  val configuration: AvroConfiguration
+class UnionDecoder (descriptor: SerialDescriptor,
+                    private val value: GenericRecord,
+                    override val serializersModule: SerializersModule,
+                    private  val configuration: AvroConfiguration
 ) : AbstractDecoder(), FieldDecoder
 {
    private enum class DecoderState(val index : Int){
@@ -28,7 +28,7 @@ class SealedClassDecoder (descriptor: SerialDescriptor,
    }
    private var currentState = DecoderState.BEFORE
 
-   var leafDescriptor : SerialDescriptor = descriptor.leavesOfSealedClasses().firstOrNull {
+   private var leafDescriptor : SerialDescriptor = descriptor.possibleSerializationSubclasses(serializersModule).firstOrNull {
       val schemaName = RecordNaming(value.schema.fullName, emptyList())
       val serialName = RecordNaming(it)
       serialName == schemaName
@@ -54,5 +54,5 @@ class SealedClassDecoder (descriptor: SerialDescriptor,
       return recordDecoder.decodeSerializableValue(deserializer)
    }
 
-   override fun decodeAny(): Any? = UnsupportedOperationException()
+   override fun decodeAny(): Any = UnsupportedOperationException()
 }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RecordEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RecordEncoder.kt
index c5f7079..3e0e3c4 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RecordEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RecordEncoder.kt
@@ -30,7 +30,7 @@ interface StructureEncoder : FieldEncoder {
          }
          StructureKind.CLASS -> RecordEncoder(fieldSchema(), serializersModule) { addValue(it) }
          StructureKind.MAP -> MapEncoder(fieldSchema(), serializersModule) { addValue(it) }
-         PolymorphicKind.SEALED -> SealedClassEncoder(fieldSchema(), serializersModule) { addValue(it) }
+         is PolymorphicKind -> UnionEncoder(fieldSchema(), serializersModule) { addValue(it) }
          else -> throw SerializationException(".beginStructure was called on a non-structure type [$descriptor]")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RootRecordEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RootRecordEncoder.kt
index f698aeb..41b7aee 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RootRecordEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/RootRecordEncoder.kt
@@ -19,7 +19,7 @@ class RootRecordEncoder(private val schema: Schema,
    override fun beginStructure(descriptor: SerialDescriptor): CompositeEncoder {
       return when (descriptor.kind) {
          is StructureKind.CLASS -> RecordEncoder(schema, serializersModule, callback)
-         is PolymorphicKind.SEALED -> SealedClassEncoder(schema,serializersModule,callback)
+         is PolymorphicKind -> UnionEncoder(schema,serializersModule,callback)
          else -> throw SerializationException("Unsupported root element passed to root record encoder")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/SealedClassEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/SealedClassEncoder.kt
deleted file mode 100644
index 4f00b0b..0000000
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/SealedClassEncoder.kt
+++ /dev/null
@@ -1,23 +0,0 @@
-package com.github.avrokotlin.avro4k.encoder
-
-import com.github.avrokotlin.avro4k.Record
-import kotlinx.serialization.ExperimentalSerializationApi
-import kotlinx.serialization.SerializationStrategy
-import kotlinx.serialization.descriptors.SerialDescriptor
-import kotlinx.serialization.encoding.AbstractEncoder
-import kotlinx.serialization.modules.SerializersModule
-import org.apache.avro.Schema
-
-@ExperimentalSerializationApi
-class SealedClassEncoder(private val schema: Schema,
-                         override val serializersModule: SerializersModule,
-                         private val callback: (Record) -> Unit): AbstractEncoder() {
-   override fun encodeString(value: String){
-      //No need to encode the string of the concrete type. This will be handled by the UnionEncoder
-   }
-
-   override fun <T : Any?> encodeSerializableValue(serializer: SerializationStrategy<T>, value: T){
-      return UnionEncoder(schema,serializersModule,callback).encodeSerializableValue(serializer,value)
-   }
-   override fun endStructure(descriptor: SerialDescriptor) {}
-}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/UnionEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/UnionEncoder.kt
index b6243e7..6c955be 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/UnionEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/encoder/UnionEncoder.kt
@@ -16,7 +16,9 @@ import org.apache.avro.Schema
 class UnionEncoder(private val unionSchema : Schema,
                    override val serializersModule: SerializersModule,
                    private val callback: (Record) -> Unit) : AbstractEncoder() {
-
+   override fun encodeString(value: String){
+      //No need to encode the name of the concrete type. The name will never be encoded in the avro schema.
+   }
    override fun beginStructure(descriptor: SerialDescriptor): CompositeEncoder {
       return when (descriptor.kind) {
          is StructureKind.CLASS, is StructureKind.OBJECT -> {
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SchemaFor.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SchemaFor.kt
index bfe3fb3..0b98451 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SchemaFor.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SchemaFor.kt
@@ -196,13 +196,13 @@ fun schemaFor(serializersModule: SerializersModule,
             namingStrategy,
             resolvedSchemas
          )
-         PolymorphicKind.SEALED -> SealedClassSchemaFor(descriptor, namingStrategy, serializersModule, resolvedSchemas)
          StructureKind.CLASS, StructureKind.OBJECT -> when (descriptor.serialName) {
             "kotlin.Pair" -> PairSchemaFor(descriptor, namingStrategy, serializersModule, resolvedSchemas)
             else -> ClassSchemaFor(descriptor, namingStrategy, serializersModule, resolvedSchemas)
          }
          StructureKind.LIST -> ListSchemaFor(descriptor, serializersModule, namingStrategy, resolvedSchemas)
          StructureKind.MAP -> MapSchemaFor(descriptor, serializersModule, namingStrategy, resolvedSchemas)
+         is PolymorphicKind -> UnionSchemaFor(descriptor, namingStrategy, serializersModule, resolvedSchemas)
          else -> throw SerializationException("Unsupported type ${descriptor.serialName} of ${descriptor.kind}")
       }
    }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SealedClassSchemaFor.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/schema/UnionSchemaFor.kt
similarity index 54%
rename from src/main/kotlin/com/github/avrokotlin/avro4k/schema/SealedClassSchemaFor.kt
rename to src/main/kotlin/com/github/avrokotlin/avro4k/schema/UnionSchemaFor.kt
index 3ac211d..0388686 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/schema/SealedClassSchemaFor.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/schema/UnionSchemaFor.kt
@@ -1,20 +1,20 @@
 package com.github.avrokotlin.avro4k.schema
 
 import com.github.avrokotlin.avro4k.RecordNaming
-import com.github.avrokotlin.avro4k.leavesOfSealedClasses
+import com.github.avrokotlin.avro4k.possibleSerializationSubclasses
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.descriptors.SerialDescriptor
 import kotlinx.serialization.modules.SerializersModule
 import org.apache.avro.Schema
 
 @ExperimentalSerializationApi
-class SealedClassSchemaFor(private val descriptor: SerialDescriptor,
-                           private val namingStrategy: NamingStrategy,
-                           private val serializersModule: SerializersModule,
-                           private val resolvedSchemas: MutableMap<RecordNaming, Schema>
+class UnionSchemaFor(private val descriptor: SerialDescriptor,
+                     private val namingStrategy: NamingStrategy,
+                     private val serializersModule: SerializersModule,
+                     private val resolvedSchemas: MutableMap<RecordNaming, Schema>
 ) : SchemaFor {
    override fun schema(): Schema {
-      val leafSerialDescriptors = descriptor.leavesOfSealedClasses()
+      val leafSerialDescriptors = descriptor.possibleSerializationSubclasses(serializersModule)
       return Schema.createUnion(
          leafSerialDescriptors.map { ClassSchemaFor(it,namingStrategy,serializersModule, resolvedSchemas).schema() }
       )


Test patch:
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/io/PolymorphicClassIoTest.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/io/PolymorphicClassIoTest.kt
new file mode 100644
index 0000000..4644842
--- /dev/null
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/io/PolymorphicClassIoTest.kt
@@ -0,0 +1,30 @@
+package com.github.avrokotlin.avro4k.io
+
+import com.github.avrokotlin.avro4k.Avro
+import com.github.avrokotlin.avro4k.schema.ReferencingPolymorphicRoot
+import com.github.avrokotlin.avro4k.schema.UnsealedChildOne
+import com.github.avrokotlin.avro4k.schema.UnsealedChildTwo
+import com.github.avrokotlin.avro4k.schema.UnsealedPolymorphicRoot
+import io.kotest.core.spec.style.StringSpec
+import io.kotest.matchers.shouldBe
+import kotlinx.serialization.modules.SerializersModule
+import kotlinx.serialization.modules.polymorphic
+import kotlinx.serialization.modules.subclass
+import org.apache.avro.generic.GenericRecord
+
+class PolymorphicClassIoTest : StringSpec({
+   "read / write nested polymorphic class" {
+      val module = SerializersModule {
+         polymorphic(UnsealedPolymorphicRoot::class) {
+            subclass(UnsealedChildOne::class)
+            subclass(UnsealedChildTwo::class)
+         }
+      }
+      val avro = Avro(serializersModule = module)
+      writeRead(ReferencingPolymorphicRoot(UnsealedChildOne("one")), ReferencingPolymorphicRoot.serializer(), avro)
+      writeRead(ReferencingPolymorphicRoot(UnsealedChildOne("one")), ReferencingPolymorphicRoot.serializer(), avro) {
+         val root = it["root"] as GenericRecord
+         root.schema shouldBe avro.schema(UnsealedChildOne.serializer())
+      }
+   }
+})
\ No newline at end of file
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/io/StreamTests.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/io/StreamTests.kt
index c59fa78..28a96d0 100644
--- a/src/test/kotlin/com/github/avrokotlin/avro4k/io/StreamTests.kt
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/io/StreamTests.kt
@@ -11,75 +11,75 @@ import org.apache.avro.generic.GenericRecord
 import org.apache.avro.io.DecoderFactory
 import java.io.ByteArrayOutputStream
 
-fun <T> writeRead(t: T, serializer: KSerializer<T>) {
-   writeData(t, serializer).apply {
-      val record = readData(this, serializer)
-      val tt = Avro.default.fromRecord(serializer, record)
+fun <T> writeRead(t: T, serializer: KSerializer<T>, avro: Avro = Avro.default) {
+   writeData(t, serializer, avro).apply {
+      val record = readData(this, serializer, avro)
+      val tt = avro.fromRecord(serializer, record)
       t shouldBe tt
    }
-   writeBinary(t, serializer).apply {
-      val record = readBinary(this, serializer)
-      val tt = Avro.default.fromRecord(serializer, record)
+   writeBinary(t, serializer, avro).apply {
+      val record = readBinary(this, serializer, avro)
+      val tt = avro.fromRecord(serializer, record)
       t shouldBe tt
    }
-   writeJson(t, serializer).apply {
-      val record = readJson(this, serializer)
-      val tt = Avro.default.fromRecord(serializer, record)
+   writeJson(t, serializer, avro).apply {
+      val record = readJson(this, serializer, avro)
+      val tt = avro.fromRecord(serializer, record)
       t shouldBe tt
    }
 }
 
-fun <T> writeRead(t: T, expected: T, serializer: KSerializer<T>) {
-   writeData(t, serializer).apply {
-      val record = readData(this, serializer)
-      val tt = Avro.default.fromRecord(serializer, record)
+fun <T> writeRead(t: T, expected: T, serializer: KSerializer<T>, avro: Avro = Avro.default) {
+   writeData(t, serializer, avro).apply {
+      val record = readData(this, serializer, avro)
+      val tt = avro.fromRecord(serializer, record)
       tt shouldBe expected
    }
-   writeBinary(t, serializer).apply {
-      val record = readBinary(this, serializer)
-      val tt = Avro.default.fromRecord(serializer, record)
+   writeBinary(t, serializer, avro).apply {
+      val record = readBinary(this, serializer, avro)
+      val tt = avro.fromRecord(serializer, record)
       tt shouldBe expected
    }
 }
 
-fun <T> writeRead(t: T, serializer: KSerializer<T>, test: (GenericRecord) -> Any) {
-   writeData(t, serializer).apply {
-      val record = readData(this, serializer)
+fun <T> writeRead(t: T, serializer: KSerializer<T>, avro: Avro = Avro.default, test: (GenericRecord) -> Any) {
+   writeData(t, serializer, avro).apply {
+      val record = readData(this, serializer, avro)
       test(record)
    }
-   writeBinary(t, serializer).apply {
-      val record = readBinary(this, serializer)
+   writeBinary(t, serializer, avro).apply {
+      val record = readBinary(this, serializer, avro)
       test(record)
    }
-   writeJson(t, serializer).apply {
-      val record = readJson(this, serializer)
+   writeJson(t, serializer, avro).apply {
+      val record = readJson(this, serializer, avro)
       test(record)
    }
 }
 
-fun <T> writeData(t: T, serializer: SerializationStrategy<T>): ByteArray {
-   val schema = Avro.default.schema(serializer)
+fun <T> writeData(t: T, serializer: SerializationStrategy<T>, avro: Avro = Avro.default): ByteArray {
+   val schema = avro.schema(serializer)
    val out = ByteArrayOutputStream()
-   val avro = Avro.default.openOutputStream(serializer) {
+   val output = avro.openOutputStream(serializer) {
       encodeFormat = AvroEncodeFormat.Data()
       this.schema = schema
    }.to(out)
-   avro.write(t)
-   avro.close()
+   output.write(t)
+   output.close()
    return out.toByteArray()
 }
 
-fun <T> readJson(bytes: ByteArray, serializer: KSerializer<T>): GenericRecord {
-   val schema = Avro.default.schema(serializer)
+fun <T> readJson(bytes: ByteArray, serializer: KSerializer<T>, avro: Avro = Avro.default): GenericRecord {
+   val schema = avro.schema(serializer)
    val datumReader = GenericDatumReader<GenericRecord>(schema)
    val decoder = DecoderFactory.get().jsonDecoder(schema, SeekableByteArrayInput(bytes))
    return datumReader.read(null, decoder)
 }
 
-fun <T> writeJson(t: T, serializer: KSerializer<T>): ByteArray {
-   val schema = Avro.default.schema(serializer)
+fun <T> writeJson(t: T, serializer: KSerializer<T>, avro: Avro = Avro.default): ByteArray {
+   val schema = avro.schema(serializer)
    val baos = ByteArrayOutputStream()
-   val output = Avro.default.openOutputStream(serializer) {
+   val output = avro.openOutputStream(serializer) {
       encodeFormat = AvroEncodeFormat.Json
       this.schema = schema
    }.to(baos)
@@ -88,28 +88,28 @@ fun <T> writeJson(t: T, serializer: KSerializer<T>): ByteArray {
    return baos.toByteArray()
 }
 
-fun <T> readData(bytes: ByteArray, serializer: KSerializer<T>): GenericRecord {
-   val schema = Avro.default.schema(serializer)
-   val avro = Avro.default.openInputStream {
+fun <T> readData(bytes: ByteArray, serializer: KSerializer<T>, avro: Avro = Avro.default): GenericRecord {
+   val schema = avro.schema(serializer)
+   val input = avro.openInputStream {
       decodeFormat = AvroDecodeFormat.Data(schema)
    }.from(bytes)
-   return avro.next() as GenericRecord
+   return input.next() as GenericRecord
 }
 
-fun <T> writeBinary(t: T, serializer: SerializationStrategy<T>): ByteArray {
-   val schema = Avro.default.schema(serializer)
+fun <T> writeBinary(t: T, serializer: SerializationStrategy<T>, avro: Avro = Avro.default): ByteArray {
+   val schema = avro.schema(serializer)
    val out = ByteArrayOutputStream()
-   val avro = Avro.default.openOutputStream(serializer) {
+   val output = avro.openOutputStream(serializer) {
       encodeFormat = AvroEncodeFormat.Binary
       this.schema = schema
    }.to(out)
-   avro.write(t)
-   avro.close()
+   output.write(t)
+   output.close()
    return out.toByteArray()
 }
 
-fun <T> readBinary(bytes: ByteArray, serializer: KSerializer<T>): GenericRecord {
-   val schema = Avro.default.schema(serializer)
+fun <T> readBinary(bytes: ByteArray, serializer: KSerializer<T>, avro: Avro = Avro.default): GenericRecord {
+   val schema = avro.schema(serializer)
    val datumReader = GenericDatumReader<GenericRecord>(schema)
    val decoder = DecoderFactory.get().binaryDecoder(SeekableByteArrayInput(bytes), null)
    return datumReader.read(null, decoder)
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicClassSchemaTest.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicClassSchemaTest.kt
new file mode 100644
index 0000000..4cf2a39
--- /dev/null
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicClassSchemaTest.kt
@@ -0,0 +1,66 @@
+package com.github.avrokotlin.avro4k.schema
+
+import com.github.avrokotlin.avro4k.Avro
+import io.kotest.core.spec.style.StringSpec
+import io.kotest.matchers.shouldBe
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.*
+import kotlinx.serialization.modules.SerializersModule
+import kotlinx.serialization.modules.polymorphic
+import kotlinx.serialization.modules.subclass
+import org.apache.avro.Schema
+
+@Serializable
+abstract class UnsealedPolymorphicRoot
+
+@Serializable
+data class UnsealedChildOne(val one: String) : UnsealedPolymorphicRoot()
+
+@Serializable
+data class UnsealedChildTwo(val two: String) : UnsealedPolymorphicRoot()
+
+@Serializable
+data class ReferencingPolymorphicRoot(
+   val root : UnsealedPolymorphicRoot,
+   val nullableRoot : UnsealedPolymorphicRoot? = null
+)
+
+class PolymorphicClassSchemaTest : StringSpec({
+   "schema for polymorphic hierarchy" {
+      val module = SerializersModule {
+         polymorphic(UnsealedPolymorphicRoot::class) {
+            subclass(UnsealedChildOne::class)
+            subclass(UnsealedChildTwo::class)
+         }
+      }
+      val schema = Avro(serializersModule = module).schema(UnsealedPolymorphicRoot.serializer())
+      val expected = Schema.Parser().parse(javaClass.getResourceAsStream("/polymorphic.json"))
+      schema shouldBe expected
+   }
+
+   "supports polymorphic references / nested fields" {
+      val module = SerializersModule {
+         polymorphic(UnsealedPolymorphicRoot::class) {
+            subclass(UnsealedChildOne::class)
+            subclass(UnsealedChildTwo::class)
+         }
+      }
+      val schema = Avro(serializersModule = module).schema(ReferencingPolymorphicRoot.serializer())
+      val expected = Schema.Parser().parse(javaClass.getResourceAsStream("/polymorphic_reference.json"))
+      schema shouldBe expected
+   }
+})
+
+object TypeSerializer : JsonTransformingSerializer<UnsealedPolymorphicRoot>(UnsealedPolymorphicRoot.serializer()) {
+   override fun transformDeserialize(element: JsonElement): JsonElement {
+      return when (element) {
+         is JsonPrimitive -> {
+            buildJsonObject {
+               put("type", element)
+            }
+         }
+         is JsonObject -> element
+         else -> throw IllegalAccessException("Unsupported json element type")
+      }
+   }
+}
\ No newline at end of file
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicSchemaTest.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicSchemaTest.kt
new file mode 100644
index 0000000..6959230
--- /dev/null
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/schema/PolymorphicSchemaTest.kt
@@ -0,0 +1,34 @@
+package com.github.avrokotlin.avro4k.schema
+
+import com.github.avrokotlin.avro4k.Avro
+import com.github.avrokotlin.avro4k.io.AvroDecodeFormat
+import com.github.avrokotlin.avro4k.io.AvroEncodeFormat
+import com.github.avrokotlin.avro4k.serializer.LocalDateSerializer
+import kotlinx.serialization.Serializable
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.time.LocalDate
+
+@Serializable
+data class BugFoo(
+    @Serializable(with = LocalDateSerializer::class)
+    val d: LocalDate,
+    val f: Float,
+    val s: String
+)
+
+fun main() {
+    val byteArrayOutputStream = ByteArrayOutputStream()
+    val value = BugFoo(LocalDate.of(2002,1,1),2.3f,"34")
+    val serializer = BugFoo.serializer()
+    Avro.default.openOutputStream(serializer) {
+        encodeFormat = AvroEncodeFormat.Binary
+    }.to(byteArrayOutputStream).write(value).flush()
+
+    val byteArrayInputStream = ByteArrayInputStream(byteArrayOutputStream.toByteArray())
+    val result = Avro.default.openInputStream(serializer) {
+        decodeFormat = AvroDecodeFormat.Binary(Avro.default.schema(serializer))
+    }.from(byteArrayInputStream).next()
+
+    println("Decoded from array stream: $result")
+}
\ No newline at end of file
diff --git a/src/test/resources/polymorphic.json b/src/test/resources/polymorphic.json
new file mode 100644
index 0000000..b5d718a
--- /dev/null
+++ b/src/test/resources/polymorphic.json
@@ -0,0 +1,24 @@
+[
+  {
+    "type": "record",
+    "name": "UnsealedChildOne",
+    "namespace": "com.github.avrokotlin.avro4k.schema",
+    "fields": [
+      {
+        "name": "one",
+        "type": "string"
+      }
+    ]
+  },
+  {
+    "type": "record",
+    "name": "UnsealedChildTwo",
+    "namespace": "com.github.avrokotlin.avro4k.schema",
+    "fields": [
+      {
+        "name": "two",
+        "type": "string"
+      }
+    ]
+  }
+]
diff --git a/src/test/resources/polymorphic_reference.json b/src/test/resources/polymorphic_reference.json
new file mode 100644
index 0000000..75f6f14
--- /dev/null
+++ b/src/test/resources/polymorphic_reference.json
@@ -0,0 +1,41 @@
+{
+  "type": "record",
+  "name": "ReferencingPolymorphicRoot",
+  "namespace": "com.github.avrokotlin.avro4k.schema",
+  "fields": [
+    {
+      "name": "root",
+      "type": [
+        {
+          "type": "record",
+          "name": "UnsealedChildOne",
+          "fields": [
+            {
+              "name": "one",
+              "type": "string"
+            }
+          ]
+        },
+        {
+          "type": "record",
+          "name": "UnsealedChildTwo",
+          "fields": [
+            {
+              "name": "two",
+              "type": "string"
+            }
+          ]
+        }
+      ]
+    },
+    {
+      "name": "nullableRoot",
+      "type": [
+        "null",
+        "UnsealedChildOne",
+        "UnsealedChildTwo"
+      ]
+    }
+  ]
+}
+