fix: cast enums to integers during qrep for mysql without binlog metadata …

flow/connectors/mysql/cdc.go

@@ -27,7 +27,6 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/model"
 	"github.com/PeerDB-io/peerdb/flow/otel_metrics"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
-	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/datatypes"
 	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
@@ -320,15 +319,10 @@ func (c *MySqlConnector) PullRecords(
 		return err
 	}
 
-	versionToCmp := mysql_validation.MySQLMinVersionForBinlogRowMetadata
-	if c.config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
-		versionToCmp = mysql_validation.MariaDBMinVersionForBinlogRowMetadata
-	}
-	cmp, err := c.CompareServerVersion(ctx, versionToCmp)
+	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to get server version: %w", err)
+		return fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
 	}
-	binlogRowMetadataSupported := cmp >= 0
 
 	syncer, mystream, gset, pos, err := c.startStreaming(ctx, req.LastOffset.Text)
 	if err != nil {

flow/connectors/mysql/mysql.go

@@ -22,6 +22,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
+	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
@@ -549,3 +550,15 @@ func (c *MySqlConnector) GetTableSizeEstimatedBytes(ctx context.Context, tableId
 	defer rs.Close()
 	return rs.GetInt(0, 0)
 }
+
+func (c *MySqlConnector) IsBinlogRowMetadataSupported(ctx context.Context) (bool, error) {
+	versionToCmp := mysql_validation.MySQLMinVersionForBinlogRowMetadata
+	if c.config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
+		versionToCmp = mysql_validation.MariaDBMinVersionForBinlogRowMetadata
+	}
+	cmp, err := c.CompareServerVersion(ctx, versionToCmp)
+	if err != nil {
+		return false, fmt.Errorf("failed to get server version: %w", err)
+	}
+	return cmp >= 0, nil
+}

flow/connectors/mysql/qrep.go

@@ -177,6 +177,33 @@ func (c *MySqlConnector) GetDefaultPartitionKeyForTables(
 	}, nil
 }
 
+func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string, isBinlogMetadataSupported bool, mirrorVersion uint32) string {
+	columns := []string{}
+	selectAsterisk := true
+	for _, col := range cols {
+		if slices.Contains(exclude, col.Name) {
+			selectAsterisk = false
+			continue
+		}
+
+		converted := common.QuoteMySQLIdentifier(col.Name)
+		if !isBinlogMetadataSupported && col.Type == string(types.QValueKindEnum) &&
+			mirrorVersion >= shared.InternalVersion_MySQLConvertEnumsToInts {
+			// if binlog metadata is not supported, we need to cast enum columns to integers to align it with cdc stream
+			converted = fmt.Sprintf("CAST(%s AS UNSIGNED) AS %s", converted, converted)
+			selectAsterisk = false
+		}
+		columns = append(columns, converted)
+	}
+
+	selectedColumns := "*"
+	if !selectAsterisk {
+		selectedColumns = strings.Join(columns, ", ")
+	}
+
+	return selectedColumns
+}
+
 func (c *MySqlConnector) PullQRepRecords(
 	ctx context.Context,
 	catalogPool shared.CatalogPool,
@@ -192,17 +219,12 @@ func (c *MySqlConnector) PullQRepRecords(
 		return 0, 0, fmt.Errorf("failed to get schema for watermark table %s: %w", config.WatermarkTable, err)
 	}
 
-	selectedColumns := "*"
-	if len(config.Exclude) != 0 {
-		quotedColumns := make([]string, 0, len(tableSchema.Columns))
-		for _, col := range tableSchema.Columns {
-			if !slices.Contains(config.Exclude, col.Name) {
-				quotedColumns = append(quotedColumns, common.QuoteMySQLIdentifier(col.Name))
-			}
-		}
-		selectedColumns = strings.Join(quotedColumns, ",")
+	isBinlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
+	if err != nil {
+		return 0, 0, fmt.Errorf("failed to determine if binlog metadata is supported: %w", err)
 	}
 
+	selectedColumns := buildSelectedColumns(tableSchema.Columns, config.Exclude, isBinlogRowMetadataSupported, config.Version)
 	parsedSrcTable, err := common.ParseTableIdentifier(config.WatermarkTable)
 	if err != nil {
 		c.logger.Error("unable to parse source table", slog.Any("error", err))

flow/connectors/mysql/qrep_test.go

@@ -0,0 +1,110 @@
+package connmysql
+
+import (
+	"testing"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
+)
+
+func TestBuildSelectedColumns(t *testing.T) {
+	testCases := []struct {
+		name                      string
+		expectedSelectedColumns   string
+		cols                      []*protos.FieldDescription
+		exclude                   []string
+		isBinlogMetadataSupported bool
+		mirrorVersion             uint32
+	}{
+		{
+			name: "no excluded columns, binlog metadata supported",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "name", Type: string(types.QValueKindString)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                   []string{},
+			isBinlogMetadataSupported: true,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts,
+			expectedSelectedColumns:   "*",
+		},
+		{
+			name: "one excluded column, binlog metadata supported",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "name", Type: string(types.QValueKindString)},
+			},
+			exclude:                   []string{"name"},
+			isBinlogMetadataSupported: true,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts,
+			expectedSelectedColumns:   "`id`",
+		},
+		{
+			name: "one enum column, binlog metadata not supported, new version",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                   []string{},
+			isBinlogMetadataSupported: false,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts,
+			expectedSelectedColumns:   "`id`, CAST(`status` AS UNSIGNED) AS `status`",
+		},
+		{
+			name: "one enum column, binlog metadata not supported, old version",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                   []string{},
+			isBinlogMetadataSupported: false,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts - 1,
+			expectedSelectedColumns:   "*",
+		},
+		{
+			name: "one enum column, binlog metadata not supported, version zero",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                   []string{},
+			isBinlogMetadataSupported: false,
+			mirrorVersion:             0,
+			expectedSelectedColumns:   "*",
+		},
+		{
+			name: "one enum column, one excluded non enum column, binlog metadata supported",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+				{Name: "created_at", Type: string(types.QValueKindTimestamp)},
+			},
+			exclude:                   []string{"created_at"},
+			isBinlogMetadataSupported: true,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts,
+			expectedSelectedColumns:   "`id`, `status`",
+		},
+		{
+			name: "enum with exclude, binlog metadata not supported, old version",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+				{Name: "created_at", Type: string(types.QValueKindTimestamp)},
+			},
+			exclude:                   []string{"created_at"},
+			isBinlogMetadataSupported: false,
+			mirrorVersion:             shared.InternalVersion_MySQLConvertEnumsToInts - 1,
+			expectedSelectedColumns:   "`id`, `status`",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			selectedColumns := buildSelectedColumns(tc.cols, tc.exclude, tc.isBinlogMetadataSupported, tc.mirrorVersion)
+			if selectedColumns != tc.expectedSelectedColumns {
+				t.Errorf("expected selected columns to be %s, but got %s", tc.expectedSelectedColumns, selectedColumns)
+			}
+		})
+	}
+}

