Repository: hamba/avro
Pull request #25 (2019-08-09 13:36:29)

Fingerprint Using


Patch:
diff --git a/pkg/crc64/crc64.go b/pkg/crc64/crc64.go
new file mode 100644
index 0000000..f97522d
--- /dev/null
+++ b/pkg/crc64/crc64.go
@@ -0,0 +1,93 @@
+// Package crc64 implements the Avro CRC-64 checksum.
+// See https://avro.apache.org/docs/current/spec.html#schema_fingerprints for information.
+package crc64
+
+import (
+	"hash"
+	"sync"
+)
+
+// Size is the of a CRC-64 checksum in bytes.
+const Size = 8
+
+// Empty is the empty checksum.
+const Empty = 0xc15d213aa4d7a795
+
+// Table is a 256-word table representing the polynomial for efficient processing.
+type Table [256]uint64
+
+func makeTable() *Table {
+	t := new(Table)
+	for i := 0; i < 256; i++ {
+		fp := uint64(i)
+		for j := 0; j < 8; j++ {
+			fp = (fp >> 1) ^ (Empty & -(fp & 1))
+		}
+		t[i] = fp
+	}
+	return t
+}
+
+var (
+	tableBuildOnce sync.Once
+	crc64Table     *Table
+)
+
+func buildTableOnce() {
+	tableBuildOnce.Do(buildTable)
+}
+
+func buildTable() {
+	crc64Table = makeTable()
+}
+
+type digest struct {
+	crc uint64
+	tab *Table
+}
+
+// New creates a new hash.Hash64 computing the Avro CRC-64 checksum.
+// Its Sum method will lay the value out in big-endian byte order.
+func New() hash.Hash64 {
+	buildTableOnce()
+
+	return &digest{
+		crc: Empty,
+		tab: crc64Table,
+	}
+}
+
+// Size returns the bytes size of the checksum.
+func (d *digest) Size() int {
+	return Size
+}
+
+// BlockSize returns the block size of the checksum.
+func (d *digest) BlockSize() int {
+	return 1
+}
+
+// Reset resets the hash instance.
+func (d *digest) Reset() {
+	d.crc = Empty
+}
+
+// Write accumulatively adds the given data to the checksum.
+func (d *digest) Write(p []byte) (n int, err error) {
+	for i := 0; i < len(p); i++ {
+		d.crc = (d.crc >> 8) ^ d.tab[(int)(byte(d.crc)^p[i])&0xff]
+	}
+
+	return len(p), nil
+}
+
+// Sum64 returns the checksum as a uint64.
+func (d *digest) Sum64() uint64 {
+	return d.crc
+}
+
+// Sum returns the checksum as a byte slice, using the given byte slice.
+func (d *digest) Sum(in []byte) []byte {
+	s := d.Sum64()
+	return append(in, byte(s>>56), byte(s>>48), byte(s>>40), byte(s>>32), byte(s>>24), byte(s>>16), byte(s>>8), byte(s))
+}
diff --git a/schema.go b/schema.go
index 4780d11..18828de 100644
--- a/schema.go
+++ b/schema.go
@@ -1,12 +1,15 @@
 package avro
 
 import (
+	"crypto/md5"
 	"crypto/sha256"
 	"errors"
 	"fmt"
+	"hash"
 	"strconv"
 	"strings"
 
+	"github.com/hamba/avro/pkg/crc64"
 	"github.com/modern-go/concurrent"
 )
 
@@ -36,6 +39,22 @@ const (
 	Null    Type = "null"
 )
 
