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

flow/connectors/bigquery/qvalue_convert.go

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

flow/connectors/mysql/cdc.go

@@ -26,7 +26,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"
@@ -42,7 +41,7 @@ func (c *MySqlConnector) GetTableSchema(
 ) (map[string]*protos.TableSchema, error) {
 	res := make(map[string]*protos.TableSchema, len(tableMappings))
 	for _, tm := range tableMappings {
-		tableSchema, err := c.getTableSchemaForTable(ctx, env, tm, system)
+		tableSchema, err := c.getTableSchemaForTable(ctx, env, tm, system, version)
 		if err != nil {
 			c.logger.Info("error fetching schema", slog.String("table", tm.SourceTableIdentifier), slog.Any("error", err))
 			return nil, err
@@ -59,6 +58,7 @@ func (c *MySqlConnector) getTableSchemaForTable(
 	env map[string]string,
 	tm *protos.TableMapping,
 	system protos.TypeSystem,
+	mirrorVersion uint32,
 ) (*protos.TableSchema, error) {
 	qualifiedTable, err := common.ParseTableIdentifier(tm.SourceTableIdentifier)
 	if err != nil {
@@ -87,8 +87,13 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		name       string
 		seqInIndex int64
 	}
-	var primaryEntries []pkEntry
 
+	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
+	}
+
+	var primaryEntries []pkEntry
 	for idx := range rs.RowNumber() {
 		columnName, err := rs.GetString(idx, 0)
 		if err != nil {
@@ -114,7 +119,7 @@ func (c *MySqlConnector) getTableSchemaForTable(
 		if err != nil {
 			return nil, err
 		}
-		qkind, err := QkindFromMysqlColumnType(dataType)
+		qkind, err := QkindFromMysqlColumnType(dataType, binlogRowMetadataSupported, mirrorVersion)
 		if err != nil {
 			return nil, err
 		}
@@ -338,15 +343,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 {
@@ -516,7 +516,7 @@ func (c *MySqlConnector) PullRecords(
 			for _, stmt := range stmts {
 				if alterTableStmt, ok := stmt.(*ast.AlterTableStmt); ok {
 					if err := c.processAlterTableQuery(
-						ctx, catalogPool, req, alterTableStmt, string(ev.Schema), binlogRowMetadataSupported); err != nil {
+						ctx, catalogPool, req, alterTableStmt, string(ev.Schema), binlogRowMetadataSupported, req.InternalVersion); err != nil {
 						return fmt.Errorf("failed to process ALTER TABLE query: %w", err)
 					}
 				}
@@ -706,7 +706,7 @@ func (c *MySqlConnector) PullRecords(
 
 func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool shared.CatalogPool,
 	req *model.PullRecordsRequest[model.RecordItems], stmt *ast.AlterTableStmt, stmtSchema string,
-	binlogRowMetadataSupported bool,
+	binlogRowMetadataSupported bool, mirrorVersion uint32,
 ) error {
 	// if ALTER TABLE doesn't have database/schema name, use one attached to event
 	var sourceSchemaName string
@@ -753,7 +753,7 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 						slog.String("tableName", sourceTableName))
 				}
 
-				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr())
+				qkind, err := QkindFromMysqlColumnType(col.Tp.InfoSchemaStr(), binlogRowMetadataSupported, mirrorVersion)
 				if err != nil {
 					return err
 				}

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"
 )
 
@@ -555,3 +556,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,31 @@ func (c *MySqlConnector) GetDefaultPartitionKeyForTables(
 	}, nil
 }
 
+func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string) string {
+	columns := []string{}
+	selectAsterisk := true
+	for _, col := range cols {
+		if slices.Contains(exclude, col.Name) {
+			selectAsterisk = false
+			continue
+		}
+
+		converted := common.QuoteMySQLIdentifier(col.Name)
+		if col.Type == string(types.QValueKindUint16Enum) {
+			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,
@@ -187,22 +212,13 @@ func (c *MySqlConnector) PullQRepRecords(
 	stream *model.QRecordStream,
 ) (int64, int64, error) {
 	tableSchema, err := c.getTableSchemaForTable(ctx, config.Env,
-		&protos.TableMapping{SourceTableIdentifier: config.WatermarkTable}, protos.TypeSystem_Q)
+		&protos.TableMapping{SourceTableIdentifier: config.WatermarkTable}, protos.TypeSystem_Q,
+		config.Version)
 	if err != nil {
 		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, ",")
-	}
-
+	selectedColumns := buildSelectedColumns(tableSchema.Columns, config.Exclude)
 	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,84 @@
+package connmysql
+
+import (
+	"testing"
+
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
+)
+
+func TestBuildSelectedColumns(t *testing.T) {
+	testCases := []struct {
+		name                    string
+		expectedSelectedColumns string
+		cols                    []*protos.FieldDescription
+		exclude                 []string
+	}{
+		{
+			name: "no excluded columns, string enums",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "name", Type: string(types.QValueKindString)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                 []string{},
+			expectedSelectedColumns: "*",
+		},
+		{
+			name: "one excluded column",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "name", Type: string(types.QValueKindString)},
+			},
+			exclude:                 []string{"name"},
+			expectedSelectedColumns: "`id`",
+		},
+		{
+			name: "uint16enum column is cast to unsigned",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindUint16Enum)},
+			},
+			exclude:                 []string{},
+			expectedSelectedColumns: "`id`, CAST(`status` AS UNSIGNED) AS `status`",
+		},
+		{
+			name: "string enum column is not cast",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindEnum)},
+			},
+			exclude:                 []string{},
+			expectedSelectedColumns: "*",
+		},
+		{
+			name: "uint16enum with exclude",
+			cols: []*protos.FieldDescription{
+				{Name: "id", Type: string(types.QValueKindInt32)},
+				{Name: "status", Type: string(types.QValueKindUint16Enum)},
+				{Name: "created_at", Type: string(types.QValueKindTimestamp)},
+			},
+			exclude:                 []string{"created_at"},
+			expectedSelectedColumns: "`id`, CAST(`status` AS UNSIGNED) AS `status`",
+		},
+		{
+			name: "string enum with exclude",
+			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"},
+			expectedSelectedColumns: "`id`, `status`",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			selectedColumns := buildSelectedColumns(tc.cols, tc.exclude)
+			if selectedColumns != tc.expectedSelectedColumns {
+				t.Errorf("expected selected columns to be %s, but got %s", tc.expectedSelectedColumns, selectedColumns)
+			}
+		})
+	}
+}

