table: Non-clustered table non-unique global index needs partid in key, v1

pkg/ddl/create_table.go

@@ -1328,6 +1328,25 @@ func BuildTableInfo(
 		existedColsMap[v.Name.L] = struct{}{}
 	}
 	foreignKeyID := tbInfo.MaxForeignKeyID
+
+	// This pre-scan is necessary because index building in the main loop may need to check
+	// tbInfo.HasClusteredIndex() before the PRIMARY KEY constraint is fully processed.
+	for _, constr := range constraints {
+		if constr.Tp == ast.ConstraintPrimaryKey {
+			isSingleIntPK := isSingleIntPKFromTableInfo(constr, tbInfo)
+
+			if ShouldBuildClusteredIndex(ctx.GetClusteredIndexDefMode(), constr.Option, isSingleIntPK) {
+				if isSingleIntPK {
+					tbInfo.PKIsHandle = true
+				} else {
+					tbInfo.IsCommonHandle = true
+					tbInfo.CommonHandleVersion = 1
+				}
+			}
+			break // Only one PRIMARY KEY possible
+		}
+	}
+
 	for _, constr := range constraints {
 		var hiddenCols []*model.ColumnInfo
 		if constr.Tp != ast.ConstraintColumnar {
@@ -1380,7 +1399,7 @@ func BuildTableInfo(
 				// all partial primary key.
 				return nil, dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs("create an primary key with partial index is not supported")
 			}
-			isSingleIntPK := isSingleIntPK(constr, lastCol)
+			isSingleIntPK := isSingleIntPKFromCol(constr, lastCol)
 			if ShouldBuildClusteredIndex(ctx.GetClusteredIndexDefMode(), constr.Option, isSingleIntPK) {
 				if isSingleIntPK {
 					tbInfo.PKIsHandle = true
@@ -1676,18 +1695,46 @@ func addIndexForForeignKey(ctx *metabuild.Context, tbInfo *model.TableInfo) erro
 	return nil
 }
 
-func isSingleIntPK(constr *ast.Constraint, lastCol *model.ColumnInfo) bool {
-	if len(constr.Keys) != 1 {
-		return false
-	}
-	switch lastCol.GetType() {
+func isIntCol(col *model.ColumnInfo) bool {
+	switch col.GetType() {
 	case mysql.TypeLong, mysql.TypeLonglong,
 		mysql.TypeTiny, mysql.TypeShort, mysql.TypeInt24:
 		return true
 	}
 	return false
 }
 
+// isSingleIntPKFromTableInfo determines if a constraint represents a single integer primary key
+// by looking up the column information from the table info. This is used during the pre-scan
+// phase before CheckPKOnGeneratedColumn is called.
+func isSingleIntPKFromTableInfo(constr *ast.Constraint, tbInfo *model.TableInfo) bool {
+	// Multi-column PKs are not single integer PKs
+	if len(constr.Keys) != 1 {
+		return false
+	}
+
+	// Expression-based PKs (e.g., PRIMARY KEY ((col+1))) are not single integer PKs
+	if constr.Keys[0].Expr != nil {
+		return false
+	}
+
+	// Find the column in the table
+	colName := constr.Keys[0].Column.Name.L
+	for _, col := range tbInfo.Columns {
+		if col.Name.L == colName {
+			return isIntCol(col)
+		}
+	}
+	return false
+}
+
+func isSingleIntPKFromCol(constr *ast.Constraint, lastCol *model.ColumnInfo) bool {
+	if len(constr.Keys) != 1 {
+		return false
+	}
+	return isIntCol(lastCol)
+}
+
 // ShouldBuildClusteredIndex is used to determine whether the CREATE TABLE statement should build a clustered index table.
 func ShouldBuildClusteredIndex(mode vardef.ClusteredIndexDefMode, opt *ast.IndexOption, isSingleIntPK bool) bool {
 	if opt == nil || opt.PrimaryKeyTp == ast.PrimaryKeyTypeDefault {

pkg/ddl/index.go

@@ -342,6 +342,36 @@ func getIndexColumnLength(col *model.ColumnInfo, colLen int, columnarIndexType m
 	}
 }
 
+// Set global index version for new global indexes.
+// Version 1 is needed for non-clustered tables to prevent collisions after
+// EXCHANGE PARTITION due to duplicate _tidb_rowid values.
+// For non-unique indexes, the handle is always encoded in the key.
+// For unique indexes with NULL values, the handle is also encoded in the key
+// (since NULL != NULL, multiple NULLs are allowed).
+// In both cases, we need the partition ID in the key to distinguish rows
+// from different partitions that may have the same _tidb_rowid.
+// Clustered tables don't have this issue and use version 0.
+func setGlobalIndexVersion(tblInfo *model.TableInfo, idxInfo *model.IndexInfo) {
+	idxInfo.GlobalIndexVersion = 0
+	if idxInfo.Global && !tblInfo.HasClusteredIndex() {
+		needPartitionInKey := !idxInfo.Unique
+		if !needPartitionInKey {
+			nullCols := getNullColInfos(tblInfo, idxInfo.Columns)
+			if len(nullCols) > 0 {
+				needPartitionInKey = true
+			}
+		}
+		if needPartitionInKey {
+			idxInfo.GlobalIndexVersion = model.GlobalIndexVersionV1
+			failpoint.Inject("SetGlobalIndexVersion", func(val failpoint.Value) {
+				if valInt, ok := val.(int); ok {
+					idxInfo.GlobalIndexVersion = uint8(valInt)
+				}
+			})
+		}
+	}
+}
+
 // decimal using a binary format that packs nine decimal (base 10) digits into four bytes.
 func calcBytesLengthForDecimal(m int) int {
 	return (m / 9 * 4) + ((m%9)+1)/2
@@ -410,6 +440,7 @@ func BuildIndexInfo(
 			idxInfo.Tp = indexOption.Tp
 		}
 		idxInfo.Global = indexOption.Global
+		setGlobalIndexVersion(tblInfo, idxInfo)
 
 		conditionString, err := CheckAndBuildIndexConditionString(tblInfo, indexOption.Condition)
 		if err != nil {
@@ -697,15 +728,15 @@ func setIndexVisibility(tblInfo *model.TableInfo, name ast.CIStr, invisible bool
 	}
 }
 
-func getNullColInfos(tblInfo *model.TableInfo, indexInfo *model.IndexInfo) ([]*model.ColumnInfo, error) {
-	nullCols := make([]*model.ColumnInfo, 0, len(indexInfo.Columns))
-	for _, colName := range indexInfo.Columns {
+func getNullColInfos(tblInfo *model.TableInfo, cols []*model.IndexColumn) []*model.ColumnInfo {
+	nullCols := make([]*model.ColumnInfo, 0, len(cols))
+	for _, colName := range cols {
 		col := model.FindColumnInfo(tblInfo.Columns, colName.Name.L)
 		if !mysql.HasNotNullFlag(col.GetFlag()) || mysql.HasPreventNullInsertFlag(col.GetFlag()) {
 			nullCols = append(nullCols, col)
 		}
 	}
-	return nullCols, nil
+	return nullCols
 }
 
 func checkPrimaryKeyNotNull(jobCtx *jobContext, w *worker, job *model.Job,
@@ -718,10 +749,7 @@ func checkPrimaryKeyNotNull(jobCtx *jobContext, w *worker, job *model.Job,
 	if err != nil {
 		return nil, err
 	}
-	nullCols, err := getNullColInfos(tblInfo, indexInfo)
-	if err != nil {
-		return nil, err
-	}
+	nullCols := getNullColInfos(tblInfo, indexInfo.Columns)
 	if len(nullCols) == 0 {
 		return nil, nil
 	}
@@ -2454,7 +2482,16 @@ func (w *baseIndexWorker) fetchRowColVals(txn kv.Transaction, taskRange reorgBac
 				if index.Meta().HasCondition() {
 					return false, dbterror.ErrUnsupportedAddPartialIndex.GenWithStackByArgs("add partial index without fast reorg")
 				}
-				idxRecord, err1 := w.getIndexRecord(index.Meta(), handle, recordKey)
+				actualHandle := handle
+				// For global indexes V1+ on partitioned tables, we need to wrap the handle
+				// with the partition ID to create a PartitionHandle.
+				// This is critical for non-clustered tables after EXCHANGE PARTITION,
+				// where duplicate _tidb_rowid values exist across partitions.
+				// Legacy indexes (version 0) don't use PartitionHandle in the key.
+				if index.Meta().Global && index.Meta().GlobalIndexVersion >= model.GlobalIndexVersionV1 {
+					actualHandle = kv.NewPartitionHandle(taskRange.physicalTable.GetPhysicalID(), handle)
+				}
+				idxRecord, err1 := w.getIndexRecord(index.Meta(), actualHandle, recordKey)
 				if err1 != nil {
 					return false, errors.Trace(err1)
 				}

pkg/ddl/partition.go

@@ -626,6 +626,7 @@ func buildTablePartitionInfo(ctx *metabuild.Context, s *ast.PartitionOptions, tb
 			dupCheck[strings.ToLower(idxUpdate.Name)] = struct{}{}
 			if idxUpdate.Option != nil && idxUpdate.Option.Global {
 				tbInfo.Indices[idxOffset].Global = true
+				setGlobalIndexVersion(tbInfo, tbInfo.Indices[idxOffset])
 			} else {
 				tbInfo.Indices[idxOffset].Global = false
 			}
@@ -3293,6 +3294,7 @@ func (w *worker) onReorganizePartition(jobCtx *jobContext, job *model.Job) (ver
 			tblInfo.Partition.DDLChangedIndex[index.ID] = false
 			tblInfo.Partition.DDLChangedIndex[newIndex.ID] = true
 			newIndex.Global = newGlobal
+			setGlobalIndexVersion(tblInfo, newIndex)
 			tblInfo.Indices = append(tblInfo.Indices, newIndex)
 		}
 		failpoint.Inject("reorgPartCancel1", func(val failpoint.Value) {

pkg/ddl/rollingback.go

@@ -29,10 +29,7 @@ import (
 
 // UpdateColsNull2NotNull changes the null option of columns of an index.
 func UpdateColsNull2NotNull(tblInfo *model.TableInfo, indexInfo *model.IndexInfo) error {
-	nullCols, err := getNullColInfos(tblInfo, indexInfo)
-	if err != nil {
-		return errors.Trace(err)
-	}
+	nullCols := getNullColInfos(tblInfo, indexInfo.Columns)
 
 	for _, col := range nullCols {
 		col.AddFlag(mysql.NotNullFlag)
@@ -66,10 +63,7 @@ func convertAddIdxJob2RollbackJob(
 	originalState := allIndexInfos[0].State
 	for _, indexInfo := range allIndexInfos {
 		if indexInfo.Primary {
-			nullCols, err := getNullColInfos(tblInfo, indexInfo)
-			if err != nil {
-				return 0, errors.Trace(err)
-			}
+			nullCols := getNullColInfos(tblInfo, indexInfo.Columns)
 			for _, col := range nullCols {
 				// Field PreventNullInsertFlag flag reset.
 				col.DelFlag(mysql.PreventNullInsertFlag)

pkg/ddl/tests/partition/BUILD.bazel

@@ -7,6 +7,7 @@ go_test(
         "db_partition_test.go",
         "error_injection_test.go",
         "exchange_partition_test.go",
+        "global_index_version_test.go",
         "main_test.go",
         "multi_domain_test.go",
         "placement_test.go",