Repository: hamba/avro
Pull request #38 (2020-02-28 16:31:04)

Logical Types
* Added logical type schemas
* Added logical type encoding and decoding


Patch:
diff --git a/.travis.yml b/.travis.yml
index 8f1fc37..2d66e90 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,7 +5,9 @@ go:
   - "1.11.x"
   - "1.12.x"
   - "1.13.x"
+  - "1.14.x"
   - "1.x"
+
 env:
   - GO111MODULE=on GOFLAGS=-mod=readonly GOPROXY=https://proxy.golang.org
 
diff --git a/README.md b/README.md
index 48354ec..6313924 100644
--- a/README.md
+++ b/README.md
@@ -62,22 +62,29 @@ More examples in the [godoc](https://godoc.org/github.com/hamba/avro).
 
 #### Types Conversions
 
-| Avro     | Go Struct                          | Go Interface              |
-| -------- | ---------------------------------- | ------------------------- |
-| `null`   | `nil`                              | `nil`                     |
-| `boolean`| `bool`                             | `bool`                    |
-| `bytes`  | `[]byte`                           | `[]byte`                  |
-| `float`  | `float32`                          | `float32`                 |
-| `double` | `float64`                          | `float64`                 |
-| `long`   | `int64`                            | `int64`                   |
-| `int`    | `int`, `int32`, `int16`, `int8`    | `int`                     |
-| `string` | `string`                           | `string`                  |
-| `array`  | `[]T`                              | `[]interface{}`           |
-| `enum`   | `string`                           | `string`                  |
-| `fixed`  | `[n]byte`                          | `[]byte`                  |
-| `map`    | `map[string]T{}`                   | `map[string]interface{}`  |
-| `record` | `struct`                           | `map[string]interface{}`  |
-| `union`  | *see below*                        | *see below*               |
+| Avro                    | Go Struct                          | Go Interface              |
+| ----------------------- | ---------------------------------- | ------------------------- |
+| `null`                  | `nil`                              | `nil`                     |
+| `boolean`               | `bool`                             | `bool`                    |
+| `bytes`                 | `[]byte`                           | `[]byte`                  |
+| `float`                 | `float32`                          | `float32`                 |
+| `double`                | `float64`                          | `float64`                 |
+| `long`                  | `int64`                            | `int64`                   |
+| `int`                   | `int`, `int32`, `int16`, `int8`    | `int`                     |
+| `string`                | `string`                           | `string`                  |
+| `array`                 | `[]T`                              | `[]interface{}`           |
+| `enum`                  | `string`                           | `string`                  |
+| `fixed`                 | `[n]byte`                          | `[]byte`                  |
+| `map`                   | `map[string]T{}`                   | `map[string]interface{}`  |
+| `record`                | `struct`                           | `map[string]interface{}`  |
+| `union`                 | *see below*                        | *see below*               |
+| `int.date`              | `time.Time`                        | `time.Time`               |
+| `int.time-millis`       | `time.Duration`                    | `time.Duration`           |
+| `long.time-micros`      | `time.Duration`                    | `time.Duration`           |
+| `long.timestamp-millis` | `time.Time`                        | `time.Time`               |
+| `long.timestamp-micros` | `time.Time`                        | `time.Time`               |
+| `bytes.decimal`         | `*big.Rat`                         | `*big.Rat`                |
+| `fixed.decimal`         | `*big.Rat`                         | `*big.Rat`                |
 
 ##### Unions
 
@@ -111,9 +118,3 @@ BenchmarkLinkedinEncode-8   	 2000000	       816 ns/op	     288 B/op	      10 al
 ```
 
 Always benchmark with your own workload. The result depends heavily on the data input.
-
-## TODO
-
-* Logical Types
-* Aliases
-* Avro Textual Form?
diff --git a/codec.go b/codec.go
index a9e4d86..d83ad39 100644
--- a/codec.go
+++ b/codec.go
@@ -2,12 +2,24 @@ package avro
 
 import (
 	"fmt"
+	"math/big"
 	"reflect"
+	"time"
 	"unsafe"
 
 	"github.com/modern-go/reflect2"
 )
 
+var (
+	timeRType      uintptr
+	ratRType       uintptr
+)
+
+func init() {
+	timeRType = reflect2.TypeOf(time.Time{}).RType()
+	ratRType = reflect2.TypeOf(big.Rat{}).RType()
+}
+
 type null struct{}
 
 // ValDecoder represents an internal value decoder.
diff --git a/codec_fixed.go b/codec_fixed.go
index fd29e89..cb08ee3 100644
--- a/codec_fixed.go
+++ b/codec_fixed.go
@@ -2,6 +2,8 @@ package avro
 
 import (
 	"fmt"
+	"math"
+	"math/big"
 	"reflect"
 	"unsafe"
 
@@ -9,20 +11,29 @@ import (
 )
 
 func createDecoderOfFixed(schema Schema, typ reflect2.Type) ValDecoder {
+	fixed := schema.(*FixedSchema)
 	switch typ.Kind() {
 	case reflect.Array:
 		arrayType := typ.(reflect2.ArrayType)
-		fixed := schema.(*FixedSchema)
 		if arrayType.Elem().Kind() != reflect.Uint8 || arrayType.Len() != fixed.Size() {
 			break
 		}
 		return &fixedCodec{arrayType: typ.(*reflect2.UnsafeArrayType)}
+
+	case reflect.Struct:
+		ls := fixed.Logical()
+		if typ.RType() != ratRType || ls == nil || ls.Type() != Decimal {
+			break
+		}
+		dec := ls.(*DecimalLogicalSchema)
+		return &fixedDecimalCodec{prec: dec.Precision(), scale: dec.Scale(), size: fixed.Size()}
 	}
 
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
 func createEncoderOfFixed(schema Schema, typ reflect2.Type) ValEncoder {
+	fixed := schema.(*FixedSchema)
 	switch typ.Kind() {
 	case reflect.Array:
 		arrayType := typ.(reflect2.ArrayType)
@@ -31,6 +42,17 @@ func createEncoderOfFixed(schema Schema, typ reflect2.Type) ValEncoder {
 			break
 		}
 		return &fixedCodec{arrayType: typ.(*reflect2.UnsafeArrayType)}
+
+	case reflect.Ptr:
+		ptrType := typ.(*reflect2.UnsafePtrType)
+		elemType := ptrType.Elem()
+
+		ls := fixed.Logical()
+		if elemType.Kind() != reflect.Struct || elemType.RType() != ratRType || ls == nil || ls.Type() != Decimal {
+			break
+		}
+		dec := ls.(*DecimalLogicalSchema)
+		return &fixedDecimalCodec{prec: dec.Precision(), scale: dec.Scale(), size: fixed.Size()}
 	}
 
 	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
@@ -52,3 +74,43 @@ func (c *fixedCodec) Encode(ptr unsafe.Pointer, w *Writer) {
 		w.writeByte(*((*byte)(bytePtr)))
 	}
 }
+
+type fixedDecimalCodec struct {
+	prec  int
+	scale int
+	size  int
+}
+
+func (c *fixedDecimalCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	b := make([]byte, c.size)
+	r.Read(b)
+	*((*big.Rat)(ptr)) = *ratFromBytes(b, c.scale)
+}
+
+func (c *fixedDecimalCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	r := *((**big.Rat)(ptr))
+	i := (&big.Int{}).Mul(r.Num(), big.NewInt(int64(math.Pow10(c.scale))))
+	i = i.Div(i, r.Denom())
+
+	var b []byte
+	switch i.Sign() {
+	case 0:
+		b = make([]byte, c.size, c.size)
+
+	case 1:
+		b = i.Bytes()
+		if b[0]&0x80 > 0 {
+			b = append([]byte{0}, b...)
+		}
+		if len(b) < c.size {
+			padded := make([]byte, c.size, c.size)
+			copy(padded[c.size-len(b):], b)
+			b = padded
+		}
+
+	case -1:
+		b = i.Add(i, (&big.Int{}).Lsh(one, uint(c.size * 8))).Bytes()
+	}
+
+	w.Write(b)
+}
diff --git a/codec_native.go b/codec_native.go
index 81a075c..fa97b10 100644
--- a/codec_native.go
+++ b/codec_native.go
@@ -2,7 +2,10 @@ package avro
 
 import (
 	"fmt"
+	"math"
+	"math/big"
 	"reflect"
+	"time"
 	"unsafe"
 
 	"github.com/modern-go/reflect2"
@@ -41,10 +44,21 @@ func createDecoderOfNative(schema Schema, typ reflect2.Type) ValDecoder {
 		return &int32Codec{}
 
 	case reflect.Int64:
-		if schema.Type() != Long {
+		st := schema.Type()
+		lt := getLogicalType(schema)
+		switch {
+		case st == Int && lt == TimeMillis: // time.Duration
+			return &timeMillisCodec{}
+
+		case st == Long && lt == TimeMicros: // time.Duration
+			return &timeMicrosCodec{}
+
+		case st == Long:
+			return &int64Codec{}
+
+		default:
 			break
 		}
-		return &int64Codec{}
 
 	case reflect.Float32:
 		if schema.Type() != Float {
@@ -69,6 +83,29 @@ func createDecoderOfNative(schema Schema, typ reflect2.Type) ValDecoder {
 			break
 		}
 		return &bytesCodec{sliceType: typ.(*reflect2.UnsafeSliceType)}
+
+	case reflect.Struct:
+		st := schema.Type()
+		ls := getLogicalSchema(schema)
+		lt := getLogicalType(schema)
+		switch {
+		case typ.RType() == timeRType && st == Int && lt == Date:
+			return &dateCodec{}
+
+		case typ.RType() == timeRType && st == Long && lt == TimestampMillis:
+			return &timestampMillisCodec{}
+
+		case typ.RType() == timeRType && st == Long && lt == TimestampMicros:
+			return &timestampMicrosCodec{}
+
+		case typ.RType() == ratRType && st == Bytes && lt == Decimal:
+			dec := ls.(*DecimalLogicalSchema)
+
+			return &bytesDecimalCodec{prec: dec.Precision(), scale: dec.Scale()}
+
+		default:
+			break
+		}
 	}
 
 	return &errorDecoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
@@ -107,10 +144,21 @@ func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 		return &int32Codec{}
 
 	case reflect.Int64:
-		if schema.Type() != Long {
+		st := schema.Type()
+		lt := getLogicalType(schema)
+		switch {
+		case st == Int && lt == TimeMillis: // time.Duration
+			return &timeMillisCodec{}
+
+		case st == Long && lt == TimeMicros: // time.Duration
+			return &timeMicrosCodec{}
+
+		case st == Long:
+			return &int64Codec{}
+
+		default:
 			break
 		}
-		return &int64Codec{}
 
 	case reflect.Float32:
 		if schema.Type() != Float {
@@ -135,6 +183,38 @@ func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 			break
 		}
 		return &bytesCodec{sliceType: typ.(*reflect2.UnsafeSliceType)}
+
+	case reflect.Struct:
+		st := schema.Type()
+		lt := getLogicalType(schema)
+		switch {
+		case typ.RType() == timeRType && st == Int && lt == Date:
+			return &dateCodec{}
+
+		case typ.RType() == timeRType && st == Long && lt == TimestampMillis:
+			return &timestampMillisCodec{}
+
+		case typ.RType() == timeRType && st == Long && lt == TimestampMicros:
+			return &timestampMicrosCodec{}
+
+		default:
+			break
+		}
+
+	case reflect.Ptr:
+		ptrType := typ.(*reflect2.UnsafePtrType)
+		elemType := ptrType.Elem()
+
+		ls := getLogicalSchema(schema)
+		if ls == nil {
+			break
+		}
+		if elemType.RType() != ratRType || schema.Type() != Bytes || ls.Type() != Decimal {
+			break
+		}
+		dec := ls.(*DecimalLogicalSchema)
+
+		return &bytesDecimalCodec{prec: dec.Precision(), scale: dec.Scale()}
 	}
 
 	if schema.Type() == Null {
@@ -144,6 +224,24 @@ func createEncoderOfNative(schema Schema, typ reflect2.Type) ValEncoder {
 	return &errorEncoder{err: fmt.Errorf("avro: %s is unsupported for Avro %s", typ.String(), schema.Type())}
 }
 
+func getLogicalSchema(schema Schema) LogicalSchema {
+	lts, ok := schema.(LogicalTypeSchema)
+	if !ok {
+		return nil
+	}
+
+	return lts.Logical()
+}
+
+func getLogicalType(schema Schema) LogicalType {
+	ls := getLogicalSchema(schema)
+	if ls == nil {
+		return ""
+	}
+
+	return ls.Type()
+}
+
 type nullCodec struct{}
 
 func (*nullCodec) Encode(ptr unsafe.Pointer, w *Writer) {}
@@ -250,3 +348,110 @@ func (c *bytesCodec) Decode(ptr unsafe.Pointer, r *Reader) {
 func (c *bytesCodec) Encode(ptr unsafe.Pointer, w *Writer) {
 	w.WriteBytes(*((*[]byte)(ptr)))
 }
+
+type dateCodec struct{}
+
+func (c *dateCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	i := r.ReadInt()
+	*((*time.Time)(ptr)) = time.Unix(0, int64(i) * int64(24*time.Hour)).UTC()
+}
+
+func (c *dateCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	t := *((*time.Time)(ptr))
+	w.WriteInt(int32(t.UnixNano() / int64(24*time.Hour)))
+}
+
+type timestampMillisCodec struct{}
+
+func (c *timestampMillisCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	i := r.ReadLong()
+	*((*time.Time)(ptr)) = time.Unix(0, i * int64(time.Millisecond)).UTC()
+}
+
+func (c *timestampMillisCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	t := *((*time.Time)(ptr))
+	w.WriteLong(t.UnixNano() / int64(time.Millisecond))
+}
+
+type timestampMicrosCodec struct{}
+
+func (c *timestampMicrosCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	i := r.ReadLong()
+	*((*time.Time)(ptr)) = time.Unix(0, i * int64(time.Microsecond)).UTC()
+}
+
+func (c *timestampMicrosCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	t := *((*time.Time)(ptr))
+	w.WriteLong(t.Unix()*1e6 + int64(t.Nanosecond()/1e3))
+}
+
+type timeMillisCodec struct{}
+
+func (c *timeMillisCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	i := r.ReadInt()
+	*((*time.Duration)(ptr)) = time.Duration(i) * time.Millisecond
+}
+
+func (c *timeMillisCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	d := *((*time.Duration)(ptr))
+	w.WriteInt(int32(d.Nanoseconds() / int64(time.Millisecond)))
+}
+
+type timeMicrosCodec struct{}
+
+func (c *timeMicrosCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	i := r.ReadLong()
+	*((*time.Duration)(ptr)) = time.Duration(i) * time.Microsecond
+}
+
+func (c *timeMicrosCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	d := *((*time.Duration)(ptr))
+	w.WriteLong(d.Nanoseconds() / int64(time.Microsecond))
+}
+
+var one = big.NewInt(1)
+
+type bytesDecimalCodec struct {
+	prec  int
+	scale int
+}
+
+func (c *bytesDecimalCodec) Decode(ptr unsafe.Pointer, r *Reader) {
+	b := r.ReadBytes()
+	i := (&big.Int{}).SetBytes(b)
+	if len(b) > 0 && b[0]&0x80 > 0 {
+		i.Sub(i, new(big.Int).Lsh(one, uint(len(b))*8))
+	}
+	*((*big.Rat)(ptr)) = *ratFromBytes(b, c.scale)
+}
+
+func ratFromBytes(b []byte, scale int) *big.Rat {
+	i := (&big.Int{}).SetBytes(b)
+	if len(b) > 0 && b[0]&0x80 > 0 {
+		i.Sub(i, new(big.Int).Lsh(one, uint(len(b))*8))
+	}
+	return big.NewRat(i.Int64(), int64(math.Pow10(scale)))
+}
+
+func (c *bytesDecimalCodec) Encode(ptr unsafe.Pointer, w *Writer) {
+	r := *((**big.Rat)(ptr))
+	i := (&big.Int{}).Mul(r.Num(), big.NewInt(int64(math.Pow10(c.scale))))
+	i = i.Div(i, r.Denom())
+
+	var b []byte
+	switch i.Sign() {
+	case 0:
+		b = []byte{0}
+
+	case 1:
+		b = i.Bytes()
+		if b[0]&0x80 > 0 {
+			b = append([]byte{0}, b...)
+		}
+
+	case -1:
+		length := uint(i.BitLen()/8+1) * 8
+		b = i.Add(i, (&big.Int{}).Lsh(one, length)).Bytes()
+	}
+	w.WriteBytes(b)
+}
diff --git a/protocol.go b/protocol.go
index a8db627..2a27e77 100644
--- a/protocol.go
+++ b/protocol.go
@@ -262,7 +262,7 @@ func parseMessage(namespace string, m map[string]interface{}, cache *SchemaCache
 		}
 	}
 
-	types := []Schema{NewPrimitiveSchema(String)}
+	types := []Schema{NewPrimitiveSchema(String, nil)}
 	if errs, ok := m["errors"].([]interface{}); ok {
 		for _, e := range errs {
 			schema, err := parseType(namespace, e, cache)
diff --git a/reader_generic.go b/reader_generic.go
index 98405c7..cd57706 100644
--- a/reader_generic.go
+++ b/reader_generic.go
@@ -2,19 +2,47 @@ package avro
 
 import (
 	"fmt"
+	"time"
 )
 
 // ReadNext reads the next Avro element as a generic interface.
 func (r *Reader) ReadNext(schema Schema) interface{} {
+	var ls LogicalSchema
+	lts, ok := schema.(LogicalTypeSchema)
+	if ok {
+		ls = lts.Logical()
+	}
+
 	switch schema.Type() {
 
 	case Boolean:
 		return r.ReadBool()
 
 	case Int:
+		if ls != nil {
+			switch ls.Type() {
+			case Date:
+				return time.Unix(0, int64(r.ReadInt())*int64(24*time.Hour)).UTC()
+
+			case TimeMillis:
+				return time.Duration(r.ReadInt()) * time.Millisecond
+			}
+		}
 		return int(r.ReadInt())
 
 	case Long:
+		if ls != nil {
+			switch ls.Type() {
+			case TimeMicros:
+				return time.Duration(r.ReadLong()) * time.Microsecond
+
+			case TimestampMillis:
+				return time.Unix(0, r.ReadLong()*int64(time.Millisecond)).UTC()
+
+			case TimestampMicros:
+				return time.Unix(0, r.ReadLong()*int64(time.Microsecond)).UTC()
+			}
+		}
 		return r.ReadLong()
 
 	case Float:
@@ -27,6 +55,10 @@ func (r *Reader) ReadNext(schema Schema) interface{} {
 		return r.ReadString()
 
 	case Bytes:
+		if ls != nil && ls.Type() == Decimal {
+			dec := ls.(*DecimalLogicalSchema)
+			return ratFromBytes(r.ReadBytes(), dec.Scale())
+		}
 		return r.ReadBytes()
 
 	case Record:
@@ -86,8 +118,13 @@ func (r *Reader) ReadNext(schema Schema) interface{} {
 		return obj
 
 	case Fixed:
-		obj := make([]byte, schema.(*FixedSchema).Size())
+		size := schema.(*FixedSchema).Size()
+		obj := make([]byte, size)
 		r.Read(obj)
+		if ls != nil && ls.Type() == Decimal {
+			dec := ls.(*DecimalLogicalSchema)
+			return ratFromBytes(obj, dec.Scale())
+		}
 		return obj
 
 	default:
diff --git a/schema.go b/schema.go
index 18828de..f64c087 100644
--- a/schema.go
+++ b/schema.go
@@ -39,6 +39,21 @@ const (
 	Null    Type = "null"
 )
 
+// LogicalType is a schema logical type.
+type LogicalType string
+
+// Schema logical type constants.
+const (
+	Decimal         LogicalType = "decimal"
+	UUID            LogicalType = "uuid"
+	Date            LogicalType = "date"
+	TimeMillis      LogicalType = "time-millis"
+	TimeMicros      LogicalType = "time-micros"
+	TimestampMillis LogicalType = "timestamp-millis"
+	TimestampMicros LogicalType = "timestamp-micros"
+	Duration        LogicalType = "duration"
+)
+
 // FingerprintType is a fingerprinting algorithm.
 type FingerprintType string
 
@@ -107,6 +122,15 @@ type Schema interface {
 	FingerprintUsing(FingerprintType) ([]byte, error)
 }
 
+// LogicalSchema represents an Avro schema with a logical type.
+type LogicalSchema interface {
+	// Type returns the type of the logical schema.
+	Type() LogicalType
+
+	// String returns the canonical form of the logical schema.
+	String() string
+}
+
 // PropertySchema represents a schema with properties.
 type PropertySchema interface {
 	// AddProp adds a property to the schema.
@@ -133,6 +157,12 @@ type NamedSchema interface {
 	FullName() string
 }
 
+// LogicalTypeSchema represents a schema that can contain a logical type.
+type LogicalTypeSchema interface {
+	// Logical returns the logical schema or nil.
+	Logical() LogicalSchema
+}
+
 type name struct {
 	name  string
 	space string
@@ -257,13 +287,15 @@ func (p *properties) Prop(name string) interface{} {
 type PrimitiveSchema struct {
 	fingerprinter
 
-	typ Type
+	typ     Type
+	logical LogicalSchema
 }
 
 // NewPrimitiveSchema creates a new PrimitiveSchema.
-func NewPrimitiveSchema(t Type) *PrimitiveSchema {
+func NewPrimitiveSchema(t Type, l LogicalSchema) *PrimitiveSchema {
 	return &PrimitiveSchema{
-		typ: t,
+		typ:     t,
+		logical: l,
 	}
 }
 
@@ -272,9 +304,18 @@ func (s *PrimitiveSchema) Type() Type {
 	return s.typ
 }
 
+// Logical returns the logical schema or nil.
+func (s *PrimitiveSchema) Logical() LogicalSchema {
+	return s.logical
+}
+
 // String returns the canonical form of the schema.
 func (s *PrimitiveSchema) String() string {
-	return `"` + string(s.typ) + `"`
+	if s.logical == nil {
+		return `"` + string(s.typ) + `"`
+	}
+
+	return `{"type":"` + string(s.typ) + `",` + s.logical.String() + `}`
 }
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
@@ -651,11 +692,12 @@ type FixedSchema struct {
 	properties
 	fingerprinter
 
-	size int
+	size    int
+	logical LogicalSchema
 }
 
 // NewFixedSchema creates a new fixed schema instance.
-func NewFixedSchema(name, namespace string, size int) (*FixedSchema, error) {
+func NewFixedSchema(name, namespace string, size int, logical LogicalSchema) (*FixedSchema, error) {
 	n, err := newName(name, namespace)
 	if err != nil {
 		return nil, err
@@ -665,6 +707,7 @@ func NewFixedSchema(name, namespace string, size int) (*FixedSchema, error) {
 		name:       n,
 		properties: properties{reserved: schemaReserved},
 		size:       size,
+		logical:    logical,
 	}, nil
 }
 
@@ -678,10 +721,21 @@ func (s *FixedSchema) Size() int {
 	return s.size
 }
 
+// Logical returns the logical schema or nil.
+func (s *FixedSchema) Logical() LogicalSchema {
+	return s.logical
+}
+
 // String returns the canonical form of the schema.
 func (s *FixedSchema) String() string {
 	size := strconv.Itoa(s.size)
-	return `{"name":"` + s.FullName() + `","type":"fixed","size":` + size + `}`
+
+	var logical string
+	if s.logical != nil {
+		logical = "," + s.logical.String()
+	}
+
+	return `{"name":"` + s.FullName() + `","type":"fixed","size":` + size + logical + `}`
 }
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
@@ -756,6 +810,68 @@ func (s *RefSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
 	return s.actual.FingerprintUsing(typ)
 }
 
+// PrimitiveLogicalSchema is a logical type with no properties.
+type PrimitiveLogicalSchema struct {
+	typ LogicalType
+}
+
+// NewPrimitiveLogicalSchema creates a new primitive logical schema instance.
+func NewPrimitiveLogicalSchema(typ LogicalType) *PrimitiveLogicalSchema {
+	return &PrimitiveLogicalSchema{
+		typ: typ,
+	}
+}
+
+// Type returns the type of the logical schema.
+func (s *PrimitiveLogicalSchema) Type() LogicalType {
+	return s.typ
+}
+
+// String returns the canonical form of the logical schema.
+func (s *PrimitiveLogicalSchema) String() string {
+	return `"logicalType":"` + string(s.typ) + `"`
+}
+
+// DecimalLogicalSchema is a decimal logical type.
+type DecimalLogicalSchema struct {
+	prec  int
+	scale int
+}
+
+// NewDecimalLogicalSchema creates a new decimal logical schema instance.
+func NewDecimalLogicalSchema(prec, scale int) *DecimalLogicalSchema {
+	return &DecimalLogicalSchema{
+		prec:  prec,
+		scale: scale,
+	}
+}
+
+// Type returns the type of the logical schema.
+func (s *DecimalLogicalSchema) Type() LogicalType {
+	return Decimal
+}
+
+// Precision returns the precision of the decimal logical schema.
+func (s *DecimalLogicalSchema) Precision() int {
+	return s.prec
+}
+
+// Scale returns the scale of the decimal logical schema.
+func (s *DecimalLogicalSchema) Scale() int {
+	return s.scale
+}
+
+// String returns the canonical form of the logical schema.
+func (s *DecimalLogicalSchema) String() string {
+	var scale string
+	if s.scale > 0 {
+		scale = `,"scale":` + strconv.Itoa(s.scale)
+	}
+	precision := strconv.Itoa(s.prec)
+
+	return `"logicalType":"` + string(Decimal) + `","precision":` + precision + scale
+}
+
 func invalidNameFirstChar(r rune) bool {
 	return (r < 'A' || r > 'Z') && (r < 'a' || r > 'z') && r != '_'
 }
diff --git a/schema_parse.go b/schema_parse.go
index 775fe35..9adccb0 100644
--- a/schema_parse.go
+++ b/schema_parse.go
@@ -4,13 +4,15 @@ import (
 	"errors"
 	"fmt"
 	"io/ioutil"
+	"math"
 	"strings"
 
 	"github.com/json-iterator/go"
 )
 
 var (
-	schemaReserved = []string{"doc", "fields", "items", "name", "namespace", "size", "symbols", "values", "type", "aliases"}
+	schemaReserved = []string{"doc", "fields", "items", "name", "namespace", "size", "symbols", "values", "type",
+		"aliases", "logicalType", "precision", "scale"}
 	fieldReserved  = []string{"default", "doc", "name", "order", "type", "aliases"}
 )
 
@@ -68,10 +70,10 @@ func parseType(namespace string, v interface{}, cache *SchemaCache) (Schema, err
 		return &NullSchema{}, nil
 
 	case string:
-		return parsePrimitive(namespace, val, cache)
+		return parsePrimitiveType(namespace, val, cache)
 
 	case map[string]interface{}:
-		return parseComplex(namespace, val, cache)
+		return parseComplexType(namespace, val, cache)
 
 	case []interface{}:
 		return parseUnion(namespace, val, cache)
@@ -80,14 +82,14 @@ func parseType(namespace string, v interface{}, cache *SchemaCache) (Schema, err
 	return nil, fmt.Errorf("avro: unknown type: %v", v)
 }
 
-func parsePrimitive(namespace, s string, cache *SchemaCache) (Schema, error) {
+func parsePrimitiveType(namespace, s string, cache *SchemaCache) (Schema, error) {
 	typ := Type(s)
 	switch typ {
 	case Null:
 		return &NullSchema{}, nil
 
 	case String, Bytes, Int, Long, Float, Double, Boolean:
-		return NewPrimitiveSchema(typ), nil
+		return parsePrimitive(typ, nil)
 
 	default:
 		schema := cache.Get(fullName(namespace, s))
@@ -99,7 +101,7 @@ func parsePrimitive(namespace, s string, cache *SchemaCache) (Schema, error) {
 	}
 }
 
-func parseComplex(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
+func parseComplexType(namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	if val, ok := m["type"].([]interface{}); ok {
 		return parseUnion(namespace, val, cache)
 	}
@@ -115,7 +117,7 @@ func parseComplex(namespace string, m map[string]interface{}, cache *SchemaCache
 		return &NullSchema{}, nil
 
 	case String, Bytes, Int, Long, Float, Double, Boolean:
-		return NewPrimitiveSchema(typ), nil
+		return parsePrimitive(typ, m)
 
 	case Record, Error:
 		return parseRecord(typ, namespace, m, cache)
@@ -137,6 +139,39 @@ func parseComplex(namespace string, m map[string]interface{}, cache *SchemaCache
 	}
 }
 
+func parsePrimitive(typ Type, m map[string]interface{}) (Schema, error) {
+	logical := parsePrimitiveLogicalType(typ, m)
+
+	return NewPrimitiveSchema(typ, logical), nil
+}
+
+func parsePrimitiveLogicalType(typ Type, m map[string]interface{}) LogicalSchema {
+	if m == nil {
+		return nil
+	}
+
+	lt, ok := m["logicalType"].(string)
+	if !ok {
+		return nil
+	}
+
+	ltyp := LogicalType(lt)
+	if (typ == String && ltyp == UUID) ||
+		(typ == Int && ltyp == Date) ||
+		(typ == Int && ltyp == TimeMillis) ||
+		(typ == Long && ltyp == TimeMicros) ||
+		(typ == Long && ltyp == TimestampMillis) ||
+		(typ == Long && ltyp == TimestampMicros) {
+		return NewPrimitiveLogicalSchema(ltyp)
+	}
+
+	if typ == Bytes && ltyp == Decimal {
+		return parseDecimalLogicalType(-1, m)
+	}
+
+	return nil
+}
+
 func parseRecord(typ Type, namespace string, m map[string]interface{}, cache *SchemaCache) (Schema, error) {
 	name, newNamespace, err := resolveFullName(m)
 	if err != nil {
@@ -319,7 +354,9 @@ func parseFixed(namespace string, m map[string]interface{}, cache *SchemaCache)
 		return nil, errors.New("avro: fixed must have a size")
 	}
 
-	fixed, err := NewFixedSchema(name, namespace, int(size))
+	logical := parseFixedLogicalType(int(size), m)
+
+	fixed, err := NewFixedSchema(name, namespace, int(size), logical)
 	if err != nil {
 		return nil, err
 	}
@@ -333,6 +370,50 @@ func parseFixed(namespace string, m map[string]interface{}, cache *SchemaCache)
 	return fixed, nil
 }
 
+func parseFixedLogicalType(size int, m map[string]interface{}) LogicalSchema {
+	lt, ok := m["logicalType"].(string)
+	if !ok {
+		return nil
+	}
+
+	ltyp := LogicalType(lt)
+	if ltyp == Duration && size == 12 {
+		return NewPrimitiveLogicalSchema(Duration)
+	}
+
+	if ltyp == Decimal {
+		return parseDecimalLogicalType(size, m)
+	}
+
+	return nil
+}
+
+func parseDecimalLogicalType(size int, m map[string]interface{}) LogicalSchema {
+	prec, ok := m["precision"].(float64)
+	if !ok || prec <= 0 {
+		return nil
+	}
+
+	if size > 0 {
+		maxPrecision := math.Round(math.Floor(math.Log10(2) * (8*float64(size) - 1)))
+		if prec > maxPrecision {
+			return nil
+		}
+	}
+
+	scale, _ := m["scale"].(float64)
+	if scale < 0 {
+		return nil
+	}
+
+	// Scale may not be bigger than precision
+	if scale > prec {
+		return nil
+	}
+
+	return NewDecimalLogicalSchema(int(prec), int(scale))
+}
+
 func fullName(namespace, name string) string {
 	if len(namespace) == 0 || strings.ContainsRune(name, '.') {
 		return name


Test patch:
diff --git a/codec_skip_internal_test.go b/codec_skip_internal_test.go
index 3ad5c80..8bfd461 100644
--- a/codec_skip_internal_test.go
+++ b/codec_skip_internal_test.go
@@ -7,7 +7,7 @@ import (
 )
 
 func TestCreateSkipDecoder_UnsupportedType(t *testing.T) {
-	schema := NewPrimitiveSchema(Type("test"))
+	schema := NewPrimitiveSchema(Type("test"), nil)
 
 	dec := createSkipDecoder(schema)
 
diff --git a/decoder_fixed_test.go b/decoder_fixed_test.go
index 74c0877..d414617 100644
--- a/decoder_fixed_test.go
+++ b/decoder_fixed_test.go
@@ -2,6 +2,7 @@ package avro_test
 
 import (
 	"bytes"
+	"math/big"
 	"testing"
 
 	"github.com/hamba/avro"
@@ -35,3 +36,62 @@ func TestDecoder_Fixed(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, [6]byte{'f', 'o', 'o', 'f', 'o', 'o'}, got)
 }
+
+func TestDecoder_FixedRat_Positive(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))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(1734, 5), got)
+}
+
+func TestDecoder_FixedRat_Negative(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0xFF, 0xFF, 0xFF, 0xFF, 0x78, 0x88}
+	schema := `{"type":"fixed", "name": "test", "size": 6, "logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(-1734, 5), got)
+}
+
+func TestDecoder_FixedRat_Zero(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
+	schema := `{"type":"fixed", "name": "test", "size": 6,"logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(0, 1), got)
+}
+
+func TestDecoder_FixedRatInvalidLogicalSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00}
+	schema := `{"type":"fixed", "name": "test", "size": 6}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.Error(t, err)
+}
diff --git a/decoder_native_test.go b/decoder_native_test.go
index 9ed7fda..9511fd0 100644
--- a/decoder_native_test.go
+++ b/decoder_native_test.go
@@ -2,7 +2,9 @@ package avro_test
 
 import (
 	"bytes"
+	"math/big"
 	"testing"
+	"time"
 
 	"github.com/hamba/avro"
 	"github.com/stretchr/testify/assert"
@@ -311,3 +313,179 @@ func TestDecoder_BytesInvalidSchema(t *testing.T) {
 
 	assert.Error(t, err)
 }
+
+func TestDecoder_Time_Date(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0xAE, 0x9D, 0x02}
+	schema := `{"type":"int","logicalType":"date"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Time
+	err = dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, time.Date(2020, 1, 2, 0, 0, 0, 0, time.UTC), got)
+}
+
+func TestDecoder_Time_TimestampMillis(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x90, 0xB2, 0xAE, 0xC3, 0xEC, 0x5B}
+	schema := `{"type":"long","logicalType":"timestamp-millis"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Time
+	err = dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC), got)
+}
+
+func TestDecoder_Time_TimestampMicros(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x80, 0xCD, 0xB7, 0xA2, 0xEE, 0xC7, 0xCD, 0x05}
+	schema := `{"type":"long","logicalType":"timestamp-micros"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Time
+	err = dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC), got)
+}
+
+func TestDecoder_TimeInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x80, 0xCD, 0xB7, 0xA2, 0xEE, 0xC7, 0xCD, 0x05}
+	schema := `{"type":"long"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Time
+	err = dec.Decode(&got)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_Duration_TimeMillis(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0xAA, 0xB4, 0xDE, 0x75}
+	schema := `{"type":"int","logicalType":"time-millis"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Duration
+	err = dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, 123456789*time.Millisecond, got)
+}
+
+func TestDecoder_Duration_TimeMicros(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07}
+	schema := `{"type":"long","logicalType":"time-micros"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Duration
+	err = dec.Decode(&got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, 123456789123*time.Microsecond, got)
+}
+
+func TestDecoder_DurationInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07}
+	schema := `{"type":"string"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	var got time.Duration
+	err = dec.Decode(&got)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_BytesRat_Positive(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x6, 0x00, 0x87, 0x78}
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(1734, 5), got)
+}
+
+func TestDecoder_BytesRat_Negative(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x6, 0xFF, 0x78, 0x88}
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(-1734, 5), got)
+}
+
+func TestDecoder_BytesRat_Zero(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x00}
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.NoError(t, err)
+	assert.Equal(t, big.NewRat(0, 1), got)
+}
+
+func TestDecoder_BytesRatInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x00}
+	schema := `{"type":"string"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.Error(t, err)
+}
+
+func TestDecoder_BytesRatInvalidLogicalSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	data := []byte{0x02, 0x00}
+	schema := `{"type":"string","logicalType":"uuid"}`
+	dec, err := avro.NewDecoder(schema, bytes.NewReader(data))
+	assert.NoError(t, err)
+
+	got := &big.Rat{}
+	err = dec.Decode(got)
+
+	assert.Error(t, err)
+}
diff --git a/decoder_test.go b/decoder_test.go
index 7275fc0..b77ae7e 100644
--- a/decoder_test.go
+++ b/decoder_test.go
@@ -21,7 +21,7 @@ func TestDecoder_DecodeUnsupportedTypeError(t *testing.T) {
 	defer ConfigTeardown()
 
 	data := []byte{0x01}
-	schema := avro.NewPrimitiveSchema(avro.Type("test"))
+	schema := avro.NewPrimitiveSchema(avro.Type("test"), nil)
 	dec := avro.NewDecoderForSchema(schema, bytes.NewReader(data))
 
 	var b bool
diff --git a/encoder_fixed_test.go b/encoder_fixed_test.go
index 44c78e9..511b772 100644
--- a/encoder_fixed_test.go
+++ b/encoder_fixed_test.go
@@ -2,6 +2,7 @@ package avro_test
 
 import (
 	"bytes"
+	"math/big"
 	"testing"
 
 	"github.com/hamba/avro"
@@ -34,3 +35,58 @@ func TestEncoder_Fixed(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, []byte{0x66, 0x6F, 0x6F, 0x66, 0x6F, 0x6F}, buf.Bytes())
 }
+
+func TestEncoder_FixedRat_Positive(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"fixed", "name": "test", "size": 6,"logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(1734, 5))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x00, 0x00, 0x00, 0x00, 0x87, 0x78}, buf.Bytes())
+}
+
+func TestEncoder_FixedRat_Negative(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"fixed", "name": "test", "size": 6, "logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(-1734, 5))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0xFF, 0xFF, 0xFF, 0xFF, 0x78, 0x88}, buf.Bytes())
+}
+
+func TestEncoder_FixedRat_Zero(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"fixed", "name": "test", "size": 6,"logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(0, 1))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00,}, buf.Bytes())
+}
+
+func TestEncoder_FixedRatInvalidLogicalSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"fixed", "name": "test", "size": 6}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(1734, 5))
+
+	assert.Error(t, err)
+}
diff --git a/encoder_native_test.go b/encoder_native_test.go
index 669940c..bbc9aba 100644
--- a/encoder_native_test.go
+++ b/encoder_native_test.go
@@ -2,7 +2,9 @@ package avro_test
 
 import (
 	"bytes"
+	"math/big"
 	"testing"
+	"time"
 
 	"github.com/hamba/avro"
 	"github.com/stretchr/testify/assert"
@@ -290,3 +292,167 @@ func TestEncoder_BytesInvalidSchema(t *testing.T) {
 
 	assert.Error(t, err)
 }
+
+func TestEncoder_Time_Date(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"int","logicalType":"date"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(time.Date(2020, 1, 2, 0, 0, 0, 0, time.UTC))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0xAE, 0x9D, 0x02}, buf.Bytes())
+}
+
+func TestEncoder_Time_TimestampMillis(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"long","logicalType":"timestamp-millis"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(time.Date(2020, 1, 2, 3, 4, 5, 6, time.UTC))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x90, 0xB2, 0xAE, 0xC3, 0xEC, 0x5B}, buf.Bytes())
+}
+
+func TestEncoder_Time_TimestampMicros(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"long","logicalType":"timestamp-micros"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(time.Date(2020, 1, 2, 3, 4, 5, 6, time.UTC))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x80, 0xCD, 0xB7, 0xA2, 0xEE, 0xC7, 0xCD, 0x05}, buf.Bytes())
+}
+
+func TestEncoder_TimeInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"long"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(time.Date(2020, 1, 2, 3, 4, 5, 6, time.UTC))
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_Duration_TimeMillis(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"int","logicalType":"time-millis"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(123456789 * time.Millisecond)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0xAA, 0xB4, 0xDE, 0x75}, buf.Bytes())
+}
+
+func TestEncoder_Duration_TimeMicros(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"long","logicalType":"time-micros"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(123456789123 * time.Microsecond)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07}, buf.Bytes())
+}
+
+func TestEncoder_DurationInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"string"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(time.Millisecond)
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_BytesRat_Positive(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(1734, 5))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x6, 0x00, 0x87, 0x78}, buf.Bytes())
+}
+
+func TestEncoder_BytesRat_Negative(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(-1734, 5))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x6, 0xFF, 0x78, 0x88}, buf.Bytes())
+}
+
+func TestEncoder_BytesRat_Zero(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(0, 1))
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0x02, 0x00}, buf.Bytes())
+}
+
+func TestEncoder_BytesRatInvalidSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"string"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(1734, 5))
+
+	assert.Error(t, err)
+}
+
+func TestEncoder_BytesRatInvalidLogicalSchema(t *testing.T) {
+	defer ConfigTeardown()
+
+	schema := `{"type":"string","logicalType":"uuid"}`
+	buf := bytes.NewBuffer([]byte{})
+	enc, err := avro.NewEncoder(schema, buf)
+	assert.NoError(t, err)
+
+	err = enc.Encode(big.NewRat(1734, 5))
+
+	assert.Error(t, err)
+}
diff --git a/encoder_test.go b/encoder_test.go
index e14fc65..cb2f8b9 100644
--- a/encoder_test.go
+++ b/encoder_test.go
@@ -20,7 +20,7 @@ func TestNewEncoder_SchemaError(t *testing.T) {
 func TestEncoder_EncodeUnsupportedType(t *testing.T) {
 	defer ConfigTeardown()
 
-	schema := avro.NewPrimitiveSchema(avro.Type("test"))
+	schema := avro.NewPrimitiveSchema(avro.Type("test"), nil)
 	buf := bytes.NewBuffer([]byte{})
 	enc := avro.NewEncoderForSchema(schema, buf)
 
diff --git a/protocol_test.go b/protocol_test.go
index f47c715..7c6f8df 100644
--- a/protocol_test.go
+++ b/protocol_test.go
@@ -26,11 +26,11 @@ func TestNewProtocol_ValidatesName(t *testing.T) {
 }
 
 func TestNewMessage(t *testing.T) {
-	field, _ := avro.NewField("test", avro.NewPrimitiveSchema(avro.String), nil)
+	field, _ := avro.NewField("test", avro.NewPrimitiveSchema(avro.String, nil), nil)
 	fields := []*avro.Field{field}
 	req, _ := avro.NewRecordSchema("test", "", fields)
-	resp := avro.NewPrimitiveSchema(avro.String)
-	types := []avro.Schema{avro.NewPrimitiveSchema(avro.String)}
+	resp := avro.NewPrimitiveSchema(avro.String, nil)
+	types := []avro.Schema{avro.NewPrimitiveSchema(avro.String, nil)}
 	errs, _ := avro.NewUnionSchema(types)
 
 	msg := avro.NewMessage(req, resp, errs, false)
diff --git a/reader_generic_test.go b/reader_generic_test.go
index 8940445..5d165ec 100644
--- a/reader_generic_test.go
+++ b/reader_generic_test.go
@@ -2,7 +2,9 @@ package avro_test
 
 import (
 	"bytes"
+	"math/big"
 	"testing"
+	"time"
 
 	"github.com/hamba/avro"
 	"github.com/stretchr/testify/assert"
@@ -31,6 +33,20 @@ func TestReader_ReadNext(t *testing.T) {
 			want:    27,
 			wantErr: false,
 		},
+		{
+			name:    "Int Date",
+			data:    []byte{0xAE, 0x9D, 0x02},
+			schema:  `{"type":"int","logicalType":"date"}`,
+			want:    time.Date(2020, 1, 2, 0, 0, 0, 0, time.UTC),
+			wantErr: false,
+		},
+		{
+			name:    "Int Time-Millis",
+			data:    []byte{0xAA, 0xB4, 0xDE, 0x75},
+			schema:  `{"type":"int","logicalType":"time-millis"}`,
+			want:    123456789 * time.Millisecond,
+			wantErr: false,
+		},
 		{
 			name:    "Long",
 			data:    []byte{0x36},
@@ -38,6 +54,27 @@ func TestReader_ReadNext(t *testing.T) {
 			want:    int64(27),
 			wantErr: false,
 		},
+		{
+			name:    "Long Time-Micros",
+			data:    []byte{0x86, 0xEA, 0xC8, 0xE9, 0x97, 0x07},
+			schema:  `{"type":"long","logicalType":"time-micros"}`,
+			want:    123456789123 * time.Microsecond,
+			wantErr: false,
+		},
+		{
+			name:    "Long Timestamp-Millis",
+			data:    []byte{0x90, 0xB2, 0xAE, 0xC3, 0xEC, 0x5B},
+			schema:  `{"type":"long","logicalType":"timestamp-millis"}`,
+			want:    time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC),
+			wantErr: false,
+		},
+		{
+			name:    "Long Timestamp-Micros",
+			data:    []byte{0x80, 0xCD, 0xB7, 0xA2, 0xEE, 0xC7, 0xCD, 0x05},
+			schema:  `{"type":"long","logicalType":"timestamp-micros"}`,
+			want:    time.Date(2020, 1, 2, 3, 4, 5, 0, time.UTC),
+			wantErr: false,
+		},
 		{
 			name:    "Float",
 			data:    []byte{0x33, 0x33, 0x93, 0x3F},
@@ -66,6 +103,13 @@ func TestReader_ReadNext(t *testing.T) {
 			want:    []byte{0xEC, 0xAB, 0x44, 0x00},
 			wantErr: false,
 		},
+		{
+			name:    "Bytes Decimal",
+			data:    []byte{0x6, 0x00, 0x87, 0x78},
+			schema:  `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`,
+			want:    big.NewRat(1734, 5),
+			wantErr: false,
+		},
 		{
 			name:    "Record",
 			data:    []byte{0x36, 0x06, 0x66, 0x6f, 0x6f},
@@ -143,6 +187,13 @@ func TestReader_ReadNext(t *testing.T) {
 			want:    []byte{'f', 'o', 'o', 'f', 'o', 'o'},
 			wantErr: false,
 		},
+		{
+			name:    "Fixed Decimal",
+			data:    []byte{0x00, 0x00, 0x00, 0x00, 0x87, 0x78},
+			schema:  `{"type":"fixed", "name": "test", "size": 6,"logicalType":"decimal","precision":4,"scale":2}`,
+			want:    big.NewRat(1734, 5),
+			wantErr: false,
+		},
 	}
 
 	for _, tt := range tests {
@@ -164,7 +215,7 @@ func TestReader_ReadNext(t *testing.T) {
 }
 
 func TestReader_ReadNextUnsupportedType(t *testing.T) {
-	schema := avro.NewPrimitiveSchema(avro.Type("test"))
+	schema := avro.NewPrimitiveSchema(avro.Type("test"), nil)
 	r := avro.NewReader(bytes.NewReader([]byte{0x01}), 10)
 
 	_ = r.ReadNext(schema)
diff --git a/schema_canonical_test.go b/schema_canonical_test.go
index 3da3e0a..9f1c9cf 100644
--- a/schema_canonical_test.go
+++ b/schema_canonical_test.go
@@ -39,6 +39,18 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `{"type":"int"}`,
 			canonical: `"int"`,
 		},
+		{
+			input:     `{"type":"int","logicalType":"date"}`,
+			canonical: `{"type":"int","logicalType":"date"}`,
+		},
+		{
+			input:     `{"type":"int","logicalType":"time-millis"}`,
+			canonical: `{"type":"int","logicalType":"time-millis"}`,
+		},
+		{
+			input:     `{"type":"int"}`,
+			canonical: `"int"`,
+		},
 		{
 			input:     `"long"`,
 			canonical: `"long"`,
@@ -47,6 +59,18 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `{"type":"long"}`,
 			canonical: `"long"`,
 		},
+		{
+			input:     `{"type":"long","logicalType":"time-micros"}`,
+			canonical: `{"type":"long","logicalType":"time-micros"}`,
+		},
+		{
+			input:     `{"type":"long","logicalType":"timestamp-millis"}`,
+			canonical: `{"type":"long","logicalType":"timestamp-millis"}`,
+		},
+		{
+			input:     `{"type":"long","logicalType":"timestamp-millis"}`,
+			canonical: `{"type":"long","logicalType":"timestamp-millis"}`,
+		},
 		{
 			input:     `"float"`,
 			canonical: `"float"`,
@@ -71,6 +95,14 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `{"type":"bytes"}`,
 			canonical: `"bytes"`,
 		},
+		{
+			input:     `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`,
+			canonical: `{"type":"bytes","logicalType":"decimal","precision":4,"scale":2}`,
+		},
+		{
+			input:     `{"type":"bytes","logicalType":"decimal","precision":4,"scale":0}`,
+			canonical: `{"type":"bytes","logicalType":"decimal","precision":4}`,
+		},
 		{
 			input:     `"string"`,
 			canonical: `"string"`,
@@ -79,6 +111,10 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `{"type":"string"}`,
 			canonical: `"string"`,
 		},
+		{
+			input:     `{"type":"string","logicalType":"uuid"}`,
+			canonical: `{"type":"string","logicalType":"uuid"}`,
+		},
 		{
 			input:     `[  ]`,
 			canonical: `[]`,
@@ -143,6 +179,18 @@ func TestSchema_Canonical(t *testing.T) {
 			input:     `{"name":"foo","type":"fixed","size":15}`,
 			canonical: `{"name":"foo","type":"fixed","size":15}`,
 		},
+		{
+			input:     `{"name":"foo","type":"fixed","logicalType":"duration","size":12}`,
+			canonical: `{"name":"foo","type":"fixed","size":12,"logicalType":"duration"}`,
+		},
+		{
+			input:     `{"name":"foo","type":"fixed","logicalType":"decimal","size":12,"precision":4,"scale":2}`,
+			canonical: `{"name":"foo","type":"fixed","size":12,"logicalType":"decimal","precision":4,"scale":2}`,
+		},
+		{
+			input:     `{"name":"foo","type":"fixed","logicalType":"decimal","size":12,"precision":4,"scale":0}`,
+			canonical: `{"name":"foo","type":"fixed","size":12,"logicalType":"decimal","precision":4}`,
+		},
 		{
 			input:     `{"namespace":"x.y.z", "type":"fixed", "name":"foo", "doc":"foo bar", "size":32}`,
 			canonical: `{"name":"x.y.z.foo","type":"fixed","size":32}`,
diff --git a/schema_internal_test.go b/schema_internal_test.go
index 1071bb5..a5b79b3 100644
--- a/schema_internal_test.go
+++ b/schema_internal_test.go
@@ -98,7 +98,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "String",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(String)
+				return NewPrimitiveSchema(String, nil)
 			},
 			def:    "test",
 			want:   "test",
@@ -107,7 +107,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "String Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(String)
+				return NewPrimitiveSchema(String, nil)
 			},
 			def:    1,
 			wantOk: false,
@@ -115,7 +115,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Bytes",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Bytes)
+				return NewPrimitiveSchema(Bytes, nil)
 			},
 			def:    "test",
 			want:   "test",
@@ -124,7 +124,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Bytes Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Bytes)
+				return NewPrimitiveSchema(Bytes, nil)
 			},
 			def:    1,
 			wantOk: false,
@@ -151,7 +151,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Fixed",
 			schemaFn: func() Schema {
-				s, _ := NewFixedSchema("foo", "", 1)
+				s, _ := NewFixedSchema("foo", "", 1, nil)
 				return s
 			},
 			def:    "test",
@@ -161,7 +161,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Fixed Invalid Type",
 			schemaFn: func() Schema {
-				s, _ := NewFixedSchema("foo", "", 1)
+				s, _ := NewFixedSchema("foo", "", 1, nil)
 				return s
 			},
 			def:    1,
@@ -170,7 +170,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Boolean",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Boolean)
+				return NewPrimitiveSchema(Boolean, nil)
 			},
 			def:    true,
 			want:   true,
@@ -179,7 +179,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Boolean Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Boolean)
+				return NewPrimitiveSchema(Boolean, nil)
 			},
 			def:    1,
 			wantOk: false,