flow/connectors/mysql/qvalue_convert.go

@@ -214,6 +214,8 @@ func QValueFromMysqlFieldValue(qkind types.QValueKind, mytype byte, fv mysql.Fie
 	case mysql.FieldValueTypeUnsigned:
 		v := fv.AsUint64()
 		switch qkind {
+		case types.QValueKindUint16Enum:
+			return types.QValueUint16Enum{Val: uint16(v)}, nil
 		case types.QValueKindBoolean:
 			return types.QValueBoolean{Val: v != 0}, nil
 		case types.QValueKindInt8:
@@ -398,6 +400,8 @@ func QValueFromMysqlRowEvent(
 				}
 			}
 			return types.QValueString{Val: strings.Join(set, ",")}, nil
+		case types.QValueKindUint16Enum:
+			return types.QValueUint16Enum{Val: uint16(val)}, nil
 		case types.QValueKindEnum: // enum
 			if val == 0 {
 				return types.QValueEnum{Val: ""}, nil

flow/connectors/mysql/schema.go

@@ -84,6 +84,11 @@ func (c *MySqlConnector) GetTablesInSchema(
 }
 
 func (c *MySqlConnector) GetColumns(ctx context.Context, version uint32, schema string, table string) (*protos.TableColumnsResponse, error) {
+	binlogRowMetadataSupported, err := c.IsBinlogRowMetadataSupported(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to determine if binlog row metadata is supported: %w", err)
+	}
+
 	rs, err := c.Execute(ctx, fmt.Sprintf(`
 		SELECT column_name, column_type, column_key
 		FROM information_schema.columns
@@ -109,7 +114,7 @@ func (c *MySqlConnector) GetColumns(ctx context.Context, version uint32, schema
 		if err != nil {
 			return nil, err
 		}
-		qkind, err := QkindFromMysqlColumnType(columnType)
+		qkind, err := QkindFromMysqlColumnType(columnType, binlogRowMetadataSupported, version)
 		if err != nil {
 			return nil, err
 		}

flow/connectors/mysql/type_conversion.go

@@ -4,10 +4,11 @@ import (
 	"fmt"
 	"strings"
 
+	"github.com/PeerDB-io/peerdb/flow/shared"
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
-func QkindFromMysqlColumnType(ct string) (types.QValueKind, error) {
+func QkindFromMysqlColumnType(ct string, binlogRowMetadataSupported bool, version uint32) (types.QValueKind, error) {
 	// https://mariadb.com/docs/server/reference/data-types/date-and-time-data-types/timestamp#tab-current-1
 	ct, _ = strings.CutSuffix(ct, " /* mariadb-5.3 */")
 	ct, _ = strings.CutSuffix(ct, " zerofill")
@@ -19,6 +20,9 @@ func QkindFromMysqlColumnType(ct string) (types.QValueKind, error) {
 	case "char", "varchar", "text", "set", "tinytext", "mediumtext", "longtext":
 		return types.QValueKindString, nil
 	case "enum":
+		if !binlogRowMetadataSupported && version >= shared.InternalVersion_MySQL5ConvertEnumsToInts {
+			return types.QValueKindUint16Enum, nil
+		}
 		return types.QValueKindEnum, nil
 	case "binary", "varbinary", "blob", "tinyblob", "mediumblob", "longblob":
 		return types.QValueKindBytes, nil

flow/connectors/postgres/qvalue_convert.go

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

flow/connectors/utils/cdc_store.go

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