flow/connectors/mysql/qvalue_convert.go

@@ -229,6 +229,8 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
 	case mysql.FieldValueTypeUnsigned:
 		v := fv.AsUint64()
 		switch qkind {
+		case types.QValueKindEnum:
+			return types.QValueEnum{Val: strconv.FormatUint(v, 10)}, nil
 		case types.QValueKindBoolean:
 			return types.QValueBoolean{Val: v != 0}, nil
 		case types.QValueKindInt8:

flow/e2e/clickhouse_mysql_test.go

@@ -274,6 +274,59 @@ func (s ClickHouseSuite) Test_MySQL_Enum() {
 	RequireEnvCanceled(s.t, env)
 }
 
+func (s ClickHouseSuite) Test_MySQL_Enum_Consistency() {
+	if _, ok := s.source.(*MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	srcTableName := "test_my_enum_consistency"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_my_enum_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
+		)
+	`, srcFullName)))
+
+	// Insert row before snapshot
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (status) VALUES ('active')`, srcFullName)))
+
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName:      s.attachSuffix(srcTableName),
+		TableNameMapping: map[string]string{srcFullName: dstTableName},
+		Destination:      s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+	flowConnConfig.DoInitialSnapshot = true
+
+	tc := NewTemporalClient(s.t)
+	env := ExecutePeerflow(s.t, tc, flowConnConfig)
+	SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
+
+	// Wait for snapshot row to appear in destination
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,status", 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)))
+
+	// Wait for CDC row
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,status", 2)
+
+	// Verify both rows have the same status value (consistency between snapshot and CDC)
+	rows, err := s.GetRows(dstTableName, "status")
+	require.NoError(s.t, err)
+	require.Len(s.t, rows.Records, 2)
+	require.Equal(s.t, rows.Records[0][0].Value(), rows.Records[1][0].Value(),
+		"snapshot and CDC enum values should be consistent")
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
 func (s ClickHouseSuite) Test_MySQL_Vector() {
 	mysource, ok := s.source.(*MySqlSource)
 	if !ok || mysource.Config.Flavor != protos.MySqlFlavor_MYSQL_MYSQL {

flow/shared/constants.go

@@ -42,6 +42,8 @@ const (
 	InternalVersion_JsonEscapeDotsInKeys
 	// MongoDB: `_id` column values stored as-is without redundant quotes
 	InternalVersion_MongoDBIdWithoutRedundantQuotes
+	// MySQL: convert enums to integers for older versions without binlog row metadata support
+	InternalVersion_MySQLConvertEnumsToInts
 
 	TotalNumberOfInternalVersions
 	InternalVersion_Latest = TotalNumberOfInternalVersions - 1