applier.go

/*
   Copyright 2025 GitHub Inc.
	 See https://github.com/github/gh-ost/blob/master/LICENSE
*/

package logic

import (
	"crypto/sha1"
	gosql "database/sql"
	"encoding/hex"
	"fmt"
	"reflect"
	"regexp"
	"strings"
	"sync/atomic"
	"time"

	"github.com/github/gh-ost/go/base"
	"github.com/github/gh-ost/go/binlog"
	"github.com/github/gh-ost/go/sql"

	"context"
	"database/sql/driver"

	"errors"
	"sync"

	"github.com/github/gh-ost/go/mysql"
	drivermysql "github.com/go-sql-driver/mysql"
	"github.com/openark/golib/sqlutils"
)

const (
	GhostChangelogTableComment = "gh-ost changelog"
	atomicCutOverMagicHint     = "ghost-cut-over-sentry"
)

// ErrNoCheckpointFound is returned when an empty checkpoint table is queried.
var ErrNoCheckpointFound = errors.New("no checkpoint found in _ghk table")

type dmlBuildResult struct {
	query     string
	args      []interface{}
	rowsDelta int64
	err       error
}

func newDmlBuildResult(query string, args []interface{}, rowsDelta int64, err error) *dmlBuildResult {
	return &dmlBuildResult{
		query:     query,
		args:      args,
		rowsDelta: rowsDelta,
		err:       err,
	}
}

func newDmlBuildResultError(err error) *dmlBuildResult {
	return &dmlBuildResult{
		err: err,
	}
}

// Applier connects and writes the applier-server, which is the server where migration
// happens. This is typically the master, but could be a replica when `--test-on-replica` or
// `--execute-on-replica` are given.
// Applier is the one to actually write row data and apply binlog events onto the ghost table.
// It is where the ghost & changelog tables get created. It is where the cut-over phase happens.
type Applier struct {
	connectionConfig  *mysql.ConnectionConfig
	db                *gosql.DB
	singletonDB       *gosql.DB
	migrationContext  *base.MigrationContext
	finishedMigrating int64
	name              string

	CurrentCoordinatesMutex sync.Mutex
	CurrentCoordinates      mysql.BinlogCoordinates

	LastIterationRangeMutex     sync.Mutex
	LastIterationRangeMinValues *sql.ColumnValues
	LastIterationRangeMaxValues *sql.ColumnValues

	dmlDeleteQueryBuilder        *sql.DMLDeleteQueryBuilder
	dmlInsertQueryBuilder        *sql.DMLInsertQueryBuilder
	dmlUpdateQueryBuilder        *sql.DMLUpdateQueryBuilder
	checkpointInsertQueryBuilder *sql.CheckpointInsertQueryBuilder

	migrationLockConn *gosql.Conn
	migrationLockDB   *gosql.DB
	migrationLockName string
	migrationLockStop chan struct{}
	migrationLockDone chan struct{}

	moveTablesTargetDB                    *gosql.DB
	moveTablesCopySelectFirstQueryBuilder *sql.MoveTableCopySelectQueryBuilder
	moveTablesCopySelectNextQueryBuilder  *sql.MoveTableCopySelectQueryBuilder
	moveTablesCopyInsertQueryBuilder      *sql.MoveTableCopyInsertQueryBuilder
}

func NewApplier(migrationContext *base.MigrationContext) *Applier {
	return &Applier{
		connectionConfig:  migrationContext.ApplierConnectionConfig,
		migrationContext:  migrationContext,
		finishedMigrating: 0,
		name:              "applier",
	}
}

// compileMigrationKeyWarningRegex compiles a regex pattern that matches duplicate key warnings
// for the migration's unique key. Duplicate warnings are formatted differently across MySQL versions,
// hence the optional table name prefix. Metacharacters in table/index names are escaped to avoid
// regex syntax errors.
func (apl *Applier) compileMigrationKeyWarningRegex() (*regexp.Regexp, error) {
	escapedTable := regexp.QuoteMeta(apl.migrationContext.GetGhostTableName())
	escapedKey := regexp.QuoteMeta(apl.migrationContext.UniqueKey.NameInGhostTable)
	migrationUniqueKeyPattern := fmt.Sprintf(`for key '(%s\.)?%s'`, escapedTable, escapedKey)
	migrationKeyRegex, err := regexp.Compile(migrationUniqueKeyPattern)
	if err != nil {
		return nil, fmt.Errorf("failed to compile migration key pattern: %w", err)
	}
	return migrationKeyRegex, nil
}

func (apl *Applier) InitDBConnections() (err error) {
	applierUri := apl.connectionConfig.GetDBUri(apl.migrationContext.DatabaseName)
	uriWithMulti := fmt.Sprintf("%s&multiStatements=true", applierUri)
	if apl.db, _, err = mysql.GetDB(apl.migrationContext.Uuid, uriWithMulti); err != nil {
		return err
	}
	singletonApplierUri := fmt.Sprintf("%s&timeout=0", applierUri)
	if apl.singletonDB, _, err = mysql.GetDB(apl.migrationContext.Uuid, singletonApplierUri); err != nil {
		return err
	}
	apl.singletonDB.SetMaxOpenConns(1)
	version, err := base.ValidateConnection(apl.db, apl.connectionConfig, apl.migrationContext, apl.name)
	if err != nil {
		return err
	}
	if _, err := base.ValidateConnection(apl.singletonDB, apl.connectionConfig, apl.migrationContext, apl.name); err != nil {
		return err
	}
	apl.migrationContext.ApplierMySQLVersion = version
	if err := apl.validateAndReadGlobalVariables(); err != nil {
		return err
	}
	if !apl.migrationContext.AliyunRDS && !apl.migrationContext.GoogleCloudPlatform && !apl.migrationContext.AzureMySQL {
		if impliedKey, err := mysql.GetInstanceKey(apl.db); err != nil {
			return err
		} else {
			apl.connectionConfig.ImpliedKey = impliedKey
		}
	}
	if err := apl.readTableColumns(); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Applier initiated on %+v, version %+v", apl.connectionConfig.ImpliedKey, apl.migrationContext.ApplierMySQLVersion)
	return nil
}

// buildMigrationLockName returns a deterministic MySQL user-level lock name
// for the given database and table, hashed if longer than MySQL's 64-char limit.
func buildMigrationLockName(db, table string) string {
	name := fmt.Sprintf("gh-ost::%s.%s", db, table)
	if len(name) <= 64 {
		return name
	}
	sum := sha1.Sum([]byte(name))
	return "gh-ost::" + hex.EncodeToString(sum[:])
}

// AcquireMigrationLock takes a user-level lock on a pinned connection,
// preventing two gh-ost processes from migrating the same table concurrently
// on the same MySQL server.
func (apl *Applier) AcquireMigrationLock(ctx context.Context) error {
	lockName := buildMigrationLockName(apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName)

	// Use a dedicated *sql.DB so the pinned connection does not consume a
	// slot in apl.db's small pool (mysql.MaxDBPoolConnections).
	lockURI := apl.connectionConfig.GetDBUri(apl.migrationContext.DatabaseName)
	lockDB, err := gosql.Open("mysql", lockURI)
	if err != nil {
		return fmt.Errorf("failed to open migration lock DB: %w", err)
	}
	lockDB.SetMaxOpenConns(1)
	lockDB.SetMaxIdleConns(1)

	conn, err := lockDB.Conn(ctx)
	if err != nil {
		lockDB.Close()
		return fmt.Errorf("failed to obtain pinned connection for migration lock: %w", err)
	}

	var lockResult gosql.NullInt64
	if err := conn.QueryRowContext(ctx, `select /* gh-ost */ get_lock(?, 0)`, lockName).Scan(&lockResult); err != nil {
		conn.Close()
		lockDB.Close()
		return fmt.Errorf("failed to execute GET_LOCK for migration lock %s: %w", lockName, err)
	}

	if !lockResult.Valid {
		conn.Close()
		lockDB.Close()
		return fmt.Errorf("GET_LOCK returned NULL while acquiring migration lock %s", lockName)
	}

	if lockResult.Int64 != 1 {
		var holderID gosql.NullInt64
		_ = conn.QueryRowContext(ctx, `select /* gh-ost */ is_used_lock(?)`, lockName).Scan(&holderID)
		conn.Close()
		lockDB.Close()
		if holderID.Valid {
			return fmt.Errorf("another gh-ost process is already migrating `%s`.`%s`: migration lock %s held by connection id %d",
				apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName, lockName, holderID.Int64)
		}
		return fmt.Errorf("another gh-ost process is already migrating `%s`.`%s`: migration lock %s is held",
			apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName, lockName)
	}

	apl.migrationLockConn = conn
	apl.migrationLockDB = lockDB
	apl.migrationLockName = lockName
	apl.migrationLockStop = make(chan struct{})
	apl.migrationLockDone = make(chan struct{})
	go apl.keepMigrationLockAlive(ctx)
	apl.migrationContext.Log.Infof("Acquired migration lock %s", lockName)
	return nil
}

// keepMigrationLockAlive pings the pinned migration-lock connection. If the
// ping fails the lock is considered lost and the migration is aborted via
// PanicAbort.
func (apl *Applier) keepMigrationLockAlive(ctx context.Context) {
	defer close(apl.migrationLockDone)
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-apl.migrationLockStop:
			return
		case <-ticker.C:
		}
		if err := apl.pingMigrationLockConn(ctx); err != nil {
			// Shutdown may have started mid-ping; don't abort if so.
			select {
			case <-apl.migrationLockStop:
				return
			default:
			}
			if ctx.Err() != nil {
				return
			}
			_ = base.SendWithContext(ctx, apl.migrationContext.PanicAbort,
				fmt.Errorf("migration lock %s connection lost: %w", apl.migrationLockName, err))
			return
		}
	}
}

