dtest · Oct 13, 2026 · Oct 9, 2026 · Oct 13, 2026 · Oct 13, 2026 · Oct 13, 2026
diff --git 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,10 @@ 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.
+				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 +178,78 @@ 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,
+}
+
+// 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 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 {
+		return val
+	}
+	if _, ok := latin1Collations[field.GetCharset()]; !ok {
+		return val
+	}
+	// 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
+	}
+
+	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 +482,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
@@ -2,6 +2,7 @@ package internal
 
 import (
 	"encoding/base64"
+	"encoding/json"
 	"testing"
 
 	psdbconnect "github.com/planetscale/airbyte-source/proto/psdbconnect/v1alpha1"
@@ -240,3 +241,162 @@ 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())
+}
+
+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},
+		{Name: "owner", Type: sqltypes.Enum, ColumnType: "enum('café''s','x')", Charset: latin1Swedish},
+	}
+	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) {
+			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)
+			owner, err := sqltypes.NewValue(query.Type_ENUM, []byte(raw[2]))
+			require.NoError(t, err)
+
+			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"))
+	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())
+}