Sync upstream 2026-03-10

dev.yml

@@ -6,7 +6,7 @@ env:
 
 up:
   - go:
-        version: "1.23.0"
+        version: "1.25.0"
   - podman
   - custom:
         name: Go Dependencies

go/base/context.go

@@ -979,9 +979,8 @@ func (this *MigrationContext) GetGhostTriggerName(triggerName string) string {
 	return triggerName + this.TriggerSuffix
 }
 
-// validateGhostTriggerLength check if the ghost trigger name length is not more than 64 characters
+// ValidateGhostTriggerLengthBelowMaxLength checks if the given trigger name (already transformed
+// by GetGhostTriggerName) does not exceed the maximum allowed length.
 func (this *MigrationContext) ValidateGhostTriggerLengthBelowMaxLength(triggerName string) bool {
-	ghostTriggerName := this.GetGhostTriggerName(triggerName)
-
-	return utf8.RuneCountInString(ghostTriggerName) <= mysql.MaxTableNameLength
+	return utf8.RuneCountInString(triggerName) <= mysql.MaxTableNameLength
 }

go/base/context_test.go

@@ -86,38 +86,69 @@ func TestGetTriggerNames(t *testing.T) {
 }
 
 func TestValidateGhostTriggerLengthBelowMaxLength(t *testing.T) {
+	// Tests simulate the real call pattern: GetGhostTriggerName first, then validate the result.
 	{
+		// Short trigger name with suffix appended: well under 64 chars
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_gho"
-		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength("my_trigger"))
+		ghostName := context.GetGhostTriggerName("my_trigger") // "my_trigger_gho" = 14 chars
+		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 	{
+		// 64-char original + "_ghost" suffix = 70 chars → exceeds limit
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_ghost"
-		require.False(t, context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4))) // 64 characters + "_ghost"
+		ghostName := context.GetGhostTriggerName(strings.Repeat("my_trigger_ghost", 4)) // 64 + 6 = 70
+		require.False(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 	{
+		// 48-char original + "_ghost" suffix = 54 chars → valid
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_ghost"
-		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 3))) // 48 characters + "_ghost"
+		ghostName := context.GetGhostTriggerName(strings.Repeat("my_trigger_ghost", 3)) // 48 + 6 = 54
+		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 	{
+		// RemoveTriggerSuffix: 64-char name ending in "_ghost" → suffix removed → 58 chars → valid
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_ghost"
 		context.RemoveTriggerSuffix = true
-		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4))) // 64 characters + "_ghost" removed
+		ghostName := context.GetGhostTriggerName(strings.Repeat("my_trigger_ghost", 4)) // suffix removed → 58
+		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 	{
+		// RemoveTriggerSuffix: name doesn't end in suffix → suffix appended → 65 + 6 = 71 chars → exceeds
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_ghost"
 		context.RemoveTriggerSuffix = true
-		require.False(t, context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)+"X")) // 65 characters + "_ghost" not removed
+		ghostName := context.GetGhostTriggerName(strings.Repeat("my_trigger_ghost", 4) + "X") // no match, appended → 71
+		require.False(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 	{
+		// RemoveTriggerSuffix: 70-char name ending in "_ghost" → suffix removed → 64 chars → exactly at limit → valid
 		context := NewMigrationContext()
 		context.TriggerSuffix = "_ghost"
 		context.RemoveTriggerSuffix = true
-		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(strings.Repeat("my_trigger_ghost", 4)+"_ghost")) // 70 characters + last "_ghost"  removed
+		ghostName := context.GetGhostTriggerName(strings.Repeat("my_trigger_ghost", 4) + "_ghost") // suffix removed → 64
+		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
+	}
+	{
+		// Edge case: exactly 64 chars after transformation → valid (boundary test)
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ght"
+		originalName := strings.Repeat("x", 60)                // 60 chars
+		ghostName := context.GetGhostTriggerName(originalName) // 60 + 4 = 64
+		require.Equal(t, 64, len(ghostName))
+		require.True(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
+	}
+	{
+		// Edge case: 65 chars after transformation → exceeds (boundary test)
+		context := NewMigrationContext()
+		context.TriggerSuffix = "_ght"
+		originalName := strings.Repeat("x", 61)                // 61 chars
+		ghostName := context.GetGhostTriggerName(originalName) // 61 + 4 = 65
+		require.Equal(t, 65, len(ghostName))
+		require.False(t, context.ValidateGhostTriggerLengthBelowMaxLength(ghostName))
 	}
 }
 

go/logic/applier.go

@@ -819,13 +819,6 @@ func (this *Applier) ReadMigrationRangeValues() error {
 // 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 (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, err error) {
-	this.LastIterationRangeMutex.Lock()
-	if this.migrationContext.MigrationIterationRangeMinValues != nil && this.migrationContext.MigrationIterationRangeMaxValues != nil {
-		this.LastIterationRangeMinValues = this.migrationContext.MigrationIterationRangeMinValues.Clone()
-		this.LastIterationRangeMaxValues = this.migrationContext.MigrationIterationRangeMaxValues.Clone()
-	}
-	this.LastIterationRangeMutex.Unlock()
-
 	for i := 0; i < 2; i++ {
 		buildFunc := sql.BuildUniqueKeyRangeEndPreparedQueryViaOffset
 		if i == 1 {
@@ -1466,10 +1459,9 @@ func (this *Applier) buildDMLEventQuery(dmlEvent *binlog.BinlogDMLEvent) []*dmlB
 				results = append(results, this.buildDMLEventQuery(dmlEvent)...)
 				return results
 			}
-			query, sharedArgs, uniqueKeyArgs, err := this.dmlUpdateQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
+			query, updateArgs, err := this.dmlUpdateQueryBuilder.BuildQuery(dmlEvent.NewColumnValues.AbstractValues(), dmlEvent.WhereColumnValues.AbstractValues())
 			args := sqlutils.Args()
-			args = append(args, sharedArgs...)
-			args = append(args, uniqueKeyArgs...)
+			args = append(args, updateArgs...)
 			return []*dmlBuildResult{newDmlBuildResult(query, args, 0, err)}
 		}
 	}
@@ -1554,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
 		}