// pingMigrationLockConn pings the pinned connection with a bounded timeout
// and propagates migrationLockStop as an early cancel so a teardown can
// interrupt a stuck ping.
func (apl *Applier) pingMigrationLockConn(parent context.Context) error {
	pingCtx, cancel := context.WithTimeout(parent, 10*time.Second)
	defer cancel()
	done := make(chan struct{})
	defer close(done)
	go func() {
		select {
		case <-apl.migrationLockStop:
			cancel()
		case <-done:
		}
	}()
	return apl.migrationLockConn.PingContext(pingCtx)
}

// releaseMigrationLock stops the keepalive goroutine, releases the user-level
// lock and closes the dedicated lock DB. Safe to call when no lock is held.
func (apl *Applier) releaseMigrationLock() {
	if apl.migrationLockConn == nil {
		return
	}
	// Stop keepalive before touching the pinned connection.
	close(apl.migrationLockStop)
	<-apl.migrationLockDone
	releaseCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if _, err := apl.migrationLockConn.ExecContext(releaseCtx, `select /* gh-ost */ release_lock(?)`, apl.migrationLockName); err != nil {
		apl.migrationContext.Log.Warningf("failed to release migration lock %s: %v", apl.migrationLockName, err)
	}
	if err := apl.migrationLockConn.Close(); err != nil {
		apl.migrationContext.Log.Warningf("failed to close migration lock connection: %v", err)
	}
	if apl.migrationLockDB != nil {
		apl.migrationLockDB.Close()
		apl.migrationLockDB = nil
	}
	apl.migrationLockConn = nil
}

func (apl *Applier) prepareQueries() (err error) {
	if apl.dmlDeleteQueryBuilder, err = sql.NewDMLDeleteQueryBuilder(
		apl.migrationContext.DatabaseName,
		apl.migrationContext.GetGhostTableName(),
		apl.migrationContext.OriginalTableColumns,
		&apl.migrationContext.UniqueKey.Columns,
	); err != nil {
		return err
	}
	if apl.dmlInsertQueryBuilder, err = sql.NewDMLInsertQueryBuilder(
		apl.migrationContext.DatabaseName,
		apl.migrationContext.GetGhostTableName(),
		apl.migrationContext.OriginalTableColumns,
		apl.migrationContext.SharedColumns,
		apl.migrationContext.MappedSharedColumns,
	); err != nil {
		return err
	}
	if apl.dmlUpdateQueryBuilder, err = sql.NewDMLUpdateQueryBuilder(
		apl.migrationContext.DatabaseName,
		apl.migrationContext.GetGhostTableName(),
		apl.migrationContext.OriginalTableColumns,
		apl.migrationContext.SharedColumns,
		apl.migrationContext.MappedSharedColumns,
		&apl.migrationContext.UniqueKey.Columns,
	); err != nil {
		return err
	}
	if apl.migrationContext.Checkpoint {
		if apl.checkpointInsertQueryBuilder, err = sql.NewCheckpointQueryBuilder(
			apl.migrationContext.DatabaseName,
			apl.migrationContext.GetCheckpointTableName(),
			&apl.migrationContext.UniqueKey.Columns,
		); err != nil {
			return err
		}
	}
	if apl.migrationContext.IsMoveTablesMode() {
		if apl.moveTablesCopySelectFirstQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
			apl.migrationContext.DatabaseName,
			apl.migrationContext.OriginalTableName,
			apl.migrationContext.SharedColumns,
			apl.migrationContext.UniqueKey.Name,
			&apl.migrationContext.UniqueKey.Columns,
			true, // <-- include start range values for first select query
		); err != nil {
			return err
		}
		if apl.moveTablesCopySelectNextQueryBuilder, err = sql.NewMoveTableCopySelectQueryBuilder(
			apl.migrationContext.DatabaseName,
			apl.migrationContext.OriginalTableName,
			apl.migrationContext.SharedColumns,
			apl.migrationContext.UniqueKey.Name,
			&apl.migrationContext.UniqueKey.Columns,
			false,
		); err != nil {
			return err
		}
		if apl.moveTablesCopyInsertQueryBuilder, err = sql.NewMoveTableCopyInsertQueryBuilder(
			apl.migrationContext.MoveTables.TargetDatabase,
			apl.migrationContext.OriginalTableName,
			apl.migrationContext.SharedColumns,
		); err != nil {
			return err
		}
	}
	return nil
}

// validateAndReadGlobalVariables potentially reads server global variables, such as the time_zone and wait_timeout.
func (apl *Applier) validateAndReadGlobalVariables() error {
	query := `select /* gh-ost */ @@global.time_zone, @@global.wait_timeout`
	if err := apl.db.QueryRow(query).Scan(
		&apl.migrationContext.ApplierTimeZone,
		&apl.migrationContext.ApplierWaitTimeout,
	); err != nil {
		return err
	}

	apl.migrationContext.Log.Infof("will use time_zone='%s' on applier", apl.migrationContext.ApplierTimeZone)
	return nil
}

// 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
// - User may allow zero dats or zero in dates
func (apl *Applier) generateSqlModeQuery() string {
	sqlModeAddendum := []string{`NO_AUTO_VALUE_ON_ZERO`}
	if !apl.migrationContext.SkipStrictMode {
		sqlModeAddendum = append(sqlModeAddendum, `STRICT_ALL_TABLES`)
	}
	sqlModeQuery := fmt.Sprintf("CONCAT(@@session.sql_mode, ',%s')", strings.Join(sqlModeAddendum, ","))
	if apl.migrationContext.AllowZeroInDate {
		sqlModeQuery = fmt.Sprintf("REPLACE(REPLACE(%s, 'NO_ZERO_IN_DATE', ''), 'NO_ZERO_DATE', '')", sqlModeQuery)
	}

	return fmt.Sprintf("sql_mode = %s", sqlModeQuery)
}

// generateInstantDDLQuery returns the SQL for this ALTER operation
// with an INSTANT assertion (requires MySQL 8.0+)
func (apl *Applier) generateInstantDDLQuery() string {
	return fmt.Sprintf(`ALTER /* gh-ost */ TABLE %s.%s %s, ALGORITHM=INSTANT`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		apl.migrationContext.AlterStatementOptions,
	)
}

// readTableColumns reads table columns on applier
func (apl *Applier) readTableColumns() (err error) {
	apl.migrationContext.Log.Infof("Examining table structure on applier")
	apl.migrationContext.OriginalTableColumnsOnApplier, _, err = mysql.GetTableColumns(apl.db, apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName)
	if err != nil {
		return err
	}
	return nil
}

// showTableStatus returns the output of `show table status like '...'` command
func (apl *Applier) showTableStatus(tableName string) (rowMap sqlutils.RowMap) {
	query := fmt.Sprintf(`show /* gh-ost */ table status from %s like '%s'`, sql.EscapeName(apl.migrationContext.DatabaseName), tableName)
	sqlutils.QueryRowsMap(apl.db, query, func(m sqlutils.RowMap) error {
		rowMap = m
		return nil
	})
	return rowMap
}

// tableExists checks if a given table exists in database
func (apl *Applier) tableExists(tableName string) (tableFound bool) {
	m := apl.showTableStatus(tableName)
	return (m != nil)
}

// ValidateOrDropExistingTables verifies ghost and changelog tables do not exist,
// or attempts to drop them if instructed to.
func (apl *Applier) ValidateOrDropExistingTables() error {
	if apl.migrationContext.InitiallyDropGhostTable {
		if err := apl.DropGhostTable(); err != nil {
			return err
		}
	}
	if apl.tableExists(apl.migrationContext.GetGhostTableName()) {
		return fmt.Errorf("table %s already exists. Panicking. Use --initially-drop-ghost-table to force dropping it, though I really prefer that you drop it or rename it away", sql.EscapeName(apl.migrationContext.GetGhostTableName()))
	}
	if apl.migrationContext.InitiallyDropOldTable {
		if err := apl.DropOldTable(); err != nil {
			return err
		}
	}
	if len(apl.migrationContext.GetOldTableName()) > mysql.MaxTableNameLength {
		apl.migrationContext.Log.Fatalf("--timestamp-old-table defined, but resulting table name (%s) is too long (only %d characters allowed)", apl.migrationContext.GetOldTableName(), mysql.MaxTableNameLength)
	}

	if apl.tableExists(apl.migrationContext.GetOldTableName()) {
		return fmt.Errorf("table %s already exists. Panicking. Use --initially-drop-old-table to force dropping it, though I really prefer that you drop it or rename it away", sql.EscapeName(apl.migrationContext.GetOldTableName()))
	}

	return nil
}

