Skip to content

Commit 7a9dca7

Browse files
zacharysierakowskizacharysierakowski
authored
Merge pull request #1703 from github/move-tables/applier-dml-events
Merge move-tables PRs onto feature branch: applier logic
2 parents 84f1c61 + e4d359d commit 7a9dca7

7 files changed

Lines changed: 821 additions & 15 deletions

File tree

‎go/base/context.go‎

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

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

287288
Log Logger
@@ -356,6 +357,9 @@ func (mctx *MigrationContext) SetConnectionConfig(storageEngine string) error {
356357
}
357358
mctx.InspectorConnectionConfig.TransactionIsolation = transactionIsolation
358359
mctx.ApplierConnectionConfig.TransactionIsolation = transactionIsolation
360+
if mctx.MoveTables.ConnectionConfig != nil {
361+
mctx.MoveTables.ConnectionConfig.TransactionIsolation = transactionIsolation
362+
}
359363
return nil
360364
}
361365

@@ -366,6 +370,9 @@ func (mctx *MigrationContext) SetConnectionCharset(charset string) {
366370

367371
mctx.InspectorConnectionConfig.Charset = charset
368372
mctx.ApplierConnectionConfig.Charset = charset
373+
if mctx.MoveTables.ConnectionConfig != nil {
374+
mctx.MoveTables.ConnectionConfig.Charset = charset
375+
}
369376
}
370377

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

‎go/logic/applier.go‎

Lines changed: 191 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,12 @@ type Applier struct {
9191
migrationLockName string
9292
migrationLockStop chan struct{}
9393
migrationLockDone chan struct{}
94+
95+
moveTablesTargetDB *gosql.DB
96+
moveTablesConnectionConfig *mysql.ConnectionConfig
97+
moveTablesCopySelectFirstQueryBuilder *sql.MoveTableCopySelectQueryBuilder
98+
moveTablesCopySelectNextQueryBuilder *sql.MoveTableCopySelectQueryBuilder
99+
moveTablesCopyInsertQueryBuilder *sql.MoveTableCopyInsertQueryBuilder
94100
}
95101

96102
func NewApplier(migrationContext *base.MigrationContext) *Applier {
@@ -99,6 +105,8 @@ func NewApplier(migrationContext *base.MigrationContext) *Applier {
99105
migrationContext: migrationContext,
100106
finishedMigrating: 0,
101107
name: "applier",
108+
109+
moveTablesConnectionConfig: migrationContext.MoveTables.ConnectionConfig,
102110
}
103111
}
104112

@@ -149,6 +157,15 @@ func (apl *Applier) InitDBConnections() (err error) {
149157
if err := apl.readTableColumns(); err != nil {
150158
return err
151159
}
160+
if apl.moveTablesConnectionConfig != nil {
161+
moveTablesURI := apl.moveTablesConnectionConfig.GetDBUri(apl.migrationContext.MoveTables.TargetDatabase) + "&multiStatements=true"
162+
if apl.moveTablesTargetDB, _, err = mysql.GetDB(apl.migrationContext.Uuid, moveTablesURI); err != nil {
163+
return err
164+
}
165+
if _, err := base.ValidateConnection(apl.moveTablesTargetDB, apl.moveTablesConnectionConfig, apl.migrationContext, apl.name); err != nil {
166+
return err
167+
}
168+
}
152169
apl.migrationContext.Log.Infof("Applier initiated on %+v, version %+v", apl.connectionConfig.ImpliedKey, apl.migrationContext.ApplierMySQLVersion)
153170
return nil
154171
}
@@ -297,26 +314,33 @@ func (apl *Applier) releaseMigrationLock() {
297314
}
298315

