Skip to content

Commit c151452

Browse files
danieljoosdanieljoos
committed
Adapt applier to support DML events for move tables
1 parent ac6f966 commit c151452

4 files changed

Lines changed: 132 additions & 32 deletions

File tree

‎go/base/context.go‎

Lines changed: 13 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -273,12 +273,13 @@ type MigrationContext struct {
273273

274274
// move tables:
275275
MoveTables struct {
276-
TableNames []string // List of table names to be moved.
277-
TargetHost string // Target hostname for the move. This must be a primary/writable host.
278-
TargetPort int // Target MySQL port for the move.
279-
TargetUser string // Target username for the move. If not specified, it will default to the source user.
280-
TargetPass string // Target password for the move. If not specified, it will default to the source password.
281-
TargetDatabase string // Target database name for the move. If not specified, it will default to the source database name.
276+
TableNames []string // List of table names to be moved.
277+
TargetHost string // Target hostname for the move. This must be a primary/writable host.
278+
TargetPort int // Target MySQL port for the move.
279+
TargetUser string // Target username for the move. If not specified, it will default to the source user.
280+
TargetPass string // Target password for the move. If not specified, it will default to the source password.
281+
TargetDatabase string // Target database name for the move. If not specified, it will default to the source database name.
282+
ConnectionConfig *mysql.ConnectionConfig
282283
}
283284

284285
Log Logger
@@ -353,6 +354,9 @@ func (mctx *MigrationContext) SetConnectionConfig(storageEngine string) error {
353354
}
354355
mctx.InspectorConnectionConfig.TransactionIsolation = transactionIsolation
355356
mctx.ApplierConnectionConfig.TransactionIsolation = transactionIsolation
357+
if mctx.MoveTables.ConnectionConfig != nil {
358+
mctx.MoveTables.ConnectionConfig.TransactionIsolation = transactionIsolation
359+
}
356360
return nil
357361
}
358362

@@ -363,6 +367,9 @@ func (mctx *MigrationContext) SetConnectionCharset(charset string) {
363367

364368
mctx.InspectorConnectionConfig.Charset = charset
365369
mctx.ApplierConnectionConfig.Charset = charset
370+
if mctx.MoveTables.ConnectionConfig != nil {
371+
mctx.MoveTables.ConnectionConfig.Charset = charset
372+
}
366373
}
367374

368375
func getSafeTableName(baseName string, suffix string) string {

‎go/logic/applier.go‎

Lines changed: 38 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,7 @@ type Applier struct {
8585
checkpointInsertQueryBuilder *sql.CheckpointInsertQueryBuilder
8686

8787
moveTablesTargetDB *gosql.DB
88+
moveTablesConnectionConfig *mysql.ConnectionConfig
8889
moveTablesCopySelectFirstQueryBuilder *sql.MoveTableCopySelectQueryBuilder
8990
moveTablesCopySelectNextQueryBuilder *sql.MoveTableCopySelectQueryBuilder
9091
moveTablesCopyInsertQueryBuilder *sql.MoveTableCopyInsertQueryBuilder
@@ -96,6 +97,8 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
9697
migrationContext: migrationContext,
9798
finishedMigrating: 0,
9899
name: "applier",
100+
101+
moveTablesConnectionConfig: migrationContext.MoveTables.ConnectionConfig,
99102
}
100103
}
101104

@@ -146,31 +149,47 @@ func (apl *Applier) InitDBConnections() (err error) {
146149
if err := apl.readTableColumns(); err != nil {
147150
return err
148151
}
152+
if apl.moveTablesConnectionConfig != nil {
153+
moveTablesURI := apl.moveTablesConnectionConfig.GetDBUri(apl.migrationContext.MoveTables.TargetDatabase) + "&multiStatements=true"
154+
if apl.moveTablesTargetDB, _, err = mysql.GetDB(apl.migrationContext.Uuid, moveTablesURI); err != nil {
155+
return err
156+
}
157+
if _, err := base.ValidateConnection(apl.moveTablesTargetDB, apl.moveTablesConnectionConfig, apl.migrationContext, apl.name); err != nil {
158+
return err
159+
}
160+
}
149161
apl.migrationContext.Log.Infof("Applier initiated on %+v, version %+v", apl.connectionConfig.ImpliedKey, apl.migrationContext.ApplierMySQLVersion)
150162
return nil
151163
}
152164

