Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Take MySQL Column Type Into Account in VStreamer #9331

Merged
merged 2 commits into from
Dec 9, 2021
Merged

Take MySQL Column Type Into Account in VStreamer #9331

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
cdb0575
Take MySQL column type into account in vstreamer
mattlord Dec 7, 2021
3ff9fbd
Detect and handle MySQL 8 specifically in vstreamer tests
mattlord Dec 8, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion go/mysql/binlog_event_json.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -468,7 +468,7 @@ func (oh opaquePlugin) getNode(typ jsonDataType, data []byte, pos int) (node *aj
precision := decimalData[0]
scale := decimalData[1]
metadata := (uint16(precision) << 8) + uint16(scale)
val, _, err := CellValue(decimalData, 2, TypeNewDecimal, metadata, querypb.Type_DECIMAL)
val, _, err := CellValue(decimalData, 2, TypeNewDecimal, metadata, &querypb.Field{Type: querypb.Type_DECIMAL})
if err != nil {
return nil, err
}
Expand Down
36 changes: 20 additions & 16 deletions go/mysql/binlog_event_rbr.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -331,12 +331,14 @@ func printTimestamp(v uint32) *bytes.Buffer {
}

// CellValue returns the data for a cell as a sqltypes.Value, and how
// many bytes it takes. It only uses the querypb.Type value for the
// signed flag.
func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Type) (sqltypes.Value, int, error) {
// many bytes it takes. It uses source type in querypb.Type and vitess type
// byte to determine general shared aspects of types and the querypb.Field to
// determine other info specifically about its underlying column (SQL column
// type, column length, charset, etc)
func CellValue(data []byte, pos int, typ byte, metadata uint16, field *querypb.Field) (sqltypes.Value, int, error) {
switch typ {
case TypeTiny:
if sqltypes.IsSigned(styp) {
if sqltypes.IsSigned(field.Type) {
return sqltypes.MakeTrusted(querypb.Type_INT8,
strconv.AppendInt(nil, int64(int8(data[pos])), 10)), 1, nil
}
Expand All @@ -352,14 +354,14 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
strconv.AppendUint(nil, uint64(data[pos])+1900, 10)), 1, nil
case TypeShort:
val := binary.LittleEndian.Uint16(data[pos : pos+2])
if sqltypes.IsSigned(styp) {
if sqltypes.IsSigned(field.Type) {
return sqltypes.MakeTrusted(querypb.Type_INT16,
strconv.AppendInt(nil, int64(int16(val)), 10)), 2, nil
}
return sqltypes.MakeTrusted(querypb.Type_UINT16,
strconv.AppendUint(nil, uint64(val), 10)), 2, nil
case TypeInt24:
if sqltypes.IsSigned(styp) && data[pos+2]&128 > 0 {
if sqltypes.IsSigned(field.Type) && data[pos+2]&128 > 0 {
// Negative number, have to extend the sign.
val := int32(uint32(data[pos]) +
uint32(data[pos+1])<<8 +
Expand All @@ -376,7 +378,7 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
strconv.AppendUint(nil, val, 10)), 3, nil
case TypeLong:
val := binary.LittleEndian.Uint32(data[pos : pos+4])
if sqltypes.IsSigned(styp) {
if sqltypes.IsSigned(field.Type) {
return sqltypes.MakeTrusted(querypb.Type_INT32,
strconv.AppendInt(nil, int64(int32(val)), 10)), 4, nil
}
Expand All @@ -399,7 +401,7 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
txt.Bytes()), 4, nil
case TypeLongLong:
val := binary.LittleEndian.Uint64(data[pos : pos+8])
if sqltypes.IsSigned(styp) {
if sqltypes.IsSigned(field.Type) {
return sqltypes.MakeTrusted(querypb.Type_INT64,
strconv.AppendInt(nil, int64(val), 10)), 8, nil
}
Expand Down Expand Up @@ -448,11 +450,11 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
return sqltypes.MakeTrusted(querypb.Type_DATETIME,
[]byte(fmt.Sprintf("%04d-%02d-%02d %02d:%02d:%02d", year, month, day, hour, minute, second))), 8, nil
case TypeVarchar, TypeVarString:
// We trust that styp is compatible with the column type
// We trust that typ is compatible with the field.Type
// Length is encoded in 1 or 2 bytes.
typeToUse := querypb.Type_VARCHAR
if styp == querypb.Type_VARBINARY || styp == querypb.Type_BINARY || styp == querypb.Type_BLOB {
typeToUse = styp
if field.Type == querypb.Type_VARBINARY || field.Type == querypb.Type_BINARY || field.Type == querypb.Type_BLOB {
typeToUse = field.Type
}
if metadata > 255 {
l := int(uint64(data[pos]) |
Expand Down Expand Up @@ -894,15 +896,17 @@ func CellValue(data []byte, pos int, typ byte, metadata uint16, styp querypb.Typ
}
l := int(data[pos])
mdata := data[pos+1 : pos+1+l]
if sqltypes.IsBinary(styp) {
if sqltypes.IsBinary(field.Type) {
// For binary(n) column types, mysql pads the data on the right with nulls. However the binlog event contains
// the data without this padding. This causes several issues:
// * if a binary(n) column is part of the sharding key, the keyspace_id() returned during the copy phase
// (where the value is the result of a mysql query) is different from the one during replication
// (where the value is the one from the binlogs)
// * mysql where clause comparisons do not do the right thing without padding
// So for fixed length binary() columns we right-pad it with nulls if necessary
if l < max {
// So for fixed length BINARY columns we right-pad it with nulls if necessary to match what MySQL returns.
// Because CHAR columns with a binary collation (e.g. utf8mb4_bin) have the same metadata as a BINARY column
// in binlog events, we also need to check for this case based on the underlying column type.
if l < max && strings.HasPrefix(strings.ToLower(field.ColumnType), "binary") {
paddedData := make([]byte, max)
copy(paddedData[:l], mdata)
mdata = paddedData
Expand Down Expand Up @@ -1094,7 +1098,7 @@ func (rs *Rows) StringValuesForTests(tm *TableMap, rowIndex int) ([]string, erro
}

// We have real data
value, l, err := CellValue(data, pos, tm.Types[c], tm.Metadata[c], querypb.Type_UINT64)
value, l, err := CellValue(data, pos, tm.Types[c], tm.Metadata[c], &querypb.Field{Type: querypb.Type_UINT64})
if err != nil {
return nil, err
}
Expand Down Expand Up @@ -1129,7 +1133,7 @@ func (rs *Rows) StringIdentifiesForTests(tm *TableMap, rowIndex int) ([]string,
}

// We have real data
value, l, err := CellValue(data, pos, tm.Types[c], tm.Metadata[c], querypb.Type_UINT64)
value, l, err := CellValue(data, pos, tm.Types[c], tm.Metadata[c], &querypb.Field{Type: querypb.Type_UINT64})
if err != nil {
return nil, err
}
Expand Down
2 changes: 1 addition & 1 deletion go/mysql/binlog_event_rbr_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -550,7 +550,7 @@ func TestCellLengthAndData(t *testing.T) {
}

// Test CellValue.
out, l, err := CellValue(padded, 1, tcase.typ, tcase.metadata, tcase.styp)
out, l, err := CellValue(padded, 1, tcase.typ, tcase.metadata, &querypb.Field{Type: tcase.styp})
if err != nil || l != len(tcase.data) || out.Type() != tcase.out.Type() || !bytes.Equal(out.Raw(), tcase.out.Raw()) {
t.Errorf("testcase cellData(%v,%v) returned unexpected result: %v %v %v, was expecting %v %v <nil>",
tcase.typ, tcase.data, out, l, err, tcase.out, len(tcase.data))
Expand Down
2 changes: 1 addition & 1 deletion go/mysql/endtoend/replication_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1105,7 +1105,7 @@ func valuesForTests(t *testing.T, rs *mysql.Rows, tm *mysql.TableMap, rowIndex i
}

// We have real data
value, l, err := mysql.CellValue(data, pos, tm.Types[c], tm.Metadata[c], querypb.Type_UINT64)
value, l, err := mysql.CellValue(data, pos, tm.Types[c], tm.Metadata[c], &querypb.Field{Type: querypb.Type_UINT64})
if err != nil {
return nil, err
}
Expand Down
4 changes: 2 additions & 2 deletions go/vt/binlog/binlog_streamer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -754,7 +754,7 @@ func writeValuesAsSQL(sql *sqlparser.TrackedBuffer, tce *tableCacheEntry, rs *my
}

// We have real data.
value, l, err := mysql.CellValue(data, pos, tce.tm.Types[c], tce.tm.Metadata[c], tce.ti.Fields[c].Type)
value, l, err := mysql.CellValue(data, pos, tce.tm.Types[c], tce.tm.Metadata[c], &querypb.Field{Type: tce.ti.Fields[c].Type})
if err != nil {
return keyspaceIDCell, nil, err
}
Expand Down Expand Up @@ -819,7 +819,7 @@ func writeIdentifiersAsSQL(sql *sqlparser.TrackedBuffer, tce *tableCacheEntry, r
sql.WriteByte('=')

// We have real data.
value, l, err := mysql.CellValue(data, pos, tce.tm.Types[c], tce.tm.Metadata[c], tce.ti.Fields[c].Type)
value, l, err := mysql.CellValue(data, pos, tce.tm.Types[c], tce.tm.Metadata[c], &querypb.Field{Type: tce.ti.Fields[c].Type})
if err != nil {
return keyspaceIDCell, nil, err
}
Expand Down
Loading