Skip to content

Commit 5c0829a

Browse files
arthurschreiberarthurschreiber
authored
Merge branch 'master' into danieljoos-dml-query-builders
2 parents ab74e99 + 30f28c2 commit 5c0829a

4 files changed

Lines changed: 190 additions & 33 deletions

File tree

‎go/logic/applier.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -666,6 +666,8 @@ func (this *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected
666666
this.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
667667
this.migrationContext.GetIteration() == 0,
668668
this.migrationContext.IsTransactionalTable(),
669+
// TODO: Don't hardcode this
670+
strings.HasPrefix(this.migrationContext.ApplierMySQLVersion, "8."),
669671
)
670672
if err != nil {
671673
return chunkSize, rowsAffected, duration, err

‎go/logic/applier_test.go‎

Lines changed: 167 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
gosql "database/sql"
1111
"strings"
1212
"testing"
13+
"time"
1314

1415
"github.com/stretchr/testify/require"
1516
"github.com/stretchr/testify/suite"
@@ -201,6 +202,30 @@ type ApplierTestSuite struct {
201202
mysqlContainer testcontainers.Container
202203
}
203204

205+
func (suite *ApplierTestSuite) getConnectionConfig(ctx context.Context) (*mysql.ConnectionConfig, error) {
206+
host, err := suite.mysqlContainer.ContainerIP(ctx)
207+
if err != nil {
208+
return nil, err
209+
}
210+
211+
config := mysql.NewConnectionConfig()
212+
config.Key.Hostname = host
213+
config.Key.Port = 3306
214+
config.User = "root"
215+
config.Password = "root-password"
216+
217+
return config, nil
218+
}
219+
220+
func (suite *ApplierTestSuite) getDb(ctx context.Context) (*gosql.DB, error) {
221+
host, err := suite.mysqlContainer.ContainerIP(ctx)
222+
if err != nil {
223+
return nil, err
224+
}
225+
226+
return gosql.Open("mysql", "root:root-password@tcp("+host+":3306)/test")
227+
}
228+
204229
func (suite *ApplierTestSuite) SetupSuite() {
205230
ctx := context.Background()
206231
req := testcontainers.ContainerRequest{
@@ -231,7 +256,7 @@ func (suite *ApplierTestSuite) SetupTest() {
231256
suite.Require().NoError(err)
232257
suite.Require().Equalf(0, rc, "failed to created database: expected exit code 0, got %d", rc)
233258

234-
rc, _, err = suite.mysqlContainer.Exec(ctx, []string{"mysql", "-uroot", "-proot-password", "-e", "CREATE TABLE test.testing (id INT, item_id INT);"})
259+
rc, _, err = suite.mysqlContainer.Exec(ctx, []string{"mysql", "-uroot", "-proot-password", "-e", "CREATE TABLE test.testing (id INT, item_id INT, PRIMARY KEY (id));"})
235260
suite.Require().NoError(err)
236261
suite.Require().Equalf(0, rc, "failed to created table: expected exit code 0, got %d", rc)
237262
}
@@ -247,15 +272,11 @@ func (suite *ApplierTestSuite) TearDownTest() {
247272
func (suite *ApplierTestSuite) TestInitDBConnections() {
248273
ctx := context.Background()
249274

250-
host, err := suite.mysqlContainer.ContainerIP(ctx)
275+
connectionConfig, err := suite.getConnectionConfig(ctx)
251276
suite.Require().NoError(err)
252277

253278
migrationContext := base.NewMigrationContext()
254-
migrationContext.ApplierConnectionConfig = mysql.NewConnectionConfig()
255-
migrationContext.ApplierConnectionConfig.Key.Hostname = host
256-
migrationContext.ApplierConnectionConfig.Key.Port = 3306
257-
migrationContext.ApplierConnectionConfig.User = "root"
258-
migrationContext.ApplierConnectionConfig.Password = "root-password"
279+
migrationContext.ApplierConnectionConfig = connectionConfig
259280
migrationContext.DatabaseName = "test"
260281
migrationContext.OriginalTableName = "testing"
261282
migrationContext.SetConnectionConfig("innodb")
@@ -276,15 +297,11 @@ func (suite *ApplierTestSuite) TestInitDBConnections() {
276297
func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
277298
ctx := context.Background()
278299

279-
host, err := suite.mysqlContainer.ContainerIP(ctx)
300+
connectionConfig, err := suite.getConnectionConfig(ctx)
280301
suite.Require().NoError(err)
281302

282303
migrationContext := base.NewMigrationContext()
283-
migrationContext.ApplierConnectionConfig = mysql.NewConnectionConfig()
284-
migrationContext.ApplierConnectionConfig.Key.Hostname = host
285-
migrationContext.ApplierConnectionConfig.Key.Port = 3306
286-
migrationContext.ApplierConnectionConfig.User = "root"
287-
migrationContext.ApplierConnectionConfig.Password = "root-password"
304+
migrationContext.ApplierConnectionConfig = connectionConfig
288305
migrationContext.DatabaseName = "test"
289306
migrationContext.OriginalTableName = "testing"
290307
migrationContext.SetConnectionConfig("innodb")
@@ -320,7 +337,7 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
320337
suite.Require().NoError(err)
321338

322339
// Check that the row was inserted
323-
db, err := gosql.Open("mysql", "root:root-password@tcp("+host+":3306)/test")
340+
db, err := suite.getDb(ctx)
324341
suite.Require().NoError(err)
325342
defer db.Close()
326343

@@ -347,15 +364,11 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueries() {
347364
func (suite *ApplierTestSuite) TestValidateOrDropExistingTables() {
348365
ctx := context.Background()
349366

350-
host, err := suite.mysqlContainer.ContainerIP(ctx)
367+
connectionConfig, err := suite.getConnectionConfig(ctx)
351368
suite.Require().NoError(err)
352369

353370
migrationContext := base.NewMigrationContext()
354-
migrationContext.ApplierConnectionConfig = mysql.NewConnectionConfig()
355-
migrationContext.ApplierConnectionConfig.Key.Hostname = host
356-
migrationContext.ApplierConnectionConfig.Key.Port = 3306
357-
migrationContext.ApplierConnectionConfig.User = "root"
358-
migrationContext.ApplierConnectionConfig.Password = "root-password"
371+
migrationContext.ApplierConnectionConfig = connectionConfig
359372
migrationContext.DatabaseName = "test"
360373
migrationContext.OriginalTableName = "testing"
361374
migrationContext.SetConnectionConfig("innodb")
@@ -374,6 +387,140 @@ func (suite *ApplierTestSuite) TestValidateOrDropExistingTables() {
374387
suite.Require().NoError(err)
375388
}
376389

390+
func (suite *ApplierTestSuite) TestApplyIterationInsertQuery() {
391+
ctx := context.Background()
392+
393+
connectionConfig, err := suite.getConnectionConfig(ctx)
394+
suite.Require().NoError(err)
395+
396+
migrationContext := base.NewMigrationContext()
397+
migrationContext.ApplierConnectionConfig = connectionConfig
398+
migrationContext.DatabaseName = "test"
399+
migrationContext.OriginalTableName = "testing"
400+
migrationContext.ChunkSize = 10
401+
migrationContext.SetConnectionConfig("innodb")
402+
403+
db, err := suite.getDb(ctx)
404+
suite.Require().NoError(err)
405+
defer db.Close()
406+
407+
_, err = db.Exec("CREATE TABLE test._testing_gho (id INT, item_id INT, PRIMARY KEY (id))")
408+
suite.Require().NoError(err)
409+
410+
// Insert some test values
411+
for i := 1; i <= 10; i++ {
412+
_, err = db.Exec("INSERT INTO test.testing (id, item_id) VALUES (?, ?)", i, i)
413+
suite.Require().NoError(err)
414+
}
415+
416+
migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "item_id"})
417+
migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "item_id"})
418+
migrationContext.UniqueKey = &sql.UniqueKey{
419+
Name: "PRIMARY",
420+
Columns: *sql.NewColumnList([]string{"id"}),
421+
}
422+
423+
migrationContext.MigrationIterationRangeMinValues = sql.ToColumnValues([]interface{}{1})
424+
migrationContext.MigrationIterationRangeMaxValues = sql.ToColumnValues([]interface{}{10})
425+
426+
applier := NewApplier(migrationContext)
427+
defer applier.Teardown()
428+
429+
err = applier.InitDBConnections()
430+
suite.Require().NoError(err)
431+
432+
chunkSize, rowsAffected, duration, err := applier.ApplyIterationInsertQuery()
433+
suite.Require().NoError(err)
434+
435+
suite.Require().Equal(migrationContext.ChunkSize, chunkSize)
436+
suite.Require().Equal(int64(10), rowsAffected)
437+
suite.Require().Greater(duration, time.Duration(0))
438+
439+
// Check that the rows were inserted
440+
rows, err := db.Query("SELECT * FROM test._testing_gho")
441+
suite.Require().NoError(err)
442+
defer rows.Close()
443+
444+
var count, id, item_id int
445+
for rows.Next() {
446+
err = rows.Scan(&id, &item_id)
447+
suite.Require().NoError(err)
448+
count += 1
449+
}
450+
suite.Require().NoError(rows.Err())
451+
452+
suite.Require().Equal(10, count)
453+
}
454+
455+
func (suite *ApplierTestSuite) TestApplyIterationInsertQueryFailsFastWhenSelectingLockedRows() {
456+
ctx := context.Background()
457+
458+
connectionConfig, err := suite.getConnectionConfig(ctx)
459+
suite.Require().NoError(err)
460+
461+
migrationContext := base.NewMigrationContext()
462+
migrationContext.ApplierConnectionConfig = connectionConfig
463+
migrationContext.DatabaseName = "test"
464+
migrationContext.OriginalTableName = "testing"
465+
migrationContext.ChunkSize = 10
466+
migrationContext.TableEngine = "innodb"
467+
migrationContext.SetConnectionConfig("innodb")
468+
469+
db, err := suite.getDb(ctx)
470+
suite.Require().NoError(err)
471+
defer db.Close()
472+
473+
_, err = db.Exec("CREATE TABLE test._testing_gho (id INT, item_id INT, PRIMARY KEY (id))")
474+
suite.Require().NoError(err)
475+
476+
// Insert some test values
477+
for i := 1; i <= 10; i++ {
478+
_, err = db.Exec("INSERT INTO test.testing (id, item_id) VALUES (?, ?)", i, i)
479+
suite.Require().NoError(err)
480+
}
481+
482+
migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "item_id"})
483+
migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "item_id"})
484+
migrationContext.UniqueKey = &sql.UniqueKey{
485+
Name: "PRIMARY",
486+
Columns: *sql.NewColumnList([]string{"id"}),
487+
}
488+
489+
migrationContext.MigrationIterationRangeMinValues = sql.ToColumnValues([]interface{}{1})
490+
migrationContext.MigrationIterationRangeMaxValues = sql.ToColumnValues([]interface{}{10})
491+
492+
applier := NewApplier(migrationContext)
493+
defer applier.Teardown()
494+
495+
err = applier.InitDBConnections()
496+
suite.Require().NoError(err)
497+
498+
// Lock one of the rows
499+
tx, err := db.Begin()
500+
suite.Require().NoError(err)
501+
defer func() {
502+
suite.Require().NoError(tx.Rollback())
503+
}()
504+
505+
_, err = tx.Exec("SELECT * FROM test.testing WHERE id = 5 FOR UPDATE")
506+
suite.Require().NoError(err)
507+
508+
chunkSize, rowsAffected, duration, err := applier.ApplyIterationInsertQuery()
509+
suite.Require().Error(err)
510+
suite.Require().EqualError(err, "Error 3572 (HY000): Statement aborted because lock(s) could not be acquired immediately and NOWAIT is set.")
511+
512+
suite.Require().Equal(migrationContext.ChunkSize, chunkSize)
513+
suite.Require().Equal(int64(0), rowsAffected)
514+
suite.Require().Equal(time.Duration(0), duration)
515+
516+
// Check that the no rows were inserted
517+
var count int
518+
err = db.QueryRow("SELECT COUNT(*) FROM test._testing_gho").Scan(&count)
519+
suite.Require().NoError(err)
520+
521+
suite.Require().Equal(0, count)
522+
}
523+
377524
func TestApplier(t *testing.T) {
378525
suite.Run(t, new(ApplierTestSuite))
379526
}

