From bb1385755ee2e42e0dee82289dbc19bb4e47aa34 Mon Sep 17 00:00:00 2001 From: Igor Konstantinov <90114057+kotigor@users.noreply.github.com> Date: Mon, 5 Oct 2026 13:53:11 +0200 Subject: [PATCH] Fix atomic cut-over on MySQL 9.1+ under utf8mb4 load On MySQL 9.1 and later (Oracle MySQL and Percona Server) the atomic cut-over never found its RENAME, and every attempt ended with Error 1205, as soon as another session ran a statement not representable in utf8mb3: an emoji in a string literal, or a binary parameter that is not valid UTF-8. Under write load the table lock of the cut-over queues exactly such statements. ExpectProcess read information_schema.processlist. Filling that table converts the statement text of every session into its utf8mb3 INFO column, and since 9.1 character set conversions into temporary tables are strict (sql_tmp_table.cc), so the long-standing MySQL Bug #87579 turned from a warning into Error 3854 that fails the whole read. The error arrives after the result set has started; sqlutils.QueryRowsMap does not check rows.Err(), so the read looked empty and the log only showed "cannot find process. Hints: metadata lock, rename". ExpectProcess now reads performance_schema.processlist (MySQL 8.0.22+) with the same conditions; it holds the statement text in the session's own character set and is not affected. Whether that table is usable is decided once when the applier initialises; information_schema.processlist remains the fallback when it is absent (MySQL 5.7, 8.0 before 8.0.22, MariaDB), empty because performance_schema is disabled, or not readable. On MySQL 9.1+ the fallback logs a warning naming the risk and the remedy. ExpectProcess also reads through QueryRow().Scan(), which reports the error of a failed read, and logs it, instead of passing it for a missing process. Fixes #1780 --- go/logic/applier.go | 79 +++++++++++--- go/logic/applier_test.go | 222 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 287 insertions(+), 14 deletions(-) diff --git a/go/logic/applier.go b/go/logic/applier.go index d1c8d4716..efd272f84 100644 --- a/go/logic/applier.go +++ b/go/logic/applier.go @@ -29,6 +29,7 @@ import ( "github.com/github/gh-ost/go/mysql" drivermysql "github.com/go-sql-driver/mysql" + version "github.com/hashicorp/go-version" "github.com/openark/golib/sqlutils" ) @@ -92,6 +93,10 @@ type Applier struct { migrationLockName string migrationLockStop chan struct{} migrationLockDone chan struct{} + + // usePerformanceSchemaProcesslist tells ExpectProcess to read performance_schema.processlist + // rather than information_schema.processlist; see chooseProcesslistTable() + usePerformanceSchemaProcesslist bool } func NewApplier(migrationContext *base.MigrationContext) *Applier { @@ -140,6 +145,7 @@ func (apl *Applier) InitDBConnections() (err error) { if err := apl.validateAndReadGlobalVariables(); err != nil { return err } + apl.chooseProcesslistTable() if !apl.migrationContext.AliyunRDS && !apl.migrationContext.GoogleCloudPlatform && !apl.migrationContext.AzureMySQL { if impliedKey, err := mysql.GetInstanceKey(apl.db); err != nil { return err @@ -351,6 +357,47 @@ func (apl *Applier) validateAndReadGlobalVariables() error { return nil } +// chooseProcesslistTable decides which table ExpectProcess reads. performance_schema.processlist is +// preferred: since MySQL 9.1, reading information_schema.processlist fails with Error 3854 while any +// other session runs a statement not representable in utf8mb3, such as one with an emoji. +// information_schema.processlist remains the fallback when performance_schema.processlist is not usable: +// absent (MySQL 5.7, 8.0 before 8.0.22, MariaDB), empty because performance_schema is disabled, or not +// readable by this account. +func (apl *Applier) chooseProcesslistTable() { + var ownSessions int64 + query := `select /* gh-ost */ count(*) from performance_schema.processlist where id = connection_id()` + err := apl.db.QueryRow(query).Scan(&ownSessions) + if err == nil && ownSessions == 1 { + apl.usePerformanceSchemaProcesslist = true + apl.migrationContext.Log.Infof("will use performance_schema.processlist on applier") + return + } + reason := "own session not listed, performance_schema may be disabled" + if err != nil { + reason = err.Error() + } + if hasStrictTmpTableCharsetConversion(apl.migrationContext.ApplierMySQLVersion) { + apl.migrationContext.Log.Warningf("performance_schema.processlist is not usable on applier (%s); falling back to information_schema.processlist. On MySQL 9.1 and later reading it fails while any other session runs a statement not representable in utf8mb3 (such as one with an emoji), and the atomic cut-over may then fail to find its RENAME. Enable performance_schema, and make sure this user can read performance_schema.processlist, to avoid this. See https://github.com/github/gh-ost/issues/1780 for details", reason) + return + } + apl.migrationContext.Log.Infof("will use information_schema.processlist on applier: performance_schema.processlist is not usable (%s)", reason) +} + +// hasStrictTmpTableCharsetConversion tells whether the server is MySQL 9.1 or later, where character +// set conversions into temporary tables are strict: information_schema.processlist then fails with +// Error 3854 instead of a warning while a session runs a statement not representable in utf8mb3 +// (MySQL Bug #87579). +func hasStrictTmpTableCharsetConversion(mysqlVersion string) bool { + if mysql.IsMariaDB(mysqlVersion) { + return false + } + serverVersion, err := version.NewVersion(mysqlVersion) + if err != nil { + return false + } + return serverVersion.Core().GreaterThanOrEqual(version.Must(version.NewVersion("9.1"))) +} + // generateSqlModeQuery return a `sql_mode = ...` query, to be wrapped with a `set session` or `set global`, // based on gh-ost configuration: // - User may skip strict mode @@ -1443,28 +1490,32 @@ func (apl *Applier) ExpectUsedLock(sessionId int64) error { return nil } -// ExpectProcess expects a process to show up in `SHOW PROCESSLIST` that has given characteristics +// ExpectProcess expects a process to show up in `SHOW PROCESSLIST` that has given characteristics. +// It reads performance_schema.processlist when usable, information_schema.processlist otherwise. func (apl *Applier) ExpectProcess(sessionId int64, stateHint, infoHint string) error { - found := false - query := ` + processlistTable := "information_schema.processlist" + if apl.usePerformanceSchemaProcesslist { + processlistTable = "performance_schema.processlist" + } + query := fmt.Sprintf(` select /* gh-ost */ id from - information_schema.processlist + %s where id != connection_id() and ? in (0, id) - and state like concat('%', ?, '%') - and info like concat('%', ?, '%')` - err := sqlutils.QueryRowsMap(apl.db, query, func(m sqlutils.RowMap) error { - found = true - return nil - }, sessionId, stateHint, infoHint) - if err != nil { - return err - } - if !found { + and state like concat('%%', ?, '%%') + and info like concat('%%', ?, '%%')`, processlistTable) + // A failed read may only report its error after the result set has started (Error 3854 does); + // sqlutils.QueryRowsMap would drop it and report the process as missing, Scan() returns it. + var id int64 + err := apl.db.QueryRow(query, sessionId, stateHint, infoHint).Scan(&id) + if errors.Is(err, gosql.ErrNoRows) { return fmt.Errorf("cannot find process. Hints: %s, %s", stateHint, infoHint) } + if err != nil { + return apl.migrationContext.Log.Errore(fmt.Errorf("failed reading %s: %w", processlistTable, err)) + } return nil } diff --git a/go/logic/applier_test.go b/go/logic/applier_test.go index f1fa28bc8..3692a3200 100644 --- a/go/logic/applier_test.go +++ b/go/logic/applier_test.go @@ -10,6 +10,7 @@ import ( gosql "database/sql" "errors" "strings" + "sync" "testing" "time" @@ -333,6 +334,78 @@ func TestClassifyAnalyzeTableResult(t *testing.T) { } } +func TestHasStrictTmpTableCharsetConversion(t *testing.T) { + tests := []struct { + version string + strict bool + }{ + {version: "5.7.41-log", strict: false}, + {version: "8.0.42", strict: false}, + {version: "8.0.41-32", strict: false}, + {version: "8.4.3", strict: false}, + {version: "9.0.1", strict: false}, + {version: "9.1.0", strict: true}, + {version: "9.1.0-commercial", strict: true}, + {version: "9.7.1-1", strict: true}, + {version: "9.7.2", strict: true}, + {version: "10.6.27-MariaDB-log", strict: false}, + {version: "11.8.8-MariaDB-ubu2404", strict: false}, + {version: "", strict: false}, + } + for _, tc := range tests { + t.Run(tc.version, func(t *testing.T) { + require.Equal(t, tc.strict, hasStrictTmpTableCharsetConversion(tc.version)) + }) + } +} + +// lockTestTableAndQueue write-locks the test table, then runs each of the given statements on a +// connection of its own and waits until it is blocked on a metadata lock behind that table lock. +// It returns the session ids of these connections, in order, and a function that releases the table +// lock and waits for the statements to finish; their outcome is not checked. +func lockTestTableAndQueue(t *testing.T, db *gosql.DB, statements ...string) (sessionIds []int64, unlock func()) { + ctx := context.Background() + + lockConn, err := db.Conn(ctx) + require.NoError(t, err) + var wg sync.WaitGroup + unlock = func() { + _, _ = lockConn.ExecContext(ctx, "UNLOCK TABLES") + _ = lockConn.Close() + wg.Wait() + } + queued := false + defer func() { + if !queued { + unlock() + } + }() + + _, err = lockConn.ExecContext(ctx, "LOCK TABLES "+getTestTableName()+" WRITE") + require.NoError(t, err) + + for _, statement := range statements { + conn, err := db.Conn(ctx) + require.NoError(t, err) + var sessionId int64 + require.NoError(t, conn.QueryRowContext(ctx, "SELECT CONNECTION_ID()").Scan(&sessionId)) + wg.Add(1) + go func() { + defer wg.Done() + defer conn.Close() + _, _ = conn.ExecContext(ctx, statement) + }() + require.Eventually(t, func() bool { + var waiting int64 + err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM performance_schema.processlist WHERE id = ? AND state LIKE '%metadata lock%'`, sessionId).Scan(&waiting) + return err == nil && waiting == 1 + }, 10*time.Second, 10*time.Millisecond, "statement is not waiting for the table lock: %s", statement) + sessionIds = append(sessionIds, sessionId) + } + queued = true + return sessionIds, unlock +} + type ApplierTestSuite struct { suite.Suite @@ -407,6 +480,7 @@ func (suite *ApplierTestSuite) TestInitDBConnections() { suite.Require().Equal(mysqlVersion, migrationContext.ApplierMySQLVersion) suite.Require().Equal(int64(28800), migrationContext.ApplierWaitTimeout) suite.Require().Equal("+00:00", migrationContext.ApplierTimeZone) + suite.Require().True(applier.usePerformanceSchemaProcesslist) suite.Require().Equal(sql.NewColumnList([]string{"id", "item_id"}), migrationContext.OriginalTableColumnsOnApplier) } @@ -631,6 +705,44 @@ func (suite *ApplierTestSuite) TestAcquireMigrationLockFailsWhenHeld() { suite.Require().Nil(applierB.migrationLockConn) } +func (suite *ApplierTestSuite) TestExpectProcess() { + ctx := context.Background() + + _, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, name VARCHAR(64))", getTestTableName())) + suite.Require().NoError(err) + + connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer) + suite.Require().NoError(err) + + migrationContext := newTestMigrationContext() + migrationContext.ApplierConnectionConfig = connectionConfig + migrationContext.SetConnectionConfig("innodb") + + applier := NewApplier(migrationContext) + defer applier.Teardown() + suite.Require().NoError(applier.InitDBConnections()) + suite.Require().True(applier.usePerformanceSchemaProcesslist) + + sessionIds, unlock := lockTestTableAndQueue(suite.T(), suite.db, + fmt.Sprintf("RENAME TABLE %s TO %s", getTestTableName(), getTestOldTableName()), + ) + defer func() { + unlock() + _, err := suite.db.ExecContext(ctx, "DROP TABLE IF EXISTS "+getTestOldTableName()) + suite.Require().NoError(err) + }() + renameSessionId := sessionIds[0] + + suite.Require().NoError(applier.ExpectProcess(renameSessionId, "metadata lock", "rename")) + suite.Require().NoError(applier.ExpectProcess(0, "metadata lock", "rename")) + suite.Require().ErrorContains(applier.ExpectProcess(renameSessionId, "metadata lock", "insert"), "cannot find process") + + // information_schema.processlist, the fallback, finds the same session + applier.usePerformanceSchemaProcesslist = false + suite.Require().NoError(applier.ExpectProcess(renameSessionId, "metadata lock", "rename")) + suite.Require().ErrorContains(applier.ExpectProcess(renameSessionId, "metadata lock", "insert"), "cannot find process") +} + func TestBuildMigrationLockName(t *testing.T) { t.Run("short name is returned verbatim", func(t *testing.T) { name := buildMigrationLockName("mydb", "mytable") @@ -1734,3 +1846,113 @@ func TestApplier(t *testing.T) { } suite.Run(t, new(ApplierTestSuite)) } + +// testMysql9ContainerImage is a MySQL release with strict character set conversions into temporary +// tables (9.1 and later) +const testMysql9ContainerImage = "mysql:9.7.2" + +// ApplierMySQL9TestSuite runs against MySQL 9.1 or later, where character set conversions into +// temporary tables are strict: information_schema.processlist cannot be read while any session runs +// a statement not representable in utf8mb3. +type ApplierMySQL9TestSuite struct { + suite.Suite + + mysqlContainer testcontainers.Container + db *gosql.DB +} + +func (suite *ApplierMySQL9TestSuite) SetupSuite() { + ctx := context.Background() + // my.cnf.test is not used: MySQL 9.7 rejects innodb_log_file_size + mysqlContainer, err := testmysql.Run(ctx, + testMysql9ContainerImage, + testmysql.WithDatabase(testMysqlDatabase), + testmysql.WithUsername(testMysqlUser), + testmysql.WithPassword(testMysqlPass), + ) + suite.Require().NoError(err) + + suite.mysqlContainer = mysqlContainer + + dsn, err := mysqlContainer.ConnectionString(ctx) + suite.Require().NoError(err) + + db, err := gosql.Open("mysql", dsn) + suite.Require().NoError(err) + + suite.db = db +} + +func (suite *ApplierMySQL9TestSuite) TearDownSuite() { + suite.Assert().NoError(suite.db.Close()) + suite.Assert().NoError(testcontainers.TerminateContainer(suite.mysqlContainer)) +} + +func (suite *ApplierMySQL9TestSuite) TearDownTest() { + ctx := context.Background() + + _, err := suite.db.ExecContext(ctx, "DROP TABLE IF EXISTS "+getTestTableName()) + suite.Require().NoError(err) + _, err = suite.db.ExecContext(ctx, "DROP TABLE IF EXISTS "+getTestOldTableName()) + suite.Require().NoError(err) +} + +func (suite *ApplierMySQL9TestSuite) TestExpectProcessWhileUtf8mb4StatementIsRunning() { + ctx := context.Background() + + _, err := suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, name VARCHAR(64))", getTestTableName())) + suite.Require().NoError(err) + + connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer) + suite.Require().NoError(err) + + migrationContext := newTestMigrationContext() + migrationContext.ApplierConnectionConfig = connectionConfig + migrationContext.SetConnectionConfig("innodb") + + applier := NewApplier(migrationContext) + defer applier.Teardown() + suite.Require().NoError(applier.InitDBConnections()) + suite.Require().True(applier.usePerformanceSchemaProcesslist) + + // A write carrying an emoji waits behind the table lock, as writes do behind the lock of the + // atomic cut-over, and so does the RENAME after it. + sessionIds, unlock := lockTestTableAndQueue(suite.T(), suite.db, + fmt.Sprintf("INSERT INTO %s (id, name) VALUES (1, '\U0001F600')", getTestTableName()), + fmt.Sprintf("RENAME TABLE %s TO %s", getTestTableName(), getTestOldTableName()), + ) + defer unlock() + renameSessionId := sessionIds[1] + + // Control: information_schema.processlist cannot be read on this server meanwhile. + var id int64 + err = suite.db.QueryRowContext(ctx, ` + select id + from + information_schema.processlist + where + id != connection_id() + and ? in (0, id) + and state like concat('%', ?, '%') + and info like concat('%', ?, '%')`, + renameSessionId, "metadata lock", "rename", + ).Scan(&id) + var mysqlErr *drivermysql.MySQLError + suite.Require().ErrorAs(err, &mysqlErr) + suite.Require().Equal(uint16(3854), mysqlErr.Number) // ER_CANNOT_CONVERT_STRING + + suite.Require().NoError(applier.ExpectProcess(renameSessionId, "metadata lock", "rename")) + + // The information_schema.processlist fallback reports the failed read, rather than a missing process. + applier.usePerformanceSchemaProcesslist = false + err = applier.ExpectProcess(renameSessionId, "metadata lock", "rename") + suite.Require().ErrorAs(err, &mysqlErr) + suite.Require().Equal(uint16(3854), mysqlErr.Number) +} + +func TestApplierMySQL9(t *testing.T) { + if testing.Short() { + t.Skip("skipping MySQL 9 applier test suite in short mode") + } + suite.Run(t, new(ApplierMySQL9TestSuite)) +}