scbuild: support DROP TYPE ... CASCADE

Recently, we've added support for constraint removal in the declarative
schema changer. This now makes it possible to support DROP TYPE ...
CASCADE statements. As a result, DROP OWNED BY now also performs
correctly when user-defined types are involved and no longer returns an
error.

This commit also extends the coverage of the declarative schema
changer's ALTER TABLE ... DROP COLUMN support, which no longer punts to
the legacy schema changer when the column to drop is referenced in
a constraint. Now, the constraint gets dropped.

The declarative schema changer still differs from Postgres behaviour
when the column to drop is part of the primary key. Whereas postgres
happily removes the primary key constraint and the column, CRDB is
unable to do so straigtforwardly. What to do here remains an open
question, there are many possible solutions, in the meantime we return
an error.

Fixes cockroachdb#51480.
Fixes cockroachdb#55908.

Release note (sql change): DROP TYPE ... CASCADE is now supported and no
longer returns an error. Consequently DROP OWNED BY no longer returns an
error when it tries to drop a type in a cascading manner.

pkg/sql/catalog/rewrite/BUILD.bazel

@@ -20,7 +20,6 @@ go_library(
         "//pkg/sql/parser",
         "//pkg/sql/pgwire/pgcode",
         "//pkg/sql/pgwire/pgerror",
-        "//pkg/sql/schemachanger/scpb",
         "//pkg/sql/schemachanger/screl",
         "//pkg/sql/sem/catid",
         "//pkg/sql/sem/tree",

pkg/sql/catalog/rewrite/rewrite.go

@@ -27,7 +27,6 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/parser"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
-	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/scpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/schemachanger/screl"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/catid"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
@@ -572,32 +571,38 @@ func rewriteSchemaChangerState(
 			err = errors.Wrap(err, "rewriting declarative schema changer state")
 		}
 	}()
