fix(mysql): pull INVISIBLE/GIPK columns during snapshot (#4553)

## Problem

MySQL/MariaDB QRep snapshot took a `SELECT *` fast path in
`buildSelectedColumns`. Wildcard expansion **omits INVISIBLE columns** —
including the generated invisible primary key (`my_row_id`) from
`sql_generate_invisible_primary_key`. But the destination table (created
from `information_schema.columns`) and CDC (binlog row images) both
**include** those columns.

Result: snapshot rows keep destination defaults forever while CDC
populates real values — silent divergence. Worst case is GIPK, where the
invisible `my_row_id` is the table's only PK, so snapshot rows arrive
with no key, breaking dedup/ordering.

## Fix

Always build an explicit column list from the fetched table schema (drop
the `SELECT *` fast path). The schema already contains invisible/GIPK
columns, and naming them explicitly bypasses wildcard invisibility, so
the snapshot pulls them — matching CDC and the destination DDL.

## Tests

- Unit: `buildSelectedColumns` never returns `*`; invisible/GIPK columns
are listed explicitly.
- E2E (`clickhouse_mysql_test.go`):
`Test_MySQL_Invisible_Column_Consistency` and
`Test_MySQL_GIPK_Consistency` compare snapshot vs CDC values
(version-gated; GIPK is MySQL-only).

🤖 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/qrep.go

@@ -244,28 +244,20 @@ func (c *MySqlConnector) GetDefaultPartitionKeyForTables(
 }
 
 func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string) string {
-	columns := []string{}
-	selectAsterisk := true
+	columns := make([]string, 0, len(cols))
 	for _, col := range cols {
 		if slices.Contains(exclude, col.Name) {
-			selectAsterisk = false
 			continue
 		}
 
 		converted := common.QuoteMySQLIdentifier(col.Name)
 		if col.Type == string(types.QValueKindUint16Enum) || col.Type == string(types.QValueKindUint64Set) {
 			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
+	return strings.Join(columns, ", ")
 }
 
 func (c *MySqlConnector) PullQRepRecords(
@@ -285,6 +277,10 @@ func (c *MySqlConnector) PullQRepRecords(
 	}
 
 	selectedColumns := buildSelectedColumns(tableSchema.Columns, config.Exclude)
+	if selectedColumns == "" {
+		return 0, 0, fmt.Errorf("no columns selected for watermark table %s (check Exclude configuration)", config.WatermarkTable)
+	}
+
 	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

@@ -27,7 +27,7 @@ func TestBuildSelectedColumns(t *testing.T) {
 				{Name: "status", Type: string(types.QValueKindEnum)},
 			},
 			exclude:                 []string{},
-			expectedSelectedColumns: "*",
+			expectedSelectedColumns: "`id`, `name`, `status`",
 		},
 		{
 			name: "one excluded column",
@@ -73,7 +73,18 @@ func TestBuildSelectedColumns(t *testing.T) {
 				{Name: "status", Type: string(types.QValueKindEnum)},
 			},
 			exclude:                 []string{},
-			expectedSelectedColumns: "*",
+			expectedSelectedColumns: "`id`, `status`",
+		},
+		{
+			// INVISIBLE / GIPK columns are present in the fetched schema but omitted by
+			// wildcard expansion; they must appear explicitly so snapshot pulls them.
+			name: "invisible/gipk column is listed explicitly",
+			cols: []*protos.FieldDescription{
+				{Name: "my_row_id", Type: string(types.QValueKindUInt64)},
+				{Name: "name", Type: string(types.QValueKindString)},
+			},
+			exclude:                 []string{},
+			expectedSelectedColumns: "`my_row_id`, `name`",
 		},
 		{
 			name: "uint16enum with exclude",

flow/e2e/clickhouse_mysql_test.go

@@ -2551,3 +2551,139 @@ func (s ClickHouseSuite) Test_MySQL_NoPrimaryKey_CDC() {
 	env.Cancel(s.t.Context())
 	RequireEnvCanceled(s.t, env)
 }
+
+func (s ClickHouseSuite) Test_MySQL_Invisible_Column_Consistency() {
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	minVersion := mysql_validation.MySQLMinVersionForInvisibleColumns
+	if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
+		minVersion = mysql_validation.MariaDBMinVersionForInvisibleColumns
+	}
+	cmp, err := mySource.CompareServerVersion(s.t.Context(), minVersion)
+	require.NoError(s.t, err)
+	if cmp < 0 {
+		s.t.Skip("server version does not support invisible columns")
+	}
+
+	srcTableName := "test_my_invisible"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_my_invisible_dst"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id INT PRIMARY KEY,
+			name VARCHAR(50) NOT NULL,
+			secret INT NOT NULL DEFAULT 0 INVISIBLE
+		)
+	`, srcFullName)))
+
+	// Pre-snapshot row with an explicit, non-default invisible value.
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (id, name, secret) VALUES (1, 'snap', 111)`, 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)
+
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "id,name,secret", 1)
+
+	// Post-snapshot CDC row with a distinct invisible value.
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (id, name, secret) VALUES (2, 'cdc', 222)`, srcFullName)))
+
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "id,name,secret", 2)
+
+	rows, err := s.GetRows(dstTableName, "id,name,secret")
+	require.NoError(s.t, err)
+	require.Len(s.t, rows.Records, 2)
+	// Snapshot row must carry the real invisible value, not the destination default.
+	require.EqualValues(s.t, 111, rows.Records[0][2].Value(), "snapshot invisible column should be pulled")
+	require.EqualValues(s.t, 222, rows.Records[1][2].Value(), "cdc invisible column should be pulled")
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}
+
+func (s ClickHouseSuite) Test_MySQL_GIPK_Consistency() {
+	mySource, ok := s.source.(*MySqlSource)
+	if !ok {
+		s.t.Skip("only applies to mysql")
+	}
+	if mySource.Config.Flavor == protos.MySqlFlavor_MYSQL_MARIA {
+		s.t.Skip("GIPK is not supported by MariaDB")
+	}
+	cmp, err := mySource.CompareServerVersion(s.t.Context(), mysql_validation.MySQLMinVersionForGeneratedInvisiblePrimaryKey)
+	require.NoError(s.t, err)
+	if cmp < 0 {
+		s.t.Skip("server version does not support generated invisible primary key")
+	}
+
+	srcTableName := "test_my_gipk"
+	srcFullName := s.attachSchemaSuffix(srcTableName)
+	dstTableName := "test_my_gipk_dst"
+
+	require.NoError(s.t, s.source.Exec(s.t.Context(), `SET SESSION sql_generate_invisible_primary_key = ON`))
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			name VARCHAR(50) NOT NULL
+		)
+	`, srcFullName)))
+	require.NoError(s.t, s.source.Exec(s.t.Context(), `SET SESSION sql_generate_invisible_primary_key = OFF`))
+
+	// Pre-snapshot row: my_row_id auto-assigns 1.
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (name) VALUES ('snap')`, srcFullName)))
+
+	// The default TableNameMapping path shards on "id", which this table lacks
+	// (its only key is the invisible my_row_id), so shard on the GIPK for the cluster suite.
+	var shardingKey string
+	if s.cluster {
+		shardingKey = "my_row_id"
+	}
+	connectionGen := FlowConnectionGenerationConfig{
+		FlowJobName: s.attachSuffix(srcTableName),
+		TableMappings: []*protos.TableMapping{{
+			SourceTableIdentifier:      srcFullName,
+			DestinationTableIdentifier: dstTableName,
+			ShardingKey:                shardingKey,
+		}},
+		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)
+
+	EnvWaitForCount(env, s, "waiting on snapshot", dstTableName, "my_row_id,name", 1)
+
+	// Post-snapshot CDC row: my_row_id auto-assigns 2.
+	require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
+		`INSERT INTO %s (name) VALUES ('cdc')`, srcFullName)))
+
+	EnvWaitForCount(env, s, "waiting on cdc", dstTableName, "my_row_id,name", 2)
+
+	rows, err := s.GetRows(dstTableName, "my_row_id,name")
+	require.NoError(s.t, err)
+	require.Len(s.t, rows.Records, 2)
+	// The snapshot row's my_row_id must be the real key value, not the destination default.
+	require.EqualValues(s.t, 1, rows.Records[0][0].Value(), "snapshot GIPK should be pulled")
+	require.Equal(s.t, "snap", rows.Records[0][1].Value())
+	require.EqualValues(s.t, 2, rows.Records[1][0].Value(), "cdc GIPK should be pulled")
+	require.Equal(s.t, "cdc", rows.Records[1][1].Value())
+
+	env.Cancel(s.t.Context())
+	RequireEnvCanceled(s.t, env)
+}

flow/pkg/mysql/validation.go

@@ -16,6 +16,9 @@ const (
 	MySQLMinVersionForBinlogRowMetadata            = "8.0.1"
 	MariaDBMinVersionForBinlogRowMetadata          = "10.5.0"
 	MySQLMinVersionForBinlogTransactionCompression = "8.0.20"
+	MySQLMinVersionForInvisibleColumns             = "8.0.23"
+	MariaDBMinVersionForInvisibleColumns           = "10.3.0"
+	MySQLMinVersionForGeneratedInvisiblePrimaryKey = "8.0.30"
 )
 
 func CompareServerVersion(conn *client.Conn, version string) (int, error) {