Fix builder's chunk order and drop panic-on-warnings row count check

- Removed count subqueries from range builders to restore the more performant approach from PR github#471
- Modified panic-on-warnings logic to trigger errors based solely on SQL warnings, not on row count mismatches
- This addresses potential race conditions where row count comparisons could produce false positives due to concurrent table modifications

Author: grodowski

go/logic/applier.go

@@ -663,47 +663,49 @@ func (this *Applier) ReadMigrationRangeValues() error {
 // 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 this done with copying row chunks)
-func (this *Applier) CalculateNextIterationRangeEndValues() (hasFurtherRange bool, expectedRowCount int64, err error) {
-	query, explodedArgs, err := sql.BuildUniqueKeyRangeEndPreparedQueryViaTemptable(
-		this.migrationContext.DatabaseName,
-		this.migrationContext.OriginalTableName,
-		&this.migrationContext.UniqueKey.Columns,
-		this.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
-		this.migrationContext.MigrationRangeMaxValues.AbstractValues(),
-		atomic.LoadInt64(&this.migrationContext.ChunkSize),
-		this.migrationContext.GetIteration() == 0,
-		fmt.Sprintf("iteration:%d", this.migrationContext.GetIteration()),
-	)
-	if err != nil {
-		return hasFurtherRange, expectedRowCount, err
-	}
-
-	rows, err := this.db.Query(query, explodedArgs...)
-	if err != nil {
-		return hasFurtherRange, expectedRowCount, err
-	}
-	defer rows.Close()
-
-	iterationRangeMaxValues := sql.NewColumnValues(this.migrationContext.UniqueKey.Len() + 1)
-	for rows.Next() {
-		if err = rows.Scan(iterationRangeMaxValues.ValuesPointers...); err != nil {
-			return hasFurtherRange, expectedRowCount, err
+func (this *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(
+			this.migrationContext.DatabaseName,
+			this.migrationContext.OriginalTableName,
+			&this.migrationContext.UniqueKey.Columns,
+			this.migrationContext.MigrationIterationRangeMinValues.AbstractValues(),
+			this.migrationContext.MigrationRangeMaxValues.AbstractValues(),
+			atomic.LoadInt64(&this.migrationContext.ChunkSize),
+			this.migrationContext.GetIteration() == 0,
+			fmt.Sprintf("iteration:%d", this.migrationContext.GetIteration()),
+		)
+		if err != nil {
+			return hasFurtherRange, err
 		}
 
-		expectedRowCount = (*iterationRangeMaxValues.ValuesPointers[len(iterationRangeMaxValues.ValuesPointers)-1].(*interface{})).(int64)
-		iterationRangeMaxValues = sql.ToColumnValues(iterationRangeMaxValues.AbstractValues()[:len(iterationRangeMaxValues.AbstractValues())-1])
+		rows, err := this.db.Query(query, explodedArgs...)
+		if err != nil {
+			return hasFurtherRange, err
+		}
+		defer rows.Close()
 
-		hasFurtherRange = expectedRowCount > 0
-	}
-	if err = rows.Err(); err != nil {
-		return hasFurtherRange, expectedRowCount, err
-	}
-	if hasFurtherRange {
-		this.migrationContext.MigrationIterationRangeMaxValues = iterationRangeMaxValues
-		return hasFurtherRange, expectedRowCount, nil
+		iterationRangeMaxValues := sql.NewColumnValues(this.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 {
+			this.migrationContext.MigrationIterationRangeMaxValues = iterationRangeMaxValues
+			return hasFurtherRange, nil
+		}
 	}
 	this.migrationContext.Log.Debugf("Iteration complete: no further range to iterate")
-	return hasFurtherRange, expectedRowCount, nil
+	return hasFurtherRange, nil
 }
 
 // ApplyIterationInsertQuery issues a chunk-INSERT query on the ghost table. It is where

go/logic/applier_test.go

@@ -563,10 +563,9 @@ func (suite *ApplierTestSuite) TestPanicOnWarningsInApplyIterationInsertQuerySuc
 	suite.Require().NoError(err)
 
 	migrationContext.SetNextIterationRangeMinValues()
-	hasFurtherRange, expectedRangeSize, err := applier.CalculateNextIterationRangeEndValues()
+	hasFurtherRange, err := applier.CalculateNextIterationRangeEndValues()
 	suite.Require().NoError(err)
 	suite.Require().True(hasFurtherRange)
-	suite.Require().Equal(int64(1), expectedRangeSize)
 
 	_, rowsAffected, _, err := applier.ApplyIterationInsertQuery()
 	suite.Require().NoError(err)

go/logic/migrator.go

@@ -1344,7 +1344,7 @@ func (this *Migrator) iterateChunks() error {
 				}
 
 				// When hasFurtherRange is false, original table might be write locked and CalculateNextIterationRangeEndValues would hangs forever
-				hasFurtherRange, expectedRangeSize, err := this.applier.CalculateNextIterationRangeEndValues()
+				hasFurtherRange, err := this.applier.CalculateNextIterationRangeEndValues()
 				if err != nil {
 					return err // wrapping call will retry
 				}
@@ -1373,10 +1373,8 @@ func (this *Migrator) iterateChunks() error {
 						for _, warning := range this.migrationContext.MigrationLastInsertSQLWarnings {
 							this.migrationContext.Log.Infof("ApplyIterationInsertQuery has SQL warnings! %s", warning)
 						}
-						if expectedRangeSize != rowsAffected {
-							joinedWarnings := strings.Join(this.migrationContext.MigrationLastInsertSQLWarnings, "; ")
-							terminateRowIteration(fmt.Errorf("ApplyIterationInsertQuery failed because of SQL warnings: [%s]", joinedWarnings))
-						}
+						joinedWarnings := strings.Join(this.migrationContext.MigrationLastInsertSQLWarnings, "; ")
+						terminateRowIteration(fmt.Errorf("ApplyIterationInsertQuery failed because of SQL warnings: [%s]", joinedWarnings))
 					}
 				}
 

go/sql/builder.go

@@ -251,6 +251,59 @@ func BuildRangeInsertPreparedQuery(databaseName, originalTableName, ghostTableNa
 	return BuildRangeInsertQuery(databaseName, originalTableName, ghostTableName, sharedColumns, mappedSharedColumns, uniqueKey, uniqueKeyColumns, rangeStartValues, rangeEndValues, rangeStartArgs, rangeEndArgs, includeRangeStartValues, transactionalTable, noWait)
 }
 
+func BuildUniqueKeyRangeEndPreparedQueryViaOffset(databaseName, tableName string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, chunkSize int64, includeRangeStartValues bool, hint string) (result string, explodedArgs []interface{}, err error) {
+	if uniqueKeyColumns.Len() == 0 {
+		return "", explodedArgs, fmt.Errorf("Got 0 columns in BuildUniqueKeyRangeEndPreparedQuery")
+	}
+	databaseName = EscapeName(databaseName)
+	tableName = EscapeName(tableName)
+
+	var startRangeComparisonSign ValueComparisonSign = GreaterThanComparisonSign
+	if includeRangeStartValues {
+		startRangeComparisonSign = GreaterThanOrEqualsComparisonSign
+	}
+	rangeStartComparison, rangeExplodedArgs, err := BuildRangePreparedComparison(uniqueKeyColumns, rangeStartArgs, startRangeComparisonSign)
+	if err != nil {
+		return "", explodedArgs, err
+	}
+	explodedArgs = append(explodedArgs, rangeExplodedArgs...)
+	rangeEndComparison, rangeExplodedArgs, err := BuildRangePreparedComparison(uniqueKeyColumns, rangeEndArgs, LessThanOrEqualsComparisonSign)
+	if err != nil {
+		return "", explodedArgs, err
+	}
+	explodedArgs = append(explodedArgs, rangeExplodedArgs...)
+
+	uniqueKeyColumnNames := duplicateNames(uniqueKeyColumns.Names())
+	uniqueKeyColumnAscending := make([]string, len(uniqueKeyColumnNames))
+	for i, column := range uniqueKeyColumns.Columns() {
+		uniqueKeyColumnNames[i] = EscapeName(uniqueKeyColumnNames[i])
+		if column.Type == EnumColumnType {
+			uniqueKeyColumnAscending[i] = fmt.Sprintf("concat(%s) asc", uniqueKeyColumnNames[i])
+		} else {
+			uniqueKeyColumnAscending[i] = fmt.Sprintf("%s asc", uniqueKeyColumnNames[i])
+		}
+	}
+	result = fmt.Sprintf(`
+		select /* gh-ost %s.%s %s */
+			%s
+		from
+			%s.%s
+		where
+			%s and %s
+		order by
+			%s
+		limit 1
+		offset %d`,
+		databaseName, tableName, hint,
+		strings.Join(uniqueKeyColumnNames, ", "),
+		databaseName, tableName,
+		rangeStartComparison, rangeEndComparison,
+		strings.Join(uniqueKeyColumnAscending, ", "),
+		(chunkSize - 1),
+	)
+	return result, explodedArgs, nil
+}
+
 func BuildUniqueKeyRangeEndPreparedQueryViaTemptable(databaseName, tableName string, uniqueKeyColumns *ColumnList, rangeStartArgs, rangeEndArgs []interface{}, chunkSize int64, includeRangeStartValues bool, hint string) (result string, explodedArgs []interface{}, err error) {
 	if uniqueKeyColumns.Len() == 0 {
 		return "", explodedArgs, fmt.Errorf("Got 0 columns in BuildUniqueKeyRangeEndPreparedQuery")
@@ -286,22 +339,8 @@ func BuildUniqueKeyRangeEndPreparedQueryViaTemptable(databaseName, tableName str
 			uniqueKeyColumnDescending[i] = fmt.Sprintf("%s desc", uniqueKeyColumnNames[i])
 		}
 	}
-
-	joinedColumnNames := strings.Join(uniqueKeyColumnNames, ", ")
 	result = fmt.Sprintf(`
-		select /* gh-ost %s.%s %s */
-			%s,
-			(select count(*) from (
-				select
-					%s
-				from
-					%s.%s
-				where
-					%s and %s
-				order by
-					%s
-				limit %d
-			) select_osc_chunk)
+		select /* gh-ost %s.%s %s */ %s
 		from (
 			select
 				%s
@@ -316,17 +355,13 @@ func BuildUniqueKeyRangeEndPreparedQueryViaTemptable(databaseName, tableName str
 		order by
 			%s
 		limit 1`,
-		databaseName, tableName, hint, joinedColumnNames,
-		joinedColumnNames, databaseName, tableName,
-		rangeStartComparison, rangeEndComparison,
-		strings.Join(uniqueKeyColumnAscending, ", "), chunkSize,
-		joinedColumnNames, databaseName, tableName,
+		databaseName, tableName, hint, strings.Join(uniqueKeyColumnNames, ", "),
+		strings.Join(uniqueKeyColumnNames, ", "), databaseName, tableName,
 		rangeStartComparison, rangeEndComparison,
 		strings.Join(uniqueKeyColumnAscending, ", "), chunkSize,
 		strings.Join(uniqueKeyColumnDescending, ", "),
 	)
-	// 2x the explodedArgs for the subquery (CTE would be possible but not supported by MySQL 5)
-	return result, append(explodedArgs, explodedArgs...), nil
+	return result, explodedArgs, nil
 }
 
 func BuildUniqueKeyMinValuesPreparedQuery(databaseName, tableName string, uniqueKey *UniqueKey) (string, error) {

go/sql/builder_test.go

@@ -325,6 +325,33 @@ func TestBuildRangeInsertPreparedQuery(t *testing.T) {
 	}
 }
 
+func TestBuildUniqueKeyRangeEndPreparedQueryViaOffset(t *testing.T) {
+	databaseName := "mydb"
+	originalTableName := "tbl"
+	var chunkSize int64 = 500
+	{
+		uniqueKeyColumns := NewColumnList([]string{"name", "position"})
+		rangeStartArgs := []interface{}{3, 17}
+		rangeEndArgs := []interface{}{103, 117}
+
+		query, explodedArgs, err := BuildUniqueKeyRangeEndPreparedQueryViaOffset(databaseName, originalTableName, uniqueKeyColumns, rangeStartArgs, rangeEndArgs, chunkSize, false, "test")
+		require.NoError(t, err)
+		expected := `
+			select /* gh-ost mydb.tbl test */
+				name, position
+			from
+				mydb.tbl
+			where
+				((name > ?) or (((name = ?)) AND (position > ?))) and ((name < ?) or (((name = ?)) AND (position < ?)) or ((name = ?) and (position = ?)))
+			order by
+				name asc, position asc
+			limit 1
+			offset 499`
+		require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
+		require.Equal(t, []interface{}{3, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)
+	}
+}
+
 func TestBuildUniqueKeyRangeEndPreparedQueryViaTemptable(t *testing.T) {
 	databaseName := "mydb"
 	originalTableName := "tbl"
@@ -338,17 +365,7 @@ func TestBuildUniqueKeyRangeEndPreparedQueryViaTemptable(t *testing.T) {
 		require.NoError(t, err)
 		expected := `
 			select /* gh-ost mydb.tbl test */
-				name, position,
-				(select count(*) from (
-					select
-						name, position
-					from
-						mydb.tbl
-					where ((name > ?) or (((name = ?)) AND (position > ?))) and ((name < ?) or (((name = ?)) AND (position < ?)) or ((name = ?) and (position = ?)))
-					order by
-						name asc, position asc
-					limit 500
-				) select_osc_chunk)
+				name, position
 			from (
 				select
 					name, position
@@ -363,7 +380,7 @@ func TestBuildUniqueKeyRangeEndPreparedQueryViaTemptable(t *testing.T) {
 				name desc, position desc
 			limit 1`
 		require.Equal(t, normalizeQuery(expected), normalizeQuery(query))
-		require.Equal(t, []interface{}{3, 3, 17, 103, 103, 117, 103, 117, 3, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)
+		require.Equal(t, []interface{}{3, 3, 17, 103, 103, 117, 103, 117}, explodedArgs)
 	}
 }