diff --git a/go/base/context.go b/go/base/context.go index 305769381..4c84302b3 100644 --- a/go/base/context.go +++ b/go/base/context.go @@ -100,8 +100,8 @@ type MoveTable struct { CreateTableStatement string // Schema, captured from the source (or from the target, on resume). In - // move-tables mode source and target schemas match, so the shared columns are - // identical to the original columns. + // move-tables mode source and target schemas match, so shared columns are the + // original columns minus generated columns, which MySQL recomputes on the target. OriginalTableColumns *sql.ColumnList OriginalTableVirtualColumns *sql.ColumnList OriginalTableUniqueKeys [](*sql.UniqueKey) diff --git a/go/logic/applier.go b/go/logic/applier.go index 0a99919ce..b41580f43 100644 --- a/go/logic/applier.go +++ b/go/logic/applier.go @@ -507,7 +507,7 @@ func (apl *Applier) prepareQueries() (err error) { if b.copySelectFirstQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder( mt.SourceDatabaseName, mt.SourceTableName, - mt.OriginalTableColumns, + mt.SharedColumns, mt.UniqueKey.Name, &mt.UniqueKey.Columns, true, // <-- include start range values for first select query @@ -517,7 +517,7 @@ func (apl *Applier) prepareQueries() (err error) { if b.copySelectNextQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder( mt.SourceDatabaseName, mt.SourceTableName, - mt.OriginalTableColumns, + mt.SharedColumns, mt.UniqueKey.Name, &mt.UniqueKey.Columns, false, @@ -527,7 +527,7 @@ func (apl *Applier) prepareQueries() (err error) { if b.copyInsertQueryBuilder, err = sql.NewMoveTableCopyInsertQueryBuilder( mt.TargetDatabaseName, mt.TargetTableName, - mt.OriginalTableColumns, + mt.SharedColumns, ); err != nil { return err } diff --git a/go/logic/applier_test.go b/go/logic/applier_test.go index f22c7d257..662bb62b1 100644 --- a/go/logic/applier_test.go +++ b/go/logic/applier_test.go @@ -2127,6 +2127,74 @@ func (suite *ApplierTestSuite) TestApplyDMLEventQueriesMoveTablesMode() { suite.Require().Equal(int64(0), migrationContext.RowsDeltaEstimate) } +func (suite *ApplierTestSuite) TestApplyDMLEventQueriesMoveTablesGeneratedColumns() { + ctx := context.Background() + createTable := "CREATE TABLE %s (id INT NOT NULL, a INT NOT NULL, virtual_sum INT AS (a + 10) VIRTUAL, b INT NOT NULL, stored_sum INT AS (a + b) STORED, PRIMARY KEY(id));" + _, err := suite.db.ExecContext(ctx, fmt.Sprintf(createTable, getTestTableName())) + suite.Require().NoError(err) + _, err = suite.otherDB.ExecContext(ctx, fmt.Sprintf(createTable, getTestOtherTableName())) + suite.Require().NoError(err) + + connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer) + suite.Require().NoError(err) + + migrationContext := newTestMigrationContext() + migrationContext.ApplierConnectionConfig = connectionConfig + migrationContext.MoveTables.ConnectionConfig = connectionConfig + migrationContext.SetConnectionConfig("innodb") + migrationContext.MoveTables.TableNames = []string{testMysqlTableName} + migrationContext.MoveTables.TargetDatabase = testMysqlDatabaseOther + migrationContext.InitMoveTableContainers() + mt := migrationContext.GetMoveTable(testMysqlTableName) + suite.Require().NotNil(mt) + mt.OriginalTableColumns = sql.NewColumnList([]string{"id", "a", "virtual_sum", "b", "stored_sum"}) + mt.SharedColumns = sql.NewColumnList([]string{"id", "a", "b"}) + mt.MappedSharedColumns = sql.NewColumnList([]string{"id", "a", "b"}) + mt.UniqueKey = &sql.UniqueKey{Name: "PRIMARY", Columns: *sql.NewColumnList([]string{"id"})} + + applier := NewApplier(migrationContext) + suite.Require().NoError(applier.prepareQueries()) + defer applier.Teardown() + suite.Require().NoError(applier.InitDBConnections()) + + err = applier.ApplyDMLEventQueries([]*binlog.BinlogDMLEvent{ + { + DatabaseName: testMysqlDatabase, + TableName: testMysqlTableName, + DML: binlog.InsertDML, + NewColumnValues: sql.ToColumnValues([]interface{}{1, 2, 12, 3, 5}), + }, + { + DatabaseName: testMysqlDatabase, + TableName: testMysqlTableName, + DML: binlog.UpdateDML, + WhereColumnValues: sql.ToColumnValues([]interface{}{1, 2, 12, 3, 5}), + NewColumnValues: sql.ToColumnValues([]interface{}{1, 7, 17, 11, 18}), + }, + }) + suite.Require().NoError(err) + + var id, a, virtualSum, b, storedSum int + err = suite.otherDB.QueryRowContext(ctx, "SELECT id, a, virtual_sum, b, stored_sum FROM "+getTestOtherTableName()).Scan(&id, &a, &virtualSum, &b, &storedSum) + suite.Require().NoError(err) + suite.Require().Equal([]int{1, 7, 17, 11, 18}, []int{id, a, virtualSum, b, storedSum}) + + err = applier.ApplyDMLEventQueries([]*binlog.BinlogDMLEvent{ + { + DatabaseName: testMysqlDatabase, + TableName: testMysqlTableName, + DML: binlog.DeleteDML, + WhereColumnValues: sql.ToColumnValues([]interface{}{1, 7, 17, 11, 18}), + }, + }) + suite.Require().NoError(err) + + var count int + err = suite.otherDB.QueryRowContext(ctx, "SELECT COUNT(*) FROM "+getTestOtherTableName()).Scan(&count) + suite.Require().NoError(err) + suite.Require().Zero(count) +} + func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueries() { ctx := context.Background() var err error @@ -2216,6 +2284,62 @@ func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueries() { suite.Require().Equal("2025-12-31 23:59:59", results[2].createdAt) } +func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueriesGeneratedColumns() { + ctx := context.Background() + createTable := "CREATE TABLE %s (id INT NOT NULL, a INT NOT NULL, virtual_sum INT AS (a + 10) VIRTUAL, b INT NOT NULL, stored_sum INT AS (a + b) STORED NOT NULL, UNIQUE KEY stored_sum_uidx (stored_sum));" + _, err := suite.db.ExecContext(ctx, fmt.Sprintf(createTable, getTestTableName())) + suite.Require().NoError(err) + _, err = suite.otherDB.ExecContext(ctx, fmt.Sprintf(createTable, getTestOtherTableName())) + suite.Require().NoError(err) + _, err = suite.db.ExecContext(ctx, "INSERT INTO "+getTestTableName()+" (id, a, b) VALUES (1, 2, 3), (2, 5, 8)") + suite.Require().NoError(err) + + connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer) + suite.Require().NoError(err) + + migrationContext := newTestMigrationContext() + migrationContext.ApplierConnectionConfig = connectionConfig + migrationContext.MoveTables.ConnectionConfig = connectionConfig + migrationContext.SetConnectionConfig("innodb") + migrationContext.MoveTables.TableNames = []string{testMysqlTableName} + migrationContext.MoveTables.TargetDatabase = testMysqlDatabaseOther + migrationContext.InitMoveTableContainers() + mt := migrationContext.GetMoveTable(testMysqlTableName) + suite.Require().NotNil(mt) + mt.OriginalTableColumns = sql.NewColumnList([]string{"id", "a", "virtual_sum", "b", "stored_sum"}) + mt.SharedColumns = sql.NewColumnList([]string{"id", "a", "b"}) + mt.MappedSharedColumns = sql.NewColumnList([]string{"id", "a", "b"}) + uniqueKeyColumns := sql.NewColumnList([]string{"stored_sum"}) + uniqueKeyColumns.GetColumn("stored_sum").IsVirtual = true + mt.UniqueKey = &sql.UniqueKey{Name: "stored_sum_uidx", Columns: *uniqueKeyColumns} + + applier := NewApplier(migrationContext) + suite.Require().NoError(applier.prepareQueries()) + defer applier.Teardown() + suite.Require().NoError(applier.InitDBConnections()) + suite.Require().NoError(applier.ReadMoveTableMigrationRangeValues(nil, mt)) + + mt.SetNextIterationRangeMinValues() + hasFurtherRange, err := applier.CalculateMoveTableNextIterationRangeEndValues(applier.db, mt) + suite.Require().NoError(err) + suite.Require().True(hasFurtherRange) + _, rowsAffected, _, err := applier.ApplyIterationMoveTableCopyQueries(applier.db, mt) + suite.Require().NoError(err) + suite.Require().Equal(int64(2), rowsAffected) + + rows, err := suite.otherDB.QueryContext(ctx, "SELECT id, a, virtual_sum, b, stored_sum FROM "+getTestOtherTableName()+" ORDER BY id") + suite.Require().NoError(err) + defer rows.Close() + var results [][]int + for rows.Next() { + var id, a, virtualSum, b, storedSum int + suite.Require().NoError(rows.Scan(&id, &a, &virtualSum, &b, &storedSum)) + results = append(results, []int{id, a, virtualSum, b, storedSum}) + } + suite.Require().NoError(rows.Err()) + suite.Require().Equal([][]int{{1, 2, 12, 3, 5}, {2, 5, 15, 8, 13}}, results) +} + func (suite *ApplierTestSuite) TestApplyIterationMoveTableCopyQueriesNoRows() { ctx := context.Background() var err error diff --git a/go/logic/migrator.go b/go/logic/migrator.go index d0d6c39c8..77059a64f 100644 --- a/go/logic/migrator.go +++ b/go/logic/migrator.go @@ -807,6 +807,47 @@ func (mgtr *Migrator) Revert() error { return nil } +func moveTablesWritableColumns(columns, virtualColumns *sql.ColumnList) *sql.ColumnList { + generatedColumnNames := make(map[string]bool, virtualColumns.Len()) + for _, columnName := range virtualColumns.Names() { + generatedColumnNames[strings.ToLower(columnName)] = true + } + + writableColumnNames := make([]string, 0, columns.Len()) + for _, columnName := range columns.Names() { + if !generatedColumnNames[strings.ToLower(columnName)] { + writableColumnNames = append(writableColumnNames, columnName) + } + } + return sql.NewColumnList(writableColumnNames) +} + +func prepareMoveTableColumnMetadata(inspector *Inspector, databaseName, tableName string, mt *base.MoveTable) error { + // Generated columns are present in row events but are not writable on the target. + // Keep separate source and target lists because query builders may mutate column metadata. + mt.SharedColumns = moveTablesWritableColumns(mt.OriginalTableColumns, mt.OriginalTableVirtualColumns) + if mt.SharedColumns.Len() == 0 { + return fmt.Errorf("move-table %s.%s has no writable columns after excluding generated columns", + sql.EscapeName(databaseName), sql.EscapeName(tableName)) + } + mt.MappedSharedColumns = moveTablesWritableColumns(mt.OriginalTableColumns, mt.OriginalTableVirtualColumns) + + // Move-tables does not perform schema conversions, but query builders still + // need type metadata to encode values such as JSON, unsigned, and binary correctly. + if err := inspector.applyColumnTypes( + databaseName, + tableName, + mt.OriginalTableColumns, + mt.SharedColumns, + mt.MappedSharedColumns, + &mt.UniqueKey.Columns, + ); err != nil { + return fmt.Errorf("failed to inspect column types for move-table %s.%s: %w", + sql.EscapeName(databaseName), sql.EscapeName(tableName), err) + } + return nil +} + // prepareMoveTablesCopyState initializes per-table runtime state for row copy in // move-tables mode (ยง2.1). Each migrated table is inspected and validated // independently into its own container (schema, unique key, row estimate, CREATE @@ -847,9 +888,9 @@ func (mgtr *Migrator) prepareMoveTablesCopyState() error { mt.OriginalTableVirtualColumns = virtualColumns mt.OriginalTableUniqueKeys = uniqueKeys mt.UniqueKey = uniqueKey - // In move-tables mode source and target schemas match, so shared columns are identical. - mt.SharedColumns = columns - mt.MappedSharedColumns = columns + if err := prepareMoveTableColumnMetadata(mgtr.inspector, mt.SourceDatabaseName, mt.SourceTableName, mt); err != nil { + return err + } mt.RowsEstimate = rowsEstimate mt.CreateTableStatement = createStatement totalRowsEstimate += rowsEstimate @@ -880,8 +921,9 @@ func (mgtr *Migrator) hydrateMoveTablesStateFromTarget() error { mt.OriginalTableVirtualColumns = virtualColumns mt.OriginalTableUniqueKeys = uniqueKeys mt.UniqueKey = uniqueKey - mt.SharedColumns = columns - mt.MappedSharedColumns = columns + if err := prepareMoveTableColumnMetadata(targetInspector, mt.TargetDatabaseName, mt.TargetTableName, mt); err != nil { + return fmt.Errorf("failed to hydrate move-table state while resuming: %w", err) + } } return nil } diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go index d6e972cc3..06bdc1b4e 100644 --- a/go/logic/migrator_test.go +++ b/go/logic/migrator_test.go @@ -34,6 +34,58 @@ import ( "github.com/testcontainers/testcontainers-go" ) +func TestMoveTablesWritableColumns(t *testing.T) { + testCases := []struct { + name string + columnNames []string + generatedNames []string + expectedWritable []string + }{ + { + name: "generated columns in middle and end", + columnNames: []string{"id", "virtual_value", "persisted_value", "stored_value"}, + generatedNames: []string{"virtual_value", "stored_value"}, + expectedWritable: []string{"id", "persisted_value"}, + }, + { + name: "generated names match case insensitively", + columnNames: []string{"ID", "Virtual_Value", "persisted_value", "Stored_Value"}, + generatedNames: []string{"virtual_value", "STORED_VALUE"}, + expectedWritable: []string{"ID", "persisted_value"}, + }, + { + name: "no generated columns", + columnNames: []string{"id", "first_value", "second_value"}, + generatedNames: nil, + expectedWritable: []string{"id", "first_value", "second_value"}, + }, + { + name: "all columns generated", + columnNames: []string{"virtual_value", "stored_value"}, + generatedNames: []string{"virtual_value", "stored_value"}, + expectedWritable: []string{}, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + columns := sql.NewColumnList(testCase.columnNames) + generatedColumns := sql.NewColumnList(testCase.generatedNames) + writableColumns := moveTablesWritableColumns(columns, generatedColumns) + + require.Equal(t, testCase.expectedWritable, writableColumns.Names()) + require.NotSame(t, columns, writableColumns) + for ordinal, columnName := range testCase.expectedWritable { + require.Equal(t, ordinal, writableColumns.Ordinals[columnName]) + require.Equal(t, columnName, writableColumns.Columns()[ordinal].Name) + } + + mappedWritableColumns := moveTablesWritableColumns(columns, generatedColumns) + require.NotSame(t, writableColumns, mappedWritableColumns) + }) + } +} + func TestMigratorOnChangelogEvent(t *testing.T) { migrationContext := base.NewMigrationContext() migrator := NewMigrator(migrationContext, "1.2.3") @@ -518,6 +570,58 @@ func (suite *MigratorTestSuite) TestMigrateEmpty() { suite.Require().Equal("_testing_del", tableName) } +func (suite *MigratorTestSuite) TestMoveTablesStateInitializesColumnMetadata() { + ctx := context.Background() + _, err := suite.db.ExecContext(ctx, fmt.Sprintf(` + CREATE TABLE %s ( + id INT PRIMARY KEY, + unsigned_value BIGINT UNSIGNED NOT NULL, + json_value JSON, + virtual_json_value VARCHAR(16) AS ( + COALESCE(JSON_UNQUOTE(JSON_EXTRACT(json_value, '$.value')), 'direct') + ) VIRTUAL, + binary_value BINARY(4) + )`, getTestTableName())) + suite.Require().NoError(err) + + newMoveTablesMigrator := func() (*Migrator, *base.MoveTable) { + migrationContext := newTestMigrationContext() + migrationContext.MoveTables.TableNames = []string{testMysqlTableName} + migrationContext.MoveTables.TargetDatabase = testMysqlDatabase + migrationContext.InitMoveTableContainers() + migrator := NewMigrator(migrationContext, "0.0.0") + return migrator, migrationContext.GetMoveTable(testMysqlTableName) + } + assertHydrated := func(mt *base.MoveTable) { + suite.Require().Equal( + []string{"id", "unsigned_value", "json_value", "binary_value"}, + mt.SharedColumns.Names(), + ) + suite.Require().Equal(sql.JSONColumnType, mt.OriginalTableColumns.GetColumnType("json_value")) + suite.Require().Equal(sql.JSONColumnType, mt.SharedColumns.GetColumnType("json_value")) + suite.Require().Equal(sql.JSONColumnType, mt.MappedSharedColumns.GetColumnType("json_value")) + suite.Require().True(mt.SharedColumns.IsUnsigned("unsigned_value")) + suite.Require().True(mt.MappedSharedColumns.IsUnsigned("unsigned_value")) + suite.Require().Equal(sql.BinaryColumnType, mt.SharedColumns.GetColumnType("binary_value")) + suite.Require().Equal(uint(4), mt.SharedColumns.GetColumn("binary_value").BinaryOctetLength) + } + + suite.Run("fresh preparation", func() { + migrator, mt := newMoveTablesMigrator() + migrator.inspector = &Inspector{db: suite.db, migrationContext: migrator.migrationContext} + suite.Require().NoError(migrator.prepareMoveTablesCopyState()) + assertHydrated(mt) + }) + + suite.Run("resume hydration", func() { + migrator, mt := newMoveTablesMigrator() + migrator.applier = NewApplier(migrator.migrationContext) + migrator.applier.moveTablesTargetDB = suite.db + suite.Require().NoError(migrator.hydrateMoveTablesStateFromTarget()) + assertHydrated(mt) + }) +} + func (suite *MigratorTestSuite) TestRetryBatchCopyWithHooks() { ctx := context.Background() diff --git a/localtests/move-tables/generated-columns/create.sql b/localtests/move-tables/generated-columns/create.sql new file mode 100644 index 000000000..c326893c3 --- /dev/null +++ b/localtests/move-tables/generated-columns/create.sql @@ -0,0 +1,36 @@ +drop table if exists gh_ost_test; +create table gh_ost_test ( + id int auto_increment, + a int not null, + virtual_sum int as (a + 10) virtual not null, + b int not null, + stored_sum int as (a + b) stored not null, + json_value json default null, + virtual_json_value varchar(16) as ( + coalesce(json_unquote(json_extract(json_value, '$.value')), 'direct') + ) virtual, + primary key(id) +) auto_increment=1; + +insert into gh_ost_test (a, b, json_value) values + (1, 2, json_object('value', 'team')), + (3, 5, json_object('value', 'project')), + (8, 13, null); + +drop event if exists gh_ost_test; +delimiter ;; +create event gh_ost_test + on schedule every 1 second + starts current_timestamp + ends current_timestamp + interval 60 second + on completion not preserve + enable + do +begin + insert into gh_ost_test (a, b, json_value) values (2, 3, json_object('value', 'team')); + insert into gh_ost_test (a, b, json_value) values (5, 8, json_object('value', 'project')); + insert into gh_ost_test (a, b, json_value) values (13, 21, null); + update gh_ost_test set a=a+1, b=b+2, json_value=json_object('value', 'updated') where id <= 3; + update gh_ost_test set b=b+1 where id > 3; + delete from gh_ost_test where id > 3 order by id limit 1; +end ;; diff --git a/localtests/move-tables/generated-columns/tables.txt b/localtests/move-tables/generated-columns/tables.txt new file mode 100644 index 000000000..11fc5eef8 --- /dev/null +++ b/localtests/move-tables/generated-columns/tables.txt @@ -0,0 +1 @@ +gh_ost_test diff --git a/localtests/move-tables/json/create.sql b/localtests/move-tables/json/create.sql new file mode 100644 index 000000000..ef1d2726b --- /dev/null +++ b/localtests/move-tables/json/create.sql @@ -0,0 +1,27 @@ +create table gh_ost_test ( + id int auto_increment, + json_value json not null, + primary key(id) +) auto_increment=1; + +insert into gh_ost_test (json_value) values + (json_object('message', 'first', 'nested', json_object('enabled', true))), + (json_object('message', 'second', 'items', json_array(1, 2, 3))), + (json_object('message', 'third', 'value', 42)); + +drop event if exists gh_ost_test; +delimiter ;; +create event gh_ost_test + on schedule every 1 second + starts current_timestamp + ends current_timestamp + interval 60 second + on completion not preserve + enable + do +begin + insert into gh_ost_test (json_value) values + (json_object('message', 'inserted', 'items', json_array('a', 'b'))); + update gh_ost_test + set json_value=json_set(json_value, '$.updated', true) + where id <= 3; +end ;; \ No newline at end of file diff --git a/localtests/move-tables/json/tables.txt b/localtests/move-tables/json/tables.txt new file mode 100644 index 000000000..11fc5eef8 --- /dev/null +++ b/localtests/move-tables/json/tables.txt @@ -0,0 +1 @@ +gh_ost_test diff --git a/localtests/move-tables/unsigned/create.sql b/localtests/move-tables/unsigned/create.sql new file mode 100644 index 000000000..2c809d7ca --- /dev/null +++ b/localtests/move-tables/unsigned/create.sql @@ -0,0 +1,32 @@ +drop table if exists gh_ost_test; +create table gh_ost_test ( + id int auto_increment, + signed_value bigint not null, + unsigned_value int unsigned not null, + unsigned_big_value bigint unsigned not null, + primary key(id) +) auto_increment=1; + +insert into gh_ost_test (signed_value, unsigned_value, unsigned_big_value) values + (-9223372036854775807, 4294967295, 18446744073709551615), + (-9223372036854775806, 4294967294, 18446744073709551614), + (-9223372036854775805, 4294967293, 18446744073709551613); + +drop event if exists gh_ost_test; +delimiter ;; +create event gh_ost_test + on schedule every 1 second + starts current_timestamp + ends current_timestamp + interval 60 second + on completion not preserve + enable + do +begin + insert into gh_ost_test (signed_value, unsigned_value, unsigned_big_value) values + (-9223372036854775804, 4294967292, 18446744073709551612); + update gh_ost_test + set signed_value=signed_value+1, + unsigned_value=unsigned_value-1, + unsigned_big_value=unsigned_big_value-1 + where id <= 3; +end ;; \ No newline at end of file diff --git a/localtests/move-tables/unsigned/tables.txt b/localtests/move-tables/unsigned/tables.txt new file mode 100644 index 000000000..11fc5eef8 --- /dev/null +++ b/localtests/move-tables/unsigned/tables.txt @@ -0,0 +1 @@ +gh_ost_test