299316
func (apl *Applier) prepareQueries() (err error) {
317+
targetDatabaseName := apl.migrationContext.DatabaseName
318+
targetTableName := apl.migrationContext.GetGhostTableName()
319+
if apl.migrationContext.IsMoveTablesMode() {
320+
targetDatabaseName = apl.migrationContext.MoveTables.TargetDatabase
321+
targetTableName = apl.migrationContext.OriginalTableName
322+
}
323+
300324
if apl.dmlDeleteQueryBuilder, err = sql.NewDMLDeleteQueryBuilder(
301-
apl.migrationContext.DatabaseName,
302-
apl.migrationContext.GetGhostTableName(),
325+
targetDatabaseName,
326+
targetTableName,
303327
apl.migrationContext.OriginalTableColumns,
304328
&apl.migrationContext.UniqueKey.Columns,
305329
); err != nil {
306330
return err
307331
}
308332
if apl.dmlInsertQueryBuilder, err = sql.NewDMLInsertQueryBuilder(
309-
apl.migrationContext.DatabaseName,
310-
apl.migrationContext.GetGhostTableName(),
333+
targetDatabaseName,
334+
targetTableName,
311335
apl.migrationContext.OriginalTableColumns,
312336
apl.migrationContext.SharedColumns,
313337
apl.migrationContext.MappedSharedColumns,
314338
); err != nil {
315339
return err
316340
}
317341
if apl.dmlUpdateQueryBuilder, err = sql.NewDMLUpdateQueryBuilder(
318-
apl.migrationContext.DatabaseName,
319-
apl.migrationContext.GetGhostTableName(),
342+
targetDatabaseName,
343+
targetTableName,
320344
apl.migrationContext.OriginalTableColumns,
321345
apl.migrationContext.SharedColumns,
322346
apl.migrationContext.MappedSharedColumns,
@@ -333,6 +357,35 @@ func (apl *Applier) prepareQueries() (err error) {
333357
return err
334358
}
335359
}
360+
if apl.migrationContext.IsMoveTablesMode() {
361+
if apl.moveTablesCopySelectFirstQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
362+
apl.migrationContext.DatabaseName,
363+
apl.migrationContext.OriginalTableName,
364+
apl.migrationContext.OriginalTableColumns,
365+
apl.migrationContext.UniqueKey.Name,
366+
&apl.migrationContext.UniqueKey.Columns,
367+
true, // <-- include start range values for first select query
368+
); err != nil {
369+
return err
370+
}
371+
if apl.moveTablesCopySelectNextQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
372+
apl.migrationContext.DatabaseName,
373+
apl.migrationContext.OriginalTableName,
374+
apl.migrationContext.OriginalTableColumns,
375+
apl.migrationContext.UniqueKey.Name,
376+
&apl.migrationContext.UniqueKey.Columns,
377+
false,
378+
); err != nil {
379+
return err
380+
}
381+
if apl.moveTablesCopyInsertQueryBuilder, err = sql.NewMoveTableCopyInsertQueryBuilder(
382+
targetDatabaseName,
383+
targetTableName,
384+
apl.migrationContext.OriginalTableColumns,
385+
); err != nil {
386+
return err
387+
}
388+
}
336389
return nil
337390
}
338391

@@ -1164,6 +1217,130 @@ func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected i
11641217
return chunkSize, rowsAffected, duration, nil
11651218
}
11661219

