mysql: convert BIT to UInt64 in binary type path for new mirrors (#4423)

## What

`qkindFromMysqlType` (the binary binlog / QRep type-conversion path)
mapped MySQL `BIT` to `Int64`, while `QkindFromMysqlColumnType` (the
`information_schema` / DDL text path) already mapped it to `UInt64`.
This made `BIT` columns that flow through the binary path inconsistent
with the rest of the system.

The binary path is reached for:
- columns discovered via `TABLE_MAP_EVENT` without a captured DDL event
(e.g. gh-ost / online-schema-change migrations) —
`processTableMapEventSchema`
- the QRep watermark column type
- the `QRecordSchemaFromMysqlFields` fallback for columns absent from
the table schema

A plain `ALTER TABLE ... ADD COLUMN bit(...)` is **not** affected — it's
captured as a DDL query event and already routes through the `UInt64`
text path.

## Change

- Map `BIT` to `UInt64` in `qkindFromMysqlType`, gated behind a new
`InternalVersion_MySQLConvertBitToUInt64`.
- Thread the mirror version through `qkindFromMysqlType` /
`QRecordSchemaFromMysqlFields` and their callers (CDC, QRep, e2e
helper).
- Add a unit test covering the version gate.

## Why gate it

The destination column type is fixed when a column is first created. An
existing mirror that already materialized a `BIT` column as `Int64` via
the binary path would break if the code started emitting `UInt64` values
into that `Int64` column on upgrade (schema deltas add columns, they
don't retype existing ones). Gating on `InternalVersion` keeps existing
mirrors on their previous behavior and gives only new mirrors the
consistent `UInt64` mapping.

## Testing

- New unit test `TestQkindFromMysqlType_Bit` asserts `Int64` below the
gate version and `UInt64` at/above it.
- `go build ./...` and the `connectors/mysql` unit tests pass.

> Note: a normal full-table BIT e2e test does **not** exercise this
path, since in-schema BIT columns resolve their type via the ungated
text path. The meaningful behavioral difference is limited to the
gh-ost/`TABLE_MAP_EVENT`-detected-column and QRep cases described above.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: dtunikov

flow/connectors/mysql/cdc.go

@@ -914,7 +914,7 @@ func (c *MySqlConnector) processTableMapEventSchema(
 				charset = uint16(collation)
 			}
 			mytype := tableMap.ColumnType[idx]
-			qkind, err := qkindFromMysqlType(mytype, unsignedMap[idx], charset)
+			qkind, err := qkindFromMysqlType(mytype, unsignedMap[idx], charset, req.InternalVersion)
 			if err != nil {
 				c.logger.Warn("Unknown MySQL type for new column, skipping",
 					slog.String("table", sourceTableName),

flow/connectors/mysql/qrep.go

@@ -141,7 +141,7 @@ func (c *MySqlConnector) GetQRepPartitions(
 	watermarkField := rs.Fields[1]
 	watermarkMyType := watermarkField.Type
 	watermarkUnsigned := (watermarkField.Flag & mysql.UNSIGNED_FLAG) != 0
-	watermarkQKind, err := qkindFromMysqlType(watermarkField.Type, watermarkUnsigned, watermarkField.Charset)
+	watermarkQKind, err := qkindFromMysqlType(watermarkField.Type, watermarkUnsigned, watermarkField.Charset, config.Version)
 	if err != nil {
 		return nil, fmt.Errorf("failed to convert mysql type to qvaluekind: %w", err)
 	}
@@ -231,7 +231,7 @@ func (c *MySqlConnector) PullQRepRecords(
 	c.deltaBytesRead.Store(0)
 	totalRecords := int64(0)
 	onResult := func(rs *mysql.Result) error {
-		schema, err := QRecordSchemaFromMysqlFields(tableSchema, rs.Fields)
+		schema, err := QRecordSchemaFromMysqlFields(tableSchema, rs.Fields, config.Version)
 		if err != nil {
 			return err
 		}

flow/connectors/mysql/qvalue_convert.go

@@ -24,7 +24,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
-func qkindFromMysqlType(mytype byte, unsigned bool, charset uint16) (types.QValueKind, error) {
+func qkindFromMysqlType(mytype byte, unsigned bool, charset uint16, version uint32) (types.QValueKind, error) {
 	switch mytype {
 	case mysql.MYSQL_TYPE_TINY:
 		if unsigned {
@@ -66,6 +66,9 @@ func qkindFromMysqlType(mytype byte, unsigned bool, charset uint16) (types.QValu
 	case mysql.MYSQL_TYPE_YEAR:
 		return types.QValueKindInt16, nil
 	case mysql.MYSQL_TYPE_BIT:
+		if version >= shared.InternalVersion_MySQLConvertBitToUInt64 {
+			return types.QValueKindUInt64, nil
+		}
 		return types.QValueKindInt64, nil
 	case mysql.MYSQL_TYPE_JSON:
 		return types.QValueKindJSON, nil
@@ -92,7 +95,7 @@ func qkindFromMysqlType(mytype byte, unsigned bool, charset uint16) (types.QValu
 	}
 }
 
-func QRecordSchemaFromMysqlFields(tableSchema *protos.TableSchema, fields []*mysql.Field) (types.QRecordSchema, error) {
+func QRecordSchemaFromMysqlFields(tableSchema *protos.TableSchema, fields []*mysql.Field, version uint32) (types.QRecordSchema, error) {
 	tableColumns := make(map[string]*protos.FieldDescription, len(tableSchema.Columns))
 	for _, col := range tableSchema.Columns {
 		tableColumns[col.Name] = col
@@ -112,7 +115,7 @@ func QRecordSchemaFromMysqlFields(tableSchema *protos.TableSchema, fields []*mys
 		} else {
 			var err error
 			unsigned := (field.Flag & mysql.UNSIGNED_FLAG) != 0
-			qkind, err = qkindFromMysqlType(field.Type, unsigned, field.Charset)
+			qkind, err = qkindFromMysqlType(field.Type, unsigned, field.Charset, version)
 			if err != nil {
 				return types.QRecordSchema{}, err
 			}

flow/connectors/mysql/qvalue_convert_test.go

@@ -4,9 +4,40 @@ import (
 	"testing"
 	"time"
 
+	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/stretchr/testify/require"
+
+	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
 )
 
+func TestQkindFromMysqlType_Bit(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		version uint32
+		want    types.QValueKind
+	}{
+		{"first version maps BIT to Int64", shared.InternalVersion_First, types.QValueKindInt64},
+		{
+			"version just before the gate maps BIT to Int64",
+			shared.InternalVersion_MySQLConvertBitToUInt64 - 1,
+			types.QValueKindInt64,
+		},
+		{
+			"gating version maps BIT to UInt64",
+			shared.InternalVersion_MySQLConvertBitToUInt64,
+			types.QValueKindUInt64,
+		},
+		{"latest version maps BIT to UInt64", shared.InternalVersion_Latest, types.QValueKindUInt64},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			qkind, err := qkindFromMysqlType(mysql.MYSQL_TYPE_BIT, false, 0, 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/e2e/mysql.go

@@ -193,7 +193,7 @@ func (s *MySqlSource) GetRows(ctx context.Context, suffix string, table string,
 		return nil, err
 	}
 
-	schema, err := connmysql.QRecordSchemaFromMysqlFields(tableSchemas[tableName], rs.Fields)
+	schema, err := connmysql.QRecordSchemaFromMysqlFields(tableSchemas[tableName], rs.Fields, shared.InternalVersion_Latest)
 	if err != nil {
 		return nil, err
 	}

flow/shared/constants.go

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