// AttemptInstantDDL attempts to use instant DDL (from MySQL 8.0, and earlier in Aurora and some others).
// If successful, the operation is only a meta-data change so a lot of time is saved!
// The risk of attempting to instant DDL when not supported is that a metadata lock may be acquired.
// This is minor, since gh-ost will eventually require a metadata lock anyway, but at the cut-over stage.
// Instant operations include:
// - Adding a column
// - Dropping a column
// - Dropping an index
// - Extending a VARCHAR column
// - Adding a virtual generated column
// It is not reliable to parse the `alter` statement to determine if it is instant or not.
// This is because the table might be in an older row format, or have some other incompatibility
// that is difficult to identify.
func (apl *Applier) AttemptInstantDDL() error {
	query := apl.generateInstantDDLQuery()
	apl.migrationContext.Log.Infof("INSTANT DDL query is: %s", query)

	// Reuse cut-over-lock-timeout from regular migration process to reduce risk
	// in situations where there may be long-running transactions.
	tableLockTimeoutSeconds := apl.migrationContext.CutOverLockTimeoutSeconds * 2
	apl.migrationContext.Log.Infof("Setting LOCK timeout as %d seconds", tableLockTimeoutSeconds)
	lockTimeoutQuery := fmt.Sprintf(`set /* gh-ost */ session lock_wait_timeout:=%d`, tableLockTimeoutSeconds)
	if _, err := apl.db.Exec(lockTimeoutQuery); err != nil {
		return err
	}
	// We don't need a trx, because for instant DDL the SQL mode doesn't matter.
	return retryOnLockWaitTimeout(func() error {
		_, err := apl.db.Exec(query)
		return err
	}, apl.migrationContext.MaxRetries(), apl.migrationContext.Log)
}

// retryOnLockWaitTimeout retries the given operation on MySQL lock wait timeout
// (errno 1205). Non-timeout errors return immediately. This is used for instant
// DDL attempts where the operation may be blocked by a long-running transaction.
func retryOnLockWaitTimeout(operation func() error, maxRetries int64, logger base.Logger) error {
	var err error
	for i := int64(0); i < maxRetries; i++ {
		if i != 0 {
			logger.Infof("Retrying after lock wait timeout (attempt %d/%d)", i+1, maxRetries)
			RetrySleepFn(time.Duration(i) * 5 * time.Second)
		}
		err = operation()
		if err == nil {
			return nil
		}
		var mysqlErr *drivermysql.MySQLError
		if !errors.As(err, &mysqlErr) || mysqlErr.Number != 1205 {
			return err
		}
	}
	return err
}

// CreateGhostTable creates the ghost table on the applier host
func (apl *Applier) CreateGhostTable() error {
	query := fmt.Sprintf(`create /* gh-ost */ table %s.%s like %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Creating ghost table %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
	)

	err := func() error {
		tx, err := apl.db.Begin()
		if err != nil {
			return err
		}
		defer tx.Rollback()

		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s'`, apl.migrationContext.ApplierTimeZone)
		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, apl.generateSqlModeQuery())

		if _, err := tx.Exec(sessionQuery); err != nil {
			return err
		}
		if _, err := tx.Exec(query); err != nil {
			return err
		}
		apl.migrationContext.Log.Infof("Ghost table created")
		if err := tx.Commit(); err != nil {
			// Neither SET SESSION nor ALTER are really transactional, so strictly speaking
			// there's no need to commit; but let's do this the legit way anyway.
			return err
		}
		return nil
	}()

	return err
}

// AlterGhost applies `alter` statement on ghost table
func (apl *Applier) AlterGhost() error {
	query := fmt.Sprintf(`alter /* gh-ost */ table %s.%s %s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		apl.migrationContext.AlterStatementOptions,
	)
	apl.migrationContext.Log.Infof("Altering ghost table %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
	)
	apl.migrationContext.Log.Debugf("ALTER statement: %s", query)

	err := func() error {
		tx, err := apl.db.Begin()
		if err != nil {
			return err
		}
		defer tx.Rollback()

		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s'`, apl.migrationContext.ApplierTimeZone)
		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, apl.generateSqlModeQuery())

		if _, err := tx.Exec(sessionQuery); err != nil {
			return err
		}
		if _, err := tx.Exec(query); err != nil {
			return err
		}
		apl.migrationContext.Log.Infof("Ghost table altered")
		if err := tx.Commit(); err != nil {
			// Neither SET SESSION nor ALTER are really transactional, so strictly speaking
			// there's no need to commit; but let's do this the legit way anyway.
			return err
		}
		return nil
	}()

	return err
}

// AlterGhost applies `alter` statement on ghost table
func (apl *Applier) AlterGhostAutoIncrement() error {
	query := fmt.Sprintf(`alter /* gh-ost */ table %s.%s AUTO_INCREMENT=%d`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		apl.migrationContext.OriginalTableAutoIncrement,
	)
	apl.migrationContext.Log.Infof("Altering ghost table AUTO_INCREMENT value %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
	)
	apl.migrationContext.Log.Debugf("AUTO_INCREMENT ALTER statement: %s", query)
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Ghost table AUTO_INCREMENT altered")
	return nil
}

// CreateChangelogTable creates the changelog table on the applier host
func (apl *Applier) CreateChangelogTable() error {
	if err := apl.DropChangelogTable(); err != nil {
		return err
	}
	query := fmt.Sprintf(`create /* gh-ost */ table %s.%s (
			id bigint unsigned auto_increment,
			last_update timestamp not null DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
			hint varchar(64) charset ascii not null,
			value varchar(4096) charset ascii not null,
			primary key(id),
			unique key hint_uidx(hint)
		) auto_increment=256 comment='%s'`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetChangelogTableName()),
		GhostChangelogTableComment,
	)
	apl.migrationContext.Log.Infof("Creating changelog table %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetChangelogTableName()),
	)
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Changelog table created")
	return nil
}

// Create the checkpoint table to store the chunk copy and applier state.
// There are two sets of columns with the same types as the shared unique key,
// one for IterationMinValues and one for IterationMaxValues.
func (apl *Applier) CreateCheckpointTable() error {
	if err := apl.DropCheckpointTable(); err != nil {
		return err
	}
	colDefs := []string{
		"`gh_ost_chk_id` bigint auto_increment primary key",
		"`gh_ost_chk_timestamp` bigint",
		"`gh_ost_chk_coords` text charset ascii",
		"`gh_ost_chk_iteration` bigint",
		"`gh_ost_rows_copied` bigint",
		"`gh_ost_dml_applied` bigint",
		"`gh_ost_is_cutover` tinyint(1) DEFAULT '0'",
	}
	for _, col := range apl.migrationContext.UniqueKey.Columns.Columns() {
		if col.MySQLType == "" {
			return fmt.Errorf("column %s has no type information. applyColumnTypes must be called", sql.EscapeName(col.Name))
		}
		minColName := sql.TruncateColumnName(col.Name, sql.MaxColumnNameLength-4) + "_min"
		colDef := fmt.Sprintf("%s %s", sql.EscapeName(minColName), col.MySQLType)
		colDefs = append(colDefs, colDef)
	}

	for _, col := range apl.migrationContext.UniqueKey.Columns.Columns() {
		maxColName := sql.TruncateColumnName(col.Name, sql.MaxColumnNameLength-4) + "_max"
		colDef := fmt.Sprintf("%s %s", sql.EscapeName(maxColName), col.MySQLType)
		colDefs = append(colDefs, colDef)
	}

	query := fmt.Sprintf("create /* gh-ost */ table %s.%s (\n %s\n)",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetCheckpointTableName()),
		strings.Join(colDefs, ",\n "),
	)
	apl.migrationContext.Log.Infof("Created checkpoint table")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	return nil
}

