Repository: hamba/avro

Issue:
Enhancement: Add support for type conversion functions before encoding
# Description:

The library currently supports predefined type conversions for certain data types, but it does not provide a way to handle custom or more complex type conversions before encoding. There are scenarios where the data type does not match the schema requirements, and the existing built-in conversions do not cover these cases. This mismatch leads to errors during encoding. Currently, there is no way to register custom type conversion logic to address this issue, requiring duplication of all of the data, converting field-by-field as necessary, prior to encoding.

# Example Use Case:

For instance, when using `json.Unmarshal`, all numeric values are unmarshalled as `float64`. If the Avro schema expects an `int32` or `int64` but the data is still a `float64`, the encoding results in an error.
Another case is when receiving data via gRPC, where the data types in its schema might not align with the data types supported by this library.

# Proposed Solution:

Provide support for custom type conversion functions that can be registered to handle mismatched data types before the encoding process, something like `avro.Register()`, but also taking in and storing a conversion function.
These functions would allow users to convert data types to the correct format based on the schema's requirements before the internal encoder applies the selected schema type.
This would enable more flexible handling of various data sources, such as JSON, gRPC, or other formats, that may present data in incompatible formats.
By allowing users to define and register their own type conversion logic, the library would become more adaptable to different data sources and use cases, ensuring proper encoding and compatibility with Avro schemas.
An alternative solution I considered was to register custom encoders, but decided against that because that would require the user to know the Avro binary encoding format, and there would be no guarantee that the encoded bytes are in the proper format required by the schema.

I'm considering adding a `TypeConverter` object, similar to `TypeResolver` (but applicable beyond just unions). This would allow for registering type conversion functions. Before creating the encoder (e.g., calling `encoderOfType()`), it would check for a matching conversion between the given data type and schema type, and if found, it would call the conversion function and use its output as if it was the original data value for encoding.

---

I wanted to use this Issue to see if this is something that would be acceptable and gather feedback on the proposed design, before starting implementing it.


Discussion:
@nrwiersma: Just to be clear, you are only proposing this for the "generic" case where you are dealing with untyped data (mostly `[]any` and `map[string]any`)?@vpapp: Yes, that is what I'm aiming for.

I see that supporting concrete types like `[]float64` to encode to `{"type":"array", "items": "int"}` would require a different approach than I was considering, since it handles slices of concrete types differently than `[]any`.
I could look at adding support for that as well, since it could be beneficial in situations like gRPC or databases, where a struct might contain a `float32` value that should be mapped to a fixed decimal in Avro, for example.@nrwiersma: Looking at the proposed solution, some thoughts that came to mind:

> This would enable more flexible handling of various data sources, such as JSON, gRPC, or other formats, that may present data in incompatible formats.

I am not entire sure that is the role this package needs to play. To play devils advocate here, how long did you spend considering changing JSON or gRPC? I suspect the answer is none, and in that case, why not them but us? For instance, trying to encoder a float64 into a gRPC varint would also not work, and requires the user to do the conversion.

> By allowing users to define and register their own type conversion logic, the library would become more adaptable to different data sources and use cases.

To be clear, the data source is outside of the purview of package, It is never seen or considered. The main goal is to efficiently encode and decode Avro.

Do you have concrete ideas on how this would be implemented? With the generic system and typed system running in the same flow, this could get tricky to separate. I wonder if this is even possible without affecting performance of the type system, which would not be acceptable. 
@vpapp: > To play devils advocate here, how long did you spend considering changing JSON or gRPC? I suspect the answer is none, and in that case, why not them but us? For instance, trying to encoder a float64 into a gRPC varint would also not work, and requires the user to do the conversion.

