Skip to content

Commit fd0b39c

Browse files
dtunikovCopilot
dtunikov
and
Copilot
authored
fix(mysql): consistent SET column replication between snapshot and cdc (#4551)
consistent mysql SET column replication when row metadata is not enabled follows existing pattern introduced for mysql enums (uint16enum q type) --------- Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
1 parent 3440261 commit fd0b39c

17 files changed

Lines changed: 146 additions & 26 deletions

‎flow/connectors/bigquery/qvalue_convert.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ func qValueKindToBigQueryType(columnDescription *protos.FieldDescription, nullab
2222
// integer types
2323
case types.QValueKindInt8, types.QValueKindInt16, types.QValueKindInt32, types.QValueKindInt64,
2424
types.QValueKindUInt8, types.QValueKindUInt16, types.QValueKindUInt32, types.QValueKindUInt64,
25-
types.QValueKindUint16Enum:
25+
types.QValueKindUint16Enum, types.QValueKindUint64Set:
2626
bqField.Type = bigquery.IntegerFieldType
2727
// decimal types
2828
case types.QValueKindFloat32, types.QValueKindFloat64:

‎flow/connectors/mysql/qrep.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -253,7 +253,7 @@ func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string) str
253253
}
254254

255255
converted := common.QuoteMySQLIdentifier(col.Name)
256-
if col.Type == string(types.QValueKindUint16Enum) {
256+
if col.Type == string(types.QValueKindUint16Enum) || col.Type == string(types.QValueKindUint64Set) {
257257
converted = fmt.Sprintf("CAST(%s AS UNSIGNED) AS %s", converted, converted)
258258
selectAsterisk = false
259259
}

‎flow/connectors/mysql/qrep_test.go‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,25 @@ func TestBuildSelectedColumns(t *testing.T) {
4747
exclude: []string{},
4848
expectedSelectedColumns: "`id`, CAST(`status` AS UNSIGNED) AS `status`",
4949
},
50+
{
51+
name: "uint64set column is cast to unsigned",
52+
cols: []*protos.FieldDescription{
53+
{Name: "id", Type: string(types.QValueKindInt32)},
54+
{Name: "perms", Type: string(types.QValueKindUint64Set)},
55+
},
56+
exclude: []string{},
57+
expectedSelectedColumns: "`id`, CAST(`perms` AS UNSIGNED) AS `perms`",
58+
},
59+
{
60+
name: "uint64set with exclude",
61+
cols: []*protos.FieldDescription{
62+
{Name: "id", Type: string(types.QValueKindInt32)},
63+
{Name: "perms", Type: string(types.QValueKindUint64Set)},
64+
{Name: "created_at", Type: string(types.QValueKindTimestamp)},
65+
},
66+
exclude: []string{"created_at"},
67+
expectedSelectedColumns: "`id`, CAST(`perms` AS UNSIGNED) AS `perms`",
68+
},
5069
{
5170
name: "string enum column is not cast",
5271
cols: []*protos.FieldDescription{

‎flow/connectors/mysql/qvalue_convert.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -270,6 +270,8 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
270270
switch qkind {
271271
case types.QValueKindUint16Enum:
272272
return types.QValueUint16Enum{Val: uint16(v)}, nil
273+
case types.QValueKindUint64Set:
274+
return types.QValueUint64Set{Val: v}, nil
273275
case types.QValueKindBoolean:
274276
return types.QValueBoolean{Val: v != 0}, nil
275277
case types.QValueKindInt8:
@@ -519,6 +521,8 @@ func QValueFromMysqlRowEvent(
519521
}
520522
}
521523
return types.QValueString{Val: strings.Join(set, ",")}, nil
524+
case types.QValueKindUint64Set:
525+
return types.QValueUint64Set{Val: uint64(val)}, nil
522526
case types.QValueKindUint16Enum:
523527
return types.QValueUint16Enum{Val: uint16(val)}, nil
524528
case types.QValueKindEnum: // enum

‎flow/connectors/mysql/qvalue_convert_test.go‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,26 @@ func TestQkindFromMysqlType_Bit(t *testing.T) {
5858
}
5959
}
6060

61+
func TestQkindFromMysqlColumnType_Set(t *testing.T) {
62+
for _, tc := range []struct {
63+
name string
64+
metadata bool
65+
version uint32
66+
want types.QValueKind
67+
}{
68+
{"with metadata maps SET to String", true, shared.InternalVersion_Latest, types.QValueKindString},
69+
{"without metadata, before gate maps SET to String", false, shared.InternalVersion_MySQL5ConvertSetsToInts - 1, types.QValueKindString},
70+
{"without metadata, gate maps SET to Uint64Set", false, shared.InternalVersion_MySQL5ConvertSetsToInts, types.QValueKindUint64Set},
71+
{"without metadata, latest maps SET to Uint64Set", false, shared.InternalVersion_Latest, types.QValueKindUint64Set},
72+
} {
73+
t.Run(tc.name, func(t *testing.T) {
74+
qkind, err := QkindFromMysqlColumnType("set('a','b','c')", tc.metadata, tc.version)
75+
require.NoError(t, err)
76+
require.Equal(t, tc.want, qkind)
77+
})
78+
}
79+
}
80+
6181
func TestProcessTime(t *testing.T) {
6282
epoch := time.Unix(0, 0).UTC()
6383
for _, ts := range []struct {

‎flow/connectors/mysql/type_conversion.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -18,10 +18,16 @@ func QkindFromMysqlColumnType(ct string, binlogRowMetadataSupported bool, versio
1818
switch strings.ToLower(ct) {
1919
case "json":
2020
return types.QValueKindJSON, nil
21-
case "char", "varchar", "text", "set", "tinytext", "mediumtext", "longtext",
21+
case "char", "varchar", "text", "tinytext", "mediumtext", "longtext",
2222
"clob", "varchar2", // maria
2323
"xmltype": // maria
2424
return types.QValueKindString, nil
25+
case "set":
26+
// keeps snapshot and cdc consistent for mysql without row metadata
27+
if !binlogRowMetadataSupported && version >= shared.InternalVersion_MySQL5ConvertSetsToInts {
28+
return types.QValueKindUint64Set, nil
29+
}
30+
return types.QValueKindString, nil
2531
case "enum":
2632
if !binlogRowMetadataSupported && version >= shared.InternalVersion_MySQL5ConvertEnumsToInts {
2733
return types.QValueKindUint16Enum, nil

‎flow/connectors/postgres/qvalue_convert.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,8 @@ func qValueKindToPostgresType(colTypeStr string) string {
5555
return "BIGINT"
5656
case types.QValueKindUint16Enum:
5757
return "INTEGER"
58+
case types.QValueKindUint64Set:
59+
return "BIGINT"
5860
case types.QValueKindFloat32:
5961
return "REAL"
6062
case types.QValueKindFloat64:

‎flow/connectors/utils/cdc_store.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -99,6 +99,7 @@ func init() {
9999
gob.Register(types.QValueString{})
100100
gob.Register(types.QValueEnum{})
101101
gob.Register(types.QValueUint16Enum{})
102+
gob.Register(types.QValueUint64Set{})
102103
gob.Register(types.QValueTimestamp{})
103104
gob.Register(types.QValueTimestampTZ{})
104105
gob.Register(types.QValueDate{})

‎flow/e2e/clickhouse_mysql_test.go‎

Lines changed: 37 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,7 @@ import (
2424
"github.com/PeerDB-io/peerdb/flow/shared/types"
2525
)
2626

27-
func mysqlEnumUsesOrdinals(source *MySqlSource) bool {
27+
func mysqlUsesOrdinals(source *MySqlSource) bool {
2828
return source.Config.Flavor == protos.MySqlFlavor_MYSQL_MYSQL &&
2929
source.Config.ReplicationMechanism == protos.MySqlReplicationMechanism_MYSQL_FILEPOS
3030
}
@@ -794,26 +794,30 @@ func (s ClickHouseSuite) Test_MySQL_Enum() {
794794
RequireEnvCanceled(s.t, env)
795795
}
796796

797-
func (s ClickHouseSuite) Test_MySQL_Enum_Consistency() {
797+
// Test_MySQL_Enum_Set_Consistency verifies that ENUM and SET columns produce identical values
798+
// via snapshot and CDC. On servers without binlog row metadata, both are emitted as integers
799+
// (enum ordinal, set bitmask); otherwise as labels.
800+
func (s ClickHouseSuite) Test_MySQL_Enum_Set_Consistency() {
798801
mySource, ok := s.source.(*MySqlSource)
799802
if !ok {
800803
s.t.Skip("only applies to mysql")
801804
}
802805

803-
srcTableName := "test_my_enum_consistency"
806+
srcTableName := "test_my_enum_set_consistency"
804807
srcFullName := s.attachSchemaSuffix(srcTableName)
805-
dstTableName := "test_my_enum_consistency_dst"
808+
dstTableName := "test_my_enum_set_consistency_dst"
806809

807810
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
808811
CREATE TABLE IF NOT EXISTS %s (
809812
id SERIAL PRIMARY KEY,
810-
status ENUM('active', 'inactive', 'pending') NOT NULL
813+
status ENUM('active', 'inactive', 'pending') NOT NULL,
814+
tags SET('a', 'b', 'c') NOT NULL
811815
)
812816
`, srcFullName)))
813817

814818
// Insert row before snapshot
815819
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
816-
`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
820+
`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
817821

818822
connectionGen := FlowConnectionGenerationConfig{
819823
FlowJobName: s.attachSuffix(srcTableName),
@@ -828,50 +832,58 @@ func (s ClickHouseSuite) Test_MySQL_Enum_Consistency() {
828832
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
829833

830834
// Wait for snapshot row to appear in destination
831-
EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status", 1)
835+
EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status,tags", 1)
832836

833837
// Insert row via CDC — on old MySQL this comes as integer from binlog
834838
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
835-
`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
839+
`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
836840

837841
// Wait for CDC row
838-
EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status", 2)
842+
EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status,tags", 2)
839843

840-
// Verify both rows have the same status value (consistency between snapshot and CDC)
841-
rows, err := s.GetRows(dstTableName, "id,status")
844+
// Verify snapshot and CDC produce the same values for both columns
845+
rows, err := s.GetRows(dstTableName, "id,status,tags")
842846
require.NoError(s.t, err)
843847
require.Len(s.t, rows.Records, 2)
844848
require.Equal(s.t, rows.Records[0][1].Value(), rows.Records[1][1].Value(),
845849
"snapshot and CDC enum values should be consistent")
846-
if mysqlEnumUsesOrdinals(mySource) {
850+
require.Equal(s.t, rows.Records[0][2].Value(), rows.Records[1][2].Value(),
851+
"snapshot and CDC set values should be consistent")
852+
if mysqlUsesOrdinals(mySource) {
847853
require.EqualValues(s.t, 1, rows.Records[0][1].Value())
854+
require.EqualValues(s.t, 3, rows.Records[0][2].Value()) // 'a,b' bitmask = 1|2
848855
} else {
849856
require.Equal(s.t, "active", rows.Records[0][1].Value())
857+
require.Equal(s.t, "a,b", rows.Records[0][2].Value())
850858
}
851859

852860
env.Cancel(s.t.Context())
853861
RequireEnvCanceled(s.t, env)
854862
}
855863

856-
func (s ClickHouseSuite) Test_MySQL_Enum_Consistency_Version0() {
864+
// Test_MySQL_Enum_Set_Consistency_Version0 documents the legacy divergence: on pre-metadata
865+
// servers, mirrors created before the enum/set-to-int versions emit labels via snapshot but
866+
// ordinals/bitmasks via CDC. Existing mirrors deliberately keep this behavior for stability.
867+
func (s ClickHouseSuite) Test_MySQL_Enum_Set_Consistency_Version0() {
857868
mySource, ok := s.source.(*MySqlSource)
858869
if !ok {
859870
s.t.Skip("only applies to mysql")
860871
}
861872

862-
srcTableName := "test_my_enum_consistency_v0"
873+
srcTableName := "test_my_enum_set_consistency_v0"
863874
srcFullName := s.attachSchemaSuffix(srcTableName)
864-
dstTableName := "test_my_enum_consistency_v0_dst"
875+
dstTableName := "test_my_enum_set_consistency_v0_dst"
865876

866877
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
867878
CREATE TABLE IF NOT EXISTS %s (
868879
id SERIAL PRIMARY KEY,
869-
status ENUM('active', 'inactive', 'pending') NOT NULL
880+
status ENUM('active', 'inactive', 'pending') NOT NULL,
881+
tags SET('a', 'b', 'c') NOT NULL
870882
)
871883
`, srcFullName)))
872884

873885
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
874-
`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
886+
`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
875887

876888
connectionGen := FlowConnectionGenerationConfig{
877889
FlowJobName: s.attachSuffix(srcTableName),
@@ -887,23 +899,26 @@ func (s ClickHouseSuite) Test_MySQL_Enum_Consistency_Version0() {
887899
env := ExecutePeerflow(s.t, tc, flowConnConfig)
888900
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
889901

890-
EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status", 1)
902+
EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status,tags", 1)
891903

892904
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
893-
`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
905+
`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
894906

895-
EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status", 2)
907+
EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status,tags", 2)
896908

897-
rows, err := s.GetRows(dstTableName, "id,status")
909+
rows, err := s.GetRows(dstTableName, "id,status,tags")
898910
require.NoError(s.t, err)
899911
require.Len(s.t, rows.Records, 2)
900912
require.EqualValues(s.t, 1, rows.Records[0][0].Value())
901913
require.EqualValues(s.t, 2, rows.Records[1][0].Value())
902914
require.Equal(s.t, "active", rows.Records[0][1].Value())
903-
if mysqlEnumUsesOrdinals(mySource) {
915+
require.Equal(s.t, "a,b", rows.Records[0][2].Value())
916+
if mysqlUsesOrdinals(mySource) {
904917
require.Equal(s.t, "1", rows.Records[1][1].Value())
918+
require.Equal(s.t, "3", rows.Records[1][2].Value())
905919
} else {
906920
require.Equal(s.t, "active", rows.Records[1][1].Value())
921+
require.Equal(s.t, "a,b", rows.Records[1][2].Value())
907922
}
908923

909924
env.Cancel(s.t.Context())

‎flow/model/qrecord_avro_size_test.go‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,14 @@ func TestAvroSizeComputation(t *testing.T) {
275275
return types.QValueUint16Enum{Val: uint16(rand.IntN(65536))}
276276
},
277277
},
278+
{
279+
name: "uint64set",
280+
kind: types.QValueKindUint64Set,
281+
numRecords: 10_000,
282+
genValue: func() types.QValue {
283+
return types.QValueUint64Set{Val: rand.Uint64()}
284+
},
285+
},
278286
{
279287
name: "timestamp",
280288
kind: types.QValueKindTimestamp,

0 commit comments

Comments
 (0)