tabledesc: improve view validation

Previously, we did not validate the DependsOn, DependsOnTyoe and
DependedOnBy cross-references. This commit adds this validation and
also fixes a few bugs:
- privileges were not validated for non-physical tables.
- sequence dependencies were improperly updated by the schema changer.

Unfortunately, there remain a number of unfortunate bugs around the
inconsistent handling of these fields:
- back-references are not systematically present.
- ColumnIDs in DependedOnBy has a different meaning depending on whether
  the table is a sequence or not.

These inconsistencies need to be ironed out by a cluster migration. This
in turn will require changes to the PostRestoreChanges descriptor
builder method. It probably also requires version-gating validation
checks.

Fixes cockroachdb#63147.

Release note: None

pkg/sql/catalog/dbdesc/database_desc.go

@@ -252,7 +252,12 @@ func (desc *immutable) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 	}
 
 	// Validate the privilege descriptor.
-	vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Database))
+	if desc.Privileges == nil {
+		vea.Report(errors.AssertionFailedf("privileges not set"))
+	} else {
+		vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Database))
+	}
+
 	// The DefaultPrivilegeDescriptor may be nil.
 	if desc.GetDefaultPrivileges() != nil {
 		// Validate the default privilege descriptor.

pkg/sql/catalog/descpb/privilege.go

@@ -368,6 +368,10 @@ func (p PrivilegeDescriptor) ValidateSuperuserPrivileges(
 	objectName string,
 	allowedSuperuserPrivileges privilege.List,
 ) error {
+	if parentID == InvalidID && objectType != privilege.Database {
+		// Special case for virtual objects.
+		return nil
+	}
 	for _, user := range []security.SQLUsername{
 		// Check "root" user.
 		security.RootUserName(),

pkg/sql/catalog/descs/validate.go

@@ -104,11 +104,13 @@ func (c collectionBackedDereferencer) DereferenceDescriptors(
 func (c collectionBackedDereferencer) fastDescLookup(
 	ctx context.Context, id descpb.ID,
 ) (catalog.Descriptor, error) {
-	leaseCacheEntry := c.tc.leased.cache.GetByID(id)
-	if leaseCacheEntry != nil {
-		return leaseCacheEntry.(lease.LeasedDescriptor).Underlying(), nil
+	if uc := c.tc.uncommitted.getByID(id); uc != nil {
+		return uc, nil
 	}
-	return c.tc.uncommitted.getByID(id), nil
+	if ld := c.tc.leased.cache.GetByID(id); ld != nil {
+		return ld.(lease.LeasedDescriptor).Underlying(), nil
+	}
+	return nil, nil
 }
 
 // DereferenceDescriptorIDs implements the validate.ValidationDereferencer

pkg/sql/catalog/schemadesc/schema_desc.go

@@ -169,8 +169,12 @@ func (desc *immutable) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 	}
 
 	// Validate the privilege descriptor.
-	vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Schema))
-	// The DefaultPrivilegeDescriptor may be nil.
+	if desc.Privileges == nil {
+		vea.Report(errors.AssertionFailedf("privileges not set"))
+	} else {
+		vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Schema))
+	}
+
 	if desc.GetDefaultPrivileges() != nil {
 		// Validate the default privilege descriptor.
 		vea.Report(catprivilege.ValidateDefaultPrivileges(*desc.GetDefaultPrivileges()))

pkg/sql/catalog/tabledesc/structured.go

@@ -97,6 +97,11 @@ type PostDeserializationTableDescriptorChanges struct {
 	// one computed column which also had a DEFAULT expression, which therefore
 	// had to be removed. See issue #72881 for details.
 	RemovedDefaultExprFromComputedColumn bool
+
+	// RemovedDuplicateIDsInRefs indicates that the table
+	// has redundant IDs in its DependsOn, DependsOnTypes and DependedOnBy
+	// references.
+	RemovedDuplicateIDsInRefs bool
 }
 
 // DescriptorType returns the type of this descriptor.
@@ -2506,3 +2511,25 @@ func LocalityConfigGlobal() catpb.LocalityConfig {
 func PrimaryKeyIndexName(tableName string) string {
 	return tableName + "_pkey"
 }
+
+// UpdateColumnsDependedOnBy creates, updates or deletes a depended-on-by column
+// reference by ID.
+func (desc *Mutable) UpdateColumnsDependedOnBy(id descpb.ID, colIDs catalog.TableColSet) {
+	ref := descpb.TableDescriptor_Reference{
+		ID:        id,
+		ColumnIDs: colIDs.Ordered(),
+		ByID:      true,
+	}
+	for i := range desc.DependedOnBy {
+		by := &desc.DependedOnBy[i]
+		if by.ID == id {
+			if colIDs.Empty() {
+				desc.DependedOnBy = append(desc.DependedOnBy[:i], desc.DependedOnBy[i+1:]...)
+				return
+			}
+			*by = ref
+			return
+		}
+	}
+	desc.DependedOnBy = append(desc.DependedOnBy, ref)
+}

pkg/sql/catalog/tabledesc/table_desc_builder.go

@@ -223,6 +223,7 @@ func maybeFillInDescriptor(
 	)
 	addedGrantOptions := catprivilege.MaybeUpdateGrantOptions(desc.Privileges)
 	changes.UpgradedPrivileges = fixedPrivileges || addedGrantOptions
+	changes.RemovedDuplicateIDsInRefs = maybeRemoveDuplicateIDsInRefs(desc)
 	return changes
 }
 
@@ -597,3 +598,55 @@ func maybeFixPrimaryIndexEncoding(idx *descpb.IndexDescriptor) (hasChanged bool)
 	idx.EncodingType = descpb.PrimaryIndexEncoding
 	return true
 }
