MySQL: Fix mixed case for schema changes (#3125)

In our handler to process alter table commands
(`processAlterTableQuery`), we use a `.String()` method to convert the
column object we get to string to add it to our state of added columns.
Turns out this method actually converts the column to lowercase before
giving it to us:

https://github.com/pingcap/tidb/blob/1962f220e9d40608867fa172c91e5ad9b67c0865/pkg/parser/ast/expressions.go#L546

This PR instead uses a different method - `.OrigColName()` - which gives
us the column name as is.

https://github.com/pingcap/tidb/blob/1962f220e9d40608867fa172c91e5ad9b67c0865/pkg/parser/ast/expressions.go#L558

Also this PR introduces a test for MySQL schema changes in
`mysql_test.go`, and it differs from the generic test in how an insert
is not performed after column addition and the quoting of identifiers is
different

Author: Amogh-Bharadwaj

flow/connectors/mysql/cdc.go

@@ -699,7 +699,7 @@ func (c *MySqlConnector) processAlterTableQuery(ctx context.Context, catalogPool
 				}
 
 				fd := &protos.FieldDescription{
-					Name:         col.Name.String(),
+					Name:         col.Name.OrigColName(),
 					Type:         string(qkind),
 					TypeModifier: typmod,
 					Nullable:     nullable,

flow/e2e/clickhouse/mysql_test.go

@@ -7,8 +7,11 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/PeerDB-io/peerdb/flow/connectors"
 	"github.com/PeerDB-io/peerdb/flow/e2e"
 	"github.com/PeerDB-io/peerdb/flow/generated/protos"
+	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/types"
 	peerflow "github.com/PeerDB-io/peerdb/flow/workflows"
 )
 
@@ -547,3 +550,120 @@ func (s ClickHouseSuite) Test_MySQL_Specific_Geometric_Types() {
 	env.Cancel(s.t.Context())
 	e2e.RequireEnvCanceled(s.t, env)
 }
+
+func (s ClickHouseSuite) Test_MySQL_Schema_Changes() {
+	if _, ok := s.source.(*e2e.MySqlSource); !ok {
+		s.t.Skip("only applies to mysql")
+	}
+
+	t := s.T()
+	destinationSchemaConnector, ok := s.DestinationConnector().(connectors.GetTableSchemaConnector)
+	if !ok {
+		t.Skip("skipping test because destination connector does not implement GetTableSchemaConnector")
+	}
+
+	srcTable := "test_mysql_schema_changes"
+	dstTable := "test_mysql_schema_changes_dst"
+	srcTableName := e2e.AttachSchema(s, srcTable)
+	dstTableName := s.DestinationTable(dstTable)
+	secondSrcTable := "test_mysql_schema_changes_second"
+	secondDstTable := "test_mysql_schema_changes_second_dst"
+	secondSrcTableName := e2e.AttachSchema(s, secondSrcTable)
+
+	require.NoError(t, s.Source().Exec(t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id SERIAL PRIMARY KEY,
+			c1 BIGINT
+		);
+	`, srcTableName)))
+	require.NoError(t, s.Source().Exec(t.Context(), fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			id SERIAL PRIMARY KEY
+		);
+	`, secondSrcTableName)))
+
+	connectionGen := e2e.FlowConnectionGenerationConfig{
+		FlowJobName:   e2e.AddSuffix(s, srcTable),
+		TableMappings: e2e.TableMappings(s, srcTable, dstTable, secondSrcTable, secondDstTable),
+		Destination:   s.Peer().Name,
+	}
+	flowConnConfig := connectionGen.GenerateFlowConnectionConfigs(s)
+
+	// wait for PeerFlowStatusQuery to finish setup
+	// and then insert and mutate schema repeatedly.
+	tc := e2e.NewTemporalClient(t)
+	env := e2e.ExecutePeerflow(t.Context(), tc, peerflow.CDCFlowWorkflow, flowConnConfig, nil)
+	e2e.SetupCDCFlowStatusQuery(t, env, flowConnConfig)
+	e2e.EnvNoError(t, env, s.Source().Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s(c1) VALUES(1)`, srcTableName)))
+
+	e2e.EnvWaitForEqualTablesWithNames(env, s, "normalize reinsert", srcTable, dstTable, "id,c1")
+
+	expectedTableSchema := &protos.TableSchema{
+		TableIdentifier: e2e.ExpectedDestinationTableName(s, dstTable),
+		Columns: []*protos.FieldDescription{
+			{
+				Name:         e2e.ExpectedDestinationIdentifier(s, "id"),
+				Type:         string(types.QValueKindNumeric),
+				TypeModifier: -1,
+			},
+			{
+				Name:         e2e.ExpectedDestinationIdentifier(s, "c1"),
+				Type:         string(types.QValueKindNumeric),
+				TypeModifier: -1,
+			},
+			{
+				Name:         "_PEERDB_IS_DELETED",
+				Type:         string(types.QValueKindBoolean),
+				TypeModifier: -1,
+			},
+			{
+				Name:         "_PEERDB_SYNCED_AT",
+				Type:         string(types.QValueKindTimestamp),
+				TypeModifier: -1,
+			},
+		},
+	}
+	output, err := destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
+		[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
+	e2e.EnvNoError(t, env, err)
+	e2e.EnvTrue(t, env, e2e.CompareTableSchemas(expectedTableSchema, output[dstTableName]))
+
+	// alter source table, add column addedColumn and insert another row.
+	e2e.EnvNoError(t, env, s.Source().Exec(t.Context(), fmt.Sprintf("ALTER TABLE %s ADD COLUMN `addedColumn` BIGINT", srcTableName)))
+	// so that the batch finishes, insert a row into the second source table.
+	e2e.EnvNoError(t, env, s.Source().Exec(t.Context(), fmt.Sprintf(`INSERT INTO %s VALUES(DEFAULT)`, secondSrcTableName)))
+	e2e.EnvWaitForEqualTablesWithNames(env, s, "normalize altered row", srcTable, dstTable, "id,c1,coalesce(`addedColumn`,0) `addedColumn`")
+	expectedTableSchema = &protos.TableSchema{
+		TableIdentifier: e2e.ExpectedDestinationTableName(s, dstTable),
+		Columns: []*protos.FieldDescription{
+			{
+				Name:         e2e.ExpectedDestinationIdentifier(s, "id"),
+				Type:         string(types.QValueKindNumeric),
+				TypeModifier: -1,
+			},
+			{
+				Name:         e2e.ExpectedDestinationIdentifier(s, "c1"),
+				Type:         string(types.QValueKindNumeric),
+				TypeModifier: -1,
+			},
+			{
+				Name:         "_PEERDB_SYNCED_AT",
+				Type:         string(types.QValueKindTimestamp),
+				TypeModifier: -1,
+			},
+			{
+				Name:         e2e.ExpectedDestinationIdentifier(s, "addedColumn"),
+				Type:         string(types.QValueKindNumeric),
+				TypeModifier: -1,
+			},
+		},
+	}
+	output, err = destinationSchemaConnector.GetTableSchema(t.Context(), nil, shared.InternalVersion_Latest, protos.TypeSystem_Q,
+		[]*protos.TableMapping{{SourceTableIdentifier: dstTableName}})
+	e2e.EnvNoError(t, env, err)
+	e2e.EnvTrue(t, env, e2e.CompareTableSchemas(expectedTableSchema, output[dstTableName]))
+	e2e.EnvEqualTablesWithNames(env, s, srcTable, dstTable, "id,c1,coalesce(`addedColumn`,0) `addedColumn`")
+
+	env.Cancel(t.Context())
+	e2e.RequireEnvCanceled(t, env)
+}