// dropTable drops a given table on the applied host
func (apl *Applier) dropTable(tableName string) error {
	query := fmt.Sprintf(`drop /* gh-ost */ table if exists %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(tableName),
	)
	apl.migrationContext.Log.Infof("Dropping table %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(tableName),
	)
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Table dropped")
	return nil
}

// StateMetadataLockInstrument checks if metadata_locks is enabled in performance_schema.
// If not it attempts to enable metadata_locks if this is allowed.
func (apl *Applier) StateMetadataLockInstrument() error {
	query := `select /*+ MAX_EXECUTION_TIME(300) */ ENABLED, TIMED from performance_schema.setup_instruments WHERE NAME = 'wait/lock/metadata/sql/mdl'`
	var enabled, timed string
	if err := apl.db.QueryRow(query).Scan(&enabled, &timed); err != nil {
		if errors.Is(err, gosql.ErrNoRows) {
			// performance_schema may be disabled.
			return nil
		}
		return apl.migrationContext.Log.Errorf("query performance_schema.setup_instruments with name wait/lock/metadata/sql/mdl error: %s", err)
	}
	if strings.EqualFold(enabled, "YES") && strings.EqualFold(timed, "YES") {
		apl.migrationContext.IsOpenMetadataLockInstruments = true
		return nil
	}
	if !apl.migrationContext.AllowSetupMetadataLockInstruments {
		return nil
	}
	apl.migrationContext.Log.Infof("instrument wait/lock/metadata/sql/mdl state: enabled %s, timed %s", enabled, timed)
	if _, err := apl.db.Exec(`UPDATE performance_schema.setup_instruments SET ENABLED = 'YES', TIMED = 'YES' WHERE NAME = 'wait/lock/metadata/sql/mdl'`); err != nil {
		return apl.migrationContext.Log.Errorf("enable instrument wait/lock/metadata/sql/mdl error: %s", err)
	}
	apl.migrationContext.IsOpenMetadataLockInstruments = true
	apl.migrationContext.Log.Infof("instrument wait/lock/metadata/sql/mdl enabled")
	return nil
}

// dropTriggers drop the triggers on the applied host
func (apl *Applier) DropTriggersFromGhost() error {
	if len(apl.migrationContext.Triggers) > 0 {
		for _, trigger := range apl.migrationContext.Triggers {
			triggerName := apl.migrationContext.GetGhostTriggerName(trigger.Name)
			query := fmt.Sprintf("drop trigger if exists %s", sql.EscapeName(triggerName))
			_, err := sqlutils.ExecNoPrepare(apl.db, query)
			if err != nil {
				return err
			}
			apl.migrationContext.Log.Infof("Trigger '%s' dropped", triggerName)
		}
	}
	return nil
}

// createTriggers creates the triggers on the applied host
func (apl *Applier) createTriggers(tableName string) error {
	if len(apl.migrationContext.Triggers) > 0 {
		for _, trigger := range apl.migrationContext.Triggers {
			triggerName := apl.migrationContext.GetGhostTriggerName(trigger.Name)
			query := fmt.Sprintf(`create /* gh-ost */ trigger %s %s %s on %s.%s for each row
		%s`,
				sql.EscapeName(triggerName),
				trigger.Timing,
				trigger.Event,
				sql.EscapeName(apl.migrationContext.DatabaseName),
				sql.EscapeName(tableName),
				trigger.Statement,
			)
			apl.migrationContext.Log.Infof("Createing trigger %s on %s.%s",
				sql.EscapeName(triggerName),
				sql.EscapeName(apl.migrationContext.DatabaseName),
				sql.EscapeName(tableName),
			)
			if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
				return err
			}
		}
		apl.migrationContext.Log.Infof("Triggers created on %s", tableName)
	}
	return nil
}

// CreateTriggers creates the original triggers on applier host
func (apl *Applier) CreateTriggersOnGhost() error {
	err := apl.createTriggers(apl.migrationContext.GetGhostTableName())
	return err
}

// DropChangelogTable drops the changelog table on the applier host
func (apl *Applier) DropChangelogTable() error {
	return apl.dropTable(apl.migrationContext.GetChangelogTableName())
}

// DropCheckpointTable drops the checkpoint table on applier host
func (apl *Applier) DropCheckpointTable() error {
	return apl.dropTable(apl.migrationContext.GetCheckpointTableName())
}

// DropOldTable drops the _Old table on the applier host
func (apl *Applier) DropOldTable() error {
	return apl.dropTable(apl.migrationContext.GetOldTableName())
}

// DropGhostTable drops the ghost table on the applier host
func (apl *Applier) DropGhostTable() error {
	return apl.dropTable(apl.migrationContext.GetGhostTableName())
}

// WriteChangelog writes a value to the changelog table.
// It returns the hint as given, for convenience
func (apl *Applier) WriteChangelog(hint, value string) (string, error) {
	explicitId := 0
	switch hint {
	case "heartbeat":
		explicitId = 1
	case "state":
		explicitId = 2
	case "throttle":
		explicitId = 3
	}
	query := fmt.Sprintf(`
		insert /* gh-ost */
		into
			%s.%s
			(id, hint, value)
		values
			(NULLIF(?, 0), ?, ?)
		on duplicate key update
			last_update=NOW(),
			value=VALUES(value)`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetChangelogTableName()),
	)
	_, err := sqlutils.ExecNoPrepare(apl.db, query, explicitId, hint, value)
	return hint, err
}

func (apl *Applier) WriteAndLogChangelog(hint, value string) (string, error) {
	apl.WriteChangelog(hint, value)
	return apl.WriteChangelog(fmt.Sprintf("%s at %d", hint, time.Now().UnixNano()), value)
}

func (apl *Applier) WriteChangelogState(value string) (string, error) {
	return apl.WriteAndLogChangelog("state", value)
}

// WriteCheckpoints writes a checkpoint to the _ghk table.
func (apl *Applier) WriteCheckpoint(chk *Checkpoint) (int64, error) {
	var insertId int64
	uniqueKeyArgs := sqlutils.Args(chk.IterationRangeMin.AbstractValues()...)
	uniqueKeyArgs = append(uniqueKeyArgs, chk.IterationRangeMax.AbstractValues()...)
	query, uniqueKeyArgs, err := apl.checkpointInsertQueryBuilder.BuildQuery(uniqueKeyArgs)
	if err != nil {
		return insertId, err
	}
	args := sqlutils.Args(chk.LastTrxCoords.String(), chk.Iteration, chk.RowsCopied, chk.DMLApplied, chk.IsCutover)
	args = append(args, uniqueKeyArgs...)
	res, err := apl.db.Exec(query, args...)
	if err != nil {
		return insertId, err
	}
	return res.LastInsertId()
}

func (apl *Applier) ReadLastCheckpoint() (*Checkpoint, error) {
	row := apl.db.QueryRow(fmt.Sprintf(`select /* gh-ost */ * from %s.%s order by gh_ost_chk_id desc limit 1`, sql.EscapeName(apl.migrationContext.DatabaseName), sql.EscapeName(apl.migrationContext.GetCheckpointTableName())))
	chk := &Checkpoint{
		IterationRangeMin: sql.NewColumnValues(apl.migrationContext.UniqueKey.Columns.Len()),
		IterationRangeMax: sql.NewColumnValues(apl.migrationContext.UniqueKey.Columns.Len()),
	}

	var coordStr string
	var timestamp int64
	ptrs := []interface{}{&chk.Id, &timestamp, &coordStr, &chk.Iteration, &chk.RowsCopied, &chk.DMLApplied, &chk.IsCutover}
	ptrs = append(ptrs, chk.IterationRangeMin.ValuesPointers...)
	ptrs = append(ptrs, chk.IterationRangeMax.ValuesPointers...)
	err := row.Scan(ptrs...)
	if err != nil {
		if errors.Is(err, gosql.ErrNoRows) {
			return nil, ErrNoCheckpointFound
		}
		return nil, err
	}
	chk.Timestamp = time.Unix(timestamp, 0)
	if apl.migrationContext.UseGTIDs {
		gtidCoords, err := mysql.NewGTIDBinlogCoordinates(coordStr)
		if err != nil {
			return nil, err
		}
		chk.LastTrxCoords = gtidCoords
	} else {
		fileCoords, err := mysql.ParseFileBinlogCoordinates(coordStr)
		if err != nil {
			return nil, err
		}
		chk.LastTrxCoords = fileCoords
	}
	return chk, nil
}

// InitiateHeartbeat creates a heartbeat cycle, writing to the changelog table.
// Apl is done asynchronously
func (apl *Applier) InitiateHeartbeat() {
	var numSuccessiveFailures int64
	injectHeartbeat := func() error {
		if atomic.LoadInt64(&apl.migrationContext.HibernateUntil) > 0 {
			return nil
		}
		if _, err := apl.WriteChangelog("heartbeat", time.Now().Format(time.RFC3339Nano)); err != nil {
			numSuccessiveFailures++
			if numSuccessiveFailures > apl.migrationContext.MaxRetries() {
				return apl.migrationContext.Log.Errore(err)
			}
		} else {
			numSuccessiveFailures = 0
		}
		return nil
	}
	injectHeartbeat()

	ticker := time.NewTicker(time.Duration(apl.migrationContext.HeartbeatIntervalMilliseconds) * time.Millisecond)
	defer ticker.Stop()
	for {
		// Check for context cancellation each iteration
		ctx := apl.migrationContext.GetContext()
		select {
		case <-ctx.Done():
			apl.migrationContext.Log.Debugf("Heartbeat injection cancelled")
			return
		case <-ticker.C:
			// Process heartbeat
		}

		if atomic.LoadInt64(&apl.finishedMigrating) > 0 {
			return
		}

		if atomic.LoadInt64(&apl.migrationContext.CleanupImminentFlag) > 0 {
			return
		}

		// Generally speaking, we would issue a goroutine, but I'd actually rather
		// have this block the loop rather than spam the master in the event something
		// goes wrong
		if throttle, _, reasonHint := apl.migrationContext.IsThrottled(); throttle && (reasonHint == base.UserCommandThrottleReasonHint) {
			continue
		}
		if err := injectHeartbeat(); err != nil {
			// Use helper to prevent deadlock if listenOnPanicAbort already exited
			_ = base.SendWithContext(apl.migrationContext.GetContext(), apl.migrationContext.PanicAbort, fmt.Errorf("injectHeartbeat writing failed %d times, last error: %w", numSuccessiveFailures, err))
			return
		}
	}
}

// ExecuteThrottleQuery executes the `--throttle-query` and returns its results.
func (apl *Applier) ExecuteThrottleQuery() (int64, error) {
	throttleQuery := apl.migrationContext.GetThrottleQuery()

	if throttleQuery == "" {
		return 0, nil
	}
	var result int64
	if err := apl.db.QueryRow(throttleQuery).Scan(&result); err != nil {
		return 0, apl.migrationContext.Log.Errore(err)
	}
	return result, nil
}

// readMigrationMinValues returns the minimum values to be iterated on rowcopy
func (apl *Applier) readMigrationMinValues(tx *gosql.Tx, uniqueKey *sql.UniqueKey) error {
	apl.migrationContext.Log.Debugf("Reading migration range according to key: %s", uniqueKey.Name)
	query, err := sql.BuildUniqueKeyMinValuesPreparedQuery(apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName, uniqueKey)
	if err != nil {
		return err
	}

	rows, err := tx.Query(query)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		apl.migrationContext.MigrationRangeMinValues = sql.NewColumnValues(uniqueKey.Len())
		if err = rows.Scan(apl.migrationContext.MigrationRangeMinValues.ValuesPointers...); err != nil {
			return err
		}
	}
	apl.migrationContext.Log.Infof("Migration min values: [%s]", apl.migrationContext.MigrationRangeMinValues)

	return rows.Err()
}

// readMigrationMaxValues returns the maximum values to be iterated on rowcopy
func (apl *Applier) readMigrationMaxValues(tx *gosql.Tx, uniqueKey *sql.UniqueKey) error {
	apl.migrationContext.Log.Debugf("Reading migration range according to key: %s", uniqueKey.Name)
	query, err := sql.BuildUniqueKeyMaxValuesPreparedQuery(apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName, uniqueKey)
	if err != nil {
		return err
	}

	rows, err := tx.Query(query)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		apl.migrationContext.MigrationRangeMaxValues = sql.NewColumnValues(uniqueKey.Len())
		if err = rows.Scan(apl.migrationContext.MigrationRangeMaxValues.ValuesPointers...); err != nil {
			return err
		}
	}
	apl.migrationContext.Log.Infof("Migration max values: [%s]", apl.migrationContext.MigrationRangeMaxValues)

	return rows.Err()
}

// ReadMigrationRangeValues reads min/max values that will be used for rowcopy.
// Before read min/max, write a changelog state into the ghc table to avoid lost data in mysql two-phase commit.
/*
Detail description of the lost data in mysql two-phase commit issue by @Fanduzi:
	When using semi-sync and setting rpl_semi_sync_master_wait_point=AFTER_SYNC,
	if an INSERT statement is being committed but blocks due to an unmet ack count,
	the data inserted by the transaction is not visible to ReadMigrationRangeValues,
	so the copy of the existing data in the table does not include the new row inserted by the transaction.
	However, the binlog event for the transaction is already written to the binlog,
	so the addDMLEventsListener only captures the binlog event after the transaction,
	and thus the transaction's binlog event is not captured, resulting in data loss.

	If write a changelog into ghc table before ReadMigrationRangeValues, and the transaction commit blocks
	because the ack is not met, then the changelog will not be able to write, so the ReadMigrationRangeValues
	will not be run. When the changelog writes successfully, the ReadMigrationRangeValues will read the
	newly inserted data, thus Avoiding data loss due to the above problem.
*/
func (apl *Applier) ReadMigrationRangeValues() error {
	if _, err := apl.WriteChangelogState(string(ReadMigrationRangeValues)); err != nil {
		return err
	}

	tx, err := apl.db.Begin()
	if err != nil {
		return err
	}
	defer tx.Rollback()

	if err := apl.readMigrationMinValues(tx, apl.migrationContext.UniqueKey); err != nil {
		return err
	}
	if err := apl.readMigrationMaxValues(tx, apl.migrationContext.UniqueKey); err != nil {
		return err
	}

	return tx.Commit()
}

// CalculateNextIterationRangeEndValues reads the next-iteration-range-end unique key values,
// which will be used for copying the next chunk of rows. Ir returns "false" if there is
// no further chunk to work through, i.e. we're past the last chunk and are done with
// iterating the range (and thus done with copying row chunks)
func (apl *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, err error) {
	for i := 0; i < 2; i++ {
		buildFunc := sql.BuildUniqueKeyRangeEndPreparedQueryViaOffset
		if i == 1 {
			buildFunc = sql.BuildUniqueKeyRangeEndPreparedQueryViaTemptable
		}
		query, explodedArgs, err := buildFunc(
			apl.migrationContext.DatabaseName,
			apl.migrationContext.OriginalTableName,
			&apl.migrationContext.UniqueKey.Columns,
			apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
			apl.migrationContext.MigrationRangeMaxValues.AbstractValues(),
			atomic.LoadInt64(&apl.migrationContext.ChunkSize),
			apl.migrationContext.GetIteration() == 0,
			fmt.Sprintf("iteration:%d", apl.migrationContext.GetIteration()),
		)
		if err != nil {
			return hasFurtherRange, err
		}

		rows, err := apl.db.Query(query, explodedArgs...)
		if err != nil {
			return hasFurtherRange, err
		}
		defer rows.Close()

		iterationRangeMaxValues := sql.NewColumnValues(apl.migrationContext.UniqueKey.Len())
		for rows.Next() {
			if err = rows.Scan(iterationRangeMaxValues.ValuesPointers...); err != nil {
				return hasFurtherRange, err
			}
			hasFurtherRange = true
		}
		if err = rows.Err(); err != nil {
			return hasFurtherRange, err
		}
		if hasFurtherRange {
			apl.migrationContext.MigrationIterationRangeMaxValues = iterationRangeMaxValues
			return hasFurtherRange, nil
		}
	}
	apl.migrationContext.Log.Debugf("Iteration complete: no further range to iterate")
	return hasFurtherRange, nil
}

// ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where
// data actually gets copied from original table.
func (apl *Applier) ApplyIterationInsertQuery() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
	startTime := time.Now()
	chunkSize = atomic.LoadInt64(&apl.migrationContext.ChunkSize)

	query, explodedArgs, err := sql.BuildRangeInsertPreparedQuery(
		apl.migrationContext.DatabaseName,
		apl.migrationContext.OriginalTableName,
		apl.migrationContext.GetGhostTableName(),
		apl.migrationContext.SharedColumns.Names(),
		apl.migrationContext.MappedSharedColumns.Names(),
		apl.migrationContext.UniqueKey.Name,
		&apl.migrationContext.UniqueKey.Columns,
		apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
		apl.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
		apl.migrationContext.GetIteration() == 0,
		apl.migrationContext.IsTransactionalTable(),
		// TODO: Don't hardcode this
		strings.HasPrefix(apl.migrationContext.ApplierMySQLVersion, "8."),
	)
	if err != nil {
		return chunkSize, rowsAffected, duration, err
	}

	sqlResult, err := func() (gosql.Result, error) {
		tx, err := apl.db.Begin()
		if err != nil {
			return nil, err
		}
		defer tx.Rollback()

		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s'`, apl.migrationContext.ApplierTimeZone)
		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, apl.generateSqlModeQuery())

		if _, err := tx.Exec(sessionQuery); err != nil {
			return nil, err
		}
		result, err := tx.Exec(query, explodedArgs...)
		if err != nil {
			return nil, err
		}

		if apl.migrationContext.PanicOnWarnings {
			rows, err := tx.Query("SHOW WARNINGS")
			if err != nil {
				return nil, err
			}
			defer rows.Close()
			if err = rows.Err(); err != nil {
				return nil, err
			}

			// Compile regex once before loop to avoid performance penalty and handle errors properly
			migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
			if err != nil {
				return nil, err
			}

			var sqlWarnings []string
			for rows.Next() {
				var level, message string
				var code int
				if err := rows.Scan(&level, &code, &message); err != nil {
					apl.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
					continue
				}
				if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
					continue
				}
				sqlWarnings = append(sqlWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
			}
			apl.migrationContext.MigrationLastInsertSQLWarnings = sqlWarnings
		}

		if err := tx.Commit(); err != nil {
			return nil, err
		}
		return result, nil
	}()

	if err != nil {
		return chunkSize, rowsAffected, duration, err
	}
	rowsAffected, _ = sqlResult.RowsAffected()
	duration = time.Since(startTime)
	apl.migrationContext.Log.Debugf(
		"Issued INSERT on range: [%s]..[%s]; iteration: %d; chunk-size: %d",
		apl.migrationContext.MigrationIterationRangeMinValues,
		apl.migrationContext.MigrationIterationRangeMaxValues,
		apl.migrationContext.GetIteration(),
		chunkSize)
	return chunkSize, rowsAffected, duration, nil
}

