*: non-unique Global Index on non-clustered removed pid from value, V2

Makefile

@@ -662,6 +662,15 @@ bazel_ci_prepare:
 
 .PHONY: bazel_ci_simple_prepare
 bazel_ci_simple_prepare:
+	@mkdir -p ./bin; \
+	if [ ! -f ./bin/tikv-server.tar.gz ]; then \
+		echo "Downloading tikv-server tarball"; \
+		wget -q -O ./bin/tikv-server.tar.gz "https://fileserver.pingcap.net/download/builds/devbuild/9652/tikv-linux-amd64.tar.gz" || true; \
+	else \
+		echo "tikv-server tarball already exists, skipping download"; \
+	fi; \
+	echo "Extracting tikv-server"; \
+	tar -xzf ./bin/tikv-server.tar.gz -C ./bin/ || true
 	bazel $(BAZEL_GLOBAL_CONFIG) run $(BAZEL_CMD_CONFIG) //:gazelle
 	bazel $(BAZEL_GLOBAL_CONFIG) run $(BAZEL_CMD_CONFIG)  \
 		--run_under="cd $(CURDIR) && " \

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

@@ -410,6 +410,20 @@ func BuildIndexInfo(
 			idxInfo.Tp = indexOption.Tp
 		}
 		idxInfo.Global = indexOption.Global
+		// Set global index version for new global indexes.
+		// Version 2 is needed for non-clustered tables with non-unique global indexes
+		// to prevent collisions after EXCHANGE PARTITION due to duplicate _tidb_rowid values.
+		// V2 encodes partition ID in the key only (not in the value) for storage efficiency.
+		// Clustered tables and unique indexes don't have this issue and use version 0.
+		idxInfo.GlobalIndexVersion = 0
+		if indexOption.Global && !idxInfo.Unique && !tblInfo.HasClusteredIndex() {
+			idxInfo.GlobalIndexVersion = model.GlobalIndexVersionV2
+			failpoint.Inject("SetGlobalIndexVersion", func(val failpoint.Value) {
+				if valInt, ok := val.(int); ok {
+					idxInfo.GlobalIndexVersion = uint8(valInt)
+				}
+			})
+		}
 
 		conditionString, err := CheckAndBuildIndexConditionString(tblInfo, indexOption.Condition)
 		if err != nil {
@@ -2454,7 +2468,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

@@ -624,8 +624,18 @@ func buildTablePartitionInfo(ctx *metabuild.Context, s *ast.PartitionOptions, tb
 				return dbterror.ErrWrongNameForIndex.GenWithStackByArgs(idxUpdate.Name)
 			}
 			dupCheck[strings.ToLower(idxUpdate.Name)] = struct{}{}
+			tbInfo.Indices[idxOffset].GlobalIndexVersion = 0
 			if idxUpdate.Option != nil && idxUpdate.Option.Global {
 				tbInfo.Indices[idxOffset].Global = true
+				// Only use V2 for non-clustered tables with non-unique global indexes
+				if !tbInfo.Indices[idxOffset].Unique && !tbInfo.HasClusteredIndex() {
+					tbInfo.Indices[idxOffset].GlobalIndexVersion = model.GlobalIndexVersionV2
+					failpoint.Inject("SetGlobalIndexVersion", func(val failpoint.Value) {
+						if valInt, ok := val.(int); ok {
+							tbInfo.Indices[idxOffset].GlobalIndexVersion = uint8(valInt)
+						}
+					})
+				}
 			} else {
 				tbInfo.Indices[idxOffset].Global = false
 			}
@@ -3293,6 +3303,16 @@ 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
+			newIndex.GlobalIndexVersion = 0
+			if newGlobal && !newIndex.Unique && !tblInfo.HasClusteredIndex() {
+				// Only use V2 for non-clustered tables with non-unique global indexes
+				newIndex.GlobalIndexVersion = model.GlobalIndexVersionV2
+				failpoint.Inject("SetGlobalIndexVersion", func(val failpoint.Value) {
+					if valInt, ok := val.(int); ok {
+						newIndex.GlobalIndexVersion = uint8(valInt)
+					}
+				})
+			}
 			tblInfo.Indices = append(tblInfo.Indices, newIndex)
 		}
 		failpoint.Inject("reorgPartCancel1", func(val failpoint.Value) {

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",