Merge #153986

153986: sql/schemachanger: detect secondary and pk index name collisions r=fqazi a=fqazi

Previously, the declarative schema changer did not properly check if secondary index names and primary index names collided. This meant that instead of getting a query at statement time, one could be hit at runtime instead. To address this, this patch checks if either secondary or primary indexes with the same name exist when creating a new secondary index.

Fixes: #153702

Release note (bug fix): Addressed a runtime error that could be hit if a new secondary index had a name collision with a primary index.

Co-authored-by: Faizan Qazi <[email protected]>

Author: craig[bot]

pkg/sql/logictest/testdata/logic_test/create_index

@@ -694,3 +694,14 @@ tbl_with_collate  CREATE TABLE public.tbl_with_collate (
                   ) WITH (schema_locked = true);
 
 subtest end
+
+
+subtest index_name_conflicts_with_primary_key
+
+statement ok
+CREATE TABLE t_index_name_conflicts_with_primary_key (a INT PRIMARY KEY, b INT);
+
+statement error pgcode 42P07 index with name \"t_index_name_conflicts_with_primary_key_pkey\" already exists
+CREATE INDEX t_index_name_conflicts_with_primary_key_pkey ON t_index_name_conflicts_with_primary_key (a);
+
+subtest end

pkg/sql/schemachanger/scbuild/internal/scbuildstmt/create_index.go

@@ -119,20 +119,34 @@ func CreateIndex(b BuildCtx, n *tree.CreateIndex) {
 			"%q is not a table or materialized view", n.Table.ObjectName))
 	}
 	// Resolve the index name and make sure it doesn't exist yet.
-	{
+	if len(n.Name) > 0 {
 		indexElements := b.ResolveIndex(idxSpec.secondary.TableID, n.Name, ResolveParams{
 			IsExistenceOptional: true,
 			RequiredPrivilege:   privilege.CREATE,
 		})
-		if _, target, sec := scpb.FindSecondaryIndex(indexElements); sec != nil {
+		skipCreation := false
+		indexElements.ForEach(func(_ scpb.Status, target scpb.TargetStatus, e scpb.Element) {
+			switch e.(type) {
+			// Names can conflict on either primary or secondary indexes.
+			case *scpb.PrimaryIndex:
+			case *scpb.SecondaryIndex:
+			default:
+				// No index element that we care about.
+				return
+			}
 			if n.IfNotExists {
+				skipCreation = true
 				return
 			}
 			if target == scpb.ToAbsent {
 				panic(pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
 					"index %q being dropped, try again later", n.Name.String()))
 			}
 			panic(pgerror.Newf(pgcode.DuplicateRelation, "index with name %q already exists", n.Name))
+		})
+		if skipCreation {
+			return
+
 		}
 	}
 	if _, _, tbl := scpb.FindTable(relationElements); tbl != nil {