That is a fair question.  Our immediate requirement is to be able to go from Avro to JSON to Avro.  In order to specify what data types the numbers in JSON are, we would need a schema (likely generated from the Avro schema, but that's minor), and a module that can use that schema to unmarshal the JSON, which is something I have not encountered before.  I mentioned gRPC as an example where it would be using a struct with defined types, so having the ability to encode into Avro from different source types that could be compatible with type conversion would be helpful, but this is not a requirement for us at this time.

> Do you have concrete ideas on how this would be implemented? With the generic system and typed system running in the same flow, this could get tricky to separate. I wonder if this is even possible without affecting performance of the type system, which would not be acceptable.

I put something together in my fork (https://github.com/vpapp/avro/tree/encoder-type-conversion) that works for the generic case but not for concrete types, so it covers our immediate requirement.  At the start of `Writer.WriteVal()`, I added a call to a function that calls `sync.Map.Load()` for the type conversion function before getting the encoder, and at the end of `Reader.ReadVal()`, I added a call to a function that does a type assertion to skip non-generic target types and a `sync.Map.Load()` in the case of generic `any` type for the type conversion function.  So there is a bit of extra overhead, but not much, and benchmarks show no extra allocations, so GC shouldn't be affected.

running the existing benchmarks before my changes:
```
$ time go test -bench .
goos: linux
goarch: amd64
pkg: github.com/hamba/avro/v2
cpu: Intel(R) Core(TM) i7-10850H CPU @ 2.70GHz
BenchmarkSuperheroDecode-12           	 1900647	       707.6 ns/op	      47 B/op	       0 allocs/op
BenchmarkSuperheroEncode-12           	 1739728	       726.4 ns/op	     112 B/op	       1 allocs/op
BenchmarkPartialSuperheroDecode-12    	 4925834	       230.0 ns/op	       9 B/op	       0 allocs/op
BenchmarkSuperheroGenericDecode-12    	  197395	      7900 ns/op	    2040 B/op	      60 allocs/op
BenchmarkSuperheroGenericEncode-12    	 1000000	      1124 ns/op	      80 B/op	       3 allocs/op
BenchmarkSuperheroWriteFlush-12       	 3729450	       300.5 ns/op	       0 B/op	       0 allocs/op
```

with my changes:
```
$ time go test -bench .
goos: linux
goarch: amd64
pkg: github.com/hamba/avro/v2
cpu: Intel(R) Core(TM) i7-10850H CPU @ 2.70GHz
BenchmarkSuperheroDecode-12           	 2051798	       705.5 ns/op	      47 B/op	       0 allocs/op
BenchmarkSuperheroEncode-12           	 1602932	       762.2 ns/op	     112 B/op	       1 allocs/op
BenchmarkPartialSuperheroDecode-12    	 5010168	       250.0 ns/op	       9 B/op	       0 allocs/op
BenchmarkSuperheroGenericDecode-12    	  195776	      7360 ns/op	    2040 B/op	      60 allocs/op
BenchmarkSuperheroGenericEncode-12    	 1000000	      1090 ns/op	      80 B/op	       3 allocs/op
BenchmarkSuperheroWriteFlush-12       	 3339801	       354.1 ns/op	       0 B/op	       0 allocs/op
```

So it's pretty close, and there may be ways to optimize it.

A different way to go about this would be to add to `encoderOfType()` and `decoderOfType()`, so they return type-converting encoders similar to current encoders, but that would need the user-provided function to be an encoder/decoder, so it would have to deal with the bytes in the Avro data, which could result in invalid data for the given schema.
@nrwiersma: For comparing the benchmarks, https://pkg.go.dev/golang.org/x/perf/cmd/benchstat can be super useful, just FYI.

Both `ReadVal` and `WriteVal` are also called in the non-generic path, so you will see see the conversion functions triggered, but only on the initial object. Could be a little weird and confusing. `encoderOfType` and `decoderOfType` are triggered on the non-generic path, but only for the first object seen, after that the de/encode functions are cached. The only parts unqiue to the generic path are `efaceDecoder` and `interfaceEncoder`, but this is deeper in and makes things more complex.@vpapp: Thanks for the benchstat pointer, will keep it in mind for the future. 👍 

> The only parts unqiue to the generic path are `efaceDecoder` and `interfaceEncoder`, but this is deeper in and makes things more complex.

Looks like I could add `frozenConfig` to them so they have access to the registered type converters, and call the conversion from their `Decode`/`Encode` instead of from `ReadVal` and `WriteVal`, and the rest should be similar to the current implementation.  I'll give that a go, and see how it looks.
@vpapp: Turns out it didn't really add complexity.  The `Reader`/`Writer` passed in to `Decode()`/`Encode()` contain `frozenConfig`, so I can access the type conversion registry from there.  It does lose out on the scenario where the Schema is a single type (e.g. `{"type":"int"}` as the whole schema) encoding from an `any`, because it looks at the underlying concrete type at the top level and doesn't go through `interfaceEncoder`.
For reference, I have 2 branches in my fork, [one](https://github.com/vpapp/avro/tree/codec-type-conversion) for adding to `ReadVal`/`WriteVal`, and [another](https://github.com/vpapp/avro/tree/generic-codec-type-conversion) for adding to `efaceDecoder`/`interfaceEncoder`.

Another thing I wanted to pass by you is the API.  Currently there are 2 registration functions:
* `RegisterDecoderTypeConversion(typ Type, fn ConversionFn)` for decoding from Avro type `typ`
* `RegisterEncoderTypeConversion(obj any, typ Type, fn ConversionFn)` for encoding from Go type of `obj` to Avro type `typ`

For the encoder one, an alternate would be `RegisterEncoderTypeConversion(typ Type, fn ConversionFn)` for encoding to Avro type `typ` for consistency with the decoder one.  The difference being that first parameter of the Go type.  With the current way, if the user wants, let's say `float32` and `float64` conversions to `avro.Int`, they would need 2 functions; 1 for each source Go type.  With the alternate, they would only need one function, but with a type switch to determine the source data type.  Which one would you prefer?@nrwiersma: > It does lose out on the scenario where the Schema is a single type (e.g. {"type":"int"} as the whole schema) encoding from an any

This seems like an edge case I can live with.

> Another thing I wanted to pass by you is the API

There is a secret option 3, which is to have a `Conversion` interface with an `Encode` and `Decode` function, and for convenience a `ConversionFuncs` that allows them to be functions. This reduce the API to be `RegisterTypeConversion(typ Type, c Conversion)`, which might make it a little more unified.@vpapp: I made another [branch](https://github.com/vpapp/avro/tree/generic-codec-type-conversion-interface) with an interface, reducing the API, but it added some overhead to the user code for required boilerplate, so I also added a struct to reduce that.  I'm ok with going ahead with this version.

If the overall design looks ok to you, I'll write up some documentation and submit a PR, where we can discuss any issues with the code itself.

Solution patch:
diff --git a/README.md b/README.md
index 7ccd9b4..2ebeae5 100644
--- a/README.md
+++ b/README.md
@@ -103,23 +103,23 @@ the data.
 
 \** Please note that when the Go type is an unsigned integer care must be taken to ensure that information is not lost
 when converting between the Avro type and Go type. For example, storing a *negative* number in Avro of `int = -100`
-would be interpreted as `uint16 = 65,436` in Go. Another example would be storing numbers in Avro `int = 256` that 
-are larger than the Go type `uint8 = 0`. 
+would be interpreted as `uint16 = 65,436` in Go. Another example would be storing numbers in Avro `int = 256` that
+are larger than the Go type `uint8 = 0`.
 
 ##### Unions
 
 The following union types are accepted: `map[string]any`, `*T` and `any`.
 
-* **map[string]any:** If the union value is `nil`, a `nil` map will be en/decoded. 
+* **map[string]any:** If the union value is `nil`, a `nil` map will be en/decoded.
 When a non-`nil` union value is encountered, a single key is en/decoded. The key is the avro
-type name, or scheam full name in the case of a named schema (enum, fixed or record).
-* ***T:** This is allowed in a "nullable" union. A nullable union is defined as a two schema union, 
-with one of the types being `null` (ie. `["null", "string"]` or `["string", "null"]`), in this case 
+type name, or schema full name in the case of a named schema (enum, fixed or record).
+* ***T:** This is allowed in a "nullable" union. A nullable union is defined as a two schema union,
+with one of the types being `null` (ie. `["null", "string"]` or `["string", "null"]`), in this case
 a `*T` is allowed, with `T` matching the conversion table above. In the case of a slice, the slice can be used
 directly.
 * **any:** An `interface` can be provided and the type or name resolved. Primitive types
-are pre-registered, but named types, maps and slices will need to be registered with the `Register` function. 
-In the case of arrays and maps the enclosed schema type or name is postfix to the type with a `:` separator, 
+are pre-registered, but named types, maps and slices will need to be registered with the `Register` function.
+In the case of arrays and maps the enclosed schema type or name is postfix to the type with a `:` separator,
 e.g `"map:string"`. Behavior when a type cannot be resolved will depend on your chosen configuation options:
 	* !Config.UnionResolutionError && !Config.PartialUnionTypeResolution: the map type above is used
 	* Config.UnionResolutionError && !Config.PartialUnionTypeResolution: an error is returned
@@ -136,11 +136,18 @@ Enums may also implement `TextMarshaler` and `TextUnmarshaler`, and must resolve
 
 ##### Identical Underlying Types
 
-One type can be [ConvertibleTo](https://go.dev/ref/spec#Conversions) another type if they have identical underlying types. 
-A non-native type is allowed be used if it can be convertible to *time.Time*, *big.Rat* or *avro.LogicalDuration* for the particular of *LogicalTypes*.
+One type can be [ConvertibleTo](https://go.dev/ref/spec#Conversions) another type if they have identical underlying types.
+A non-native type is allowed to be used if it can be convertible to *time.Time*, *big.Rat* or *avro.LogicalDuration* for the particular of *LogicalTypes*.
 
 Ex.: `type Timestamp time.Time`
 
+##### Custom Type Conversion
+
+In case of incompatible types, custom type conversion functions can be registered with the `RegisterTypeConverters` function.
+This requires the use of `map[string]any` or `[]any`.
+The type conversion for encoding will receive the original value that is to be encoded, and must return a data type that is compatible with the schema, as specified in the table above.
+The type conversion for decoding will receive the decoded value with a data type that is compatible with the schema, and its return value will be used as the final decoded value.
+
 ##### Untrusted Input With Bytes and Strings
 
 For security reasons, the configuration `Config.MaxByteSliceSize` restricts the maximum size of `bytes` and `string` types created
@@ -247,7 +254,7 @@ Note that this variable is global, so ideally you'd need to unset it after you'r
 ## Go Version Support
 
 This library supports the last two versions of Go. While the minimum Go version is
-not guaranteed to increase along side Go, it may jump from time to time to support 
+not guaranteed to increase along side Go, it may jump from time to time to support
 additional features. This will be not be considered a breaking change.
 
 ## Who uses hamba/avro?
diff --git a/codec_dynamic.go b/codec_dynamic.go
index f14a04e..644113b 100644
--- a/codec_dynamic.go
+++ b/codec_dynamic.go
@@ -26,6 +26,15 @@ func newEfaceDecoder(d *decoderContext, schema Schema) *efaceDecoder {
 
 func (d *efaceDecoder) Decode(ptr unsafe.Pointer, r *Reader) {
 	pObj := (*any)(ptr)
+
+	defer func() {
+		obj, err := r.cfg.typeConverters.DecodeTypeConvert(*pObj, d.schema)
+		if err != nil {
+			r.Error = err
+		}
+		*pObj = obj
+	}()
+
 	if *pObj == nil {
 		*pObj = genericDecode(d.typ, d.dec, r)
 		return
@@ -55,5 +64,12 @@ type interfaceEncoder struct {
 
 func (e *interfaceEncoder) Encode(ptr unsafe.Pointer, w *Writer) {
 	obj := e.typ.UnsafeIndirect(ptr)
+
+	obj, err := w.cfg.typeConverters.EncodeTypeConvert(obj, e.schema)
+	if err != nil {
+		w.Error = err
+		return
+	}
+
 	w.WriteVal(e.schema, obj)
 }
diff --git a/codec_fixed.go b/codec_fixed.go
index d6a642b..c0c28db 100644
--- a/codec_fixed.go
+++ b/codec_fixed.go
@@ -29,8 +29,8 @@ func createDecoderOfFixed(fixed *FixedSchema, typ reflect2.Type) ValDecoder {
 		elemType := ptrType.Elem()
 
 		ls := fixed.Logical()
-		tpy1 := elemType.Type1()
-		if elemType.Kind() != reflect.Struct || !tpy1.ConvertibleTo(ratType) || ls == nil ||
+		typ1 := elemType.Type1()
+		if elemType.Kind() != reflect.Struct || !typ1.ConvertibleTo(ratType) || ls == nil ||
 			ls.Type() != Decimal {
 			break
 		}
@@ -72,8 +72,8 @@ func createEncoderOfFixed(fixed *FixedSchema, typ reflect2.Type) ValEncoder {
 		elemType := ptrType.Elem()
 
 		ls := fixed.Logical()
-		tpy1 := elemType.Type1()
-		if elemType.Kind() != reflect.Struct || !tpy1.ConvertibleTo(ratType) || ls == nil ||
+		typ1 := elemType.Type1()
+		if elemType.Kind() != reflect.Struct || !typ1.ConvertibleTo(ratType) || ls == nil ||
 			ls.Type() != Decimal {
 			break
 		}
diff --git a/codec_native.go b/codec_native.go
index 85612a6..bf89f2b 100644
--- a/codec_native.go
+++ b/codec_native.go
@@ -164,12 +164,12 @@ func createDecoderOfNative(schema *PrimitiveSchema, typ reflect2.Type) ValDecode
 	case reflect.Ptr:
 		ptrType := typ.(*reflect2.UnsafePtrType)
 		elemType := ptrType.Elem()
-		tpy1 := elemType.Type1()
+		typ1 := elemType.Type1()
 		ls := getLogicalSchema(schema)
 		if ls == nil {
 			break
 		}
-		if !tpy1.ConvertibleTo(ratType) || schema.Type() != Bytes || ls.Type() != Decimal {
+		if !typ1.ConvertibleTo(ratType) || schema.Type() != Bytes || ls.Type() != Decimal {
 			break
 		}
 		dec := ls.(*DecimalLogicalSchema)
@@ -310,12 +310,12 @@ func createEncoderOfNative(schema *PrimitiveSchema, typ reflect2.Type) ValEncode
 	case reflect.Ptr:
 		ptrType := typ.(*reflect2.UnsafePtrType)
 		elemType := ptrType.Elem()
-		tpy1 := elemType.Type1()
+		typ1 := elemType.Type1()
 		ls := getLogicalSchema(schema)
 		if ls == nil {
 			break
 		}
-		if !tpy1.ConvertibleTo(ratType) || schema.Type() != Bytes || ls.Type() != Decimal {
+		if !typ1.ConvertibleTo(ratType) || schema.Type() != Bytes || ls.Type() != Decimal {
 			break
 		}
 		dec := ls.(*DecimalLogicalSchema)
diff --git a/config.go b/config.go
index 57b2769..a744b69 100644
--- a/config.go
+++ b/config.go
@@ -61,8 +61,9 @@ type Config struct {
 // Freeze makes the configuration immutable.
 func (c Config) Freeze() API {
 	api := &frozenConfig{
-		config:   c,
-		resolver: NewTypeResolver(),
+		config:         c,
+		resolver:       NewTypeResolver(),
+		typeConverters: NewTypeConverters(),
 	}
 
 	api.readerPool = &sync.Pool{
@@ -109,10 +110,13 @@ type API interface {
 	DecoderOf(schema Schema, typ reflect2.Type) ValDecoder
 
 	// EncoderOf returns the value encoder for a given schema and type.
-	EncoderOf(schema Schema, tpy reflect2.Type) ValEncoder
+	EncoderOf(schema Schema, typ reflect2.Type) ValEncoder
 
 	// Register registers names to their types for resolution. All primitive types are pre-registered.
 	Register(name string, obj any)
+
+	// RegisterTypeConverters registers type conversion functions.
+	RegisterTypeConverters(conv ...TypeConverter)
 }
 
 type frozenConfig struct {
@@ -125,6 +129,8 @@ type frozenConfig struct {
 	writerPool *sync.Pool
 
 	resolver *TypeResolver
+
+	typeConverters *TypeConverters
 }
 
 func (c *frozenConfig) Marshal(schema Schema, v any) ([]byte, error) {
@@ -204,6 +210,10 @@ func (c *frozenConfig) Register(name string, obj any) {
 	c.resolver.Register(name, obj)
 }
 
+func (c *frozenConfig) RegisterTypeConverters(convs ...TypeConverter) {
+	c.typeConverters.RegisterTypeConverters(convs...)
+}
+
 type cacheKey struct {
 	fingerprint [32]byte
 	rtype       uintptr
diff --git a/typeconverter.go b/typeconverter.go
new file mode 100644
index 0000000..274356d
--- /dev/null
+++ b/typeconverter.go
@@ -0,0 +1,116 @@
+package avro
+
+import (
+	"sync"
+)
+
+// TypeConverter represents a data type converter.
+type TypeConverter interface {
+	// Type returns the Avro data type of the type converter.
+	Type() Type
+
+	// LogicalType returns the Avro logical data type of the type converter.
+	LogicalType() LogicalType
+
+	// EncodeTypeConvert is the type conversion function called before encoding to Avro.
+	EncodeTypeConvert(in any) (any, error)
+
+	// DecodeTypeConvert is the type conversion function called after decoding from Avro.
+	DecodeTypeConvert(in any) (any, error)
+}
+
+type specificType struct {
+	typ Type
+	lt  LogicalType
+}
+
+// TypeConverters holds the user-provided type conversion functions.
+type TypeConverters struct {
+	convs sync.Map // map[specificType]TypeConverter
+}
+
+// NewTypeConverters creates a new type converter.
+func NewTypeConverters() *TypeConverters {
+	return &TypeConverters{}
+}
+
+// RegisterTypeConverters registers type converters for converting the data types during encoding and decoding.
+func (c *TypeConverters) RegisterTypeConverters(convs ...TypeConverter) {
+	for _, conv := range convs {
+		if typ := conv.Type(); len(typ) == 0 {
+			continue
+		}
+		c.convs.Store(specificType{typ: conv.Type(), lt: conv.LogicalType()}, conv)
+	}
+}
+
+// EncodeTypeConvert runs the encode type conversion function for the given value and schema.
+func (c *TypeConverters) EncodeTypeConvert(in any, schema Schema) (any, error) {
+	conv, ok := c.getTypeConverter(schema)
+	if !ok {
+		return in, nil
+	}
+
+	return conv.EncodeTypeConvert(in)
+}
+
+// DecodeTypeConvert runs the decode type conversion function for the given value and schema.
+func (c *TypeConverters) DecodeTypeConvert(in any, schema Schema) (any, error) {
+	conv, ok := c.getTypeConverter(schema)
+	if !ok {
+		return in, nil
+	}
+
+	return conv.DecodeTypeConvert(in)
+}
+
+func (c *TypeConverters) getTypeConverter(schema Schema) (TypeConverter, bool) {
+	typ := schema.Type()
+	lt := getLogicalType(schema)
+	conv, ok := c.convs.Load(specificType{typ: typ, lt: lt})
+	if !ok {
+		return nil, false
+	}
+	return conv.(TypeConverter), ok
+}
+
+// RegisterTypeConverters registers type converters for encoding and decoding the data type specified in the converter.
+func RegisterTypeConverters(convs ...TypeConverter) {
+	DefaultConfig.RegisterTypeConverters(convs...)
+}
+
+// TypeConversionFuncs is a helper struct to reduce boilerplate in user code.
+//
+// Implements the TypeConverter interface.
+type TypeConversionFuncs struct {
+	AvroType              Type
+	AvroLogicalType       LogicalType
+	EncoderTypeConversion func(in any) (any, error)
+	DecoderTypeConversion func(in any) (any, error)
+}
+
+// Type returns the Avro data type of the type converter.
+func (c TypeConversionFuncs) Type() Type {
+	return c.AvroType
+}
+
+// LogicalType returns the Avro data type of the type converter.
+func (c TypeConversionFuncs) LogicalType() LogicalType {
+	return c.AvroLogicalType
+}
+
+// EncodeTypeConvert runs the converter's encoder type conversion function if it's set.
+func (c TypeConversionFuncs) EncodeTypeConvert(in any) (any, error) {
+	if c.EncoderTypeConversion == nil {
+		return in, nil
+	}
+	return c.EncoderTypeConversion(in)
+}
+
+// DecodeTypeConvert runs the converter's decoder type conversion function if it's set.
+func (c TypeConversionFuncs) DecodeTypeConvert(in any) (any, error) {
+	if c.DecoderTypeConversion == nil {
+		return in, nil
+	}
+	return c.DecoderTypeConversion(in)
+}



Test patch:
diff --git a/typeconverter_test.go b/typeconverter_test.go
new file mode 100644
index 0000000..ca2b44f
--- /dev/null
+++ b/typeconverter_test.go
@@ -0,0 +1,278 @@
+package avro_test
+
+import (
+	"bytes"
+	"errors"
+	"fmt"
+	"math/big"
+	"testing"
+
+	"github.com/hamba/avro/v2"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	boolConverter = avro.TypeConversionFuncs{
+		AvroType: avro.Boolean,
+		DecoderTypeConversion: func(in any) (any, error) {
+			b := in.(bool)
+			if b {
+				return "yes", nil
+			} else {
+				return "no", nil
+			}
+		},
+	}
+
+	intConverter = avro.TypeConversionFuncs{
+		AvroType: avro.Int,
+		EncoderTypeConversion: func(in any) (any, error) {
+			switch v := in.(type) {
+			case float32:
+				if float32(int(v)) != v {
+					return 0, fmt.Errorf("%v is not an integer", in)
+				}
+				return int(v), nil
+			case float64:
+				if float64(int(v)) != v {
+					return 0, fmt.Errorf("%v is not an integer", in)
+				}
+				return int(v), nil
+			}
+			return in, nil
+		},
+	}
+
+	fixedDecimalConverter = avro.TypeConversionFuncs{
+		AvroType:        avro.Fixed,
+		AvroLogicalType: avro.Decimal,
+		EncoderTypeConversion: func(in any) (any, error) {
+			switch v := in.(type) {
+			case string:
+				val, _ := new(big.Rat).SetString(v)
+				return val, nil
+			}
+			return in, nil
+		},
+		DecoderTypeConversion: func(in any) (any, error) {
+			r := in.(*big.Rat)
+			f, _ := r.Float64()
+			return f, nil
+		},
+	}
+)
+
+func nonConverter(typ avro.Type) avro.TypeConversionFuncs {
+	return avro.TypeConversionFuncs{
+		AvroType: typ,
+	}
+}
+
+func errorConverter(typ avro.Type, err error) avro.TypeConversionFuncs {
+	return avro.TypeConversionFuncs{
+		AvroType: typ,
+		EncoderTypeConversion: func(in any) (any, error) {
+			return nil, err
+		},
+		DecoderTypeConversion: func(in any) (any, error) {
+			return nil, err
+		},
+	}
+}
+
+func TestEncoderTypeConverter_Array(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"array", "items":"int"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(intConverter)
+
+	val := []any{
+		float32(27),
+		float64(28),
+	}
+	err = enc.Encode(val)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x03, 0x04, 0x36, 0x38, 0x0}, buf.Bytes())
+}
+
+func TestEncoderTypeConverter_RecordMap(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"record", "name":"test", "fields":[{"name":"a", "type":"int"},{"name":"b", "type": {"type":"fixed", "name":"fixed", "size":6, "logicalType":"decimal", "precision":4, "scale":2}}]}`
+	buf := &bytes.Buffer{}
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(intConverter, fixedDecimalConverter)
+
+	val := map[string]any{
+		"a": float64(27),
+		"b": "346.8",
+	}
+	err = enc.Encode(val)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x36, 0x00, 0x00, 0x00, 0x00, 0x87, 0x78}, buf.Bytes())
+}
+
+func TestEncoderTypeConverter_RecordStruct(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"record", "name":"test", "fields":[{"name":"a", "type":"int"},{"name":"b", "type": {"type":"fixed", "name":"fixed", "size":6, "logicalType":"decimal", "precision":4, "scale":2}}]}`
+	buf := &bytes.Buffer{}
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(intConverter, fixedDecimalConverter)
+
+	type TestRecord struct {
+		A any `avro:"a"`
+		B any `avro:"b"`
+	}
+
+	val := TestRecord{
+		A: float64(27),
+		B: "346.8",
+	}
+	err = enc.Encode(val)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x36, 0x00, 0x00, 0x00, 0x00, 0x87, 0x78}, buf.Bytes())
+}
+
+func TestEncoderTypeConverter_UnionInterfaceUnregisteredArray(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `["int", {"type":"array", "items":"int"}]`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(intConverter)
+
+	var val any = map[string]any{
+		"array": []any{
+			float32(27),
+			float64(28),
+		},
+	}
+	err = enc.Encode(val)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x03, 0x04, 0x36, 0x38, 0x00}, buf.Bytes())
+}
+
+func TestEncoderTypeConverter_NotSet(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"array", "items":"int"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(nonConverter(avro.Int))
+
+	val := []any{
+		27,
+		28,
+	}
+	err = enc.Encode(val)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte{0x03, 0x04, 0x36, 0x38, 0x0}, buf.Bytes())
+}
+
+func TestEncoderTypeConverter_Error(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"array", "items":"int"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	require.NoError(t, err)
+
+	testError := errors.New("test error")
+	avro.RegisterTypeConverters(errorConverter(avro.Int, testError))
+
+	val := []any{
+		float32(27.1),
+	}
+	err = enc.Encode(val)
+
+	assert.ErrorIs(t, err, testError)
+	assert.Empty(t, buf.Bytes())
+}
+
+func TestDecoderTypeConverter_Single(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x01}
+	schema := `{"type":"boolean"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(boolConverter)
+
+	var b any
+	err = dec.Decode(&b)
+
+	require.NoError(t, err)
+	assert.Equal(t, "yes", b)
+}
+
+func TestDecoderTypeConverter_FixedRat(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x00, 0x00, 0x00, 0x00, 0x87, 0x78}
+	schema := `{"type":"fixed", "name": "test", "size": 6,"logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(fixedDecimalConverter)
+
+	var got any
+	err = dec.Decode(&got)
+
+	require.NoError(t, err)
+	assert.Equal(t, float64(346.8), got)
+}
+
+func TestDecoderTypeConverter_NotSet(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x01}
+	schema := `{"type":"boolean"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	avro.RegisterTypeConverters(nonConverter(avro.Boolean))
+
+	var b any
+	err = dec.Decode(&b)
+
+	require.NoError(t, err)
+	assert.Equal(t, true, b)
+}
+
+func TestDecoderTypeConverter_Error(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x36}
+	schema := `{"type":"int"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	require.NoError(t, err)
+
+	testError := errors.New("test error")
+	avro.RegisterTypeConverters(errorConverter(avro.Int, testError))
+
+	var got any
+	err = dec.Decode(&got)
+
+	assert.ErrorIs(t, err, testError)
+	assert.Nil(t, got)
+}