153165
func (apl *Applier) prepareQueries() (err error) {
166+
targetDatabaseName := apl.migrationContext.DatabaseName
167+
targetTableName := apl.migrationContext.GetGhostTableName()
168+
if apl.migrationContext.IsMoveTablesMode() {
169+
targetDatabaseName = apl.migrationContext.MoveTables.TargetDatabase
170+
targetTableName = apl.migrationContext.OriginalTableName
171+
}
172+
154173
if apl.dmlDeleteQueryBuilder, err = sql.NewDMLDeleteQueryBuilder(
155-
apl.migrationContext.DatabaseName,
156-
apl.migrationContext.GetGhostTableName(),
174+
targetDatabaseName,
175+
targetTableName,
157176
apl.migrationContext.OriginalTableColumns,
158177
&apl.migrationContext.UniqueKey.Columns,
159178
); err != nil {
160179
return err
161180
}
162181
if apl.dmlInsertQueryBuilder, err = sql.NewDMLInsertQueryBuilder(
163-
apl.migrationContext.DatabaseName,
164-
apl.migrationContext.GetGhostTableName(),
182+
targetDatabaseName,
183+
targetTableName,
165184
apl.migrationContext.OriginalTableColumns,
166185
apl.migrationContext.SharedColumns,
167186
apl.migrationContext.MappedSharedColumns,
168187
); err != nil {
169188
return err
170189
}
171190
if apl.dmlUpdateQueryBuilder, err = sql.NewDMLUpdateQueryBuilder(
172-
apl.migrationContext.DatabaseName,
173-
apl.migrationContext.GetGhostTableName(),
191+
targetDatabaseName,
192+
targetTableName,
174193
apl.migrationContext.OriginalTableColumns,
175194
apl.migrationContext.SharedColumns,
176195
apl.migrationContext.MappedSharedColumns,
@@ -191,7 +210,7 @@ func (apl *Applier) prepareQueries() (err error) {
191210
if apl.moveTablesCopySelectFirstQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
192211
apl.migrationContext.DatabaseName,
193212
apl.migrationContext.OriginalTableName,
194-
apl.migrationContext.SharedColumns,
213+
apl.migrationContext.OriginalTableColumns,
195214
apl.migrationContext.UniqueKey.Name,
196215
&apl.migrationContext.UniqueKey.Columns,
197216
true, // <-- include start range values for first select query
@@ -201,17 +220,17 @@ func (apl *Applier) prepareQueries() (err error) {
201220
if apl.moveTablesCopySelectNextQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
202221
apl.migrationContext.DatabaseName,
203222
apl.migrationContext.OriginalTableName,
204-
apl.migrationContext.SharedColumns,
223+
apl.migrationContext.OriginalTableColumns,
205224
apl.migrationContext.UniqueKey.Name,
206225
&apl.migrationContext.UniqueKey.Columns,
207226
false,
208227
); err != nil {
209228
return err
210229
}
211230
if apl.moveTablesCopyInsertQueryBuilder, err = sql.NewMoveTableCopyInsertQueryBuilder(
212-
apl.migrationContext.MoveTables.TargetDatabase,
213-
apl.migrationContext.OriginalTableName,
214-
apl.migrationContext.SharedColumns,
231+
targetDatabaseName,
232+
targetTableName,
233+
apl.migrationContext.OriginalTableColumns,
215234
); err != nil {
216235
return err
217236
}
@@ -1781,7 +1800,11 @@ func (apl *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent)) e
17811800
ctx := context.Background()
17821801

17831802
err := func() error {
1784-
conn, err := apl.db.Conn(ctx)
1803+
db := apl.db
1804+
if apl.migrationContext.IsMoveTablesMode() {
1805+
db = apl.moveTablesTargetDB
1806+
}
1807+
conn, err := db.Conn(ctx)
17851808
if err != nil {
17861809
return err
17871810
}
@@ -1885,6 +1908,9 @@ func (apl *Applier) Teardown() {
18851908
apl.migrationContext.Log.Debugf("Tearing down...")
18861909
apl.db.Close()
18871910
apl.singletonDB.Close()
1911+
if apl.moveTablesTargetDB != nil {
1912+
apl.moveTablesTargetDB.Close()
1913+
}
18881914
atomic.StoreInt64(&apl.finishedMigrating, 1)
18891915
}
18901916

‎go/logic/applier_test.go‎

Lines changed: 69 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -332,6 +332,8 @@ func (suite *ApplierTestSuite) TearDownTest() {
332332
suite.Require().NoError(err)
333333
_, err = suite.db.ExecContext(ctx, "DROP TABLE IF EXISTS "+getTestGhostTableName())
334334
suite.Require().NoError(err)
335+
_, err = suite.otherDB.ExecContext(ctx, "DROP TABLE IF EXISTS "+getTestOtherTableName())
336+
suite.Require().NoError(err)
335337
}
336338

337339
func (suite *ApplierTestSuite) TestInitDBConnections() {
@@ -1561,6 +1563,72 @@ func (suite *ApplierTestSuite) TestMultipleDMLEventsInBatch() {
15611563
// Critically: id=2 (bob@example.com) is NOT present, proving event #3 was rolled back
15621564
}
15631565

1566+
func (suite *ApplierTestSuite) TestApplyDMLEventQueriesMoveTablesMode() {
1567+
ctx := context.Background()
1568+
var err error
1569+
1570+
_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT, item_id INT);", getTestTableName()))
1571+
suite.Require().NoError(err)
1572+
_, err = suite.otherDB.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT, item_id INT);", getTestOtherTableName()))
1573+
suite.Require().NoError(err)
1574+
1575+
connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
1576+
suite.Require().NoError(err)
1577+
1578+
migrationContext := newTestMigrationContext()
1579+
migrationContext.ApplierConnectionConfig = connectionConfig
1580+
migrationContext.MoveTables.ConnectionConfig = connectionConfig
1581+
migrationContext.SetConnectionConfig("innodb")
1582+
1583+
migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "item_id"})
1584+
migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "item_id"})
1585+
migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "item_id"})
1586+
migrationContext.UniqueKey = &sql.UniqueKey{
1587+
Name: "primary_key",
1588+
Columns: *sql.NewColumnList([]string{"id"}),
1589+
}
1590+
migrationContext.MoveTables.TableNames = []string{testMysqlTableName}
1591+
migrationContext.MoveTables.TargetDatabase = testMysqlDatabaseOther
1592+
1593+
applier := NewApplier(migrationContext)
1594+
suite.Require().NoError(applier.prepareQueries())
1595+
defer applier.Teardown()
1596+
1597+
err = applier.InitDBConnections()
1598+
suite.Require().NoError(err)
1599+
1600+
dmlEvents := []*binlog.BinlogDMLEvent{
1601+
{
1602+
DatabaseName: testMysqlDatabase,
1603+
TableName: testMysqlTableName,
1604+
DML: binlog.InsertDML,
1605+
NewColumnValues: sql.ToColumnValues([]interface{}{123456, 42}),
1606+
},
1607+
}
1608+
err = applier.ApplyDMLEventQueries(dmlEvents)
1609+
suite.Require().NoError(err)
1610+
1611+
// Check that the row was inserted into the ghost table via moveTablesTargetDB
1612+
rows, err := suite.otherDB.Query("SELECT * FROM " + getTestOtherTableName())
1613+
suite.Require().NoError(err)
1614+
defer rows.Close()
1615+
1616+
var count, id, item_id int
1617+
for rows.Next() {
1618+
err = rows.Scan(&id, &item_id)
1619+
suite.Require().NoError(err)
1620+
count += 1
1621+
}
1622+
suite.Require().NoError(rows.Err())
1623+
1624+
suite.Require().Equal(1, count)
1625+
suite.Require().Equal(123456, id)
1626+
suite.Require().Equal(42, item_id)
1627+
1628+
suite.Require().Equal(int64(1), migrationContext.TotalDMLEventsApplied)
1629+
suite.Require().Equal(int64(0), migrationContext.RowsDeltaEstimate)
1630+
}
1631+
15641632
func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueries() {
15651633
ctx := context.Background()
15661634
var err error
@@ -1577,6 +1645,7 @@ func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueries() {
15771645

15781646
migrationContext := newTestMigrationContext()
15791647
migrationContext.ApplierConnectionConfig = connectionConfig
1648+
migrationContext.MoveTables.ConnectionConfig = connectionConfig
15801649
migrationContext.SetConnectionConfig("innodb")
15811650
migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "name", "created_at"})
15821651
migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "name", "created_at"})
@@ -1595,8 +1664,6 @@ func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueries() {
15951664
err = applier.InitDBConnections()
15961665
suite.Require().NoError(err)
15971666

1598-
applier.moveTablesTargetDB = suite.otherDB
1599-
16001667
err = applier.CreateChangelogTable()
16011668
suite.Require().NoError(err)
16021669

‎go/sql/builder.go‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -336,14 +336,14 @@ type MoveTableCopySelectQueryBuilder struct {
336336
argsCount int
337337
}
338338

339-
func NewMoveTableCopySelectQueryBuilder(sourceDatabaseName, sourceTableName string, sharedColumns *ColumnList, uniqueKey string, uniqueKeyColumns *ColumnList, includeRangeStartValues bool) (*MoveTableCopySelectQueryBuilder, error) {
339+
func NewMoveTableCopySelectQueryBuilder(sourceDatabaseName, sourceTableName string, columns *ColumnList, uniqueKey string, uniqueKeyColumns *ColumnList, includeRangeStartValues bool) (*MoveTableCopySelectQueryBuilder, error) {
340340
sourceDatabaseName = EscapeName(sourceDatabaseName)
341341
sourceTableName = EscapeName(sourceTableName)
342-
sharedColumnsNames := sharedColumns.Names()
343-
for i := range sharedColumnsNames {
344-
sharedColumnsNames[i] = EscapeName(sharedColumnsNames[i])
342+
columnNames := columns.Names()
343+
for i := range columnNames {
344+
columnNames[i] = EscapeName(columnNames[i])
345345
}
346-
sharedColumnsListing := strings.Join(sharedColumnsNames, ", ")
346+
sharedColumnsListing := strings.Join(columnNames, ", ")
347347
uniqueKey = EscapeName(uniqueKey)
348348
var minRangeComparisonSign = GreaterThanComparisonSign
349349
if includeRangeStartValues {
@@ -422,16 +422,16 @@ type MoveTableCopyInsertQueryBuilder struct {
422422
valueListSize int
423423
}
424424

425-
func NewMoveTableCopyInsertQueryBuilder(targetDatabaseName, targetTableName string, sharedColumns *ColumnList) (*MoveTableCopyInsertQueryBuilder, error) {
425+
func NewMoveTableCopyInsertQueryBuilder(targetDatabaseName, targetTableName string, columns *ColumnList) (*MoveTableCopyInsertQueryBuilder, error) {
426426
targetDatabaseName = EscapeName(targetDatabaseName)
427427
targetTableName = EscapeName(targetTableName)
428-
sharedColumnsNames := sharedColumns.Names()
429-
for i := range sharedColumnsNames {
430-
sharedColumnsNames[i] = EscapeName(sharedColumnsNames[i])
428+
columnsNames := columns.Names()
429+
for i := range columnsNames {
430+
columnsNames[i] = EscapeName(columnsNames[i])
431431
}
432-
sharedColumnsListing := strings.Join(sharedColumnsNames, ", ")
433-
valueListPlaceholder := "(" + strings.Join(buildColumnsPreparedValues(sharedColumns), ", ") + ")"
434-
valueListSize := len(sharedColumnsNames)
432+
sharedColumnsListing := strings.Join(columnsNames, ", ")
433+
valueListPlaceholder := "(" + strings.Join(buildColumnsPreparedValues(columns), ", ") + ")"
434+
valueListSize := len(columnsNames)
435435
stmt := fmt.Sprintf(`
436436
insert /* gh-ost %s.%s */ ignore
437437
into

0 commit comments

Comments
 (0)