fix(mysql): consistent SET column replication between snapshot and cdc

flow/connectors/bigquery/qvalue_convert.go

@@ -22,7 +22,7 @@ func qValueKindToBigQueryType(columnDescription *protos.FieldDescription, nullab
 	// integer types
 	case types.QValueKindInt8, types.QValueKindInt16, types.QValueKindInt32, types.QValueKindInt64,
 		types.QValueKindUInt8, types.QValueKindUInt16, types.QValueKindUInt32, types.QValueKindUInt64,
-		types.QValueKindUint16Enum:
+		types.QValueKindUint16Enum, types.QValueKindUint64Set:
 		bqField.Type = bigquery.IntegerFieldType
 	// decimal types
 	case types.QValueKindFloat32, types.QValueKindFloat64:

flow/connectors/mysql/qrep.go

@@ -253,7 +253,7 @@ func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string) str
 		}
 
 		converted := common.QuoteMySQLIdentifier(col.Name)
-		if col.Type == string(types.QValueKindUint16Enum) {
+		if col.Type == string(types.QValueKindUint16Enum) || col.Type == string(types.QValueKindUint64Set) {
 			converted = fmt.Sprintf("CAST(%s AS UNSIGNED) AS %s", converted, converted)
 			selectAsterisk = false
 		}

flow/connectors/mysql/qvalue_convert.go

@@ -260,6 +260,8 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
 		switch qkind {
 		case types.QValueKindUint16Enum:
 			return types.QValueUint16Enum{Val: uint16(v)}, nil
+		case types.QValueKindUint64Set:
+			return types.QValueUint64Set{Val: v}, nil
 		case types.QValueKindBoolean:
 			return types.QValueBoolean{Val: v != 0}, nil
 		case types.QValueKindInt8:
@@ -508,6 +510,8 @@ func QValueFromMysqlRowEvent(
 				}
 			}
 			return types.QValueString{Val: strings.Join(set, ",")}, nil
+		case types.QValueKindUint64Set:
+			return types.QValueUint64Set{Val: uint64(val)}, nil
 		case types.QValueKindUint16Enum:
 			return types.QValueUint16Enum{Val: uint16(val)}, nil
 		case types.QValueKindEnum: // enum

flow/connectors/mysql/qvalue_convert_test.go

@@ -58,6 +58,26 @@ func TestQkindFromMysqlType_Bit(t *testing.T) {
 	}
 }
 