// ApplyIterationMoveTableCopyQueries issues a SELECT query on the original table and an INSERT query on the target table,
// copying a chunk of rows. It is used when `--move-table` is specified, instead of ApplyIterationInsertQuery.
func (apl *Applier) ApplyIterationMoveTableCopyQueries() (chunkSize int64, rowsAffected int64, duration time.Duration, err error) {
	startTime := time.Now()
	chunkSize = atomic.LoadInt64(&apl.migrationContext.ChunkSize)

	// First, select data from the source database:
	rows, err := func() ([]*sql.ColumnValues, error) {
		var qb *sql.MoveTableCopySelectQueryBuilder
		if apl.migrationContext.GetIteration() == 0 {
			qb = apl.moveTablesCopySelectFirstQueryBuilder
		} else {
			qb = apl.moveTablesCopySelectNextQueryBuilder
		}
		query, explodedArgs, err := qb.BuildQuery(
			apl.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
			apl.migrationContext.MigrationIterationRangeMaxValues.AbstractValues(),
		)
		if err != nil {
			return nil, err
		}
		sqlRows, err := apl.db.Query(query, explodedArgs...)
		if err != nil {
			return nil, err
		}
		defer sqlRows.Close()
		chunkRows := make([]*sql.ColumnValues, 0, chunkSize)
		for sqlRows.Next() {
			row := sql.NewColumnValues(apl.migrationContext.SharedColumns.Len())
			err := sqlRows.Scan(row.ValuesPointers...)
			if err != nil {
				return nil, err
			}
			chunkRows = append(chunkRows, row)
		}
		return chunkRows, nil
	}()
	if err != nil {
		return chunkSize, rowsAffected, duration, err
	}

	// Then, insert data into the destination database:
	sqlResult, err := func() (gosql.Result, error) {
		query, explodedArgs, err := apl.moveTablesCopyInsertQueryBuilder.BuildQuery(rows)
		if err != nil {
			return nil, err
		}
		tx, err := apl.moveTablesTargetDB.Begin()
		if err != nil {
			return nil, err
		}
		defer tx.Rollback()

		sessionQuery := fmt.Sprintf(`SET SESSION time_zone = '%s', %s`,
			apl.migrationContext.ApplierTimeZone,
			apl.generateSqlModeQuery())
		if _, err := tx.Exec(sessionQuery); err != nil {
			return nil, err
		}

		sqlResult, err := tx.Exec(query, explodedArgs...)
		if err != nil {
			return nil, err
		}

		if apl.migrationContext.PanicOnWarnings {
			rows, err := tx.Query("SHOW WARNINGS")
			if err != nil {
				return nil, err
			}
			defer rows.Close()
			if err = rows.Err(); err != nil {
				return nil, err
			}
			migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
			if err != nil {
				return nil, err
			}
			var sqlWarnings []string
			for rows.Next() {
				var level, message string
				var code int
				if err := rows.Scan(&level, &code, &message); err != nil {
					apl.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
					continue
				}
				if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
					continue
				}
				sqlWarnings = append(sqlWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
			}
			apl.migrationContext.MigrationLastInsertSQLWarnings = sqlWarnings
		}

		if err := tx.Commit(); err != nil {
			return nil, err
		}
		return sqlResult, nil
	}()
	if err != nil {
		return chunkSize, rowsAffected, duration, err
	}
	rowsAffected, _ = sqlResult.RowsAffected()
	duration = time.Since(startTime)
	apl.migrationContext.Log.Debugf(
		"Issued SELECT+INSERT on range: [%s]..[%s]; iteration: %d; chunk-size: %d",
		apl.migrationContext.MigrationIterationRangeMinValues,
		apl.migrationContext.MigrationIterationRangeMaxValues,
		apl.migrationContext.GetIteration(),
		chunkSize,
	)

	return chunkSize, rowsAffected, duration, nil
}