@@ -187,7 +187,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    1,
 			want:   1,
@@ -196,7 +196,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int Int8",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    int8(1),
 			want:   1,
@@ -205,7 +205,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int Int16",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    int16(1),
 			want:   1,
@@ -214,7 +214,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int Int32",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    int32(1),
 			want:   1,
@@ -223,7 +223,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int Float64",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    float64(1),
 			want:   1,
@@ -232,7 +232,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Int Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Int)
+				return NewPrimitiveSchema(Int, nil)
 			},
 			def:    "test",
 			wantOk: false,
@@ -240,7 +240,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Long",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Long)
+				return NewPrimitiveSchema(Long, nil)
 			},
 			def:    int64(1),
 			want:   int64(1),
@@ -249,7 +249,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Long Float64",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Long)
+				return NewPrimitiveSchema(Long, nil)
 			},
 			def:    float64(1),
 			want:   int64(1),
@@ -258,7 +258,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Long Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Long)
+				return NewPrimitiveSchema(Long, nil)
 			},
 			def:    "test",
 			wantOk: false,
@@ -266,7 +266,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Float",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Float)
+				return NewPrimitiveSchema(Float, nil)
 			},
 			def:    float32(1),
 			want:   float32(1),
@@ -275,7 +275,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Float Float64",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Float)
+				return NewPrimitiveSchema(Float, nil)
 			},
 			def:    float64(1),
 			want:   float32(1),