+
+// maybeRemoveDuplicateIDsInRefs ensures that IDs in references to other tables
+// are not duplicated.
+func maybeRemoveDuplicateIDsInRefs(d *descpb.TableDescriptor) (hasChanged bool) {
+	// Strip duplicates from DependsOn.
+	if s := cleanedIDs(d.DependsOn); len(s) < len(d.DependsOn) {
+		d.DependsOn = s
+		hasChanged = true
+	}
+	// Do the same for DependsOnTypes.
+	if s := cleanedIDs(d.DependsOnTypes); len(s) < len(d.DependsOnTypes) {
+		d.DependsOnTypes = s
+		hasChanged = true
+	}
+	// Do the same for column IDs in DependedOnBy table references.
+	for i := range d.DependedOnBy {
+		ref := &d.DependedOnBy[i]
+		s := catalog.MakeTableColSet(ref.ColumnIDs...).Ordered()
+		// Also strip away O-IDs, which may have made their way in here in the past.
+		// But only strip them if they're not the only ID. Otherwise this will
+		// make for an even more confusing validation failure (we check that IDs
+		// are not zero).
+		if len(s) > 1 && s[0] == 0 {
+			s = s[1:]
+		}
+		if len(s) < len(ref.ColumnIDs) {
+			ref.ColumnIDs = s
+			hasChanged = true
+		}
+	}
+	// Do the same in columns for sequence refs.
+	for i := range d.Columns {
+		col := &d.Columns[i]
+		if s := cleanedIDs(col.UsesSequenceIds); len(s) < len(col.UsesSequenceIds) {
+			col.UsesSequenceIds = s
+			hasChanged = true
+		}
+		if s := cleanedIDs(col.OwnsSequenceIds); len(s) < len(col.OwnsSequenceIds) {
+			col.OwnsSequenceIds = s
+			hasChanged = true
+		}
+	}
+	return hasChanged
+}
+
+func cleanedIDs(input []descpb.ID) []descpb.ID {
+	s := catalog.MakeDescriptorIDSet(input...).Ordered()
+	if len(s) == 0 {
+		return nil
+	}
+	return s
+}

pkg/sql/catalog/tabledesc/validate.go

@@ -159,6 +159,20 @@ func (desc *wrapper) ValidateCrossReferences(
 		}
 	}
 
+	// For views, check dependent relations.
+	if desc.IsView() {
+		for _, id := range desc.DependsOn {
+			vea.Report(desc.validateOutboundTableRef(id, vdg))
+		}
+		for _, id := range desc.DependsOnTypes {
+			vea.Report(desc.validateOutboundTypeRef(id, vdg))
+		}
+	}
+
+	for _, by := range desc.DependedOnBy {
+		vea.Report(desc.validateInboundTableRef(by, vdg))
+	}
+
 	// Check foreign keys.
 	for i := range desc.OutboundFKs {
 		vea.Report(desc.validateOutboundFK(&desc.OutboundFKs[i], vdg))
@@ -184,6 +198,78 @@ func (desc *wrapper) ValidateCrossReferences(
 	}
 }
 
