From 63565d1a419b4105ec444d2df51558bd7c599f70 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Wed, 8 Jul 2026 18:21:15 +0200 Subject: [PATCH 1/3] fix(mysql): pull INVISIBLE/GIPK columns during snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SELECT * snapshot fast path relied on wildcard expansion, which omits INVISIBLE columns (and the generated invisible primary key, my_row_id). The destination table and CDC both include those columns, so snapshot rows kept destination defaults forever while CDC populated real values — silent divergence. For GIPK the invisible my_row_id is the only PK, so snapshot rows arrived with no key, breaking dedup/ordering. Always build an explicit column list from the fetched table schema so invisible columns are named explicitly (bypassing wildcard invisibility). Add INVISIBLE and GIPK e2e tests comparing snapshot and CDC. Co-Authored-By: Claude Opus 4.8 (1M context) --- flow/connectors/mysql/qrep.go | 12 +-- flow/connectors/mysql/qrep_test.go | 15 +++- flow/e2e/clickhouse_mysql_test.go | 129 +++++++++++++++++++++++++++++ flow/pkg/mysql/validation.go | 3 + 4 files changed, 147 insertions(+), 12 deletions(-) diff --git a/flow/connectors/mysql/qrep.go b/flow/connectors/mysql/qrep.go index 8c0e5532c..a41a0cbd0 100644 --- a/flow/connectors/mysql/qrep.go +++ b/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) { 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( diff --git a/flow/connectors/mysql/qrep_test.go b/flow/connectors/mysql/qrep_test.go index 57a26c85a..3b62de1ac 100644 --- a/flow/connectors/mysql/qrep_test.go +++ b/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", @@ -54,7 +54,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", diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 3433ec9cd..77177a33c 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -2378,3 +2378,132 @@ 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" + + // GIPK is controlled by a session variable; the connector holds a single persistent connection, + // so the setting sticks for the subsequent CREATE. Reset it afterwards to avoid leaking into + // other tables in the suite. + 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))) + + 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, "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) +} diff --git a/flow/pkg/mysql/validation.go b/flow/pkg/mysql/validation.go index 26f265d72..b87ee4539 100644 --- a/flow/pkg/mysql/validation.go +++ b/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) { From ddc7c164c232d87b6c7188935f08e80ab64cc5c6 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 9 Jul 2026 10:53:29 +0200 Subject: [PATCH 2/3] fail if selectedColumns == "" --- flow/connectors/mysql/qrep.go | 4 ++++ flow/e2e/clickhouse_mysql_test.go | 3 --- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/flow/connectors/mysql/qrep.go b/flow/connectors/mysql/qrep.go index a41a0cbd0..fcc77106b 100644 --- a/flow/connectors/mysql/qrep.go +++ b/flow/connectors/mysql/qrep.go @@ -277,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)) diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index 77177a33c..ccf605bb8 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -2460,9 +2460,6 @@ func (s ClickHouseSuite) Test_MySQL_GIPK_Consistency() { srcFullName := s.attachSchemaSuffix(srcTableName) dstTableName := "test_my_gipk_dst" - // GIPK is controlled by a session variable; the connector holds a single persistent connection, - // so the setting sticks for the subsequent CREATE. Reset it afterwards to avoid leaking into - // other tables in the suite. 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 ( From dfd46e6d7b7a0e1e5afc4097fdb8a86d18f84932 Mon Sep 17 00:00:00 2001 From: Dmitrii Tunikov Date: Thu, 9 Jul 2026 11:31:53 +0200 Subject: [PATCH 3/3] fix ch cluster mode tests --- flow/e2e/clickhouse_mysql_test.go | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/flow/e2e/clickhouse_mysql_test.go b/flow/e2e/clickhouse_mysql_test.go index ccf605bb8..bccc12457 100644 --- a/flow/e2e/clickhouse_mysql_test.go +++ b/flow/e2e/clickhouse_mysql_test.go @@ -2472,10 +2472,20 @@ func (s ClickHouseSuite) Test_MySQL_GIPK_Consistency() { 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), - TableNameMapping: map[string]string{srcFullName: dstTableName}, - Destination: s.Peer().Name, + FlowJobName: s.attachSuffix(srcTableName), + TableMappings: []*protos.TableMapping{{ + SourceTableIdentifier: srcFullName, + DestinationTableIdentifier: dstTableName, + ShardingKey: shardingKey, + }}, + Destination: s.Peer().Name, } flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s) flowConnConfig.DoInitialSnapshot = true