1220+
// ApplyIterationMoveTableCopyQueries issues a SELECT query on the original table and an INSERT query on the target table,
1221+
// copying a chunk of rows. It is used when `--move-tables` is specified, instead of ApplyIterationInsertQuery.
1222+
func (apl *Applier) ApplyIterationMoveTableCopyQueries() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
1223+
startTime := time.Now()
1224+
chunkSize = atomic.LoadInt64(&apl.migrationContext.ChunkSize)
1225+
1226+
// First, select data from the source database:
1227+
rows, err := func() ([]*sql.ColumnValues, error) {
1228+
var qb *sql.MoveTableCopySelectQueryBuilder
1229+
if apl.migrationContext.GetIteration() == 0 {
1230+
qb = apl.moveTablesCopySelectFirstQueryBuilder
1231+
} else {
1232+
qb = apl.moveTablesCopySelectNextQueryBuilder
1233+
}
1234+
query, explodedArgs, err := qb.BuildQuery(
1235+
apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
1236+
apl.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
1237+
)
1238+
if err != nil {
1239+
return nil, err
1240+
}
1241+
sqlRows, err := apl.db.Query(query, explodedArgs...)
1242+
if err != nil {
1243+
return nil, err
1244+
}
1245+
defer sqlRows.Close()
1246+
chunkRows := make([]*sql.ColumnValues, 0, chunkSize)
1247+
for sqlRows.Next() {
1248+
row := sql.NewColumnValues(apl.migrationContext.SharedColumns.Len())
1249+
err := sqlRows.Scan(row.ValuesPointers...)
1250+
if err != nil {
1251+
return nil, err
1252+
}
1253+
chunkRows = append(chunkRows, row)
1254+
}
1255+
if rowsErr := sqlRows.Err(); rowsErr != nil {
1256+
return nil, rowsErr
1257+
}
1258+
return chunkRows, nil
1259+
}()
1260+
if err != nil {
1261+
return chunkSize, rowsAffected, duration, err
1262+
}
1263+
1264+
// no need to INSERT if there are no rows to copy:
1265+
if len(rows) == 0 {
1266+
duration = time.Since(startTime)
1267+
return chunkSize, 0, duration, nil
1268+
}
1269+
1270+
// Then, insert data into the destination database:
1271+
sqlResult, err := func() (gosql.Result, error) {
1272+
query, explodedArgs, err := apl.moveTablesCopyInsertQueryBuilder.BuildQuery(rows)
1273+
if err != nil {
1274+
return nil, err
1275+
}
1276+
tx, err := apl.moveTablesTargetDB.Begin()
1277+
if err != nil {
1278+
return nil, err
1279+
}
1280+
defer tx.Rollback()
1281+
1282+
sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s', %s`,
1283+
apl.migrationContext.ApplierTimeZone,
1284+
apl.generateSqlModeQuery())
1285+
if _, err := tx.Exec(sessionQuery); err != nil {
1286+
return nil, err
1287+
}
1288+
1289+
sqlResult, err := tx.Exec(query, explodedArgs...)
1290+
if err != nil {
1291+
return nil, err
1292+
}
1293+
1294+
if apl.migrationContext.PanicOnWarnings {
1295+
rows, err := tx.Query("SHOW WARNINGS")
1296+
if err != nil {
1297+
return nil, err
1298+
}
1299+
defer rows.Close()
1300+
if err = rows.Err(); err != nil {
1301+
return nil, err
1302+
}
1303+
migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
1304+
if err != nil {
1305+
return nil, err
1306+
}
1307+
var sqlWarnings []string
1308+
for rows.Next() {
1309+
var level, message string
1310+
var code int
1311+
if err := rows.Scan(&level, &code, &message); err != nil {
1312+
apl.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
1313+
continue
1314+
}
1315+
if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
1316+
continue
1317+
}
1318+
sqlWarnings = append(sqlWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
1319+
}
1320+
apl.migrationContext.MigrationLastInsertSQLWarnings = sqlWarnings
1321+
}
1322+
1323+
if err := tx.Commit(); err != nil {
1324+
return nil, err
1325+
}
1326+
return sqlResult, nil
1327+
}()
1328+
if err != nil {
1329+
return chunkSize, rowsAffected, duration, err
1330+
}
1331+
rowsAffected, _ = sqlResult.RowsAffected()
1332+
duration = time.Since(startTime)
1333+
apl.migrationContext.Log.Debugf(
1334+
"Issued SELECT+INSERT on range: [%s]..[%s]; iteration: %d; chunk-size: %d",
1335+
apl.migrationContext.MigrationIterationRangeMinValues,
1336+
apl.migrationContext.MigrationIterationRangeMaxValues,
1337+
apl.migrationContext.GetIteration(),
1338+
chunkSize,
1339+
)
1340+
1341+
return chunkSize, rowsAffected, duration, nil
1342+
}
1343+
11671344
// LockOriginalTable places a write lock on the original table
11681345
func (apl *Applier) LockOriginalTable() error {
11691346
query := fmt.Sprintf(`lock /* gh-ost */ tables %s.%s write`,
@@ -1783,7 +1960,11 @@ func (apl *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent)) e
17831960
ctx := context.Background()
17841961

17851962
err := func() error {
1786-
conn, err := apl.db.Conn(ctx)
1963+
db := apl.db
1964+
if apl.migrationContext.IsMoveTablesMode() {
1965+
db = apl.moveTablesTargetDB
1966+
}
1967+
conn, err := db.Conn(ctx)
17871968
if err != nil {
17881969
return err
17891970
}
@@ -1888,6 +2069,9 @@ func (apl *Applier) Teardown() {
18882069
apl.releaseMigrationLock()
18892070
apl.db.Close()
18902071
apl.singletonDB.Close()
2072+
if apl.moveTablesTargetDB != nil {
2073+
apl.moveTablesTargetDB.Close()
2074+
}
18912075
atomic.StoreInt64(&apl.finishedMigrating, 1)
18922076
}
18932077

0 commit comments

Comments
 (0)