+func (desc *wrapper) validateOutboundTableRef(
+	id descpb.ID, vdg catalog.ValidationDescGetter,
+) error {
+	referencedTable, err := vdg.GetTableDescriptor(id)
+	if err != nil {
+		return errors.NewAssertionErrorWithWrappedErrf(err, "invalid depends-on relation reference")
+	}
+	for _, by := range referencedTable.TableDesc().DependedOnBy {
+		if by.ID == desc.GetID() {
+			return nil
+		}
+	}
+	return errors.AssertionFailedf("depends-on relation %q (%d) has no corresponding depended-on-by back reference",
+		referencedTable.GetName(), id)
+}
+
+func (desc *wrapper) validateOutboundTypeRef(id descpb.ID, vdg catalog.ValidationDescGetter) error {
+	_, err := vdg.GetTypeDescriptor(id)
+	if err != nil {
+		return errors.NewAssertionErrorWithWrappedErrf(err, "invalid depends-on type reference")
+	}
+	// TODO(postamar): maintain back-references in type, and validate these.
+	return nil
+}
+
+func (desc *wrapper) validateInboundTableRef(
+	by descpb.TableDescriptor_Reference, vdg catalog.ValidationDescGetter,
+) error {
+	backReferencedTable, err := vdg.GetTableDescriptor(by.ID)
+	if err != nil {
+		return errors.NewAssertionErrorWithWrappedErrf(err, "invalid depended-on-by relation back reference")
+	}
+
+	if desc.IsSequence() {
+		// The ColumnIDs field takes a different meaning when the validated
+		// descriptor is for a sequence. In this case, they refer to the columns
+		// in the referenced descriptor instead.
+		for _, colID := range by.ColumnIDs {
+			col, _ := backReferencedTable.FindColumnWithID(colID)
+			if col == nil {
+				return errors.AssertionFailedf("depended-on-by relation %q (%d) does not have a column with ID %d",
+					backReferencedTable.GetName(), by.ID, colID)
+			}
+			var found bool
+			for i := 0; i < col.NumUsesSequences(); i++ {
+				if col.GetUsesSequenceID(i) == desc.GetID() {
+					found = true
+					break
+				}
+			}
+			if found {
+				continue
+			}
+			return errors.AssertionFailedf(
+				"depended-on-by relation %q (%d) has no reference to this sequence in column %q (%d)",
+				backReferencedTable.GetName(), by.ID, col.GetName(), col.GetID())
+		}
+	}
+
+	// View back-references need corresponding forward reference.
+	if !backReferencedTable.IsView() {
+		return nil
+	}
+	for _, id := range backReferencedTable.TableDesc().DependsOn {
+		if id == desc.GetID() {
+			return nil
+		}
+	}
+	return errors.AssertionFailedf("depended-on-by view %q (%d) has no corresponding depends-on forward reference",
+		backReferencedTable.GetName(), by.ID)
+}
+
 func (desc *wrapper) validateOutboundFK(
 	fk *descpb.ForeignKeyConstraint, vdg catalog.ValidationDescGetter,
 ) error {
@@ -322,10 +408,64 @@ func (desc *wrapper) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 		vea.Report(errors.AssertionFailedf("invalid parent ID %d", desc.GetParentID()))
 	}
 
+	// Validate the privilege descriptor.
+	if desc.Privileges == nil {
+		vea.Report(errors.AssertionFailedf("privileges not set"))
+	} else {
+		vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Table))
+	}
+
+	// Validate that the depended-on-by references are well-formed.
+	for _, ref := range desc.DependedOnBy {
+		if ref.ID == descpb.InvalidID {
+			vea.Report(errors.AssertionFailedf(
+				"invalid relation ID %d in depended-on-by references",
+				ref.ID))
+		}
+		if len(ref.ColumnIDs) > catalog.MakeTableColSet(ref.ColumnIDs...).Len() {
+			vea.Report(errors.AssertionFailedf("duplicate column IDs found in depended-on-by references: %v",
+				ref.ColumnIDs))
+		}
+	}
+
+	if !desc.IsView() {
+		if len(desc.DependsOn) > 0 {
+			vea.Report(errors.AssertionFailedf(
+				"has depends-on references despite not being a view"))
+		}
+		if len(desc.DependsOnTypes) > 0 {
+			vea.Report(errors.AssertionFailedf(
+				"has depends-on-types references despite not being a view"))
+		}
+	}
+
 	if desc.IsSequence() {
 		return
 	}
 
+	// Validate the depended-on-by references to this table's columns and indexes
+	// for non-sequence relations.
+	// The ColumnIDs field takes a different meaning when the validated
+	// descriptor is for a sequence, in that case they refer to the columns of the
+	// referenced relation. This case is handled during cross-reference
+	// validation.
+	for _, ref := range desc.DependedOnBy {
+		if ref.IndexID != 0 {
+			if idx, _ := desc.FindIndexWithID(ref.IndexID); idx == nil {
+				vea.Report(errors.AssertionFailedf(
+					"index ID %d found in depended-on-by references, no such index in this relation",
+					ref.IndexID))
+			}
+		}
+		for _, colID := range ref.ColumnIDs {
+			if col, _ := desc.FindColumnWithID(colID); col == nil {
+				vea.Report(errors.AssertionFailedf(
+					"column ID %d found in depended-on-by references, no such column in this relation",
+					colID))
+			}
+		}
+	}
+
 	if len(desc.Columns) == 0 {
 		vea.Report(ErrMissingColumns)
 		return
@@ -338,7 +478,6 @@ func (desc *wrapper) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 		return
 	}
 
-	// TODO(dt, nathan): virtual descs don't validate (missing privs, PK, etc).
 	if desc.IsVirtualTable() {
 		return
 	}
@@ -399,9 +538,6 @@ func (desc *wrapper) ValidateSelf(vea catalog.ValidationErrorAccumulator) {
 		}
 	}
 
-	// Validate the privilege descriptor.
-	vea.Report(catprivilege.Validate(*desc.Privileges, desc, privilege.Table))
-
 	// Ensure that mutations cannot be queued if a primary key change or
 	// an alter column type schema change has either been started in
 	// this transaction, or is currently in progress.