‎go/sql/builder.go‎

Lines changed: 11 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -182,7 +182,7 @@ func BuildRangePreparedComparison(columns *ColumnList, args []interface{}, compa
182182
return BuildRangeComparison(columns.Names(), values, args, comparisonSign)
183183
}
184184

185-
func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartValues, rangeEndValues []string, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool) (result string, explodedArgs []interface{}, err error) {
185+
func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartValues, rangeEndValues []string, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool, noWait bool) (result string, explodedArgs []interface{}, err error) {
186186
if len(sharedColumns) == 0 {
187187
return "", explodedArgs, fmt.Errorf("Got 0 shared columns in BuildRangeInsertQuery")
188188
}
@@ -212,15 +212,19 @@ func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName strin
212212
return "", explodedArgs, err
213213
}
214214
explodedArgs = append(explodedArgs, rangeExplodedArgs...)
215+
transactionalClause := ""
216+
if transactionalTable {
217+
if noWait {
218+
transactionalClause = "for share nowait"
219+
} else {
220+
transactionalClause = "lock in share mode"
221+
}
222+
}
215223
rangeEndComparison, rangeExplodedArgs, err := BuildRangeComparison(uniqueKeyColumns.Names(), rangeEndValues, rangeEndArgs, LessThanOrEqualsComparisonSign)
216224
if err != nil {
217225
return "", explodedArgs, err
218226
}
219227
explodedArgs = append(explodedArgs, rangeExplodedArgs...)
220-
transactionalClause := ""
221-
if transactionalTable {
222-
transactionalClause = "lock in share mode"
223-
}
224228
result = fmt.Sprintf(`
225229
insert /* gh-ost %s.%s */ ignore
226230
into
@@ -241,10 +245,10 @@ func BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName strin
241245
return result, explodedArgs, nil
242246
}
243247

244-
func BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool) (result string, explodedArgs []interface{}, err error) {
248+
func BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName string, sharedColumns []string, mappedSharedColumns []string, uniqueKey string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, includeRangeStartValues bool, transactionalTable bool, noWait bool) (result string, explodedArgs []interface{}, err error) {
245249
rangeStartValues := buildColumnsPreparedValues(uniqueKeyColumns)
246250
rangeEndValues := buildColumnsPreparedValues(uniqueKeyColumns)
247-
return BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, includeRangeStartValues, transactionalTable)
251+
return BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, includeRangeStartValues, transactionalTable, noWait)
248252
}
249253

250254
func BuildUniqueKeyRangeEndPreparedQueryViaOffset(databaseName, tableName string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, chunkSize int64, includeRangeStartValues bool, hint string) (result string, explodedArgs []interface{}, err error) {

‎go/sql/builder_test.go‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,7 @@ func TestBuildRangeInsertQuery(t *testing.T) {
171171
rangeStartArgs := []interface{}{3}
172172
rangeEndArgs := []interface{}{103}
173173

174-
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, false)
174+
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, true, true)
175175
require.NoError(t, err)
176176
expected := `
177177
insert /* gh-ost mydb.tbl */ ignore
@@ -186,6 +186,7 @@ func TestBuildRangeInsertQuery(t *testing.T) {
186186
where
187187
(((id > @v1s) or ((id = @v1s)))
188188
and ((id < @v1e) or ((id = @v1e))))
189+
for share nowait
189190
)`
190191
require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
191192
require.Equal(t, []interface{}{3, 3, 103, 103}, explodedArgs)
@@ -198,7 +199,7 @@ func TestBuildRangeInsertQuery(t *testing.T) {
198199
rangeStartArgs := []interface{}{3, 17}
199200
rangeEndArgs := []interface{}{103, 117}
200201

201-
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, false)
202+
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, true, true)
202203
require.NoError(t, err)
203204
expected := `
204205
insert /* gh-ost mydb.tbl */ ignore
@@ -219,6 +220,7 @@ func TestBuildRangeInsertQuery(t *testing.T) {
219220
or (((name = @v1e))
220221
AND (position < @v2e))
221222
or ((name = @v1e) and (position = @v2e))))
223+
for share nowait
222224
)`
223225
require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
224226
require.Equal(t, []interface{}{3, 3, 17, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)
@@ -239,7 +241,7 @@ func TestBuildRangeInsertQueryRenameMap(t *testing.T) {
239241
rangeStartArgs := []interface{}{3}
240242
rangeEndArgs := []interface{}{103}
241243

242-
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, false)
244+
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, true, true)
243245
require.NoError(t, err)
244246
expected := `
245247
insert /* gh-ost mydb.tbl */ ignore
@@ -255,6 +257,7 @@ func TestBuildRangeInsertQueryRenameMap(t *testing.T) {
255257
(((id > @v1s) or ((id = @v1s)))
256258
and
257259
((id < @v1e) or ((id = @v1e))))
260+
for share nowait
258261
)`
259262
require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
260263
require.Equal(t, []interface{}{3, 3, 103, 103}, explodedArgs)
@@ -267,7 +270,7 @@ func TestBuildRangeInsertQueryRenameMap(t *testing.T) {
267270
rangeStartArgs := []interface{}{3, 17}
268271
rangeEndArgs := []interface{}{103, 117}
269272

270-
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, false)
273+
query, explodedArgs, err := BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, true, true, true)
271274
require.NoError(t, err)
272275
expected := `
273276
insert /* gh-ost mydb.tbl */ ignore
@@ -284,6 +287,7 @@ func TestBuildRangeInsertQueryRenameMap(t *testing.T) {
284287
AND (position > @v2s)) or ((name = @v1s) and (position = @v2s)))
285288
and ((name < @v1e) or (((name = @v1e)) AND (position < @v2e))
286289
or ((name = @v1e) and (position = @v2e))))
290+
for share nowait
287291
)`
288292
require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
289293
require.Equal(t, []interface{}{3, 3, 17, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)
@@ -301,7 +305,7 @@ func TestBuildRangeInsertPreparedQuery(t *testing.T) {
301305
rangeStartArgs := []interface{}{3, 17}
302306
rangeEndArgs := []interface{}{103, 117}
303307

304-
query, explodedArgs, err := BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartArgs, rangeEndArgs, true, true)
308+
query, explodedArgs, err := BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableName, sharedColumns, sharedColumns, uniqueKey, uniqueKeyColumns, rangeStartArgs, rangeEndArgs, true, true, true)
305309
require.NoError(t, err)
306310
expected := `
307311
insert /* gh-ost mydb.tbl */ ignore
@@ -314,7 +318,7 @@ func TestBuildRangeInsertPreparedQuery(t *testing.T) {
314318
mydb.tbl
315319
force index (name_position_uidx)
316320
where (((name > ?) or (((name = ?)) AND (position > ?)) or ((name = ?) and (position = ?))) and ((name < ?) or (((name = ?)) AND (position < ?)) or ((name = ?) and (position = ?))))
317-
lock in share mode
321+
for share nowait
318322
)`
319323
require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
320324
require.Equal(t, []interface{}{3, 3, 17, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)

0 commit comments

Comments
 (0)