From 7d4d4d19e9c4e839cebb9d8447ff1bfe0754ce58 Mon Sep 17 00:00:00 2001 From: Derek Downey Date: Fri, 9 Oct 2026 15:50:41 +0000 Subject: [PATCH 1/5] Convert latin1 VStream text to UTF-8 before JSON encoding --- cmd/internal/types.go | 87 ++++++++++++++++++++++++++++++++++++-- cmd/internal/types_test.go | 69 ++++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 3 deletions(-) diff --git a/cmd/internal/types.go b/cmd/internal/types.go index a24a099..a118cb7 100644 --- a/cmd/internal/types.go +++ b/cmd/internal/types.go @@ -7,6 +7,7 @@ import ( "strconv" "strings" "time" + "unicode/utf8" "github.com/pkg/errors" psdbconnect "github.com/planetscale/airbyte-source/proto/psdbconnect/v1alpha1" @@ -158,7 +159,11 @@ func QueryResultToRecords(qr *sqltypes.Result, ps *PlanetScaleSource) []map[stri record := make(map[string]interface{}) for idx, val := range row { if idx < len(columns) { - parsedValue := parseValue(val, qr.Fields[idx].GetColumnType(), qr.Fields[idx].GetType(), ps) + field := qr.Fields[idx] + // Convert before parseValue so latin1 text is valid UTF-8 + // before JSON encoding, while enum/set index mapping still sees ASCII digits. + val = convertTextToUTF8(val, field) + parsedValue := parseValue(val, field.GetColumnType(), field.GetType(), ps) if parsedValue.isBool { record[columns[idx]] = parsedValue.boolValue } else if parsedValue.isNull { @@ -174,6 +179,82 @@ func QueryResultToRecords(qr *sqltypes.Result, ps *PlanetScaleSource) []map[stri return data } +// MySQL latin1 collation IDs. MySQL's latin1 is Windows-1252, not ISO-8859-1. +// VStream Field.Charset is the column collation ID (e.g. 8 = latin1_swedish_ci). +var latin1Collations = map[uint32]struct{}{ + 5: {}, // latin1_german1_ci + 8: {}, // latin1_swedish_ci + 15: {}, // latin1_danish_ci + 31: {}, // latin1_german2_ci + 47: {}, // latin1_bin + 48: {}, // latin1_general_ci + 49: {}, // latin1_general_cs + 94: {}, // latin1_spanish_ci +} + +// cp1252Map is the Unicode code point for each byte 0x80-0x9F in Windows-1252. +// All other latin1 bytes map directly to the same Unicode code point. +var cp1252Map = [32]rune{ + 0x20AC, 0x0081, 0x201A, 0x0192, 0x201E, 0x2026, 0x2020, 0x2021, + 0x02C6, 0x2030, 0x0160, 0x2039, 0x0152, 0x008D, 0x017D, 0x008F, + 0x0090, 0x2018, 0x2019, 0x201C, 0x201D, 0x2022, 0x2013, 0x2014, + 0x02DC, 0x2122, 0x0161, 0x203A, 0x0153, 0x009D, 0x017E, 0x0178, +} + +func isTextColumn(t query.Type) bool { + switch t { + case query.Type_CHAR, query.Type_VARCHAR, query.Type_TEXT, query.Type_ENUM, query.Type_SET: + return true + default: + return false + } +} + +// convertTextToUTF8 rewrites latin1 text values to UTF-8 so JSON encoding +// does not replace non-ASCII bytes with U+FFFD or truncate the string. +// Binary columns (charset 63) and already-UTF-8 text are left unchanged. +func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { + if val.IsNull() || field == nil || !isTextColumn(field.GetType()) { + return val + } + if _, ok := latin1Collations[field.GetCharset()]; !ok { + return val + } + + converted := decodeWindows1252(val.Raw()) + newVal, err := sqltypes.NewValue(val.Type(), converted) + if err != nil { + return val + } + return newVal +} + +func decodeWindows1252(src []byte) []byte { + ascii := true + for _, b := range src { + if b >= 0x80 { + ascii = false + break + } + } + if ascii { + return src + } + + dst := make([]byte, 0, len(src)+len(src)/2) + for _, b := range src { + var r rune + switch { + case b < 0x80, b >= 0xA0: + r = rune(b) + default: + r = cp1252Map[b-0x80] + } + dst = utf8.AppendRune(dst, r) + } + return dst +} + type Value struct { sqlValue sqltypes.Value boolValue bool @@ -406,8 +487,8 @@ type AirbyteStreamState struct { } type AirbyteState struct { - Type string `json:"type"` - Stream *AirbyteStreamState `json:"stream,omitempty"` + Type string `json:"type"` + Stream *AirbyteStreamState `json:"stream,omitempty"` } type StreamDescriptor struct { diff --git a/cmd/internal/types_test.go b/cmd/internal/types_test.go index 00fd66f..38c827f 100644 --- a/cmd/internal/types_test.go +++ b/cmd/internal/types_test.go @@ -2,6 +2,7 @@ package internal import ( "encoding/base64" + "encoding/json" "testing" psdbconnect "github.com/planetscale/airbyte-source/proto/psdbconnect/v1alpha1" @@ -240,3 +241,71 @@ func TestCanLeadDecimalWithZero(t *testing.T) { assert.Equal(t, "-0.66", row["d6"].(sqltypes.Value).ToString()) assert.Equal(t, "-0.77", row["d7"].(sqltypes.Value).ToString()) } + +func TestConvertsLatin1TextToUTF8(t *testing.T) { + // HEX 53E36F… is latin1 for "São". 0xED is í, 0xE9 is é. + sao, err := sqltypes.NewValue(query.Type_VARCHAR, []byte{0x53, 0xE3, 0x6F, 0x20, 0x50, 0x61, 0x75, 0x6C, 0x6F}) + require.NoError(t, err) + brasilia, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("Bras\xEDlia")) + require.NoError(t, err) + nordAm, err := sqltypes.NewValue(query.Type_TEXT, []byte("nord-am\xE9ricain")) + require.NoError(t, err) + ascii, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("Portland")) + require.NoError(t, err) + + const latin1Swedish = 8 + input := sqltypes.Result{ + Fields: []*query.Field{ + {Name: "location", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: latin1Swedish}, + {Name: "check_in_timezone", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: latin1Swedish}, + {Name: "label", Type: sqltypes.Text, ColumnType: "text", Charset: latin1Swedish}, + {Name: "city", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: latin1Swedish}, + }, + Rows: [][]sqltypes.Value{ + {sao, brasilia, nordAm, ascii}, + }, + } + + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + row := output[0] + assert.Equal(t, "São Paulo", row["location"].(sqltypes.Value).ToString()) + assert.Equal(t, "Brasília", row["check_in_timezone"].(sqltypes.Value).ToString()) + assert.Equal(t, "nord-américain", row["label"].(sqltypes.Value).ToString()) + assert.Equal(t, "Portland", row["city"].(sqltypes.Value).ToString()) + + encoded, err := json.Marshal(row) + require.NoError(t, err) + assert.Contains(t, string(encoded), `"São Paulo"`) + assert.Contains(t, string(encoded), `"Brasília"`) + assert.Contains(t, string(encoded), `"nord-américain"`) + assert.NotContains(t, string(encoded), "\ufffd") +} + +func TestLeavesUTF8AndBinaryColumnsUnchanged(t *testing.T) { + utf8Sao, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("São Paulo")) + require.NoError(t, err) + // Same latin1 bytes as "São", but this is a binary column so they must not be transcoded. + rawLatin1, err := sqltypes.NewValue(query.Type_VARBINARY, []byte{0x53, 0xE3, 0x6F}) + require.NoError(t, err) + + const ( + utf8mb4 = 255 + binaryColl = 63 + ) + input := sqltypes.Result{ + Fields: []*query.Field{ + {Name: "location", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: utf8mb4}, + {Name: "payload", Type: sqltypes.VarBinary, ColumnType: "varbinary(16)", Charset: binaryColl}, + }, + Rows: [][]sqltypes.Value{ + {utf8Sao, rawLatin1}, + }, + } + + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + row := output[0] + assert.Equal(t, "São Paulo", row["location"].(sqltypes.Value).ToString()) + assert.Equal(t, []byte{0x53, 0xE3, 0x6F}, row["payload"].(sqltypes.Value).Raw()) +} From 37445f97da7673abd2398101ab369110a66cdf1e Mon Sep 17 00:00:00 2001 From: Derek Downey Date: Tue, 13 Oct 2026 08:41:02 -0400 Subject: [PATCH 2/5] Update cmd/internal/types.go Co-authored-by: Lizz --- cmd/internal/types.go | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/cmd/internal/types.go b/cmd/internal/types.go index a118cb7..650eb33 100644 --- a/cmd/internal/types.go +++ b/cmd/internal/types.go @@ -160,8 +160,7 @@ func QueryResultToRecords(qr *sqltypes.Result, ps *PlanetScaleSource) []map[stri for idx, val := range row { if idx < len(columns) { field := qr.Fields[idx] - // Convert before parseValue so latin1 text is valid UTF-8 - // before JSON encoding, while enum/set index mapping still sees ASCII digits. + // Convert before parseValue so latin1 text is valid UTF-8 before JSON encoding. val = convertTextToUTF8(val, field) parsedValue := parseValue(val, field.GetColumnType(), field.GetType(), ps) if parsedValue.isBool { From ecccedf71c1f617a16d65d75bf44ba31ec24b65a Mon Sep 17 00:00:00 2001 From: Derek Downey Date: Tue, 13 Oct 2026 08:41:15 -0400 Subject: [PATCH 3/5] Update cmd/internal/types.go Co-authored-by: Lizz --- cmd/internal/types.go | 35 ++++++++++++++++++++++++----------- 1 file changed, 24 insertions(+), 11 deletions(-) diff --git a/cmd/internal/types.go b/cmd/internal/types.go index 650eb33..4415bcf 100644 --- a/cmd/internal/types.go +++ b/cmd/internal/types.go @@ -200,25 +200,21 @@ var cp1252Map = [32]rune{ 0x02DC, 0x2122, 0x0161, 0x203A, 0x0153, 0x009D, 0x017E, 0x0178, } -func isTextColumn(t query.Type) bool { - switch t { - case query.Type_CHAR, query.Type_VARCHAR, query.Type_TEXT, query.Type_ENUM, query.Type_SET: - return true - default: - return false - } -} - // convertTextToUTF8 rewrites latin1 text values to UTF-8 so JSON encoding -// does not replace non-ASCII bytes with U+FFFD or truncate the string. +// does not replace non-ASCII bytes with U+FFFD. // Binary columns (charset 63) and already-UTF-8 text are left unchanged. func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { - if val.IsNull() || field == nil || !isTextColumn(field.GetType()) { + // Gate on collation, not type: latin1_bin columns arrive typed BINARY/VARBINARY/BLOB. + if val.IsNull() || field == nil { return val } if _, ok := latin1Collations[field.GetCharset()]; !ok { return val } + // Binlog ENUM/SET values are already UTF-8 labels taken from column_type (vitess v21+). + if t := field.GetType(); (t == query.Type_ENUM || t == query.Type_SET) && isDeclaredLabel(val.Raw(), field.GetColumnType()) { + return val + } converted := decodeWindows1252(val.Raw()) newVal, err := sqltypes.NewValue(val.Type(), converted) @@ -228,6 +224,23 @@ func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { return newVal } +// isDeclaredLabel reports whether raw is valid UTF-8 made up only of labels declared in columnType. +func isDeclaredLabel(raw []byte, columnType string) bool { + if !utf8.Valid(raw) { + return false + } + labels := map[string]struct{}{} + for _, l := range parseEnumOrSetValues(columnType) { + labels[l] = struct{}{} + } + for _, part := range strings.Split(string(raw), ",") { + if _, ok := labels[part]; !ok { + return false + } + } + return true +} + func decodeWindows1252(src []byte) []byte { ascii := true for _, b := range src { From e1d3ebfd03b01c650441cd0774e2942af7bf3d93 Mon Sep 17 00:00:00 2001 From: Derek Downey Date: Tue, 13 Oct 2026 08:41:25 -0400 Subject: [PATCH 4/5] Update cmd/internal/types_test.go Co-authored-by: Lizz --- cmd/internal/types_test.go | 67 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/cmd/internal/types_test.go b/cmd/internal/types_test.go index 38c827f..d14d95f 100644 --- a/cmd/internal/types_test.go +++ b/cmd/internal/types_test.go @@ -309,3 +309,70 @@ func TestLeavesUTF8AndBinaryColumnsUnchanged(t *testing.T) { assert.Equal(t, "São Paulo", row["location"].(sqltypes.Value).ToString()) assert.Equal(t, []byte{0x53, 0xE3, 0x6F}, row["payload"].(sqltypes.Value).Raw()) } + +func TestConvertsLatin1BinColumnsToUTF8(t *testing.T) { + // MySQL sets BINARY_FLAG on _bin collations, so latin1_bin columns arrive typed VARBINARY. + kaka, err := sqltypes.NewValue(query.Type_VARBINARY, []byte("Kak\xE1")) + require.NoError(t, err) + + const latin1Bin = 47 + input := sqltypes.Result{ + Fields: []*query.Field{ + {Name: "tag", Type: sqltypes.VarBinary, ColumnType: "varchar(8)", Charset: latin1Bin}, + }, + Rows: [][]sqltypes.Value{ + {kaka}, + }, + } + + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + assert.Equal(t, "Kaká", output[0]["tag"].(sqltypes.Value).ToString()) +} + +func TestConvertsLatin1EnumAndSetInBothVStreamPhases(t *testing.T) { + // The copy phase sends latin1 label bytes; the binlog phase sends UTF-8 labels from column_type. + const latin1Swedish = 8 + fields := []*query.Field{ + {Name: "city", Type: sqltypes.Enum, ColumnType: "enum('Brasília','Portland')", Charset: latin1Swedish}, + {Name: "tags", Type: sqltypes.Set, ColumnType: "set('café','bar')", Charset: latin1Swedish}, + } + phases := map[string][2]string{ + "copy": {"Bras\xEDlia", "caf\xE9,bar"}, + "binlog": {"Brasília", "café,bar"}, + } + for phase, raw := range phases { + t.Run(phase, func(t *testing.T) { + city, err := sqltypes.NewValue(query.Type_ENUM, []byte(raw[0])) + require.NoError(t, err) + tags, err := sqltypes.NewValue(query.Type_SET, []byte(raw[1])) + require.NoError(t, err) + + input := sqltypes.Result{Fields: fields, Rows: [][]sqltypes.Value{{city, tags}}} + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + assert.Equal(t, "Brasília", output[0]["city"].(sqltypes.Value).ToString()) + assert.Equal(t, "café,bar", output[0]["tags"].(sqltypes.Value).ToString()) + }) + } +} + +func TestConvertsWindows1252ControlRange(t *testing.T) { + // MySQL latin1 is cp1252; its undefined bytes (e.g. 0x81, 0x9D) map to C1 controls. + note, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("\x80\x91\x92\x93\x94\x81\x9D")) + require.NoError(t, err) + + const latin1Swedish = 8 + input := sqltypes.Result{ + Fields: []*query.Field{ + {Name: "note", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: latin1Swedish}, + }, + Rows: [][]sqltypes.Value{ + {note}, + }, + } + + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + assert.Equal(t, "€‘’“”\u0081\u009d", output[0]["note"].(sqltypes.Value).ToString()) +} From ca09b9f65918a5f36b7b49803f363216f91433c5 Mon Sep 17 00:00:00 2001 From: Liz van Dijk Date: Tue, 13 Oct 2026 16:14:39 +0100 Subject: [PATCH 5/5] Pass valid UTF-8 through instead of matching ENUM/SET labels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Skip conversion whenever the value is already valid UTF-8, and drop isDeclaredLabel. With this, the output only differs from main where main emitted U+FFFD; anything main passed through intact is unchanged. It also fixes two cases the label check got wrong: - Binlog ENUM labels containing a quote were double-encoded. column_type stores 'café''s' and vitess unescapes it, but parseEnumOrSetValues does not, so the match failed and "café's" became "café's". - UTF-8 bytes stored in latin1 columns (written through a latin1 connection) now come through as "São" again instead of "São". The cost is latin1 text whose high bytes happen to form valid UTF-8 (e.g. "é" stored as C3 A9), which stays as main shows it. The copy phase does deliver raw latin1 bytes (rowstreamer runs set names 'binary'), so the label check was only ever needed for the binlog phase. Verified end to end against vttestserver in both phases. Co-Authored-By: Claude Opus 5.5 --- cmd/internal/types.go | 23 +++-------------------- cmd/internal/types_test.go | 32 ++++++++++++++++++++++++++++---- 2 files changed, 31 insertions(+), 24 deletions(-) diff --git a/cmd/internal/types.go b/cmd/internal/types.go index 4415bcf..67cc8f1 100644 --- a/cmd/internal/types.go +++ b/cmd/internal/types.go @@ -202,7 +202,7 @@ var cp1252Map = [32]rune{ // convertTextToUTF8 rewrites latin1 text values to UTF-8 so JSON encoding // does not replace non-ASCII bytes with U+FFFD. -// Binary columns (charset 63) and already-UTF-8 text are left unchanged. +// Binary columns (charset 63) and values that are already valid UTF-8 are left unchanged. func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { // Gate on collation, not type: latin1_bin columns arrive typed BINARY/VARBINARY/BLOB. if val.IsNull() || field == nil { @@ -211,8 +211,8 @@ func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { if _, ok := latin1Collations[field.GetCharset()]; !ok { return val } - // Binlog ENUM/SET values are already UTF-8 labels taken from column_type (vitess v21+). - if t := field.GetType(); (t == query.Type_ENUM || t == query.Type_SET) && isDeclaredLabel(val.Raw(), field.GetColumnType()) { + // Valid UTF-8 passes through unchanged: binlog ENUM/SET labels (vitess v21+) and UTF-8 stored in latin1 columns. + if utf8.Valid(val.Raw()) { return val } @@ -224,23 +224,6 @@ func convertTextToUTF8(val sqltypes.Value, field *query.Field) sqltypes.Value { return newVal } -// isDeclaredLabel reports whether raw is valid UTF-8 made up only of labels declared in columnType. -func isDeclaredLabel(raw []byte, columnType string) bool { - if !utf8.Valid(raw) { - return false - } - labels := map[string]struct{}{} - for _, l := range parseEnumOrSetValues(columnType) { - labels[l] = struct{}{} - } - for _, part := range strings.Split(string(raw), ",") { - if _, ok := labels[part]; !ok { - return false - } - } - return true -} - func decodeWindows1252(src []byte) []byte { ascii := true for _, b := range src { diff --git a/cmd/internal/types_test.go b/cmd/internal/types_test.go index d14d95f..9b9b87d 100644 --- a/cmd/internal/types_test.go +++ b/cmd/internal/types_test.go @@ -336,10 +336,11 @@ func TestConvertsLatin1EnumAndSetInBothVStreamPhases(t *testing.T) { fields := []*query.Field{ {Name: "city", Type: sqltypes.Enum, ColumnType: "enum('Brasília','Portland')", Charset: latin1Swedish}, {Name: "tags", Type: sqltypes.Set, ColumnType: "set('café','bar')", Charset: latin1Swedish}, + {Name: "owner", Type: sqltypes.Enum, ColumnType: "enum('café''s','x')", Charset: latin1Swedish}, } - phases := map[string][2]string{ - "copy": {"Bras\xEDlia", "caf\xE9,bar"}, - "binlog": {"Brasília", "café,bar"}, + phases := map[string][3]string{ + "copy": {"Bras\xEDlia", "caf\xE9,bar", "caf\xE9's"}, + "binlog": {"Brasília", "café,bar", "café's"}, } for phase, raw := range phases { t.Run(phase, func(t *testing.T) { @@ -347,16 +348,39 @@ func TestConvertsLatin1EnumAndSetInBothVStreamPhases(t *testing.T) { require.NoError(t, err) tags, err := sqltypes.NewValue(query.Type_SET, []byte(raw[1])) require.NoError(t, err) + owner, err := sqltypes.NewValue(query.Type_ENUM, []byte(raw[2])) + require.NoError(t, err) - input := sqltypes.Result{Fields: fields, Rows: [][]sqltypes.Value{{city, tags}}} + input := sqltypes.Result{Fields: fields, Rows: [][]sqltypes.Value{{city, tags, owner}}} output := QueryResultToRecords(&input, &PlanetScaleSource{}) require.Len(t, output, 1) assert.Equal(t, "Brasília", output[0]["city"].(sqltypes.Value).ToString()) assert.Equal(t, "café,bar", output[0]["tags"].(sqltypes.Value).ToString()) + assert.Equal(t, "café's", output[0]["owner"].(sqltypes.Value).ToString()) }) } } +func TestLeavesValidUTF8InLatin1ColumnsUnchanged(t *testing.T) { + // UTF-8 bytes stored in a latin1 column (written through a latin1 connection) pass through unchanged. + sao, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("S\xC3\xA3o")) + require.NoError(t, err) + + const latin1Swedish = 8 + input := sqltypes.Result{ + Fields: []*query.Field{ + {Name: "location", Type: sqltypes.VarChar, ColumnType: "varchar(255)", Charset: latin1Swedish}, + }, + Rows: [][]sqltypes.Value{ + {sao}, + }, + } + + output := QueryResultToRecords(&input, &PlanetScaleSource{}) + require.Len(t, output, 1) + assert.Equal(t, "São", output[0]["location"].(sqltypes.Value).ToString()) +} + func TestConvertsWindows1252ControlRange(t *testing.T) { // MySQL latin1 is cp1252; its undefined bytes (e.g. 0x81, 0x9D) map to C1 controls. note, err := sqltypes.NewValue(query.Type_VARCHAR, []byte("\x80\x91\x92\x93\x94\x81\x9D"))