diff --git a/go.mod b/go.mod index a3b976e77..8dc97cd5a 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,6 @@ require ( github.com/puzpuzpuz/xsync/v3 v3.5.1 github.com/rs/zerolog v1.34.0 github.com/stretchr/testify v1.8.1 - github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc github.com/vmihailenco/msgpack/v5 v5.4.1 ) diff --git a/go.sum b/go.sum index b7d7cbe13..135588a6b 100644 --- a/go.sum +++ b/go.sum @@ -32,8 +32,6 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc h1:9lRDQMhESg+zvGYmW5DyG0UqvY96Bu5QYsTLvCHdrgo= -github.com/tmthrgd/go-hex v0.0.0-20190904060850-447a3041c3bc/go.mod h1:bciPuU6GHm1iF1pBvUfxfsH0Wmnc2VbpgvbI9ZWuIRs= github.com/vmihailenco/msgpack/v5 v5.4.1 h1:cQriyiUvjTwOHg8QZaPihLWeRAAVoCpE00IUPn0Bjt8= github.com/vmihailenco/msgpack/v5 v5.4.1/go.mod h1:GaZTsDaehaPpQVyxrf5mtQlH+pc21PIudVV/E3rRQok= github.com/vmihailenco/tagparser/v2 v2.0.0 h1:y09buUbR+b5aycVFQs/g70pqKVZNBmxwAhO7/IwNM9g= diff --git a/internal/dbtest/msgpack_test.go b/internal/dbtest/msgpack_test.go new file mode 100644 index 000000000..486d098b5 --- /dev/null +++ b/internal/dbtest/msgpack_test.go @@ -0,0 +1,40 @@ +package dbtest_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/require" + "github.com/uptrace/bun" +) + +// TestMsgpackRoundTrip covers the reported symptom directly: a msgpack column is +// written as a byte literal, and each dialect spells those differently. Writing +// PostgreSQL's '\x..' form everywhere stored the value as text on the others, +// where it could no longer be decoded. +func TestMsgpackRoundTrip(t *testing.T) { + type Item struct { + Something int `msgpack:"something"` + } + + type Model struct { + bun.BaseModel `bun:"table:msgpack_models"` + + ID int64 `bun:",pk,autoincrement"` + Encoded Item `bun:",msgpack"` + } + + testEachDB(t, func(t *testing.T, dbName string, db *bun.DB) { + ctx := context.Background() + mustResetModel(t, ctx, db, (*Model)(nil)) + + inserted := &Model{Encoded: Item{Something: 1}} + _, err := db.NewInsert().Model(inserted).Exec(ctx) + require.NoError(t, err) + + selected := new(Model) + err = db.NewSelect().Model(selected).Where("id = ?", inserted.ID).Scan(ctx) + require.NoError(t, err) + require.Equal(t, inserted.Encoded, selected.Encoded) + }) +} diff --git a/internal/hex.go b/internal/hex.go deleted file mode 100644 index 6fae2bb78..000000000 --- a/internal/hex.go +++ /dev/null @@ -1,43 +0,0 @@ -package internal - -import ( - fasthex "github.com/tmthrgd/go-hex" -) - -type HexEncoder struct { - b []byte - written bool -} - -func NewHexEncoder(b []byte) *HexEncoder { - return &HexEncoder{ - b: b, - } -} - -func (enc *HexEncoder) Bytes() []byte { - return enc.b -} - -func (enc *HexEncoder) Write(b []byte) (int, error) { - if !enc.written { - enc.b = append(enc.b, '\'') - enc.b = append(enc.b, `\x`...) - enc.written = true - } - - i := len(enc.b) - enc.b = append(enc.b, make([]byte, fasthex.EncodedLen(len(b)))...) - fasthex.Encode(enc.b[i:], b) - - return len(b), nil -} - -func (enc *HexEncoder) Close() error { - if enc.written { - enc.b = append(enc.b, '\'') - } else { - enc.b = append(enc.b, "NULL"...) - } - return nil -} diff --git a/schema/append_value.go b/schema/append_value.go index e1dce123c..c80eca14c 100644 --- a/schema/append_value.go +++ b/schema/append_value.go @@ -1,12 +1,14 @@ package schema import ( + "bytes" "database/sql/driver" "fmt" "net" "reflect" "strconv" "strings" + "sync" "time" "github.com/puzpuzpuz/xsync/v3" @@ -289,22 +291,68 @@ func addrAppender(fn AppenderFunc) AppenderFunc { } } +// maxPooledMsgpackBuffer bounds what the pool keeps alive between queries. +// bytes.Buffer.Reset keeps its backing array, so a writer grown past this by one +// oversized value is dropped instead of retained. +const maxPooledMsgpackBuffer = 64 << 10 + +var msgpackWriterPool = sync.Pool{ + New: func() any { return new(msgpackWriter) }, +} + +// msgpackWriter stages encoder output and records whether the encoder wrote at +// all. A msgpack.CustomEncoder may return successfully without touching the +// writer, which is SQL NULL, and that is a different outcome from an encoder +// that wrote zero bytes, which is an empty value. Buffer emptiness cannot tell +// the two apart, so the bit is tracked explicitly. +type msgpackWriter struct { + buf bytes.Buffer + written bool +} + +func (w *msgpackWriter) Write(b []byte) (int, error) { + w.written = true + return w.buf.Write(b) +} + +func (w *msgpackWriter) reset() { + w.buf.Reset() + w.written = false +} + func appendMsgpack(gen QueryGen, b []byte, v reflect.Value) []byte { - hexEnc := internal.NewHexEncoder(b) + // The dialect needs the whole encoded value to wrap it in its own byte + // literal syntax, so the msgpack output is staged rather than streamed + // straight into the query. + w := msgpackWriterPool.Get().(*msgpackWriter) + defer func() { + if w.buf.Cap() <= maxPooledMsgpackBuffer { + w.reset() + msgpackWriterPool.Put(w) + } + }() enc := msgpack.GetEncoder() defer msgpack.PutEncoder(enc) - enc.Reset(hexEnc) + enc.Reset(w) if err := enc.EncodeValue(v); err != nil { return dialect.AppendError(b, err) } - if err := hexEnc.Close(); err != nil { - return dialect.AppendError(b, err) + if !w.written { + return dialect.AppendNull(b) + } + + // The encoder did write, so this is a value even when it wrote no bytes. + // Bytes() is still nil after a zero-length write, and AppendBytes maps nil + // to NULL, so spell the empty case explicitly. + bs := w.buf.Bytes() + if bs == nil { + bs = []byte{} } - return hexEnc.Bytes() + return gen.Dialect().AppendBytes(b, bs) } func AppendQueryAppender(gen QueryGen, b []byte, app QueryAppender) []byte { diff --git a/schema/appendmsgpack_test.go b/schema/appendmsgpack_test.go new file mode 100644 index 000000000..1e6846215 --- /dev/null +++ b/schema/appendmsgpack_test.go @@ -0,0 +1,224 @@ +package schema + +import ( + "errors" + "reflect" + "testing" + + "github.com/vmihailenco/msgpack/v5" +) + +// blobDialect stands in for any dialect whose byte literals are not spelled the +// PostgreSQL way. SQLite and MySQL use X'..' and MSSQL uses 0x.., so a msgpack +// column must be emitted through Dialect.AppendBytes rather than with a +// hard-coded syntax. It also mirrors the real dialects' nil => NULL rule, so a +// value that must never be NULL cannot pass here by accident. +type blobDialect struct { + *nopDialect +} + +func (blobDialect) AppendBytes(b, bs []byte) []byte { + if bs == nil { + return append(b, "NULL"...) + } + + b = append(b, `X'`...) + const digits = "0123456789abcdef" + for _, c := range bs { + b = append(b, digits[c>>4], digits[c&0x0f]) + } + return append(b, '\'') +} + +type item struct { + Something int `msgpack:"something"` +} + +// emptyMsgpack marshals to no bytes. msgpack still hands the result to the +// writer, so this is an empty value rather than a missing one. +type emptyMsgpack struct{} + +func (emptyMsgpack) MarshalMsgpack() ([]byte, error) { return nil, nil } + +// unwrittenMsgpack encodes successfully without ever invoking the writer, which +// the previous implementation spelled as SQL NULL. +type unwrittenMsgpack struct{} + +func (unwrittenMsgpack) EncodeMsgpack(*msgpack.Encoder) error { return nil } + +// failingMsgpack writes before it fails, so the staging writer is left dirty +// when the encode returns an error. +type failingMsgpack struct{} + +func (failingMsgpack) EncodeMsgpack(enc *msgpack.Encoder) error { + if err := enc.EncodeString("partial output"); err != nil { + return err + } + return errors.New("encode failed") +} + +func TestAppendMsgpack(t *testing.T) { + value := item{Something: 1} + + encoded, err := msgpack.Marshal(value) + if err != nil { + t.Fatalf("msgpack.Marshal: %v", err) + } + + t.Run("uses the dialect's byte literal syntax", func(t *testing.T) { + gen := NewQueryGen(blobDialect{newNopDialect()}) + + got := string(appendMsgpack(gen, nil, reflect.ValueOf(value))) + + want := string(blobDialect{}.AppendBytes(nil, encoded)) + if got != want { + t.Errorf("appendMsgpack = %q, want %q", got, want) + } + }) + + t.Run("appends to the existing query", func(t *testing.T) { + gen := NewQueryGen(blobDialect{newNopDialect()}) + + got := string(appendMsgpack(gen, []byte("VALUES ("), reflect.ValueOf(value))) + + want := "VALUES (" + string(blobDialect{}.AppendBytes(nil, encoded)) + if got != want { + t.Errorf("appendMsgpack = %q, want %q", got, want) + } + }) + + // PostgreSQL is the one dialect the previous hard-coded form happened to + // match, so it is the one that must not move. + t.Run("still emits PostgreSQL syntax for the base dialect", func(t *testing.T) { + gen := NewQueryGen(newNopDialect()) + + got := string(appendMsgpack(gen, nil, reflect.ValueOf(value))) + + want := string(BaseDialect{}.AppendBytes(nil, encoded)) + if got != want { + t.Errorf("appendMsgpack = %q, want %q", got, want) + } + if got[:3] != `'\x` { + t.Errorf("appendMsgpack = %q, want a '\\x.. literal", got) + } + }) + + // An encoder that never writes and one that writes nothing are different + // outcomes: the first has no value, the second has an empty one. SQL NULL and + // an empty blob are not interchangeable for constraints, comparisons, or + // scanning, so the staging writer has to record whether it was written to at + // all rather than inferring it from an empty buffer. + t.Run("distinguishes an unwritten encoder from a zero-length write", func(t *testing.T) { + gen := NewQueryGen(newNopDialect()) + + if got := string(appendMsgpack(gen, nil, reflect.ValueOf(unwrittenMsgpack{}))); got != "NULL" { + t.Errorf("encoder that never wrote = %q, want NULL", got) + } + if got := string(appendMsgpack(gen, nil, reflect.ValueOf(emptyMsgpack{}))); got != `'\x'` { + t.Errorf("marshaler that wrote no bytes = %q, want %q", got, `'\x'`) + } + }) + + // A failing encoder writes before it fails, so its half-written output must + // not reach the query. Comparing the whole result against the caller's prefix + // plus the marker covers that in one assertion: any staged bytes would appear + // between them, hex-encoded and therefore invisible to a substring check. + // What happens to the writer afterwards is covered by TestMsgpackWriterReset, + // which does not depend on pool reuse. + t.Run("a failed encode yields only an error marker", func(t *testing.T) { + gen := NewQueryGen(newNopDialect()) + + got := string(appendMsgpack(gen, []byte("VALUES ("), reflect.ValueOf(failingMsgpack{}))) + + if want := "VALUES (?!(encode failed)"; got != want { + t.Errorf("appendMsgpack = %q, want %q", got, want) + } + }) + + // Zero-length outcomes have to be spelled by the dialect too, or a branch + // that special-cased them could hard-code PostgreSQL output and still pass + // every test above. + t.Run("routes zero-length outcomes through the dialect", func(t *testing.T) { + gen := NewQueryGen(blobDialect{newNopDialect()}) + + if got := string(appendMsgpack(gen, nil, reflect.ValueOf(emptyMsgpack{}))); got != "X''" { + t.Errorf("marshaler that wrote no bytes = %q, want %q", got, "X''") + } + if got := string(appendMsgpack(gen, nil, reflect.ValueOf(unwrittenMsgpack{}))); got != "NULL" { + t.Errorf("encoder that never wrote = %q, want NULL", got) + } + }) + + // The whole path only matters if the struct tag actually routes here. + t.Run("is selected by the msgpack struct tag", func(t *testing.T) { + type tagged struct { + Encoded item `bun:",msgpack"` + } + + d := blobDialect{newNopDialect()} + table := NewTables(d).Get(reflect.TypeOf(tagged{})) + + var field *Field + for _, f := range table.Fields { + if f.GoName == "Encoded" { + field = f + } + } + if field == nil { + t.Fatal("Encoded field not found on the table") + } + + model := tagged{Encoded: value} + got := string(field.AppendValue(NewQueryGen(d), nil, reflect.ValueOf(model))) + + want := string(blobDialect{}.AppendBytes(nil, encoded)) + if got != want { + t.Errorf("AppendValue = %q, want %q", got, want) + } + }) +} + +// TestMsgpackWriterReset checks the writer directly rather than through the +// pool: sync.Pool may hand back a different object, so a test that encodes twice +// and inspects the second result cannot prove anything about reuse. +func TestMsgpackWriterReset(t *testing.T) { + var w msgpackWriter + + if _, err := w.Write([]byte("stale")); err != nil { + t.Fatalf("Write: %v", err) + } + if !w.written { + t.Fatal("Write did not record that the writer was used") + } + + w.reset() + + if w.written { + t.Error("reset left the written flag set") + } + if w.buf.Len() != 0 { + t.Errorf("reset left %d bytes behind", w.buf.Len()) + } +} + +func BenchmarkAppendMsgpack(b *testing.B) { + type payload struct { + ID int + Name string + Tags []string + Bytes []byte + } + value := payload{ + ID: 1234567890, + Name: "representative-msgpack-tagged-column", + Tags: []string{"alpha", "beta", "gamma"}, + Bytes: make([]byte, 512), + } + rv := reflect.ValueOf(value) + gen := NewQueryGen(newNopDialect()) + + b.ReportAllocs() + for i := 0; i < b.N; i++ { + _ = appendMsgpack(gen, make([]byte, 0, 4096), rv) + } +} diff --git a/schema/table.go b/schema/table.go index 71021c718..1dd8205a1 100644 --- a/schema/table.go +++ b/schema/table.go @@ -12,6 +12,7 @@ import ( "github.com/jinzhu/inflection" "github.com/uptrace/bun/dialect/feature" + "github.com/uptrace/bun/dialect/sqltype" "github.com/uptrace/bun/internal" "github.com/uptrace/bun/internal/tagparser" ) @@ -565,6 +566,13 @@ func (t *Table) newField(sf reflect.StructField, tag tagparser.Tag) *Field { field.UserSQLType = s } field.DiscoveredSQLType = DiscoverSQLType(field.IndirectType) + if field.Tag.HasOption("msgpack") { + // A msgpack column holds the encoded bytes, not the Go type's own + // representation, so it needs a binary column whatever the field is + // declared as. Without this the type is discovered from the struct and + // the column ends up textual, which cannot store arbitrary msgpack. + field.DiscoveredSQLType = sqltype.Blob + } field.Append = FieldAppender(t.dialect, field) field.Scan = FieldScanner(t.dialect, field) field.IsZero = zeroChecker(field.StructField.Type)