// LockOriginalTable places a write lock on the original table
func (apl *Applier) LockOriginalTable() error {
	query := fmt.Sprintf(`lock /* gh-ost */ tables %s.%s write`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Locking %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.LockTablesStartTime = time.Now()
	if _, err := sqlutils.ExecNoPrepare(apl.singletonDB, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Table locked")
	return nil
}

// UnlockTables makes tea. No wait, it unlocks tables.
func (apl *Applier) UnlockTables() error {
	query := `unlock /* gh-ost */ tables`
	apl.migrationContext.Log.Infof("Unlocking tables")
	if _, err := sqlutils.ExecNoPrepare(apl.singletonDB, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Tables unlocked")
	return nil
}

// SwapTablesQuickAndBumpy issues a two-step swap table operation:
// - rename original table to _old
// - rename ghost table to original
// There is a point in time in between where the table does not exist.
func (apl *Applier) SwapTablesQuickAndBumpy() error {
	query := fmt.Sprintf(`alter /* gh-ost */ table %s.%s rename %s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
	)
	apl.migrationContext.Log.Infof("Renaming original table")
	apl.migrationContext.RenameTablesStartTime = time.Now()
	if _, err := sqlutils.ExecNoPrepare(apl.singletonDB, query); err != nil {
		return err
	}
	query = fmt.Sprintf(`alter /* gh-ost */ table %s.%s rename %s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Renaming ghost table")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.RenameTablesEndTime = time.Now()

	apl.migrationContext.Log.Infof("Tables renamed")
	return nil
}

// RenameTablesRollback renames back both table: original back to ghost,
// _old back to original. This is used by `--test-on-replica`
func (apl *Applier) RenameTablesRollback() (renameError error) {
	// Restoring tables to original names.
	// We prefer the single, atomic operation:
	query := fmt.Sprintf(`rename /* gh-ost */ table %s.%s to %s.%s, %s.%s to %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Renaming back both tables")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err == nil {
		return nil
	}
	// But, if for some reason the above was impossible to do, we rename one by one.
	query = fmt.Sprintf(`rename /* gh-ost */ table %s.%s to %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
	)
	apl.migrationContext.Log.Infof("Renaming back to ghost table")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		renameError = err
	}
	query = fmt.Sprintf(`rename /* gh-ost */ table %s.%s to %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Renaming back to original table")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		renameError = err
	}
	return apl.migrationContext.Log.Errore(renameError)
}

// StopSlaveIOThread is applicable with --test-on-replica; it stops the IO thread, duh.
// We need to keep the SQL thread active so as to complete processing received events,
// and have them written to the binary log, so that we can then read them via streamer.
func (apl *Applier) StopSlaveIOThread() error {
	replicaTerm := mysql.ReplicaTermFor(apl.migrationContext.ApplierMySQLVersion, `slave`)
	query := fmt.Sprintf("stop /* gh-ost */ %s io_thread", replicaTerm)
	apl.migrationContext.Log.Infof("Stopping replication IO thread")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Replication IO thread stopped")
	return nil
}

// StartSlaveIOThread is applicable with --test-on-replica
func (apl *Applier) StartSlaveIOThread() error {
	replicaTerm := mysql.ReplicaTermFor(apl.migrationContext.ApplierMySQLVersion, `slave`)
	query := fmt.Sprintf("start /* gh-ost */ %s io_thread", replicaTerm)
	apl.migrationContext.Log.Infof("Starting replication IO thread")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Replication IO thread started")
	return nil
}

// StopSlaveSQLThread is applicable with --test-on-replica
func (apl *Applier) StopSlaveSQLThread() error {
	replicaTerm := mysql.ReplicaTermFor(apl.migrationContext.ApplierMySQLVersion, `slave`)
	query := fmt.Sprintf("stop /* gh-ost */ %s sql_thread", replicaTerm)
	apl.migrationContext.Log.Infof("Verifying SQL thread is stopped")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("SQL thread stopped")
	return nil
}

// StartSlaveSQLThread is applicable with --test-on-replica
func (apl *Applier) StartSlaveSQLThread() error {
	replicaTerm := mysql.ReplicaTermFor(apl.migrationContext.ApplierMySQLVersion, `slave`)
	query := fmt.Sprintf("start /* gh-ost */ %s sql_thread", replicaTerm)
	apl.migrationContext.Log.Infof("Verifying SQL thread is running")
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("SQL thread started")
	return nil
}

// StopReplication is used by `--test-on-replica` and stops replication.
func (apl *Applier) StopReplication() error {
	if err := apl.StopSlaveIOThread(); err != nil {
		return err
	}
	if err := apl.StopSlaveSQLThread(); err != nil {
		return err
	}

	readBinlogCoordinates, executeBinlogCoordinates, err := mysql.GetReplicationBinlogCoordinates(apl.migrationContext.ApplierMySQLVersion, apl.db, apl.migrationContext.UseGTIDs)
	if err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Replication IO thread at %+v. SQL thread is at %+v", readBinlogCoordinates, executeBinlogCoordinates)
	return nil
}

// StartReplication is used by `--test-on-replica` on cut-over failure
func (apl *Applier) StartReplication() error {
	if err := apl.StartSlaveIOThread(); err != nil {
		return err
	}
	if err := apl.StartSlaveSQLThread(); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Replication started")
	return nil
}

// GetSessionLockName returns a name for the special hint session voluntary lock
func (apl *Applier) GetSessionLockName(sessionId int64) string {
	return fmt.Sprintf("gh-ost.%d.lock", sessionId)
}

// ExpectUsedLock expects the special hint voluntary lock to exist on given session
func (apl *Applier) ExpectUsedLock(sessionId int64) error {
	var result int64
	query := `select /* gh-ost */ is_used_lock(?)`
	lockName := apl.GetSessionLockName(sessionId)
	apl.migrationContext.Log.Infof("Checking session lock: %s", lockName)
	if err := apl.db.QueryRow(query, lockName).Scan(&result); err != nil || result != sessionId {
		return fmt.Errorf("session lock %s expected to be found but wasn't", lockName)
	}
	return nil
}

// ExpectProcess expects a process to show up in `SHOW PROCESSLIST` that has given characteristics
func (apl *Applier) ExpectProcess(sessionId int64, stateHint, infoHint string) error {
	found := false
	query := `
		select /* gh-ost */ id
		from
			information_schema.processlist
		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 {
		return fmt.Errorf("cannot find process. Hints: %s, %s", stateHint, infoHint)
	}
	return nil
}

// DropAtomicCutOverSentryTableIfExists checks if the "old" table name
// happens to be a cut-over magic table; if so, it drops it.
func (apl *Applier) DropAtomicCutOverSentryTableIfExists() error {
	apl.migrationContext.Log.Infof("Looking for magic cut-over table")
	tableName := apl.migrationContext.GetOldTableName()
	rowMap := apl.showTableStatus(tableName)
	if rowMap == nil {
		// Table does not exist
		return nil
	}
	if rowMap["Comment"].String != atomicCutOverMagicHint {
		return fmt.Errorf("expected magic comment on %s, did not find it", tableName)
	}
	apl.migrationContext.Log.Infof("Dropping magic cut-over table")
	return apl.dropTable(tableName)
}

// CreateAtomicCutOverSentryTable
func (apl *Applier) CreateAtomicCutOverSentryTable() error {
	if err := apl.DropAtomicCutOverSentryTableIfExists(); err != nil {
		return err
	}
	tableName := apl.migrationContext.GetOldTableName()

	query := fmt.Sprintf(`
		create /* gh-ost */ table %s.%s (
			id int auto_increment primary key
		) engine=%s comment='%s'`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(tableName),
		apl.migrationContext.TableEngine,
		atomicCutOverMagicHint,
	)
	apl.migrationContext.Log.Infof("Creating magic cut-over table %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(tableName),
	)
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		return err
	}
	apl.migrationContext.Log.Infof("Magic cut-over table created")

	return nil
}

// InitAtomicCutOverWaitTimeout sets the cut-over session wait_timeout in order to reduce the
// time an unresponsive (but still connected) gh-ost process can hold the cut-over lock.
func (apl *Applier) InitAtomicCutOverWaitTimeout(tx *gosql.Tx) error {
	cutOverWaitTimeoutSeconds := apl.migrationContext.CutOverLockTimeoutSeconds * 3
	apl.migrationContext.Log.Infof("Setting cut-over idle timeout as %d seconds", cutOverWaitTimeoutSeconds)
	query := fmt.Sprintf(`set /* gh-ost */ session wait_timeout:=%d`, cutOverWaitTimeoutSeconds)
	_, err := tx.Exec(query)
	return err
}

// RevertAtomicCutOverWaitTimeout restores the original wait_timeout for the applier session post-cut-over.
func (apl *Applier) RevertAtomicCutOverWaitTimeout() {
	apl.migrationContext.Log.Infof("Reverting cut-over idle timeout to %d seconds", apl.migrationContext.ApplierWaitTimeout)
	query := fmt.Sprintf(`set /* gh-ost */ session wait_timeout:=%d`, apl.migrationContext.ApplierWaitTimeout)
	if _, err := sqlutils.ExecNoPrepare(apl.db, query); err != nil {
		apl.migrationContext.Log.Errorf("failed to restore applier wait_timeout to %d seconds: %v",
			apl.migrationContext.ApplierWaitTimeout, err,
		)
	}
}

// AtomicCutOverMagicLock
func (apl *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocked chan<- error, okToUnlockTable <-chan bool, tableUnlocked chan<- error, renameLockSessionId *int64) error {
	tx, err := apl.db.Begin()
	if err != nil {
		tableLocked <- err
		return err
	}
	defer func() {
		sessionIdChan <- -1
		tableLocked <- fmt.Errorf("unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads")
		tableUnlocked <- fmt.Errorf("unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads")
		tx.Rollback()
		apl.DropAtomicCutOverSentryTableIfExists()
	}()

	var sessionId int64
	if err := tx.QueryRow(`select /* gh-ost */ connection_id()`).Scan(&sessionId); err != nil {
		tableLocked <- err
		return err
	}
	sessionIdChan <- sessionId

	lockResult := 0
	query := `select /* gh-ost */ get_lock(?, 0)`
	lockName := apl.GetSessionLockName(sessionId)
	apl.migrationContext.Log.Infof("Grabbing voluntary lock: %s", lockName)
	if err := tx.QueryRow(query, lockName).Scan(&lockResult); err != nil || lockResult != 1 {
		err := fmt.Errorf("unable to acquire lock %s", lockName)
		tableLocked <- err
		return err
	}

	tableLockTimeoutSeconds := apl.migrationContext.CutOverLockTimeoutSeconds * 2
	apl.migrationContext.Log.Infof("Setting LOCK timeout as %d seconds", tableLockTimeoutSeconds)
	query = fmt.Sprintf(`set /* gh-ost */ session lock_wait_timeout:=%d`, tableLockTimeoutSeconds)
	if _, err := tx.Exec(query); err != nil {
		tableLocked <- err
		return err
	}

	if err := apl.CreateAtomicCutOverSentryTable(); err != nil {
		tableLocked <- err
		return err
	}

	if err := apl.InitAtomicCutOverWaitTimeout(tx); err != nil {
		tableLocked <- err
		return err
	}
	defer apl.RevertAtomicCutOverWaitTimeout()

	query = fmt.Sprintf(`lock /* gh-ost */ tables %s.%s write, %s.%s write`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
	)
	apl.migrationContext.Log.Infof("Locking %s.%s, %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
	)
	apl.migrationContext.LockTablesStartTime = time.Now()
	if _, err := tx.Exec(query); err != nil {
		tableLocked <- err
		return err
	}
	apl.migrationContext.Log.Infof("Tables locked")
	tableLocked <- nil // No error.

	// From this point on, we are committed to UNLOCK TABLES. No matter what happens,
	// the UNLOCK must execute (or, alternatively, this connection dies, which gets the same impact)

	// The cut-over phase will proceed to apply remaining backlog onto ghost table,
	// and issue RENAME. We wait here until told to proceed.
	<-okToUnlockTable
	apl.migrationContext.Log.Infof("Will now proceed to drop magic table and unlock tables")

	// The magic table is here because we locked it. And we are the only ones allowed to drop it.
	// And in fact, we will:
	apl.migrationContext.Log.Infof("Dropping magic cut-over table")
	query = fmt.Sprintf(`drop /* gh-ost */ table if exists %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
	)

	if _, err := tx.Exec(query); err != nil {
		apl.migrationContext.Log.Errore(err)
		// We DO NOT return here because we must `UNLOCK TABLES`!
	}

	apl.migrationContext.Log.Infof("Session renameLockSessionId is %+v", *renameLockSessionId)
	// Checking the lock is held by rename session
	if *renameLockSessionId > 0 && apl.migrationContext.IsOpenMetadataLockInstruments && !apl.migrationContext.SkipMetadataLockCheck {
		sleepDuration := time.Duration(10*apl.migrationContext.CutOverLockTimeoutSeconds) * time.Millisecond
		for i := 1; i <= 100; i++ {
			err := apl.ExpectMetadataLock(*renameLockSessionId)
			if err == nil {
				apl.migrationContext.Log.Infof("Rename session is pending lock on the origin table !")
				break
			} else {
				time.Sleep(sleepDuration)
			}
		}
	}
	// Tables still locked
	apl.migrationContext.Log.Infof("Releasing lock from %s.%s, %s.%s",
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
	)
	query = `unlock /* gh-ost */ tables`
	if _, err := tx.Exec(query); err != nil {
		tableUnlocked <- err
		return apl.migrationContext.Log.Errore(err)
	}
	apl.migrationContext.Log.Infof("Tables unlocked")
	tableUnlocked <- nil
	return nil
}

// AtomicCutoverRename
func (apl *Applier) AtomicCutoverRename(sessionIdChan chan int64, tablesRenamed chan<- error) error {
	tx, err := apl.db.Begin()
	if err != nil {
		return err
	}
	defer func() {
		tx.Rollback()
		sessionIdChan <- -1
		tablesRenamed <- fmt.Errorf("unexpected error in AtomicCutoverRename(), injected to release blocking channel reads")
	}()
	var sessionId int64
	if err := tx.QueryRow(`select /* gh-ost */ connection_id()`).Scan(&sessionId); err != nil {
		return err
	}
	sessionIdChan <- sessionId

	apl.migrationContext.Log.Infof("Setting RENAME timeout as %d seconds", apl.migrationContext.CutOverLockTimeoutSeconds)
	query := fmt.Sprintf(`set /* gh-ost */ session lock_wait_timeout:=%d`, apl.migrationContext.CutOverLockTimeoutSeconds)
	if _, err := tx.Exec(query); err != nil {
		return err
	}

	query = fmt.Sprintf(`rename /* gh-ost */ table %s.%s to %s.%s, %s.%s to %s.%s`,
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetOldTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.GetGhostTableName()),
		sql.EscapeName(apl.migrationContext.DatabaseName),
		sql.EscapeName(apl.migrationContext.OriginalTableName),
	)
	apl.migrationContext.Log.Infof("Issuing and expecting this to block: %s", query)
	if _, err := tx.Exec(query); err != nil {
		tablesRenamed <- err
		return apl.migrationContext.Log.Errore(err)
	}
	tablesRenamed <- nil
	apl.migrationContext.Log.Infof("Tables renamed")
	return nil
}

func (apl *Applier) ShowStatusVariable(variableName string) (result int64, err error) {
	query := fmt.Sprintf(`show /* gh-ost */ global status like '%s'`, variableName)
	if err := apl.db.QueryRow(query).Scan(&variableName, &result); err != nil {
		return 0, err
	}
	return result, nil
}

// updateModifiesUniqueKeyColumns checks whether a UPDATE DML event actually
// modifies values of the migration's unique key (the iterated key). This will call
// for special handling.
func (apl *Applier) updateModifiesUniqueKeyColumns(dmlEvent *binlog.BinlogDMLEvent) (modifiedColumn string, isModified bool) {
	for _, column := range apl.migrationContext.UniqueKey.Columns.Columns() {
		tableOrdinal := apl.migrationContext.OriginalTableColumns.Ordinals[column.Name]
		whereColumnValue := dmlEvent.WhereColumnValues.AbstractValues()[tableOrdinal]
		newColumnValue := dmlEvent.NewColumnValues.AbstractValues()[tableOrdinal]

		if !reflect.DeepEqual(whereColumnValue, newColumnValue) {
			return column.Name, true
		}
	}
	return "", false
}

// buildDMLEventQuery creates a query to operate on the ghost table, based on an intercepted binlog
// event entry on the original table.
func (apl *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) []*dmlBuildResult {
	switch dmlEvent.DML {
	case binlog.DeleteDML:
		{
			query, uniqueKeyArgs, err := apl.dmlDeleteQueryBuilder.BuildQuery(dmlEvent.WhereColumnValues.AbstractValues())
			return []*dmlBuildResult{newDmlBuildResult(query, uniqueKeyArgs, -1, err)}
		}
	case binlog.InsertDML:
		{
			query, sharedArgs, err := apl.dmlInsertQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues())
			return []*dmlBuildResult{newDmlBuildResult(query, sharedArgs, 1, err)}
		}
	case binlog.UpdateDML:
		{
			if _, isModified := apl.updateModifiesUniqueKeyColumns(dmlEvent); isModified {
				results := make([]*dmlBuildResult, 0, 2)
				dmlEvent.DML = binlog.DeleteDML
				results = append(results, apl.buildDMLEventQuery(dmlEvent)...)
				dmlEvent.DML = binlog.InsertDML
				results = append(results, apl.buildDMLEventQuery(dmlEvent)...)
				return results
			}
			query, updateArgs, err := apl.dmlUpdateQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
			args := sqlutils.Args()
			args = append(args, updateArgs...)
			return []*dmlBuildResult{newDmlBuildResult(query, args, 0, err)}
		}
	}
	return []*dmlBuildResult{newDmlBuildResultError(fmt.Errorf("unknown dml event type: %+v", dmlEvent.DML))}
}

// executeBatchWithWarningChecking executes a batch of DML statements with SHOW WARNINGS
// interleaved after each statement to detect warnings from any statement in the batch.
// This is used when PanicOnWarnings is enabled to ensure warnings from middle statements
// are not lost (SHOW WARNINGS only shows warnings from the last statement in a multi-statement batch).
func (apl *Applier) executeBatchWithWarningChecking(ctx context.Context, tx *gosql.Tx, buildResults []*dmlBuildResult) (int64, error) {
	// Build query with interleaved SHOW WARNINGS: stmt1; SHOW WARNINGS; stmt2; SHOW WARNINGS; ...
	var queryBuilder strings.Builder
	args := make([]interface{}, 0)

	for _, buildResult := range buildResults {
		queryBuilder.WriteString(buildResult.query)
		queryBuilder.WriteString(";\nSHOW WARNINGS;\n")
		args = append(args, buildResult.args...)
	}

	query := queryBuilder.String()

	// Execute the multi-statement query
	rows, err := tx.QueryContext(ctx, query, args...)
	if err != nil {
		return 0, fmt.Errorf("%w; query=%s; args=%+v", err, query, args)
	}
	defer rows.Close()

	var totalDelta int64

	// QueryContext with multi-statement queries returns rows positioned at the first result set
	// that produces rows (i.e., the first SHOW WARNINGS), automatically skipping DML results.
	// Verify we're at a SHOW WARNINGS result set (should have 3 columns: Level, Code, Message)
	cols, err := rows.Columns()
	if err != nil {
		return 0, fmt.Errorf("failed to get columns: %w", err)
	}

	// If somehow we're not at a result set with columns, try to advance
	if len(cols) == 0 {
		if !rows.NextResultSet() {
			return 0, fmt.Errorf("expected SHOW WARNINGS result set after first statement")
		}
	}

	// Compile regex once before loop to avoid performance penalty and handle errors properly
	migrationKeyRegex, err := apl.compileMigrationKeyWarningRegex()
	if err != nil {
		return 0, err
	}

	// Iterate through SHOW WARNINGS result sets.
	// DML statements don't create navigable result sets, so we move directly between SHOW WARNINGS.
	// Pattern: [at SHOW WARNINGS #1] -> read warnings -> NextResultSet() -> [at SHOW WARNINGS #2] -> ...
	for i := 0; i < len(buildResults); i++ {
		// We can't get exact rows affected with QueryContext (needed for reading SHOW WARNINGS).
		// Use the theoretical delta (+1 for INSERT, -1 for DELETE, 0 for UPDATE) as an approximation.
		// This may be inaccurate (e.g., INSERT IGNORE with duplicate affects 0 rows but we count +1).
		totalDelta += buildResults[i].rowsDelta

		// Read warnings from this statement's SHOW WARNINGS result set
		var sqlWarnings []string
		for rows.Next() {
			var level, message string
			var code int
			if err := rows.Scan(&level, &code, &message); err != nil {
				// Scan failure means we cannot reliably read warnings.
				// Since PanicOnWarnings is a safety feature, we must fail hard rather than silently skip.
				return 0, fmt.Errorf("failed to scan SHOW WARNINGS for statement %d: %w", i+1, err)
			}

			if strings.Contains(message, "Duplicate entry") && migrationKeyRegex.MatchString(message) {
				// Duplicate entry on migration unique key is expected during binlog replay
				// (row was already copied during bulk copy phase)
				continue
			}
			sqlWarnings = append(sqlWarnings, fmt.Sprintf("%s: %s (%d)", level, message, code))
		}

		// Check for errors that occurred while iterating through warnings
		if err := rows.Err(); err != nil {
			return 0, fmt.Errorf("error reading SHOW WARNINGS result set for statement %d: %w", i+1, err)
		}

		if len(sqlWarnings) > 0 {
			return 0, fmt.Errorf("warnings detected in statement %d of %d: %v", i+1, len(buildResults), sqlWarnings)
		}

		// Move to the next statement's SHOW WARNINGS result set
		// For the last statement, there's no next result set
		// DML statements don't create result sets, so we only need one NextResultSet call
		// to move from SHOW WARNINGS #N to SHOW WARNINGS #(N+1)
		if i < len(buildResults)-1 {
			if !rows.NextResultSet() {
				if err := rows.Err(); err != nil {
					return 0, fmt.Errorf("error moving to SHOW WARNINGS for statement %d: %w", i+2, err)
				}
				return 0, fmt.Errorf("expected SHOW WARNINGS result set for statement %d", i+2)
			}
		}
	}

	return totalDelta, nil
}

// ApplyDMLEventQueries applies multiple DML queries onto the _ghost_ table
func (apl *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent)) error {
	var totalDelta int64
	ctx := context.Background()

	err := func() error {
		conn, err := apl.db.Conn(ctx)
		if err != nil {
			return err
		}
		defer conn.Close()

		sessionQuery := "SET /* gh-ost */ SESSION time_zone = '+00:00'"
		sessionQuery = fmt.Sprintf("%s, %s", sessionQuery, apl.generateSqlModeQuery())
		if _, err := conn.ExecContext(ctx, sessionQuery); err != nil {
			return err
		}

		tx, err := conn.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		rollback := func(err error) error {
			tx.Rollback()
			return err
		}

		buildResults := make([]*dmlBuildResult, 0, len(dmlEvents))
		nArgs := 0
		for _, dmlEvent := range dmlEvents {
			for _, buildResult := range apl.buildDMLEventQuery(dmlEvent) {
				if buildResult.err != nil {
					return rollback(buildResult.err)
				}
				nArgs += len(buildResult.args)
				buildResults = append(buildResults, buildResult)
			}
		}

		// When PanicOnWarnings is enabled, we need to check warnings after each statement
		// in the batch. SHOW WARNINGS only shows warnings from the last statement in a
		// multi-statement query, so we interleave SHOW WARNINGS after each DML statement.
		if apl.migrationContext.PanicOnWarnings {
			totalDelta, err = apl.executeBatchWithWarningChecking(ctx, tx, buildResults)
			if err != nil {
				return rollback(err)
			}
		} else {
			// Fast path: batch together DML queries into multi-statements to minimize network trips.
			// We use the raw driver connection to access the rows affected for each statement.
			execErr := conn.Raw(func(driverConn any) error {
				ex := driverConn.(driver.ExecerContext)
				nvc := driverConn.(driver.NamedValueChecker)

				multiArgs := make([]driver.NamedValue, 0, nArgs)
				multiQueryBuilder := strings.Builder{}
				for _, buildResult := range buildResults {
					for _, arg := range buildResult.args {
						nv := driver.NamedValue{Value: driver.Value(arg)}
						nvc.CheckNamedValue(&nv)
						multiArgs = append(multiArgs, nv)
					}

					multiQueryBuilder.WriteString(buildResult.query)
					multiQueryBuilder.WriteString(";\n")
				}

				res, err := ex.ExecContext(ctx, multiQueryBuilder.String(), multiArgs)
				if err != nil {
					err = fmt.Errorf("%w; query=%s; args=%+v", err, multiQueryBuilder.String(), multiArgs)
					return err
				}

				mysqlRes := res.(drivermysql.Result)

				// each DML is either a single insert (delta +1), update (delta +0) or delete (delta -1).
				// multiplying by the rows actually affected (either 0 or 1) will give an accurate row delta for this DML event
				for i, rowsAffected := range mysqlRes.AllRowsAffected() {
					totalDelta += buildResults[i].rowsDelta * rowsAffected
				}
				return nil
			})

			if execErr != nil {
				return rollback(execErr)
			}
		}

		if err := tx.Commit(); err != nil {
			return err
		}
		return nil
	}()

	if err != nil {
		return apl.migrationContext.Log.Errore(err)
	}
	// no error
	atomic.AddInt64(&apl.migrationContext.TotalDMLEventsApplied, int64(len(dmlEvents)))
	if apl.migrationContext.CountTableRows {
		atomic.AddInt64(&apl.migrationContext.RowsDeltaEstimate, totalDelta)
	}
	apl.migrationContext.Log.Debugf("ApplyDMLEventQueries() applied %d events in one transaction", len(dmlEvents))
	return nil
}

func (apl *Applier) Teardown() {
	apl.migrationContext.Log.Debugf("Tearing down...")
	apl.releaseMigrationLock()
	apl.db.Close()
	apl.singletonDB.Close()
	atomic.StoreInt64(&apl.finishedMigrating, 1)
}

func (apl *Applier) ExpectMetadataLock(sessionId int64) error {
	found := false
	query := `
		select /* gh-ost */ m.owner_thread_id
			from performance_schema.metadata_locks m join performance_schema.threads t 
			on m.owner_thread_id=t.thread_id
			where m.object_type = 'TABLE' and m.object_schema = ? and m.object_name = ? 
			and m.lock_type = 'EXCLUSIVE' and m.lock_status = 'PENDING' 
			and t.processlist_id = ?
	`
	err := sqlutils.QueryRowsMap(apl.db, query, func(m sqlutils.RowMap) error {
		found = true
		return nil
	}, apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName, sessionId)
	if err != nil {
		return err
	}
	if !found {
		err = fmt.Errorf("cannot find PENDING metadata lock on original table: `%s`.`%s`", apl.migrationContext.DatabaseName, apl.migrationContext.OriginalTableName)
		return apl.migrationContext.Log.Errore(err)
	}
	return nil
}