Skip to content

Commit ae8a9a2

Browse files
ilidemiilidemi
committed
816 read invisible columns during snapshot
1 parent 13ec4d2 commit ae8a9a2

3 files changed

Lines changed: 169 additions & 11 deletions

File tree

‎flow/connectors/mysql/qrep.go‎

Lines changed: 1 addition & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -179,27 +179,19 @@ func (c *MySqlConnector) GetDefaultPartitionKeyForTables(
179179

180180
func buildSelectedColumns(cols []*protos.FieldDescription, exclude []string) string {
181181
columns := []string{}
182-
selectAsterisk := true
183182
for _, col := range cols {
184183
if slices.Contains(exclude, col.Name) {
185-
selectAsterisk = false
186184
continue
187185
}
188186

189187
converted := common.QuoteMySQLIdentifier(col.Name)
190188
if col.Type == string(types.QValueKindUint16Enum) {
191189
converted = fmt.Sprintf("CAST(%s AS UNSIGNED) AS %s", converted, converted)
192-
selectAsterisk = false
193190
}
194191
columns = append(columns, converted)
195192
}
196193

197-
selectedColumns := "*"
198-
if !selectAsterisk {
199-
selectedColumns = strings.Join(columns, ", ")
200-
}
201-
202-
return selectedColumns
194+
return strings.Join(columns, ", ")
203195
}
204196

205197
func (c *MySqlConnector) PullQRepRecords(

‎flow/connectors/mysql/qrep_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@ func TestBuildSelectedColumns(t *testing.T) {
2222
{Name: "status", Type: string(types.QValueKindEnum)},
2323
},
2424
exclude: []string{},
25-
expectedSelectedColumns: "*",
25+
expectedSelectedColumns: "`id`, `name`, `status`",
2626
},
2727
{
2828
name: "one excluded column",
@@ -49,7 +49,7 @@ func TestBuildSelectedColumns(t *testing.T) {
4949
{Name: "status", Type: string(types.QValueKindEnum)},
5050
},
5151
exclude: []string{},
52-
expectedSelectedColumns: "*",
52+
expectedSelectedColumns: "`id`, `status`",
5353
},
5454
{
5555
name: "uint16enum with exclude",

‎flow/e2e/clickhouse_mysql_test.go‎

Lines changed: 166 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,23 @@ func mysqlEnumUsesOrdinals() bool {
2626
return internal.MySQLTestVersionIsMySQLPos()
2727
}
2828

29+
func (s ClickHouseSuite) requireMySQLVersionAtLeast(version string) *MySqlSource {
30+
s.t.Helper()
31+
32+
mySource, ok := s.source.(*MySqlSource)
33+
if !ok || mySource.Config.Flavor != protos.MySqlFlavor_MYSQL_MYSQL {
34+
s.t.Skip("only applies to mysql")
35+
}
36+
37+
cmp, err := mySource.CompareServerVersion(s.t.Context(), version)
38+
require.NoError(s.t, err)
39+
if cmp < 0 {
40+
s.t.Skipf("requires MySQL %s+", version)
41+
}
42+
43+
return mySource
44+
}
45+
2946
func (s ClickHouseSuite) Test_UnsignedMySQL() {
3047
if _, ok := s.source.(*MySqlSource); !ok {
3148
s.t.Skip("only applies to mysql")
@@ -592,6 +609,155 @@ func (s ClickHouseSuite) Test_MySQL_Enum_Consistency_Version0() {
592609
RequireEnvCanceled(s.t, env)
593610
}
594611

612+
func (s ClickHouseSuite) Test_MySQL_InvisibleColumn_Consistency() {
613+
s.requireMySQLVersionAtLeast("8.0.23")
614+
615+
srcTableName := "test_invisible_column"
616+
srcFullName := s.attachSchemaSuffix(srcTableName)
617+
dstTableName := "test_invisible_column_dst"
618+
619+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
620+
CREATE TABLE IF NOT EXISTS %s (
621+
id INT PRIMARY KEY,
622+
vis INT NOT NULL,
623+
invis INT INVISIBLE NOT NULL
624+
)
625+
`, srcFullName)))
626+
627+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
628+
`INSERT INTO %s (id, vis, invis) VALUES (1, 10, 100), (2, 20, 200)`, srcFullName)))
629+
630+
connectionGen := FlowConnectionGenerationConfig{
631+
FlowJobName: s.attachSuffix(srcTableName),
632+
TableNameMapping: map[string]string{srcFullName: dstTableName},
633+
Destination: s.Peer().Name,
634+
}
635+
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
636+
flowConnConfig.DoInitialSnapshot = true
637+
638+
tc := NewTemporalClient(s.t)
639+
env := ExecutePeerflow(s.t, tc, flowConnConfig)
640+
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
641+
642+
EnvWaitForEqualTablesWithNames(env, s, "waiting on invisible snapshot",
643+
srcTableName, dstTableName, "id,vis,invis")
644+
645+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
646+
`INSERT INTO %s (id, vis, invis) VALUES (3, 30, 300)`, srcFullName)))
647+
648+
EnvWaitForEqualTablesWithNames(env, s, "waiting on invisible cdc",
649+
srcTableName, dstTableName, "id,vis,invis")
650+
651+
env.Cancel(s.t.Context())
652+
RequireEnvCanceled(s.t, env)
653+
}
654+
655+
func (s ClickHouseSuite) Test_MySQL_GeneratedInvisiblePrimaryKey_Consistency() {
656+
s.requireMySQLVersionAtLeast("8.0.30")
657+
658+
srcTableName := "test_gipk"
659+
srcFullName := s.attachSchemaSuffix(srcTableName)
660+
dstTableName := "test_gipk_dst"
661+
662+
require.NoError(s.t, s.source.Exec(s.t.Context(), `SET SESSION sql_generate_invisible_primary_key=ON`))
663+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
664+
CREATE TABLE IF NOT EXISTS %s (
665+
val INT NOT NULL
666+
)
667+
`, srcFullName)))
668+
require.NoError(s.t, s.source.Exec(s.t.Context(), `SET SESSION sql_generate_invisible_primary_key=OFF`))
669+
670+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
671+
`INSERT INTO %s (val) VALUES (10), (20)`, srcFullName)))
672+
673+
connectionGen := FlowConnectionGenerationConfig{
674+
FlowJobName: s.attachSuffix(srcTableName),
675+
TableNameMapping: map[string]string{srcFullName: dstTableName},
676+
Destination: s.Peer().Name,
677+
}
678+
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
679+
flowConnConfig.DoInitialSnapshot = true
680+
681+
tc := NewTemporalClient(s.t)
682+
env := ExecutePeerflow(s.t, tc, flowConnConfig)
683+
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
684+
685+
EnvWaitForEqualTablesWithNames(env, s, "waiting on gipk snapshot",
686+
srcTableName, dstTableName, "my_row_id AS id,val")
687+
688+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
689+
`INSERT INTO %s (val) VALUES (30)`, srcFullName)))
690+
691+
EnvWaitForEqualTablesWithNames(env, s, "waiting on gipk cdc",
692+
srcTableName, dstTableName, "my_row_id AS id,val")
693+
694+
ch, err := connclickhouse.Connect(s.t.Context(), nil, s.Peer().GetClickhouseConfig())
695+
require.NoError(s.t, err)
696+
defer ch.Close()
697+
698+
var rowCount, missingRowIDCount uint64
699+
require.NoError(s.t, ch.QueryRow(s.t.Context(), fmt.Sprintf(
700+
`SELECT count(), countIf(my_row_id IS NULL OR my_row_id = 0)
701+
FROM %s FINAL WHERE _peerdb_is_deleted = 0`,
702+
clickhouse.QuoteIdentifier(dstTableName),
703+
)).Scan(&rowCount, &missingRowIDCount))
704+
require.Equal(s.t, uint64(3), rowCount)
705+
require.Zero(s.t, missingRowIDCount)
706+
707+
env.Cancel(s.t.Context())
708+
RequireEnvCanceled(s.t, env)
709+
}
710+
711+
func (s ClickHouseSuite) Test_MySQL_InvisibleColumn_WithExcludedColumn() {
712+
s.requireMySQLVersionAtLeast("8.0.23")
713+
714+
srcTableName := "test_invisible_with_exclude"
715+
srcFullName := s.attachSchemaSuffix(srcTableName)
716+
dstTableName := "test_invisible_with_exclude_dst"
717+
718+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(`
719+
CREATE TABLE IF NOT EXISTS %s (
720+
id INT PRIMARY KEY,
721+
visible INT NOT NULL,
722+
skipped INT NOT NULL,
723+
invis INT INVISIBLE NOT NULL
724+
)
725+
`, srcFullName)))
726+
727+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
728+
`INSERT INTO %s (id, visible, skipped, invis) VALUES (1, 10, 1000, 100), (2, 20, 2000, 200)`,
729+
srcFullName)))
730+
731+
connectionGen := FlowConnectionGenerationConfig{
732+
FlowJobName: s.attachSuffix(srcTableName),
733+
TableMappings: []*protos.TableMapping{{
734+
SourceTableIdentifier: srcFullName,
735+
DestinationTableIdentifier: dstTableName,
736+
ShardingKey: "id",
737+
Exclude: []string{"skipped"},
738+
}},
739+
Destination: s.Peer().Name,
740+
}
741+
flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
742+
flowConnConfig.DoInitialSnapshot = true
743+
744+
tc := NewTemporalClient(s.t)
745+
env := ExecutePeerflow(s.t, tc, flowConnConfig)
746+
SetupCDCFlowStatusQuery(s.t, env, flowConnConfig)
747+
748+
EnvWaitForEqualTablesWithNames(env, s, "waiting on excluded invisible snapshot",
749+
srcTableName, dstTableName, "id,visible,invis")
750+
751+
require.NoError(s.t, s.source.Exec(s.t.Context(), fmt.Sprintf(
752+
`INSERT INTO %s (id, visible, skipped, invis) VALUES (3, 30, 3000, 300)`, srcFullName)))
753+
754+
EnvWaitForEqualTablesWithNames(env, s, "waiting on excluded invisible cdc",
755+
srcTableName, dstTableName, "id,visible,invis")
756+
757+
env.Cancel(s.t.Context())
758+
RequireEnvCanceled(s.t, env)
759+
}
760+
595761
func (s ClickHouseSuite) Test_MySQL_Vector() {
596762
mysource, ok := s.source.(*MySqlSource)
597763
if !ok || mysource.Config.Flavor != protos.MySqlFlavor_MYSQL_MYSQL {

0 commit comments

Comments
 (0)