Unify PKE selection semantics across schemadiff consumers

changelog/25.0/25.0.0/summary.md

@@ -19,6 +19,7 @@
 - **[Minor Changes](#minor-changes)**
     - **[VReplication](#minor-changes-vreplication)**
         - [Default data protection for `_reverse` workflow cancel/complete](#vreplication-reverse-workflow-data-protection)
+        - [Unified Primary Key Equivalent (PKE) selection](#vreplication-unified-pke-selection)
     - **[VTGate](#minor-changes-vtgate)**
         - [Ingress bytes in query LogStats](#vtgate-logstats-ingress-bytes)
         - [New controls for cross-keyspace reads](#vtgate-cross-keyspace-reads)
@@ -152,6 +153,8 @@ Both compatibility behaviors will be removed in v26, along with the `SelectStrea
 
 ## <a id="minor-changes"/>Minor Changes</a>
 
+### <a id="minor-changes-vreplication"/>VReplication</a>
+
 #### <a id="vreplication-reverse-workflow-data-protection"/>Default data protection for `_reverse` workflow cancel/complete</a>
 
 When calling `cancel` or `complete` on an auto-generated `_reverse` workflow without explicitly providing `--keep-data=false`, the system now defaults to keeping data and returns a warning. This prevents accidental deletion of production tables on the original source side, where the `_reverse` workflow's target is actually your production keyspace.
@@ -168,6 +171,20 @@ The `--keep-data` flag help text has been updated to note this default explicitl
 
 See [#19906](https://github.com/vitessio/vitess/pull/19906) for details.
 
+#### <a id="vreplication-unified-pke-selection"/>Unified Primary Key Equivalent (PKE) selection</a>
+
+VReplication workflows (MoveTables, Reshard, Materialize, VDiff) and Online DDL now share a single implementation for selecting a Primary Key Equivalent (PKE) — the unique, non-NULLable key used to iterate and identify rows in tables that have no defined PRIMARY KEY. The PKE is now determined by parsing the table's `CREATE TABLE` statement (via the `schemadiff` package) instead of querying `information_schema`, and both code paths rank candidate keys using the same data-type cost model. Exact cost ties are broken deterministically by the lexicographically smallest index name.
+
+**Behavior changes:**
+
+- Online DDL's unique key prioritization now ranks candidate keys by the total storage cost of all columns in the key, rather than by properties of the key's first column only. For some tables this changes which unique key an Online DDL migration iterates over. The selected key is still a valid unique, non-NULLable key in all cases; only the preference order among multiple candidates has changed.
+- Unique keys containing functional (expression) key parts are no longer PKE candidates. Previously, the `information_schema`-based selection could pick such a key and produce a broken column list that could make VReplication identify rows by a non-unique set of columns; such keys are now skipped in favor of the next valid candidate, or all columns when no valid candidate exists.
+- If a no-PK table's `CREATE TABLE` statement cannot be parsed (for example, it uses table options unknown to the Vitess SQL parser, such as `SECONDARY_ENGINE`), VReplication and VDiff now log a warning and use all columns as the substitute PK instead of failing the workflow.
+
+**Upgrading with in-flight workflows:** the PKE is re-derived whenever a workflow (re)starts, so the key selected for a no-PK table can change across an upgrade while that table's copy phase is in progress. The source tablet now validates that a resumed copy's `lastpk` value was generated using the currently selected key columns and fails with a clear error rather than resuming from the wrong position. If you hit this error, restart the copy for the affected table (for example, by recreating the workflow) and run VDiff afterwards. Where possible, let the copy phase of tables without a primary key complete before upgrading.
+
+See [#10259](https://github.com/vitessio/vitess/issues/10259) for details.
+
 ### <a id="minor-changes-vtgate"/>VTGate</a>
 
 #### <a id="vtgate-logstats-ingress-bytes"/>Ingress bytes in query LogStats</a>

go/cmd/vttablet/cli/cli.go

@@ -172,7 +172,7 @@ func run(cmd *cobra.Command, args []string) error {
 		UpdateStream:        binlog.NewUpdateStream(ts, tablet.Keyspace, tabletAlias.Cell, qsc.SchemaEngine(), env.Parser()),
 		VREngine:            vreplication.NewEngine(env, config, ts, tabletAlias.Cell, mysqld, qsc.LagThrottler()),
 		SemiSyncMonitor:     semisyncmonitor.NewMonitor(config, qsc.Exporter()),
-		VDiffEngine:         vdiff.NewEngine(ts, tablet, env.CollationEnv(), env.Parser()),
+		VDiffEngine:         vdiff.NewEngine(ts, tablet, env),
 	}
 	if err := tm.Start(tablet, config); err != nil {
 		return fmt.Errorf("failed to parse --tablet-path or initialize DB credentials: %w", err)

go/vt/mysqlctl/schema.go

@@ -578,21 +578,14 @@ func (mysqld *Mysqld) ApplySchemaChange(ctx context.Context, dbName string, chan
 	return &tabletmanagerdatapb.SchemaChangeResult{BeforeSchema: beforeSchema, AfterSchema: afterSchema}, nil
 }
 
-// GetPrimaryKeyEquivalentColumns can be used if the table has
-// no defined PRIMARY KEY. It will return the columns in a
-// viable PRIMARY KEY equivalent (PKE) -- a NON-NULL UNIQUE
-// KEY -- along with that index's name in the specified table.
-// When multiple PKE indexes are available it will attempt to
-// choose the most efficient one based on the column data types
-// and the number of columns in the index. See here for the data
-// type storage sizes:
-//
-//	https://dev.mysql.com/doc/refman/en/storage-requirements.html
-//
-// If this function is used on a table that DOES have a
-// defined PRIMARY KEY then it may return the columns for
-// that index if it is likely the most efficient one amongst
-// the available PKE indexes on the table.
+// GetPrimaryKeyEquivalentColumns returns the columns and name of the
+// best Primary Key Equivalent (PKE) index -- a unique key with no
+// NULLable columns and no functional (expression) key parts -- using an
+// information_schema query. Prefer
+// schemadiff.GetPrimaryKeyEquivalent, which implements the same
+// selection at the SQL parser level; this database-backed variant is
+// retained as the fallback for CREATE TABLE statements the parser
+// cannot handle (e.g. ones using unsupported table options).
 func GetPrimaryKeyEquivalentColumns(ctx context.Context, exec func(string, int, bool) (*sqltypes.Result, error), dbName, table string) ([]string, string, error) {
 	// We use column name aliases to guarantee lower case for our named results.
 	sql := `
@@ -628,7 +621,7 @@ func GetPrimaryKeyEquivalentColumns(ctx context.Context, exec func(string, int,
                 WHERE stats.TABLE_SCHEMA = %s AND stats.TABLE_NAME = %s AND stats.INDEX_NAME NOT IN
                 (
                     SELECT DISTINCT INDEX_NAME FROM information_schema.STATISTICS
-                    WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND (NON_UNIQUE = 1 OR NULLABLE = 'YES')
+                    WHERE TABLE_SCHEMA = %s AND TABLE_NAME = %s AND (NON_UNIQUE = 1 OR NULLABLE = 'YES' OR COLUMN_NAME IS NULL)
                 )
                 GROUP BY INDEX_NAME ORDER BY type_cost ASC, col_count ASC LIMIT 1
             ) AS pke ON index_cols.INDEX_NAME = pke.INDEX_NAME

go/vt/schemadiff/column.go

@@ -347,6 +347,11 @@ func (c *ColumnDefinitionEntity) HasBlobTypeStorage() bool {
 	return BlobTypeStorage(c.Type()) != 0
 }
 
+// TypeCost returns the relative type cost of this column for PKE ranking.
+func (c *ColumnDefinitionEntity) TypeCost() int {
+	return TypeCost(c.Type())
+}
+
 // Charset returns the column's charset
 func (c *ColumnDefinitionEntity) Charset() string {
 	return c.ColumnDefinition.Type.Charset.Name

go/vt/schemadiff/env.go

@@ -39,3 +39,9 @@ func NewEnv(env *vtenv.Environment, defaultColl collations.ID) *Environment {
 		DefaultColl: defaultColl,
 	}
 }
+
+// NewEnvWithDefaults creates a new Environment using the default connection
+// charset from the given vtenv.Environment.
+func NewEnvWithDefaults(env *vtenv.Environment) *Environment {
+	return NewEnv(env, env.CollationEnv().DefaultConnectionCharset())
+}

go/vt/schemadiff/key.go

@@ -108,6 +108,16 @@ func (i *IndexDefinitionEntity) HasColumnPrefix() bool {
 	return false
 }
 
+// TypeCost returns the sum of type costs for all columns in the index,
+// used for ranking PKE indexes.
+func (i *IndexDefinitionEntity) TypeCost() int {
+	cost := 0
+	for _, col := range i.ColumnList.Entities {
+		cost += col.TypeCost()
+	}
+	return cost
+}
+
 // ColumnNames returns the names of the columns in the index.
 func (i *IndexDefinitionEntity) ColumnNames() []string {
 	names := make([]string, 0, len(i.IndexDefinition.Columns))

go/vt/schemadiff/mysql.go

@@ -16,6 +16,8 @@ limitations under the License.
 
 package schemadiff
 
+import "strings"
+
 var engineCasing = map[string]string{
 	"INNODB": "InnoDB",
 	"MYISAM": "MyISAM",
@@ -116,3 +118,43 @@ func IsExpandingDataType(sourceType string, targetType string) bool {
 	_, ok := expandedDataTypes[sourceType+":"+targetType]
 	return ok
 }
+
+const defaultTypeCost = 1000
+
+// typeCost maps MySQL data types to a relative cost for ranking Primary
+// Key Equivalent (PKE) candidate keys; lower is preferred. It matches the
+// ranking in mysqlctl.GetPrimaryKeyEquivalentColumns.
+var typeCost = map[string]int{
+	"enum":      0,
+	"tinyint":   1,
+	"year":      2,
+	"smallint":  3,
+	"date":      4,
+	"mediumint": 5,
+	"time":      6,
+	"int":       7,
+	"set":       8,
+	"timestamp": 9,
+	"bigint":    10,
+	"float":     11,
+	"double":    12,
+	"decimal":   13,
+	"datetime":  14,
+	"binary":    30,
+	"char":      31,
+	"varbinary": 60,
+	"varchar":   61,
+	"tinyblob":  80,
+	"tinytext":  81,
+}
+
+// TypeCost returns the relative cost of a MySQL column type for PKE
+// ranking. columnType must be a bare MySQL-canonical type name as emitted
+// by SHOW CREATE TABLE (e.g. "varchar", not "varchar(255)"); unknown
+// types get a high cost so they rank last.
+func TypeCost(columnType string) int {
+	if cost, ok := typeCost[strings.ToLower(columnType)]; ok {
+		return cost
+	}
+	return defaultTypeCost
+}

go/vt/schemadiff/onlineddl.go

@@ -192,37 +192,69 @@ func PrioritizedUniqueKeys(createTableEntity *CreateTableEntity) *IndexDefinitio
 			// Prefix comes last
 			return false
 		}
-		iFirstColEntity := uniqueKeys[i].ColumnList.Entities[0]
-		jFirstColEntity := uniqueKeys[j].ColumnList.Entities[0]
-		if iFirstColEntity.IsIntegralType() && !jFirstColEntity.IsIntegralType() {
-			// Prioritize integers
-			return true
+		if costDiff := uniqueKeys[i].TypeCost() - uniqueKeys[j].TypeCost(); costDiff != 0 {
+			return costDiff < 0
 		}
-		if !iFirstColEntity.IsIntegralType() && jFirstColEntity.IsIntegralType() {
-			// Prioritize integers
-			return false
+		if lenDiff := len(uniqueKeys[i].ColumnList.Entities) - len(uniqueKeys[j].ColumnList.Entities); lenDiff != 0 {
+			return lenDiff < 0
 		}
-		if !iFirstColEntity.HasBlobTypeStorage() && jFirstColEntity.HasBlobTypeStorage() {
-			return true
+		return false
+	})
+	return NewIndexDefinitionEntityList(uniqueKeys)
+}
+
+// IsValidPKEquivalent returns true if the key is a valid Primary Key Equivalent:
+// a unique, non-nullable key. Unlike IsValidIterationKey, this does not exclude
+// floating point types or prefix keys.
+func IsValidPKEquivalent(key *IndexDefinitionEntity) bool {
+	if key == nil {
+		return false
+	}
+	if !key.IsUnique() {
+		return false
+	}
+	if key.HasNullable() {
+		return false
+	}
+	return true
+}
+
+// GetPrimaryKeyEquivalent returns the lowest-cost valid PKE in a CREATE TABLE statement.
+// Exact (type cost, column count) ties are broken by the lexicographically smallest
+// index name so that the selection is deterministic across restarts and upgrades.
+func GetPrimaryKeyEquivalent(createTableEntity *CreateTableEntity) (columns []string, indexName string) {
+	var bestKey *IndexDefinitionEntity
+	for _, key := range createTableEntity.IndexDefinitionEntities() {
+		if key.HasExpression() {
+			continue
 		}
-		if iFirstColEntity.HasBlobTypeStorage() && !jFirstColEntity.HasBlobTypeStorage() {
-			return false
+		if !IsValidPKEquivalent(key) {
+			continue
 		}
-		if !iFirstColEntity.IsTextual() && jFirstColEntity.IsTextual() {
-			return true
+		if bestKey == nil {
+			bestKey = key
+			continue
 		}
-		if iFirstColEntity.IsTextual() && !jFirstColEntity.IsTextual() {
-			return false
+		if costDiff := key.TypeCost() - bestKey.TypeCost(); costDiff != 0 {
+			if costDiff < 0 {
+				bestKey = key
+			}
+			continue
 		}
-		if storageDiff := IntegralTypeStorage(iFirstColEntity.Type()) - IntegralTypeStorage(jFirstColEntity.Type()); storageDiff != 0 {
-			return storageDiff < 0
+		if lenDiff := len(key.ColumnList.Entities) - len(bestKey.ColumnList.Entities); lenDiff != 0 {
+			if lenDiff < 0 {
+				bestKey = key
+			}
+			continue
 		}
-		if lenDiff := len(uniqueKeys[i].ColumnList.Entities) - len(uniqueKeys[j].ColumnList.Entities); lenDiff != 0 {
-			return lenDiff < 0
+		if key.Name() < bestKey.Name() {
+			bestKey = key
 		}
-		return false
-	})
-	return NewIndexDefinitionEntityList(uniqueKeys)
+	}
+	if bestKey != nil {
+		return bestKey.ColumnNames(), bestKey.Name()
+	}
+	return nil, ""
 }
 
 // RemovedForeignKeyNames returns the names of removed foreign keys, ignoring mere name changes