move-tables: support generated and JSON columns

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)

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
 		}

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

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
 }

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()