+	selfRewrite := descriptorRewrites[d.GetID()]
 	for i := 0; i < len(state.Targets); i++ {
 		t := &state.Targets[i]
 		if err := screl.WalkDescIDs(t.Element(), func(id *descpb.ID) error {
-			if *id == descpb.InvalidID {
+			switch *id {
+			case descpb.InvalidID:
 				// Some descriptor ID fields in elements may be deliberately unset.
 				// Skip these as they are not subject to rewrite.
 				return nil
+			case d.GetID():
+				*id = selfRewrite.ID
+				return nil
+			case d.GetParentSchemaID():
+				*id = selfRewrite.ParentSchemaID
+				return nil
+			case d.GetParentID():
+				*id = selfRewrite.ParentID
+				return nil
 			}
-			rewrite, ok := descriptorRewrites[*id]
-			if !ok {
-				return errors.Errorf("missing rewrite for id %d in %s", *id, screl.ElementString(t.Element()))
+			if rewrite, ok := descriptorRewrites[*id]; ok {
+				*id = rewrite.ID
+				return nil
 			}
-			*id = rewrite.ID
-			return nil
+			return errors.Errorf("missing rewrite for id %d in %s", *id, screl.ElementString(t.Element()))
 		}); err != nil {
-			// We'll permit this in the special case of a schema parent element.
-			switch el := t.Element().(type) {
-			case *scpb.SchemaParent:
-				_, scExists := descriptorRewrites[el.SchemaID]
-				if !scExists && state.CurrentStatuses[i] == scpb.Status_ABSENT {
-					state.Targets = append(state.Targets[:i], state.Targets[i+1:]...)
-					state.CurrentStatuses = append(state.CurrentStatuses[:i], state.CurrentStatuses[i+1:]...)
-					state.TargetRanks = append(state.TargetRanks[:i], state.TargetRanks[i+1:]...)
-					i--
-					continue
-				}
+			// We'll permit this in the special case of a satisfied target.
+			if t.TargetStatus == state.CurrentStatuses[i] {
+				state.Targets = append(state.Targets[:i], state.Targets[i+1:]...)
+				state.CurrentStatuses = append(state.CurrentStatuses[:i], state.CurrentStatuses[i+1:]...)
+				state.TargetRanks = append(state.TargetRanks[:i], state.TargetRanks[i+1:]...)
+				i--
+				continue
 			}
 			return errors.Wrap(err, "rewriting descriptor ids")
 		}

pkg/sql/catalog/tabledesc/structured.go

@@ -276,6 +276,12 @@ func ForEachExprStringInTableDesc(descI catalog.TableDescriptor, f func(expr *st
 	doCheck := func(c *descpb.TableDescriptor_CheckConstraint) error {
 		return f(&c.Expr)
 	}
+	doUniqueWithoutIndex := func(c *descpb.UniqueWithoutIndexConstraint) error {
+		if !c.IsPartial() {
+			return nil
+		}
+		return f(&c.Predicate)
+	}
 
 	// Process columns.
 	for i := range desc.Columns {
@@ -300,17 +306,30 @@ func ForEachExprStringInTableDesc(descI catalog.TableDescriptor, f func(expr *st
 		}
 	}
 
+	// Process unique without index constraints.
+	for i := range desc.UniqueWithoutIndexConstraints {
+		if err := doUniqueWithoutIndex(&desc.UniqueWithoutIndexConstraints[i]); err != nil {
+			return err
+		}
+	}
+
 	// Process all non-index mutations.
 	for _, mut := range desc.Mutations {
 		if c := mut.GetColumn(); c != nil {
 			if err := doCol(c); err != nil {
 				return err
 			}
 		}
-		if c := mut.GetConstraint(); c != nil &&
-			c.ConstraintType == descpb.ConstraintToUpdate_CHECK {
-			if err := doCheck(&c.Check); err != nil {
-				return err
+		if c := mut.GetConstraint(); c != nil {
+			switch c.ConstraintType {
+			case descpb.ConstraintToUpdate_CHECK:
+				if err := doCheck(&c.Check); err != nil {
+					return err
+				}
+			case descpb.ConstraintToUpdate_UNIQUE_WITHOUT_INDEX:
+				if err := doUniqueWithoutIndex(&c.UniqueWithoutIndexConstraint); err != nil {
+					return err
+				}
 			}
 		}
 	}

pkg/sql/logictest/testdata/logic_test/alter_table

@@ -1742,7 +1742,7 @@ uwi_child  uwi_child_d_fkey  FOREIGN KEY  FOREIGN KEY (d) REFERENCES unique_with
 uwi_child  uwi_child_pkey    PRIMARY KEY  PRIMARY KEY (rowid ASC)                                   true
 
 # Attempting to drop a column with a foreign key reference fails.
-statement error pq: "unique_d" is referenced by foreign key from table "uwi_child"
+statement error pgcode 2BP01 is referenced by foreign key from table "uwi_child"
 ALTER TABLE unique_without_index DROP COLUMN d
 
 # It succeeds if we use CASCADE, and also drops the fk reference.

pkg/sql/logictest/testdata/logic_test/drop_type

@@ -449,7 +449,9 @@ CREATE TABLE t (k schema_to_drop.typ PRIMARY KEY);
 statement ok
 CREATE TABLE schema_to_drop.t (k schema_to_drop.typ PRIMARY KEY);
 
-statement error pgcode 0A000 unimplemented: cannot drop type "test.schema_to_drop.(_)?typ" because other objects \(\[test\.public\.t\]\) still depend on it
+skipif config local-legacy-schema-changer
+skipif config local-mixed-22.2-23.1
+statement error pgcode 42P10 column "k" is referenced by the primary key
 DROP SCHEMA schema_to_drop CASCADE;
 
 statement ok

pkg/sql/logictest/testdata/logic_test/new_schema_changer

@@ -364,6 +364,20 @@ CREATE VIEW v7_dep AS (SELECT i FROM t6@idx WHERE k < 'a'::typ6)
 statement error annot drop type "typ6" because other objects \(\[test.public.t6 test.public.v7_dep\]\) still depend on it
 DROP TYPE typ6
 
+statement ok
+DROP TYPE typ6 CASCADE
+
+statement ok
+DROP TYPE typ5 CASCADE
+
+statement ok
+DROP TYPE typ4 CASCADE
+
+statement ok
+DROP TYPE typ3 CASCADE
+
+statement ok
+DROP TYPE typ CASCADE
 
 subtest view_sanity
 

pkg/sql/pgwire/testdata/pgtest/notice

@@ -55,7 +55,7 @@ Query {"String": "DROP INDEX t_x_idx"}
 until crdb_only
 CommandComplete
 ----
-{"Severity":"NOTICE","SeverityUnlocalized":"NOTICE","Code":"00000","Message":"the data for dropped indexes is reclaimed asynchronously","Detail":"","Hint":"The reclamation delay can be customized in the zone configuration for the table.","Position":0,"InternalPosition":0,"InternalQuery":"","Where":"","SchemaName":"","TableName":"","ColumnName":"","DataTypeName":"","ConstraintName":"","File":"drop_index.go","Line":66,"Routine":"DropIndex","UnknownFields":null}
+{"Severity":"NOTICE","SeverityUnlocalized":"NOTICE","Code":"00000","Message":"the data for dropped indexes is reclaimed asynchronously","Detail":"","Hint":"The reclamation delay can be customized in the zone configuration for the table.","Position":0,"InternalPosition":0,"InternalQuery":"","Where":"","SchemaName":"","TableName":"","ColumnName":"","DataTypeName":"","ConstraintName":"","File":"drop_index.go","Line":65,"Routine":"DropIndex","UnknownFields":null}
 {"Type":"CommandComplete","CommandTag":"DROP INDEX"}
 
 until noncrdb_only

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

@@ -180,7 +180,7 @@ func alterPrimaryKey(b BuildCtx, tn *tree.TableName, tbl *scpb.Table, t alterPri
 	// Drop the rowid column, if applicable.
 	if rowidToDrop != nil {
 		elts := b.QueryByID(rowidToDrop.TableID).Filter(hasColumnIDAttrFilter(rowidToDrop.ColumnID))
-		dropColumn(b, tn, tbl, t.n, rowidToDrop, elts, tree.DropRestrict)
+		dropColumn(b, t.n, tbl, rowidToDrop, elts, tree.DropRestrict)
 	}
 
 	// Construct and add elements for a unique secondary index created on

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

@@ -40,7 +40,7 @@ func alterTableDropColumn(
 	}
 	checkRowLevelTTLColumn(b, tn, tbl, n, col)
 	checkColumnNotInaccessible(col, n)
-	dropColumn(b, tn, tbl, n, col, elts, n.DropBehavior)
+	dropColumn(b, n, tbl, col, elts, n.DropBehavior)
 	b.LogEventForExistingTarget(col)
 }
 
@@ -173,9 +173,8 @@ func checkColumnNotInaccessible(col *scpb.Column, n *tree.AlterTableDropColumn)
 
 func dropColumn(
 	b BuildCtx,
-	tn *tree.TableName,
-	tbl *scpb.Table,
 	n tree.NodeFormatter,
+	tbl *scpb.Table,
 	col *scpb.Column,
 	colElts ElementResultSet,
 	behavior tree.DropBehavior,
@@ -194,7 +193,7 @@ func dropColumn(
 				_, _, computedColName := scpb.FindColumnName(elts.Filter(publicTargetFilter))
 				panic(sqlerrors.NewColumnReferencedByComputedColumnError(cn.Name, computedColName.Name))
 			}
-			dropColumn(b, tn, tbl, n, e, elts, behavior)
+			dropColumn(b, n, tbl, e, elts, behavior)
 		case *scpb.PrimaryIndex:
 			tableElts := b.QueryByID(e.TableID).Filter(publicTargetFilter)
 			scpb.ForEachIndexColumn(tableElts, func(_ scpb.Status, _ scpb.TargetStatus, ic *scpb.IndexColumn) {
@@ -205,32 +204,17 @@ func dropColumn(
 				}
 			})
 		case *scpb.SecondaryIndex:
-			indexElts := b.QueryByID(e.TableID).Filter(hasIndexIDAttrFilter(e.IndexID))
-			_, _, indexName := scpb.FindIndexName(indexElts.Filter(publicTargetFilter))
-			name := tree.TableIndexName{
-				Table: *tn,
-				Index: tree.UnrestrictedName(indexName.Name),
-			}
-			dropSecondaryIndex(b, &name, behavior, e)
+			dropSecondaryIndex(b, n, behavior, e)
 		case *scpb.UniqueWithoutIndexConstraint:
-			// TODO(ajwerner): Support dropping UNIQUE WITHOUT INDEX constraints.
-			panic(errors.Wrap(scerrors.NotImplementedError(n),
-				"dropping of UNIQUE WITHOUT INDEX constraints not supported"))
+			dropConstraintByID(b, e.TableID, e.ConstraintID)
 		case *scpb.CheckConstraint:
-			// TODO(ajwerner): Support dropping CHECK constraints.
-			// We might need to extend and add check constraint to dep-rule
-			// "column constraint removed right before column reaches delete only"
-			// in addition to just `b.Drop(e)`. Read its comment for more details.
-			panic(errors.Wrap(scerrors.NotImplementedError(n),
-				"dropping of CHECK constraints not supported"))
+			dropConstraintByID(b, e.TableID, e.ConstraintID)
 		case *scpb.ForeignKeyConstraint:
-			if e.TableID != col.TableID && behavior != tree.DropCascade {
-				panic(pgerror.Newf(pgcode.DependentObjectsStillExist,
-					"cannot drop column %s because other objects depend on it", cn.Name))
+			if col.TableID != e.TableID && behavior != tree.DropCascade {
+				tn := simpleName(b, e.TableID)
+				panic(sqlerrors.NewUniqueConstraintReferencedByForeignKeyError(cn.Name, tn))
 			}
-			// TODO(ajwerner): Support dropping FOREIGN KEY constraints.
-			panic(errors.Wrap(scerrors.NotImplementedError(n),
-				"dropping of FOREIGN KEY constraints not supported"))
+			dropConstraintByID(b, e.TableID, e.ConstraintID)
 		case *scpb.View:
 			if behavior != tree.DropCascade {
 				_, _, ns := scpb.FindNamespace(b.QueryByID(col.TableID))
@@ -245,7 +229,7 @@ func dropColumn(
 					"cannot drop column %q because view %q depends on it",
 					cn.Name, nsDep.Name))
 			}
-			dropCascadeDescriptor(b, e.ViewID)
+			dropCascadeDescriptor(b, n, e.ViewID)
 		case *scpb.Sequence:
 			// Find all the sequences owned by this column and drop them either restrict
 			// or cascade. Then, we'll need to check whether these sequences have any
@@ -259,7 +243,7 @@ func dropColumn(
 			//  2BP01: cannot drop column i of table t because other objects depend on it
 			//
 			if behavior == tree.DropCascade {
-				dropCascadeDescriptor(b, e.SequenceID)
+				dropCascadeDescriptor(b, n, e.SequenceID)
 			} else {
 				dropRestrictDescriptor(b, e.SequenceID)
 				undroppedSeqBackrefsToCheck.Add(e.SequenceID)
@@ -272,7 +256,7 @@ func dropColumn(
 					cn.Name, fnName.Name),
 				)
 			}
-			dropCascadeDescriptor(b, e.FunctionID)
+			dropCascadeDescriptor(b, n, e.FunctionID)
 		default:
 			b.Drop(e)
 		}

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

@@ -43,11 +43,11 @@ func alterTableDropConstraint(
 	// Dropping UNIQUE constraint: error out as not implemented.
 	droppingUniqueConstraintNotImplemented(constraintElems, t)
 
-	constraintElems.ForEachElementStatus(func(
-		_ scpb.Status, _ scpb.TargetStatus, e scpb.Element,
-	) {
-		b.Drop(e)
-	})
+	constraintElems.Filter(notAbsentTargetFilter).ForEachElementStatus(
+		func(_ scpb.Status, _ scpb.TargetStatus, e scpb.Element) {
+			b.Drop(e)
+		},
+	)
 }
 
 func fallBackIfDroppingPrimaryKey(

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

@@ -35,7 +35,7 @@ func DropDatabase(b BuildCtx, n *tree.DropDatabase) {
 	b.IncrementSchemaChangeDropCounter("database")
 	// Perform explicit or implicit DROP DATABASE CASCADE.
 	if n.DropBehavior == tree.DropCascade || (n.DropBehavior == tree.DropDefault && !b.SessionData().SafeUpdates) {
-		dropCascadeDescriptor(b, db.DatabaseID)
+		dropCascadeDescriptor(b, n, db.DatabaseID)
 		b.LogEventForExistingTarget(db)
 		return
 	}