+// FingerprintType is a fingerprinting algorithm.
+type FingerprintType string
+
+// Fingerprint type constants.
+const (
+	CRC64Avro FingerprintType = "CRC64-AVRO"
+	MD5       FingerprintType = "MD5"
+	SHA256    FingerprintType = "SHA256"
+)
+
+var fingerprinters = map[FingerprintType]hash.Hash{
+	CRC64Avro: crc64.New(),
+	MD5:       md5.New(),
+	SHA256:    sha256.New(),
+}
+
 // SchemaCache is a cache of schemas.
 type SchemaCache struct {
 	cache concurrent.Map // map[string]Schema
@@ -83,6 +102,9 @@ type Schema interface {
 
 	// Fingerprint returns the SHA256 fingerprint of the schema.
 	Fingerprint() [32]byte
+
+	// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+	FingerprintUsing(FingerprintType) ([]byte, error)
 }
 
 // PropertySchema represents a schema with properties.
@@ -158,6 +180,7 @@ func (n name) FullName() string {
 
 type fingerprinter struct {
 	fingerprint [32]byte
+	cache       map[FingerprintType][]byte
 }
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
@@ -170,6 +193,28 @@ func (f *fingerprinter) Fingerprint(stringer fmt.Stringer) [32]byte {
 	return f.fingerprint
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (f *fingerprinter) FingerprintUsing(typ FingerprintType, stringer fmt.Stringer) ([]byte, error) {
+	if f.cache == nil {
+		f.cache = map[FingerprintType][]byte{}
+	}
+
+	if b, ok := f.cache[typ]; ok {
+		return b, nil
+	}
+
+	h, ok := fingerprinters[typ]
+	if !ok {
+		return nil, fmt.Errorf("avro: unknown fingerprint alogrithm %s", f)
+	}
+
+	h.Reset()
+	h.Write([]byte(stringer.String()))
+	fingerprint := h.Sum(make([]byte, 0, h.Size()))
+	f.cache[typ] = fingerprint
+	return fingerprint, nil
+}
+
 type properties struct {
 	reserved []string
 	props    map[string]interface{}
@@ -237,6 +282,11 @@ func (s *PrimitiveSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *PrimitiveSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // RecordSchema is an Avro record type schema.
 type RecordSchema struct {
 	name
@@ -314,6 +364,11 @@ func (s *RecordSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *RecordSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // Field is an Avro record type field.
 type Field struct {
 	properties
@@ -433,6 +488,11 @@ func (s *EnumSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *EnumSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // ArraySchema is an Avro array type schema.
 type ArraySchema struct {
 	properties
@@ -469,6 +529,11 @@ func (s *ArraySchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *ArraySchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // MapSchema is an Avro map type schema.
 type MapSchema struct {
 	properties
@@ -495,14 +560,19 @@ func (s *MapSchema) Values() Schema {
 	return s.values
 }
 
+// String returns the canonical form of the schema.
+func (s *MapSchema) String() string {
+	return `{"type":"map","values":` + s.values.String() + `}`
+}
+
 // Fingerprint returns the SHA256 fingerprint of the schema.
 func (s *MapSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
-// String returns the canonical form of the schema.
-func (s *MapSchema) String() string {
-	return `{"type":"map","values":` + s.values.String() + `}`
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *MapSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
 }
 
 // UnionSchema is an Avro union type schema.
@@ -570,6 +640,11 @@ func (s *UnionSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *UnionSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // FixedSchema is an Avro fixed type schema.
 type FixedSchema struct {
 	name
@@ -614,8 +689,15 @@ func (s *FixedSchema) Fingerprint() [32]byte {
 	return s.fingerprinter.Fingerprint(s)
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *FixedSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
+}
+
 // NullSchema is an Avro null type schema.
-type NullSchema struct{}
+type NullSchema struct {
+	fingerprinter
+}
 
 // Type returns the type of the schema.
 func (s *NullSchema) Type() Type {
@@ -629,7 +711,12 @@ func (s *NullSchema) String() string {
 
 // Fingerprint returns the SHA256 fingerprint of the schema.
 func (s *NullSchema) Fingerprint() [32]byte {
-	return [32]byte{0xf0, 0x72, 0xcb, 0xec, 0x3b, 0xf8, 0x84, 0x18, 0x71, 0xd4, 0x28, 0x42, 0x30, 0xc5, 0xe9, 0x83, 0xdc, 0x21, 0x1a, 0x56, 0x83, 0x7a, 0xed, 0x86, 0x24, 0x87, 0x14, 0x8f, 0x94, 0x7d, 0x1a, 0x1f}
+	return s.fingerprinter.Fingerprint(s)
+}
+
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *NullSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.fingerprinter.FingerprintUsing(typ, s)
 }
 
 // RefSchema is a reference to a named Avro schema.
@@ -664,6 +751,11 @@ func (s *RefSchema) Fingerprint() [32]byte {
 	return s.actual.Fingerprint()
 }
 
+// FingerprintUsing returns the fingerprint of the schema using the given algorithm or an error.
+func (s *RefSchema) FingerprintUsing(typ FingerprintType) ([]byte, error) {
+	return s.actual.FingerprintUsing(typ)
+}
+
 func invalidNameFirstChar(r rune) bool {
 	return (r < 'A' || r > 'Z') && (r < 'a' || r > 'z') && r != '_'
 }


Test patch:
diff --git a/pkg/crc64/crc64_test.go b/pkg/crc64/crc64_test.go
new file mode 100644
index 0000000..0f2c033
--- /dev/null
+++ b/pkg/crc64/crc64_test.go
@@ -0,0 +1,115 @@
+package crc64
+
+import (
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestGolden(t *testing.T) {
+	tests := []struct {
+		in   string
+		want uint64
+	}{
+		{
+			in:   `"null"`,
+			want: 7195948357588979594,
+		},
+		{
+			in:   `{"name":"foo","type":"fixed","size":15}`,
+			want: 1756455273707447556,
+		},
+		{
+			in:   `{"name":"foo","type":"record","fields":[{"name":"f1","type":"boolean"}]}`,
+			want: 7843277075252814651,
+		},
+	}
+
+	hash := New()
+
+	for i, tt := range tests {
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			hash.Reset()
+			hash.Write([]byte(tt.in))
+
+			got := hash.Sum64()
+
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestGoldenBytes(t *testing.T) {
+	tests := []struct {
+		in   string
+		want []byte
+	}{
+		{
+			in:   `"null"`,
+			want: []byte{0x63, 0xdd, 0x24, 0xe7, 0xcc, 0x25, 0x8f, 0x8a},
+		},
+		{
+			in:   `{"name":"foo","type":"fixed","size":15}`,
+			want: []byte{0x18, 0x60, 0x2e, 0xc3, 0xed, 0x31, 0xa5, 0x4},
+		},
+		{
+			in:   `{"name":"foo","type":"record","fields":[{"name":"f1","type":"boolean"}]}`,
+			want: []byte{0x6c, 0xd8, 0xea, 0xf1, 0xc9, 0x68, 0xa3, 0x3b},
+		},
+	}
+
+	hash := New()
+
+	for i, tt := range tests {
+		t.Run(strconv.Itoa(i), func(t *testing.T) {
+			hash.Reset()
+			hash.Write([]byte(tt.in))
+
+			got := make([]byte, 0, hash.Size())
+			got = hash.Sum(got)
+
+			assert.Equal(t, tt.want, got)
+		})
+	}
+}
+
+func TestDigest_BlockSize(t *testing.T) {
+	hash := New()
+
+	assert.Equal(t, 1, hash.BlockSize())
+}
+
+func bench(b *testing.B, size int64) {
+	b.SetBytes(size)
+
+	h := New()
+	in := make([]byte, 0, h.Size())
+
+	data := make([]byte, size)
+	for i := range data {
+		data[i] = byte(i)
+	}
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		h.Reset()
+		_, _ = h.Write(data)
+		h.Sum(in)
+
+		in = in[:0]
+	}
+}
+
+func BenchmarkCrc64(b *testing.B) {
+	b.Run("64KB", func(b *testing.B) {
+		bench(b, 64<<10)
+	})
+	b.Run("4KB", func(b *testing.B) {
+		bench(b, 4<<10)
+	})
+	b.Run("1KB", func(b *testing.B) {
+		bench(b, 1<<10)
+	})
+}
diff --git a/schema_internal_test.go b/schema_internal_test.go
index 8ffbe41..1071bb5 100644
--- a/schema_internal_test.go
+++ b/schema_internal_test.go
@@ -325,3 +325,14 @@ func TestRecursionError_Error(t *testing.T) {
 
 	assert.Equal(t, "", err.Error())
 }
+
+func TestSchema_FingerprintUsingCaches(t *testing.T) {
+	schema := NewPrimitiveSchema(String)
+
+	want, _ := schema.FingerprintUsing(CRC64Avro)
+
+	got, _ := schema.FingerprintUsing(CRC64Avro)
+
+	assert.Equal(t, want, schema.cache[CRC64Avro])
+	assert.Equal(t, want, got)
+}
diff --git a/schema_test.go b/schema_test.go
index 3b74c68..c67ad05 100644
--- a/schema_test.go
+++ b/schema_test.go
@@ -774,6 +774,115 @@ func TestFixedSchema_HandlesProps(t *testing.T) {
 	assert.Equal(t, "bar", s.(*avro.FixedSchema).Prop("foo"))
 }
 
+func TestSchema_FingerprintUsing(t *testing.T) {
+	tests := []struct {
+		name   string
+		schema string
+		typ    avro.FingerprintType
+		want   []byte
+	}{
+
+		{
+			name:   "Null CRC64",
+			schema: "null",
+			typ:    avro.CRC64Avro,
+			want:   []byte{0x63, 0xdd, 0x24, 0xe7, 0xcc, 0x25, 0x8f, 0x8a},
+		},
+		{
+			name:   "Null MD5",
+			schema: "null",
+			typ:    avro.MD5,
+			want:   []byte{0x9b, 0x41, 0xef, 0x67, 0x65, 0x1c, 0x18, 0x48, 0x8a, 0x8b, 0x8, 0xbb, 0x67, 0xc7, 0x56, 0x99},
+		},
+		{
+			name:   "Null SHA256",
+			schema: "null",
+			typ:    avro.SHA256,
+			want:   []byte{0xf0, 0x72, 0xcb, 0xec, 0x3b, 0xf8, 0x84, 0x18, 0x71, 0xd4, 0x28, 0x42, 0x30, 0xc5, 0xe9, 0x83, 0xdc, 0x21, 0x1a, 0x56, 0x83, 0x7a, 0xed, 0x86, 0x24, 0x87, 0x14, 0x8f, 0x94, 0x7d, 0x1a, 0x1f},
+		},
+		{
+			name:   "Primitive CRC64",
+			schema: "string",
+			typ:    avro.CRC64Avro,
+			want:   []byte{0x8f, 0x1, 0x48, 0x72, 0x63, 0x45, 0x3, 0xc7},
+		},
+		{
+			name:   "Record CRC64",
+			schema: `{"type":"record", "name":"test", "namespace": "org.hamba.avro", "doc": "docs", "fields":[{"name": "field", "type": "int"}]}`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0xaf, 0x30, 0x30, 0xf0, 0x1c, 0x99, 0x76, 0xda},
+		},
+		{
+			name:   "Enum CRC64",
+			schema: `{"type":"enum", "name":"test", "namespace": "org.hamba.avro", "symbols":["TEST"]}`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0xc, 0xb0, 0xa2, 0xa6, 0x5f, 0x96, 0x8, 0xd1},
+		},
+		{
+			name:   "Array CRC64",
+			schema: `{"type":"array", "items": "int"}`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0x52, 0x2b, 0x81, 0x4f, 0xc9, 0x63, 0xb4, 0xbe},
+		},
+		{
+			name:   "Map CRC64",
+			schema: `{"type":"map", "values": "int"}`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0xdb, 0x39, 0xe2, 0xc2, 0x53, 0x4c, 0x89, 0x73},
+		},
+		{
+			name:   "Union CRC64",
+			schema: `["null", "int"]`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0xd5, 0x1c, 0xc0, 0x92, 0x2b, 0x46, 0xb1, 0xd7},
+		},
+		{
+			name:   "Fixed CRC64",
+			schema: `{"type":"fixed", "name":"test", "namespace": "org.hamba.avro", "size": 12}`,
+			typ:    avro.CRC64Avro,
+			want:   []byte{0x1, 0x7c, 0x1f, 0x7f, 0xa7, 0x6d, 0xa0, 0xa1},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			schema := avro.MustParse(tt.schema)
+			got, err := schema.FingerprintUsing(tt.typ)
+
+			if assert.NoError(t, err) {
+				assert.Equal(t, tt.want, got)
+			}
+		})
+	}
+}
+
+func TestSchema_FingerprintUsingReference(t *testing.T) {
+	schema := avro.MustParse(`
+{
+   "type": "record",
+   "name": "valid_name",
+   "namespace": "org.hamba.avro",
+   "fields": [
+       {"name": "intField", "type": "int"},
+       {"name": "Ref", "type": "valid_name"}
+   ]
+}
+`)
+
+	got, err := schema.(*avro.RecordSchema).Fields()[1].Type().FingerprintUsing(avro.CRC64Avro)
+
+	assert.NoError(t, err)
+	assert.Equal(t, []byte{0xe1, 0xd6, 0x1e, 0x7c, 0x2f, 0xe3, 0x3c, 0x2b}, got)
+}
+
+func TestSchema_FingerprintUsingInvalidType(t *testing.T) {
+	schema := avro.MustParse("string")
+
+	_, err := schema.FingerprintUsing("test")
+
+	assert.Error(t, err)
+}
+
 func TestSchema_Interop(t *testing.T) {
 	schm := `
 {