+func TestQkindFromMysqlColumnType_Set(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		metadata bool
+		version  uint32
+		want     types.QValueKind
+	}{
+		{"with metadata maps SET to String", true, shared.InternalVersion_Latest, types.QValueKindString},
+		{"without metadata, before gate maps SET to String", false, shared.InternalVersion_MySQL5ConvertSetsToInts - 1, types.QValueKindString},
+		{"without metadata, gate maps SET to Uint64Set", false, shared.InternalVersion_MySQL5ConvertSetsToInts, types.QValueKindUint64Set},
+		{"without metadata, latest maps SET to Uint64Set", false, shared.InternalVersion_Latest, types.QValueKindUint64Set},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			qkind, err := QkindFromMysqlColumnType("set('a','b','c')", tc.metadata, tc.version)
+			require.NoError(t, err)
+			require.Equal(t, tc.want, qkind)
+		})
+	}
+}
+
 func TestProcessTime(t *testing.T) {
 	epoch := time.Unix(0, 0).UTC()
 	for _, ts := range []struct {

flow/connectors/mysql/type_conversion.go

@@ -18,10 +18,16 @@ func QkindFromMysqlColumnType(ct string, binlogRowMetadataSupported bool, versio
 	switch strings.ToLower(ct) {
 	case "json":
 		return types.QValueKindJSON, nil
-	case "char", "varchar", "text", "set", "tinytext", "mediumtext", "longtext",
+	case "char", "varchar", "text", "tinytext", "mediumtext", "longtext",
 		"clob", "varchar2", // maria
 		"xmltype": // maria
 		return types.QValueKindString, nil
+	case "set":
+		// keeps snapshot and cdc consistent for mysql without rows metadata
+		if !binlogRowMetadataSupported && version >= shared.InternalVersion_MySQL5ConvertSetsToInts {
+			return types.QValueKindUint64Set, nil
+		}
+		return types.QValueKindString, nil
 	case "enum":
 		if !binlogRowMetadataSupported && version >= shared.InternalVersion_MySQL5ConvertEnumsToInts {
 			return types.QValueKindUint16Enum, nil

flow/connectors/postgres/qvalue_convert.go

@@ -55,6 +55,8 @@ func qValueKindToPostgresType(colTypeStr string) string {
 		return "BIGINT"
 	case types.QValueKindUint16Enum:
 		return "INTEGER"
+	case types.QValueKindUint64Set:
+		return "BIGINT"
 	case types.QValueKindFloat32:
 		return "REAL"
 	case types.QValueKindFloat64:

flow/connectors/utils/cdc_store.go

@@ -99,6 +99,7 @@ func init() {
 	gob.Register(types.QValueString{})
 	gob.Register(types.QValueEnum{})
 	gob.Register(types.QValueUint16Enum{})
+	gob.Register(types.QValueUint64Set{})
 	gob.Register(types.QValueTimestamp{})
 	gob.Register(types.QValueTimestampTZ{})
 	gob.Register(types.QValueDate{})

flow/e2e/clickhouse_mysql_test.go

@@ -24,7 +24,9 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
-func mysqlEnumUsesOrdinals(source *MySqlSource) bool {
+// mysqlUsesOrdinals reports whether the source lacks binlog row metadata, in which case CDC
+// sees enum ordinals and set bitmasks as integers rather than labels.
+func mysqlUsesOrdinals(source *MySqlSource) bool {
 	return source.Config.Flavor == protos.MySqlFlavor_MYSQL_MYSQL &&
 		source.Config.ReplicationMechanism == protos.MySqlReplicationMechanism_MYSQL_FILEPOS
 }
@@ -636,26 +638,30 @@ func (s ClickHouseSuite) Test_MySQL_Enum() {
 	RequireEnvCanceled(s.t, env)
 }
 
-func (s ClickHouseSuite) Test_MySQL_Enum_Consistency() {
+// Test_MySQL_Enum_Set_Consistency verifies that ENUM and SET columns produce identical values
+// via snapshot and CDC. On servers without binlog row metadata, both are emitted as integers
+// (enum ordinal, set bitmask); otherwise as labels.
+func (s ClickHouseSuite) Test_MySQL_Enum_Set_Consistency() {
 	mySource, ok := s.source.(*MySqlSource)
 	if !ok {
 		s.t.Skip("only applies to mysql")
 	}
 
-	srcTableName := "test_my_enum_consistency"
+	srcTableName := "test_my_enum_set_consistency"
 	srcFullName := s.attachSchemaSuffix(srcTableName)
-	dstTableName := "test_my_enum_consistency_dst"
+	dstTableName := "test_my_enum_set_consistency_dst"
 
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
 		CREATE TABLE IF NOT EXISTS %s (
 			id SERIAL PRIMARY KEY,
-			status ENUM('active', 'inactive', 'pending') NOT NULL
+			status ENUM('active', 'inactive', 'pending') NOT NULL,
+			tags SET('a', 'b', 'c') NOT NULL
 		)
 	`, srcFullName)))
 
 	// Insert row before snapshot
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
-		`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
+		`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
 
 	connectionGen := FlowConnectionGenerationConfig{
 		FlowJobName:      s.attachSuffix(srcTableName),
@@ -670,50 +676,58 @@ func (s ClickHouseSuite) Test_MySQL_Enum_Consistency() {
 	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
 
 	// Wait for snapshot row to appear in destination
-	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status", 1)
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status,tags", 1)
 
 	// Insert row via CDC — on old MySQL this comes as integer from binlog
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
-		`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
+		`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
 
 	// Wait for CDC row
-	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status", 2)
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status,tags", 2)
 
-	// Verify both rows have the same status value (consistency between snapshot and CDC)
-	rows, err := s.GetRows(dstTableName, "id,status")
+	// Verify snapshot and CDC produce the same values for both columns
+	rows, err := s.GetRows(dstTableName, "id,status,tags")
 	require.NoError(s.t, err)
 	require.Len(s.t, rows.Records, 2)
 	require.Equal(s.t, rows.Records[0][1].Value(), rows.Records[1][1].Value(),
 		"snapshot and CDC enum values should be consistent")
-	if mysqlEnumUsesOrdinals(mySource) {
+	require.Equal(s.t, rows.Records[0][2].Value(), rows.Records[1][2].Value(),
+		"snapshot and CDC set values should be consistent")
+	if mysqlUsesOrdinals(mySource) {
 		require.EqualValues(s.t, 1, rows.Records[0][1].Value())
+		require.EqualValues(s.t, 3, rows.Records[0][2].Value()) // 'a,b' bitmask = 1|2
 	} else {
 		require.Equal(s.t, "active", rows.Records[0][1].Value())
+		require.Equal(s.t, "a,b", rows.Records[0][2].Value())
 	}
 
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
 
-func (s ClickHouseSuite) Test_MySQL_Enum_Consistency_Version0() {
+// Test_MySQL_Enum_Set_Consistency_Version0 documents the legacy divergence: on pre-metadata
+// servers, mirrors created before the enum/set-to-int versions emit labels via snapshot but
+// ordinals/bitmasks via CDC. Existing mirrors deliberately keep this behavior for stability.
+func (s ClickHouseSuite) Test_MySQL_Enum_Set_Consistency_Version0() {
 	mySource, ok := s.source.(*MySqlSource)
 	if !ok {
 		s.t.Skip("only applies to mysql")
 	}
 
-	srcTableName := "test_my_enum_consistency_v0"
+	srcTableName := "test_my_enum_set_consistency_v0"
 	srcFullName := s.attachSchemaSuffix(srcTableName)
-	dstTableName := "test_my_enum_consistency_v0_dst"
+	dstTableName := "test_my_enum_set_consistency_v0_dst"
 
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
 		CREATE TABLE IF NOT EXISTS %s (
 			id SERIAL PRIMARY KEY,
-			status ENUM('active', 'inactive', 'pending') NOT NULL
+			status ENUM('active', 'inactive', 'pending') NOT NULL,
+			tags SET('a', 'b', 'c') NOT NULL
 		)
 	`, srcFullName)))
 
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
-		`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
+		`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
 
 	connectionGen := FlowConnectionGenerationConfig{
 		FlowJobName:      s.attachSuffix(srcTableName),
@@ -729,23 +743,26 @@ func (s ClickHouseSuite) Test_MySQL_Enum_Consistency_Version0() {
 	env := ExecutePeerflow(s.t, tc, flowConnConfig)
 	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
 
-	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status", 1)
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status,tags", 1)
 
 	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
-		`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
+		`INSERT INTO %s (status, tags) VALUES ('active', 'a,b')`, srcFullName)))
 
-	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status", 2)
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status,tags", 2)
 
-	rows, err := s.GetRows(dstTableName, "id,status")
+	rows, err := s.GetRows(dstTableName, "id,status,tags")
 	require.NoError(s.t, err)
 	require.Len(s.t, rows.Records, 2)
 	require.EqualValues(s.t, 1, rows.Records[0][0].Value())
 	require.EqualValues(s.t, 2, rows.Records[1][0].Value())
 	require.Equal(s.t, "active", rows.Records[0][1].Value())
-	if mysqlEnumUsesOrdinals(mySource) {
+	require.Equal(s.t, "a,b", rows.Records[0][2].Value())
+	if mysqlUsesOrdinals(mySource) {
 		require.Equal(s.t, "1", rows.Records[1][1].Value())
+		require.Equal(s.t, "3", rows.Records[1][2].Value())
 	} else {
 		require.Equal(s.t, "active", rows.Records[1][1].Value())
+		require.Equal(s.t, "a,b", rows.Records[1][2].Value())
 	}
 
 	env.Cancel(s.t.Context())

flow/model/qrecord_avro_size_test.go

@@ -275,6 +275,14 @@ func TestAvroSizeComputation(t *testing.T) {
 				return types.QValueUint16Enum{Val: uint16(rand.IntN(65536))}
 			},
 		},
+		{
+			name:       "uint64set",
+			kind:       types.QValueKindUint64Set,
+			numRecords: 10_000,
+			genValue: func() types.QValue {
+				return types.QValueUint64Set{Val: rand.Uint64()}
+			},
+		},
 		{
 			name:       "timestamp",
 			kind:       types.QValueKindTimestamp,

flow/model/qrecord_copy_from_source.go

@@ -94,6 +94,8 @@ func (src *QRecordCopyFromSource) Values() ([]any, error) {
 			values[i] = v.Val
 		case types.QValueUint16Enum:
 			values[i] = v.Val
+		case types.QValueUint64Set:
+			values[i] = v.Val
 		case types.QValueCIDR:
 			values[i] = v.Val
 		case types.QValueINET:

flow/model/qvalue/avro_converter.go

@@ -76,7 +76,7 @@ func GetAvroSchemaFromQValueKind(
 		return avro.NewPrimitiveSchema(avro.String, nil), nil
 	case types.QValueKindInt8, types.QValueKindInt16, types.QValueKindInt32, types.QValueKindInt64,
 		types.QValueKindUInt8, types.QValueKindUInt16, types.QValueKindUInt32, types.QValueKindUInt64,
-		types.QValueKindUint16Enum:
+		types.QValueKindUint16Enum, types.QValueKindUint64Set:
 		return avro.NewPrimitiveSchema(avro.Long, nil), nil
 	case types.QValueKindFloat32:
 		if targetDWH == protos.DBType_BIGQUERY {
@@ -294,6 +294,8 @@ func QValueToAvro(
 		return c.processNullableUnion(int64(v.Val)), varIntSize(int64(v.Val), sizeOpt), nil
 	case types.QValueUint16Enum:
 		return c.processNullableUnion(int64(v.Val)), varIntSize(int64(v.Val), sizeOpt), nil
+	case types.QValueUint64Set:
+		return c.processNullableUnion(int64(v.Val)), varIntSize(int64(v.Val), sizeOpt), nil
 	case types.QValueUInt32:
 		return c.processNullableUnion(int64(v.Val)), varIntSize(int64(v.Val), sizeOpt), nil
 	case types.QValueUInt64:

flow/model/qvalue/equals.go

@@ -82,6 +82,11 @@ func Equals(qv types.QValue, other types.QValue) bool {
 			return q.Val == otherVal.Val
 		}
 		return false
+	case types.QValueUint64Set:
+		if otherVal, ok := other.(types.QValueUint64Set); ok {
+			return q.Val == otherVal.Val
+		}
+		return false
 	case types.QValueINET:
 		return compareString(q.Val, otherValue)
 	case types.QValueCIDR:

flow/pua/peerdb.go

@@ -267,6 +267,8 @@ func LuaRowNewIndex(ls *lua.LState) int {
 		newqv = types.QValueEnum{Val: lua.LVAsString(val)}
 	case types.QValueKindUint16Enum:
 		newqv = types.QValueUint16Enum{Val: uint16(lua.LVAsNumber(val))}
+	case types.QValueKindUint64Set:
+		newqv = types.QValueUint64Set{Val: uint64(lua.LVAsNumber(val))}
 	case types.QValueKindTimestamp:
 		newqv = types.QValueTimestamp{Val: LVAsTime(ls, val)}
 	case types.QValueKindTimestampTZ:

flow/shared/constants.go

@@ -39,6 +39,8 @@ const (
 	InternalVersion_MySQL5ConvertEnumsToInts
 	// MySQL: convert BIT to UInt64
 	InternalVersion_MySQLConvertBitToUInt64
+	// MySQL: convert sets to integers for older versions without binlog row metadata support
+	InternalVersion_MySQL5ConvertSetsToInts
 
 	TotalNumberOfInternalVersions
 	InternalVersion_Latest = TotalNumberOfInternalVersions - 1

flow/shared/types/kind.go

@@ -25,6 +25,7 @@ const (
 	QValueKindString      QValueKind = "string"
 	QValueKindEnum        QValueKind = "enum"
 	QValueKindUint16Enum  QValueKind = "uint16enum"
+	QValueKindUint64Set   QValueKind = "uint64set"
 	QValueKindTimestamp   QValueKind = "timestamp"
 	QValueKindTimestampTZ QValueKind = "timestamptz"
 	QValueKindDate        QValueKind = "date"
@@ -85,6 +86,7 @@ var QValueKindToSnowflakeTypeMap = map[QValueKind]string{
 	QValueKindString:      "STRING",
 	QValueKindEnum:        "STRING",
 	QValueKindUint16Enum:  "INTEGER",
+	QValueKindUint64Set:   "INTEGER",
 	QValueKindJSON:        "VARIANT",
 	QValueKindJSONB:       "VARIANT",
 	QValueKindTimestamp:   "TIMESTAMP_NTZ",
@@ -138,6 +140,7 @@ var QValueKindToClickHouseTypeMap = map[QValueKind]string{
 	QValueKindString:      "String",
 	QValueKindEnum:        "LowCardinality(String)",
 	QValueKindUint16Enum:  "UInt16",
+	QValueKindUint64Set:   "UInt64",
 	QValueKindJSON:        "String",
 	QValueKindTimestamp:   "DateTime64(6)",
 	QValueKindTimestampTZ: "DateTime64(6)",

flow/shared/types/qvalue.go

@@ -321,6 +321,22 @@ func (v QValueUint16Enum) LValue(ls *lua.LState) lua.LValue {
 	return lua.LNumber(v.Val)
 }
 
+type QValueUint64Set struct {
+	Val uint64
+}
+
+func (QValueUint64Set) Kind() QValueKind {
+	return QValueKindUint64Set
+}
+
+func (v QValueUint64Set) Value() any {
+	return v.Val
+}
+
+func (v QValueUint64Set) LValue(ls *lua.LState) lua.LValue {
+	return lua.LNumber(v.Val)
+}
+
 type QValueTimestamp struct {
 	Val time.Time
 }