Fix data loss when inserting duplicate values during a migration (#1633)

* Fix data loss when PanicOnWarnings encounters duplicate keys on non-migration indexes

gh-ost could silently lose data when adding a unique index to a column during
a migration, even with the PanicOnWarnings flag enabled. This occurred when:

1. A migration adds a unique index to a column (e.g., email)
2. Rows with duplicate values are inserted into the original table after the
    bulk copy phase completes (during postponed cutover)
3. These duplicate rows are applied via binlog replay to the ghost table

**Expected behavior:** Migration fails with clear error
**Actual behavior:** Original rows with duplicate values silently
deleted, data lost

Original table: id PRIMARY KEY, email (no unique constraint)
Ghost table:    id PRIMARY KEY, email UNIQUE (being added)

Initial state (after bulk copy):
- Ghost table: (id=1, email='bob@example.com')
          (id=2, email='alice@example.com')

During postponed cutover:
- INSERT (id=3, email='bob@example.com') into original table

Binlog replay attempts:
- INSERT (id=3, email='bob@example.com') into ghost table
- Duplicate email='bob@example.com' (conflicts with id=1)
- Row with id=1 silently deleted → DATA LOSS

The DMLInsertQueryBuilder used `REPLACE` statements for binlog event replay:

```sql
REPLACE INTO ghost_table (id, email) VALUES (3, 'bob@example.com');

REPLACE behavior:
- If duplicate PRIMARY KEY: deletes old row, inserts new row (no warning/error)
- If duplicate on OTHER unique index: deletes conflicting row, inserts new row
- NO warnings or errors generated, so PanicOnWarnings cannot detect the issue

The original design assumed REPLACE was needed to handle timing edge cases
where binlog replay might process a row before bulk copy, but this caused
silent data corruption when other unique indexes had duplicates.

Changed DMLInsertQueryBuilder to use INSERT IGNORE instead of REPLACE:

INSERT IGNORE INTO ghost_table (id, email) VALUES (3, 'bob@example.com');

INSERT IGNORE behavior:
- If duplicate on ANY unique index: skip insert, generate WARNING
- Does not delete existing rows

Added warning detection to ApplyDMLEventQueries() when PanicOnWarnings is
enabled:
- Checks SHOW WARNINGS after batch execution
- Ignores duplicates on migration unique key (expected - row already copied)
- FAILS migration for duplicates on other unique indexes
- Transaction rollback ensures no partial state

Edge Case: DELETE+INSERT Conversion

When an UPDATE modifies the migration unique key (e.g., PRIMARY KEY), gh-ost
converts it to DELETE+INSERT within a single transaction:

BEGIN;
DELETE FROM ghost WHERE id=2;
INSERT IGNORE INTO ghost VALUES (3, 'bob@example.com');
COMMIT;

If the INSERT encounters a duplicate on a non-migration unique index:
- With PanicOnWarnings: Warning detected, transaction rolled back, both
DELETE and INSERT undone → no data loss ✓
- Without PanicOnWarnings: DELETE succeeds, INSERT silently skips →
data loss. This further reinforces that PanicOnWarnings should default
to on.

* Address linter issue

* Clarify test comment

* Consolidate test files

* Add execinquery linter comment

---------

Co-authored-by: meiji163 <meiji163@github.com>

Author: ggilder

go/logic/applier.go

@@ -1546,6 +1546,43 @@ func (this *Applier) ApplyDMLEventQueries(dmlEvents [](*binlog.BinlogDMLEvent))
 		if execErr != nil {
 			return rollback(execErr)
 		}
+
+		// Check for warnings when PanicOnWarnings is enabled
+		if this.migrationContext.PanicOnWarnings {
+			//nolint:execinquery
+			rows, err := tx.Query("SHOW WARNINGS")
+			if err != nil {
+				return rollback(err)
+			}
+			defer rows.Close()
+			if err = rows.Err(); err != nil {
+				return rollback(err)
+			}
+
+			var sqlWarnings []string
+			for rows.Next() {
+				var level, message string
+				var code int
+				if err := rows.Scan(&level, &code, &message); err != nil {
+					this.migrationContext.Log.Warningf("Failed to read SHOW WARNINGS row")
+					continue
+				}
+				// Duplicate warnings are formatted differently across mysql versions, hence the optional table name prefix
+				migrationUniqueKeyExpression := fmt.Sprintf("for key '(%s\\.)?%s'", this.migrationContext.GetGhostTableName(), this.migrationContext.UniqueKey.NameInGhostTable)
+				matched, _ := regexp.MatchString(migrationUniqueKeyExpression, message)
+				if strings.Contains(message, "Duplicate entry") && matched {
+					// 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))
+			}
+			if len(sqlWarnings) > 0 {
+				warningMsg := fmt.Sprintf("Warnings detected during DML event application: %v", sqlWarnings)
+				return rollback(errors.New(warningMsg))
+			}
+		}
+
 		if err := tx.Commit(); err != nil {
 			return err
 		}

go/logic/applier_test.go

@@ -147,7 +147,7 @@ func TestApplierBuildDMLEventQuery(t *testing.T) {
 		require.Len(t, res, 1)
 		require.NoError(t, res[0].err)
 		require.Equal(t,
-			`replace /* gh-ost `+"`test`.`_test_gho`"+` */
+			`insert /* gh-ost `+"`test`.`_test_gho`"+` */ ignore
 		into
 			`+"`test`.`_test_gho`"+`
 			`+"(`id`, `item_id`)"+`
@@ -724,6 +724,262 @@ func (suite *ApplierTestSuite) TestWriteCheckpoint() {
 	suite.Require().Equal(chk.IsCutover, gotChk.IsCutover)
 }
 
+func (suite *ApplierTestSuite) TestPanicOnWarningsWithDuplicateKeyOnNonMigrationIndex() {
+	ctx := context.Background()
+
+	var err error
+
+	// Create table with id and email columns, where id is the primary key
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100));", getTestTableName()))
+	suite.Require().NoError(err)
+
+	// Create ghost table with same schema plus a new unique index on email
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100), UNIQUE KEY email_unique (email));", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
+	suite.Require().NoError(err)
+
+	migrationContext := newTestMigrationContext()
+	migrationContext.ApplierConnectionConfig = connectionConfig
+	migrationContext.SetConnectionConfig("innodb")
+
+	migrationContext.PanicOnWarnings = true
+
+	migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:             "PRIMARY",
+		NameInGhostTable: "PRIMARY",
+		Columns:          *sql.NewColumnList([]string{"id"}),
+	}
+
+	applier := NewApplier(migrationContext)
+	suite.Require().NoError(applier.prepareQueries())
+	defer applier.Teardown()
+
+	err = applier.InitDBConnections()
+	suite.Require().NoError(err)
+
+	// Insert initial rows into ghost table (simulating bulk copy phase)
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, email) VALUES (1, 'user1@example.com'), (2, 'user2@example.com'), (3, 'user3@example.com');", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	// Simulate binlog event: try to insert a row with duplicate email
+	// This should fail with a warning because the ghost table has a unique index on email
+	dmlEvents := []*binlog.BinlogDMLEvent{
+		{
+			DatabaseName:    testMysqlDatabase,
+			TableName:       testMysqlTableName,
+			DML:             binlog.InsertDML,
+			NewColumnValues: sql.ToColumnValues([]interface{}{4, "user2@example.com"}), // duplicate email
+		},
+	}
+
+	// This should return an error when PanicOnWarnings is enabled
+	err = applier.ApplyDMLEventQueries(dmlEvents)
+	suite.Require().Error(err)
+	suite.Require().Contains(err.Error(), "Duplicate entry")
+
+	// Verify that the ghost table still has only 3 rows (no data loss)
+	rows, err := suite.db.Query("SELECT * FROM " + getTestGhostTableName() + " ORDER BY id")
+	suite.Require().NoError(err)
+	defer rows.Close()
+
+	var count int
+	for rows.Next() {
+		var id int
+		var email string
+		err = rows.Scan(&id, &email)
+		suite.Require().NoError(err)
+		count += 1
+	}
+	suite.Require().NoError(rows.Err())
+
+	// All 3 original rows should still be present
+	suite.Require().Equal(3, count)
+}
+
+// TestUpdateModifyingUniqueKeyWithDuplicateOnOtherIndex tests the scenario where:
+// 1. An UPDATE modifies the unique key (converted to DELETE+INSERT)
+// 2. The INSERT would create a duplicate on a NON-migration unique index
+// 3. Without warning detection: DELETE succeeds, INSERT IGNORE skips = DATA LOSS
+// 4. With PanicOnWarnings: Warning detected, transaction rolled back, no data loss
+// This test verifies that PanicOnWarnings correctly prevents the data loss scenario.
+func (suite *ApplierTestSuite) TestUpdateModifyingUniqueKeyWithDuplicateOnOtherIndex() {
+	ctx := context.Background()
+
+	var err error
+
+	// Create table with id (PRIMARY) and email (NO unique constraint yet)
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100));", getTestTableName()))
+	suite.Require().NoError(err)
+
+	// Create ghost table with id (PRIMARY) AND email unique index (being added)
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100), UNIQUE KEY email_unique (email));", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
+	suite.Require().NoError(err)
+
+	migrationContext := newTestMigrationContext()
+	migrationContext.ApplierConnectionConfig = connectionConfig
+	migrationContext.SetConnectionConfig("innodb")
+
+	migrationContext.PanicOnWarnings = true
+
+	migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:             "PRIMARY",
+		NameInGhostTable: "PRIMARY",
+		Columns:          *sql.NewColumnList([]string{"id"}),
+	}
+
+	applier := NewApplier(migrationContext)
+	suite.Require().NoError(applier.prepareQueries())
+	defer applier.Teardown()
+
+	err = applier.InitDBConnections()
+	suite.Require().NoError(err)
+
+	// Setup: Insert initial rows into ghost table
+	// Row 1: id=1, email='bob@example.com'
+	// Row 2: id=2, email='charlie@example.com'
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, email) VALUES (1, 'bob@example.com'), (2, 'charlie@example.com');", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	// Simulate binlog event: UPDATE that changes BOTH PRIMARY KEY and email
+	// From: id=2, email='charlie@example.com'
+	// To:   id=3, email='bob@example.com'  (duplicate email with id=1)
+	// This will be converted to DELETE (id=2) + INSERT (id=3, 'bob@example.com')
+	// With INSERT IGNORE, the INSERT will skip because email='bob@example.com' already exists in id=1
+	// Result: id=2 deleted, id=3 never inserted = DATA LOSS
+	dmlEvents := []*binlog.BinlogDMLEvent{
+		{
+			DatabaseName:      testMysqlDatabase,
+			TableName:         testMysqlTableName,
+			DML:               binlog.UpdateDML,
+			NewColumnValues:   sql.ToColumnValues([]interface{}{3, "bob@example.com"}),     // new: id=3, email='bob@example.com'
+			WhereColumnValues: sql.ToColumnValues([]interface{}{2, "charlie@example.com"}), // old: id=2, email='charlie@example.com'
+		},
+	}
+
+	// First verify this would be converted to DELETE+INSERT
+	buildResults := applier.buildDMLEventQuery(dmlEvents[0])
+	suite.Require().Len(buildResults, 2, "UPDATE modifying unique key should be converted to DELETE+INSERT")
+
+	// Apply the event - this should FAIL because INSERT will have duplicate email warning
+	err = applier.ApplyDMLEventQueries(dmlEvents)
+	suite.Require().Error(err, "Should fail when DELETE+INSERT causes duplicate on non-migration unique key")
+	suite.Require().Contains(err.Error(), "Duplicate entry", "Error should mention duplicate entry")
+
+	// Verify that BOTH rows still exist (transaction rolled back)
+	rows, err := suite.db.Query("SELECT id, email FROM " + getTestGhostTableName() + " ORDER BY id")
+	suite.Require().NoError(err)
+	defer rows.Close()
+
+	var count int
+	var ids []int
+	var emails []string
+	for rows.Next() {
+		var id int
+		var email string
+		err = rows.Scan(&id, &email)
+		suite.Require().NoError(err)
+		ids = append(ids, id)
+		emails = append(emails, email)
+		count++
+	}
+	suite.Require().NoError(rows.Err())
+
+	// Transaction should have rolled back, so original 2 rows should still be there
+	suite.Require().Equal(2, count, "Should still have 2 rows after failed transaction")
+	suite.Require().Equal([]int{1, 2}, ids, "Should have original ids")
+	suite.Require().Equal([]string{"bob@example.com", "charlie@example.com"}, emails)
+}
+
+// TestNormalUpdateWithPanicOnWarnings tests that normal UPDATEs (not modifying unique key) work correctly
+func (suite *ApplierTestSuite) TestNormalUpdateWithPanicOnWarnings() {
+	ctx := context.Background()
+
+	var err error
+
+	// Create table with id (PRIMARY) and email
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100));", getTestTableName()))
+	suite.Require().NoError(err)
+
+	// Create ghost table with same schema plus unique index on email
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("CREATE TABLE %s (id INT PRIMARY KEY, email VARCHAR(100), UNIQUE KEY email_unique (email));", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	connectionConfig, err := getTestConnectionConfig(ctx, suite.mysqlContainer)
+	suite.Require().NoError(err)
+
+	migrationContext := newTestMigrationContext()
+	migrationContext.ApplierConnectionConfig = connectionConfig
+	migrationContext.SetConnectionConfig("innodb")
+
+	migrationContext.PanicOnWarnings = true
+
+	migrationContext.OriginalTableColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.SharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.MappedSharedColumns = sql.NewColumnList([]string{"id", "email"})
+	migrationContext.UniqueKey = &sql.UniqueKey{
+		Name:             "PRIMARY",
+		NameInGhostTable: "PRIMARY",
+		Columns:          *sql.NewColumnList([]string{"id"}),
+	}
+
+	applier := NewApplier(migrationContext)
+	suite.Require().NoError(applier.prepareQueries())
+	defer applier.Teardown()
+
+	err = applier.InitDBConnections()
+	suite.Require().NoError(err)
+
+	// Setup: Insert initial rows into ghost table
+	_, err = suite.db.ExecContext(ctx, fmt.Sprintf("INSERT INTO %s (id, email) VALUES (1, 'alice@example.com'), (2, 'bob@example.com');", getTestGhostTableName()))
+	suite.Require().NoError(err)
+
+	// Simulate binlog event: Normal UPDATE that only changes email (not PRIMARY KEY)
+	// This should use UPDATE query, not DELETE+INSERT
+	dmlEvents := []*binlog.BinlogDMLEvent{
+		{
+			DatabaseName:      testMysqlDatabase,
+			TableName:         testMysqlTableName,
+			DML:               binlog.UpdateDML,
+			NewColumnValues:   sql.ToColumnValues([]interface{}{2, "robert@example.com"}), // update email only
+			WhereColumnValues: sql.ToColumnValues([]interface{}{2, "bob@example.com"}),
+		},
+	}
+
+	// Verify this generates a single UPDATE query (not DELETE+INSERT)
+	buildResults := applier.buildDMLEventQuery(dmlEvents[0])
+	suite.Require().Len(buildResults, 1, "Normal UPDATE should generate single UPDATE query")
+
+	// Apply the event - should succeed
+	err = applier.ApplyDMLEventQueries(dmlEvents)
+	suite.Require().NoError(err)
+
+	// Verify the update was applied correctly
+	rows, err := suite.db.Query("SELECT id, email FROM " + getTestGhostTableName() + " WHERE id = 2")
+	suite.Require().NoError(err)
+	defer rows.Close()
+
+	var id int
+	var email string
+	suite.Require().True(rows.Next(), "Should find updated row")
+	err = rows.Scan(&id, &email)
+	suite.Require().NoError(err)
+	suite.Require().Equal(2, id)
+	suite.Require().Equal("robert@example.com", email)
+	suite.Require().False(rows.Next(), "Should only have one row")
+	suite.Require().NoError(rows.Err())
+}
+
 func TestApplier(t *testing.T) {
 	suite.Run(t, new(ApplierTestSuite))
 }

go/sql/builder.go

@@ -566,7 +566,7 @@ func NewDMLInsertQueryBuilder(databaseName, tableName string, tableColumns, shar
 	preparedValues := buildColumnsPreparedValues(mappedSharedColumns)
 
 	stmt := fmt.Sprintf(`
-		replace /* gh-ost %s.%s */
+		insert /* gh-ost %s.%s */ ignore
 		into
 			%s.%s
 			(%s)

go/sql/builder_test.go

@@ -538,7 +538,7 @@ func TestBuildDMLInsertQuery(t *testing.T) {
 		query, sharedArgs, err := builder.BuildQuery(args)
 		require.NoError(t, err)
 		expected := `
-			replace /* gh-ost mydb.tbl */
+			insert /* gh-ost mydb.tbl */ ignore
 				into mydb.tbl
 					(id, name, position, age)
 				values
@@ -554,7 +554,7 @@ func TestBuildDMLInsertQuery(t *testing.T) {
 		query, sharedArgs, err := builder.BuildQuery(args)
 		require.NoError(t, err)
 		expected := `
-			replace /* gh-ost mydb.tbl */
+			insert /* gh-ost mydb.tbl */ ignore
 				into mydb.tbl
 					(position, name, age, id)
 				values
@@ -589,7 +589,7 @@ func TestBuildDMLInsertQuerySignedUnsigned(t *testing.T) {
 		query, sharedArgs, err := builder.BuildQuery(args)
 		require.NoError(t, err)
 		expected := `
-			replace /* gh-ost mydb.tbl */
+			insert /* gh-ost mydb.tbl */ ignore
 				into mydb.tbl
 					(id, name, position, age)
 				values
@@ -607,7 +607,7 @@ func TestBuildDMLInsertQuerySignedUnsigned(t *testing.T) {
 		query, sharedArgs, err := builder.BuildQuery(args)
 		require.NoError(t, err)
 		expected := `
-			replace /* gh-ost mydb.tbl */
+			insert /* gh-ost mydb.tbl */ ignore
 				into mydb.tbl
 					(id, name, position, age)
 				values
@@ -625,7 +625,7 @@ func TestBuildDMLInsertQuerySignedUnsigned(t *testing.T) {
 		query, sharedArgs, err := builder.BuildQuery(args)
 		require.NoError(t, err)
 		expected := `
-			replace /* gh-ost mydb.tbl */
+			insert /* gh-ost mydb.tbl */ ignore
 				into mydb.tbl
 					(id, name, position, age)
 				values