(1) restore addDisplayNameColumn.

(2) only remove deprecated indexes

(3) debug backfilling pipeline_versions

Signed-off-by: kaikaila <[email protected]>

Author: kaikaila

backend/src/apiserver/client_manager/client_manager.go

@@ -488,9 +488,13 @@ func runLegacyUpgradeFlow(db *gorm.DB, dialect SQLDialect) error {
 		return fmt.Errorf("preflight length check failed: %w", err)
 	}
 
-	// Step 3: drop all indexes and constraints except primary key which blocks shrinking columns
-	if err := DropAllConstraintsAndIndexes(db, dialect.Name); err != nil {
-		return fmt.Errorf("drop constraints/indexes failed: %w", err)
+	// Step 3: drop all foreign key constraints which can block shrinking columns
+	if err := dropAllFKConstraints(db, dialect.Name); err != nil {
+		return fmt.Errorf("drop foreign key constraints failed: %w", err)
+	}
+
+	if err := dropLegacyIndexes(db, dialect); err != nil {
+		return fmt.Errorf("drop legacy indexes failed: %w", err)
 	}
 
 	// Step 4: shrink fields to meet new length constraints
@@ -536,8 +540,21 @@ func runLegacyUpgradeFlow(db *gorm.DB, dialect SQLDialect) error {
 		return fmt.Errorf("failed to backfill experiment UUID in run_details table: %s", err)
 	}
 
-	if err := db.Migrator().AlterColumn(&model.Pipeline{}, "Description"); err != nil {
-		return fmt.Errorf("failed to update pipeline description type. Error: %s", err)
+	// Data backfill in pipelines table for DisplayName column
+	if db.Migrator().HasTable(&model.Pipeline{}) {
+		if !db.Migrator().HasColumn(&model.Pipeline{}, "DisplayName") {
+			if err := addDisplayNameColumn(db, &model.Pipeline{}, dialect); err != nil {
+				return fmt.Errorf("failed to add DisplayName column to the %s table: %w", model.Pipeline{}.TableName(), err)
+			}
+		}
+	}
+	// Data backfill in pipeline_versions table for DisplayName column
+	if db.Migrator().HasTable(&model.PipelineVersion{}) {
+		if !db.Migrator().HasColumn(&model.PipelineVersion{}, "DisplayName") {
+			if err := addDisplayNameColumn(db, &model.PipelineVersion{}, dialect); err != nil {
+				return fmt.Errorf("failed to add DisplayName column to the %s table: %w", model.PipelineVersion{}.TableName(), err)
+			}
+		}
 	}
 
 	return nil
@@ -649,20 +666,22 @@ func FieldMeta(db *gorm.DB, mdl interface{}, field string) (table string, dbCol
 	return stmt.Table, f.DBName, nil
 }
 
-func DropAllConstraintsAndIndexes(db *gorm.DB, driverName string) error {
+// dropAllFKConstraints drops all foreign key constraints from all tables for the given driver.
+// This is the minimal set of DDL needed to safely shrink indexed columns in legacy upgrades.
+func dropAllFKConstraints(db *gorm.DB, driverName string) error {
 	switch driverName {
 	case "mysql":
-		return dropAllMySQLConstraintsAndIndexes(db)
+		return dropAllMySQLFKConstraints(db)
 	case "pgx":
-		// PostgreSQL not yet supported. No-op for now.
+		// PostgreSQL legacy path not implemented; no-op for now.
 		return nil
 	default:
-		return fmt.Errorf("DropAllConstraintsAndIndexes not supported for driver: %s", driverName)
+		return fmt.Errorf("DropAllFKConstraints not supported for driver: %s", driverName)
 	}
 }
 
-// dropAllMySQLConstraintsAndIndexes drops all foreign key constraints, unique constraints (except PRIMARY), and non-primary indexes from all tables in the current MySQL database.
-func dropAllMySQLConstraintsAndIndexes(db *gorm.DB) error {
+// dropAllMySQLFKConstraints drops all foreign key constraints from all tables in the current MySQL database.
+func dropAllMySQLFKConstraints(db *gorm.DB) error {
 	tables := []string{}
 	if err := db.Raw("SHOW TABLES").Scan(&tables).Error; err != nil {
 		return fmt.Errorf("failed to list tables: %w", err)
@@ -695,57 +714,132 @@ func dropAllMySQLConstraintsAndIndexes(db *gorm.DB) error {
 			}
 		}
 	}
+	return nil
+}
 
-	// Drop UNIQUE constraints except PRIMARY KEY
-	rows2, err := db.Raw(`
-		SELECT constraint_name, table_name
-		FROM information_schema.table_constraints
-		WHERE constraint_schema = DATABASE()
-		  AND constraint_type = 'UNIQUE'
-		  AND constraint_name != 'PRIMARY'
-	`).Rows()
-	if err != nil {
-		return fmt.Errorf("failed to list unique constraints: %w", err)
+// dropLegacyIndexes removes a small, explicit set of legacy indexes that
+// conflict/duplicate with GORM tag definitions. MySQL only; PostgreSQL is no-op.
+func dropLegacyIndexes(db *gorm.DB, dialect SQLDialect) error {
+	switch dialect.Name {
+	case "mysql":
+		return dropLegacyIndexesMySQL(db)
+	case "pgx":
+		// No legacy cleanup needed for PostgreSQL per upstream note.
+		return nil
+	default:
+		return fmt.Errorf("dropLegacyIndexes: unsupported dialect %q", dialect.Name)
 	}
-	defer rows2.Close()
+}
 
-	for rows2.Next() {
-		var constraintName, tableName string
-		if err := rows2.Scan(&constraintName, &tableName); err != nil {
-			return fmt.Errorf("failed to scan unique constraint row: %w", err)
+// dropLegacyIndexesMySQL removes a small, explicit set of legacy indexes on MySQL:
+// See https://github.com/kubeflow/pipelines/pull/12013
+// The function is idempotent: it queries information_schema/SHOW INDEX first and only drops/renames when needed.
+func dropLegacyIndexesMySQL(db *gorm.DB) error {
+	ixLogPrefix := "[legacy-index-cleanup] "
+
+	type idxRow struct {
+		IndexName string `gorm:"column:INDEX_NAME"`
+	}
+
+	// --- Rules 1 & 2: drop single-column unique index on Name for experiments/pipelines ---
+	for _, table := range []string{"experiments", "pipelines"} {
+		var rows []idxRow
+		// Find indexes that are UNIQUE, non-PRIMARY, and exactly one column which is Name.
+		q := fmt.Sprintf(`
+			SELECT INDEX_NAME
+			FROM information_schema.STATISTICS
+			WHERE TABLE_SCHEMA = DATABASE()
+			  AND TABLE_NAME = '%s'
+			GROUP BY INDEX_NAME
+			HAVING
+			  COUNT(*) = 1
+			  AND SUM(CASE WHEN COLUMN_NAME='Name' THEN 1 ELSE 0 END) = 1
+			  AND MAX(NON_UNIQUE) = 0
+			  AND MAX(INDEX_NAME <> 'PRIMARY') = 1;
+		`, table)
+
+		if err := db.Raw(q).Scan(&rows).Error; err != nil {
+			return fmt.Errorf("query single-column unique Name indexes on %s failed: %w", table, err)
 		}
-		dropStmt := fmt.Sprintf("ALTER TABLE `%s` DROP INDEX `%s`", tableName, constraintName)
-		glog.Infof("Dropping unique constraint: %s", dropStmt)
-		if err := db.Exec(dropStmt).Error; err != nil {
-			return fmt.Errorf("failed to drop unique constraint %s on table %s: %w", constraintName, tableName, err)
+		for _, r := range rows {
+			glog.Infof("%s dropping single-column unique index %q on %s(Name)", ixLogPrefix, r.IndexName, table)
+			stmt := fmt.Sprintf("DROP INDEX `%s` ON `%s`", r.IndexName, table)
+			if err := db.Exec(stmt).Error; err != nil {
+				return fmt.Errorf("drop index %s on %s failed: %w", r.IndexName, table, err)
+			}
 		}
 	}
 
-	// Drop non-primary indexes
-	for _, table := range tables {
-		var indexes []struct {
-			KeyName string `gorm:"column:Key_name"`
+	// --- Rule 3: drop legacy composite unique index on pipeline_versions if present ---
+	{
+		if db.Migrator().HasTable(&model.PipelineVersion{}) {
+			rows, err := db.Raw(`SHOW INDEX FROM pipeline_versions WHERE Key_name='idx_pipeline_version_uuid_name'`).Rows()
+			if err != nil {
+				return fmt.Errorf("query pipeline_versions indices failed: %w", err)
+			}
+			has := false
+			if err := rows.Err(); err != nil {
+				_ = rows.Close()
+				return fmt.Errorf("iterate pipeline_versions indices failed: %w", err)
+			}
+			if rows.Next() {
+				has = true
+			}
+			_ = rows.Close()
+			if has {
+				glog.Infof("%s dropping legacy composite unique index idx_pipeline_version_uuid_name on pipeline_versions", ixLogPrefix)
+				if err := db.Exec("DROP INDEX `idx_pipeline_version_uuid_name` ON `pipeline_versions`").Error; err != nil {
+					return fmt.Errorf("drop idx_pipeline_version_uuid_name on pipeline_versions failed: %w", err)
+				}
+			}
+		} else {
+			glog.Infof("%s skip pipeline_versions cleanup: table not found (fresh-introduced case)", ixLogPrefix)
 		}
-		err = db.Raw(fmt.Sprintf("SHOW INDEX FROM `%s`", table)).Scan(&indexes).Error
-		if err != nil {
-			glog.Warningf("failed to list indexes for table %s: %v", table, err)
-			continue
+	}
+
+	// --- Rule 4: normalize pipelines(Name,Namespace) unique index to keep only `namespace_name` ---
+	{
+		var rows []idxRow
+		q := `
+			SELECT INDEX_NAME
+			FROM information_schema.STATISTICS
+			WHERE TABLE_SCHEMA = DATABASE()
+			  AND TABLE_NAME = 'pipelines'
+			GROUP BY INDEX_NAME
+			HAVING
+			  COUNT(*) = 2
+			  AND SUM(CASE WHEN SEQ_IN_INDEX=1 AND COLUMN_NAME='Name' THEN 1 ELSE 0 END) = 1
+			  AND SUM(CASE WHEN SEQ_IN_INDEX=2 AND COLUMN_NAME='Namespace' THEN 1 ELSE 0 END) = 1
+			  AND MAX(NON_UNIQUE) = 0;
+		`
+		if err := db.Raw(q).Scan(&rows).Error; err != nil {
+			return fmt.Errorf("query composite unique (Name,Namespace) indexes on pipelines failed: %w", err)
+		}
+		// Set of existing names for this composite index
+		has := map[string]bool{}
+		for _, r := range rows {
+			has[r.IndexName] = true
 		}
-		seen := make(map[string]bool)
-		for _, idx := range indexes {
-			if idx.KeyName == "PRIMARY" || seen[idx.KeyName] {
-				continue
-			}
-			seen[idx.KeyName] = true
 
-			glog.Infof("Dropping index %s on table %s", idx.KeyName, table)
-			if err := db.Exec(fmt.Sprintf(
-				"DROP INDEX `%s` ON `%s`", idx.KeyName, table,
-			)).Error; err != nil {
-				return fmt.Errorf("failed to drop index %s on table %s: %w", idx.KeyName, table, err)
+		// Case A: both exist -> drop legacy name_namespace_index, keep namespace_name
+		if has["namespace_name"] && has["name_namespace_index"] {
+			glog.Infof("%s dropping duplicate unique index name_namespace_index on pipelines(Name,Namespace), keeping namespace_name", ixLogPrefix)
+			if err := db.Exec("DROP INDEX `name_namespace_index` ON `pipelines`").Error; err != nil {
+				return fmt.Errorf("drop name_namespace_index on pipelines failed: %w", err)
+			}
+		} else if !has["namespace_name"] && has["name_namespace_index"] {
+			// Case B: only legacy exists -> try rename to avoid rebuild; fallback to drop.
+			glog.Infof("%s renaming name_namespace_index -> namespace_name on pipelines(Name,Namespace)", ixLogPrefix)
+			if err := db.Exec("ALTER TABLE `pipelines` RENAME INDEX `name_namespace_index` TO `namespace_name`").Error; err != nil {
+				glog.Warningf("%s rename failed (%v); dropping name_namespace_index; AutoMigrate will recreate namespace_name", ixLogPrefix, err)
+				if err2 := db.Exec("DROP INDEX `name_namespace_index` ON `pipelines`").Error; err2 != nil {
+					return fmt.Errorf("drop name_namespace_index on pipelines failed after rename attempt: %w", err2)
+				}
 			}
 		}
+		// Case C/D: only namespace_name exists or none exist -> no-op
 	}
+
 	return nil
 }
 
@@ -793,6 +887,60 @@ func ensureColumnLength(db *gorm.DB, spec validation.ColLenSpec) error {
 	return nil
 }
 
+// addDisplayNameColumn ensures the DisplayName column exists on the given model's table,
+// backfills it from Name where missing, and then enforces NOT NULL.
+// It is safe to call multiple times (idempotent).
+func addDisplayNameColumn(db *gorm.DB, mdl interface{}, dialect SQLDialect) error {
+
+	table, dbCol, err := FieldMeta(db, mdl, "DisplayName")
+	if err != nil {
+		return fmt.Errorf("resolve meta for %T.DisplayName failed: %w", mdl, err)
+	}
+	// Only allow these tables to have DisplayName added to prevent accidental schema changes.
+	allowed := map[string]bool{
+		model.Pipeline{}.TableName():        true,
+		model.PipelineVersion{}.TableName(): true,
+	}
+	if !allowed[table] {
+		return fmt.Errorf("table %q is not allowed for DisplayName migration", table)
+	}
+
+	if db.Migrator().HasColumn(mdl, dbCol) {
+		return nil
+	}
+
+	q := dialect.QuoteIdentifier
+	quotedTable := q(table)
+	glog.Info("Adding DisplayName column to " + quotedTable)
+
+	return db.Transaction(func(tx *gorm.DB) error {
+		var stmts []string
+		switch dialect.Name {
+		case "mysql":
+			stmts = []string{
+				"ALTER TABLE " + quotedTable + " ADD COLUMN " + q(dbCol) + " VARCHAR(255) NULL;",
+				"UPDATE " + quotedTable + " SET " + q(dbCol) + " = " + q("Name") + " WHERE " + q(dbCol) + " IS NULL;",
+				"ALTER TABLE " + quotedTable + " MODIFY COLUMN " + q(dbCol) + " VARCHAR(255) NOT NULL;",
+			}
+		case "pgx":
+			stmts = []string{
+				"ALTER TABLE " + quotedTable + " ADD COLUMN " + q(dbCol) + " VARCHAR(255);",
+				"UPDATE " + quotedTable + " SET " + q(dbCol) + " = " + q("Name") + " WHERE " + q(dbCol) + " IS NULL;",
+				"ALTER TABLE " + quotedTable + " ALTER COLUMN " + q(dbCol) + " SET NOT NULL;",
+			}
+		default:
+			return fmt.Errorf("unsupported driver: %s", dialect.Name)
+		}
+
+		for _, s := range stmts {
+			if err := tx.Exec(s).Error; err != nil {
+				return fmt.Errorf("exec failed for %q: %w", s, err)
+			}
+		}
+		return nil
+	})
+}
+
 func initMinioClient(ctx context.Context, initConnectionTimeout time.Duration) storage.ObjectStoreInterface {
 	// Create minio client.
 	minioServiceHost := common.GetStringConfigWithDefault(
@@ -878,8 +1026,15 @@ func initPipelineVersionsFromPipelines(db *gorm.DB) {
 	// pipeline version API is introduced will have different Ids; and the minio
 	// file will be put directly into the directories for pipeline versions.
 	tx.Exec(`INSERT INTO
-	pipeline_versions (UUID, Name, CreatedAtInSec, Parameters, Status, PipelineId)
-	SELECT UUID, Name, CreatedAtInSec, Parameters, Status, UUID FROM pipelines;`)
+  pipeline_versions (UUID, Name, CreatedAtInSec, Parameters, Status, PipelineId)
+SELECT
+  p.UUID,
+  p.Name,
+  p.CreatedAtInSec,
+  COALESCE(p.Parameters, '{}') AS Parameters,
+  p.Status,
+  p.UUID AS PipelineId
+FROM pipelines p;`)
 
 	// Step 2: modifiy pipelines table after pipeline_versions are populated.
 	tx.Exec("update pipelines set DefaultVersionId=UUID;")