@@ -284,7 +284,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Float Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Float)
+				return NewPrimitiveSchema(Float, nil)
 			},
 			def:    "test",
 			wantOk: false,
@@ -292,7 +292,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Double",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Double)
+				return NewPrimitiveSchema(Double, nil)
 			},
 			def:    float64(1),
 			want:   float64(1),
@@ -301,7 +301,7 @@ func TestIsValidDefault(t *testing.T) {
 		{
 			name: "Double Invalid Type",
 			schemaFn: func() Schema {
-				return NewPrimitiveSchema(Double)
+				return NewPrimitiveSchema(Double, nil)
 			},
 			def:    "test",
 			wantOk: false,
@@ -327,7 +327,7 @@ func TestRecursionError_Error(t *testing.T) {
 }
 
 func TestSchema_FingerprintUsingCaches(t *testing.T) {
-	schema := NewPrimitiveSchema(String)
+	schema := NewPrimitiveSchema(String, nil)
 
 	want, _ := schema.FingerprintUsing(CRC64Avro)
 
diff --git a/schema_test.go b/schema_test.go
index c67ad05..8b51ac7 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -774,6 +774,199 @@ func TestFixedSchema_HandlesProps(t *testing.T) {
 	assert.Equal(t, "bar", s.(*avro.FixedSchema).Prop("foo"))
 }
 
+func TestSchema_LogicalTypes(t *testing.T) {
+	tests := []struct {
+		name            string
+		schema          string
+		wantType        avro.Type
+		wantLogical     bool
+		wantLogicalType avro.LogicalType
+		assertFn        func(t *testing.T, ls avro.LogicalSchema)
+	}{
+		{
+			name:        "Invalid",
+			schema:      `{"type": "int", "logicalType": "test"}`,
+			wantType:    avro.Int,
+			wantLogical: false,
+		},
+		{
+			name:            "Date",
+			schema:          `{"type": "int", "logicalType": "date"}`,
+			wantType:        avro.Int,
+			wantLogical:     true,
+			wantLogicalType: avro.Date,
+		},
+		{
+			name:            "Time Millis",
+			schema:          `{"type": "int", "logicalType": "time-millis"}`,
+			wantType:        avro.Int,
+			wantLogical:     true,
+			wantLogicalType: avro.TimeMillis,
+		},
+		{
+			name:            "Time Micros",
+			schema:          `{"type": "long", "logicalType": "time-micros"}`,
+			wantType:        avro.Long,
+			wantLogical:     true,
+			wantLogicalType: avro.TimeMicros,
+		},
+		{
+			name:            "Timestamp Millis",
+			schema:          `{"type": "long", "logicalType": "timestamp-millis"}`,
+			wantType:        avro.Long,
+			wantLogical:     true,
+			wantLogicalType: avro.TimestampMillis,
+		},
+		{
+			name:            "Timestamp Micros",
+			schema:          `{"type": "long", "logicalType": "timestamp-micros"}`,
+			wantType:        avro.Long,
+			wantLogical:     true,
+			wantLogicalType: avro.TimestampMicros,
+		},
+		{
+			name:            "UUID",
+			schema:          `{"type": "string", "logicalType": "uuid"}`,
+			wantType:        avro.String,
+			wantLogical:     true,
+			wantLogicalType: avro.UUID,
+		},
+		{
+			name:            "Duration",
+			schema:          `{"type": "fixed", "name":"test", "size": 12, "logicalType": "duration"}`,
+			wantType:        avro.Fixed,
+			wantLogical:     true,
+			wantLogicalType: avro.Duration,
+		},
+		{
+			name:        "Invalid Duration",
+			schema:      `{"type": "fixed", "name":"test", "size": 11, "logicalType": "duration"}`,
+			wantType:    avro.Fixed,
+			wantLogical: false,
+		},
+		{
+			name:            "Bytes Decimal",
+			schema:          `{"type": "bytes", "logicalType": "decimal", "precision": 4, "scale": 2}`,
+			wantType:        avro.Bytes,
+			wantLogical:     true,
+			wantLogicalType: avro.Decimal,
+			assertFn: func(t *testing.T, ls avro.LogicalSchema) {
+				dec, ok := ls.(*avro.DecimalLogicalSchema)
+				if assert.True(t, ok) {
+					assert.Equal(t, 4, dec.Precision())
+					assert.Equal(t, 2, dec.Scale())
+				}
+			},
+		},
+		{
+			name:            "Bytes Decimal No Scale",
+			schema:          `{"type": "bytes", "logicalType": "decimal", "precision": 4}`,
+			wantType:        avro.Bytes,
+			wantLogical:     true,
+			wantLogicalType: avro.Decimal,
+			assertFn: func(t *testing.T, ls avro.LogicalSchema) {
+				dec, ok := ls.(*avro.DecimalLogicalSchema)
+				if assert.True(t, ok) {
+					assert.Equal(t, 4, dec.Precision())
+					assert.Equal(t, 0, dec.Scale())
+				}
+			},
+		},
+		{
+			name:            "Bytes Decimal Negative Precision",
+			schema:          `{"type": "bytes", "logicalType": "decimal", "precision": 0}`,
+			wantType:        avro.Bytes,
+			wantLogical:     false,
+		},
+		{
+			name:            "Bytes Decimal Negative Scale",
+			schema:          `{"type": "bytes", "logicalType": "decimal", "precision": 1, "scale": -1}`,
+			wantType:        avro.Bytes,
+			wantLogical:     false,
+		},
+		{
+			name:            "Bytes Decimal Scale Larger Than Precision",
+			schema:          `{"type": "bytes", "logicalType": "decimal", "precision": 4, "scale": 6}`,
+			wantType:        avro.Bytes,
+			wantLogical:     false,
+		},
+		{
+			name:            "Fixed Decimal",
+			schema:          `{"type": "fixed", "name":"test", "size": 12, "logicalType": "decimal", "precision": 4, "scale": 2}`,
+			wantType:        avro.Fixed,
+			wantLogical:     true,
+			wantLogicalType: avro.Decimal,
+			assertFn: func(t *testing.T, ls avro.LogicalSchema) {
+				dec, ok := ls.(*avro.DecimalLogicalSchema)
+				if assert.True(t, ok) {
+					assert.Equal(t, 4, dec.Precision())
+					assert.Equal(t, 2, dec.Scale())
+				}
+			},
+		},
+		{
+			name:            "Fixed Decimal No Scale",
+			schema:          `{"type": "fixed", "name":"test", "size": 12, "logicalType": "decimal", "precision": 4}`,
+			wantType:        avro.Fixed,
+			wantLogical:     true,
+			wantLogicalType: avro.Decimal,
+			assertFn: func(t *testing.T, ls avro.LogicalSchema) {
+				dec, ok := ls.(*avro.DecimalLogicalSchema)
+				if assert.True(t, ok) {
+					assert.Equal(t, 4, dec.Precision())
+					assert.Equal(t, 0, dec.Scale())
+				}
+			},
+		},
+		{
+			name:            "Fixed Decimal Negative Precision",
+			schema:          `{"type": "fixed", "name":"test", "size": 12, "logicalType": "decimal", "precision": 0}`,
+			wantType:        avro.Fixed,
+			wantLogical:     false,
+		},
+		{
+			name:            "Fixed Decimal Precision Too Large",
+			schema:          `{"type": "fixed", "name":"test", "size": 4, "logicalType": "decimal", "precision": 10}`,
+			wantType:        avro.Fixed,
+			wantLogical:     false,
+		},
+		{
+			name:            "Fixed Decimal Scale Larger Than Precision",
+			schema:          `{"type": "fixed", "name":"test", "size": 12, "logicalType": "decimal", "precision": 4, "scale": 6}`,
+			wantType:        avro.Fixed,
+			wantLogical:     false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			schema, err := avro.Parse(tt.schema)
+
+			assert.NoError(t, err)
+			assert.Equal(t, tt.wantType, schema.Type())
+
+			lts, ok := schema.(avro.LogicalTypeSchema)
+			if !ok {
+				assert.Fail(t, "logical type schema expected")
+				return
+			}
+
+			ls := lts.Logical()
+			if assert.Equal(t, tt.wantLogical, ls != nil) {
+				if !tt.wantLogical {
+					return
+				}
+
+				assert.Equal(t, tt.wantLogicalType, ls.Type())
+
+				if tt.assertFn != nil {
+					tt.assertFn(t, ls)
+				}
+			}
+		})
+	}
+}
+
 func TestSchema_FingerprintUsing(t *testing.T) {
 	tests := []struct {
 		name   string