Merge cockroachdb#59978

59978: sql: include invalid type descriptors in crdb_internal.invalid_objects r=otan a=arulajmani Previously, `crdb_internal.invalid_objects` only validated table descriptors. This patch adds type descriptors to the list of objects that are validated. As this table is used to validate all descriptors after logic tests run, going forward, validation inconsistencies will be checked for and reported after logic test runs. Release note (sql change): `crdb_internal.invalid_objects` also includes invalid type descriptors. Co-authored-by: arulajmani <[email protected]>
tbg · Feb 8, 2021 · e3a3da3 · e3a3da3
2 parents 0e98670 + 20fc00e
commit e3a3da3
Show file tree

Hide file tree

Showing 2 changed files with 63 additions and 3 deletions.
diff --git a/pkg/sql/crdb_internal.go b/pkg/sql/crdb_internal.go
@@ -3784,14 +3784,12 @@ CREATE TABLE crdb_internal.invalid_objects (
 	) error {
 		// The internalLookupContext will only have descriptors in the current
 		// database. To deal with this, we fall through.
-		// TODO(spaskob): we can also validate type descriptors. Add a new function
-		// `forEachTypeDescAllWithTableLookup` and the results to this table.
 		descs, err := catalogkv.GetAllDescriptorsUnvalidated(ctx, p.txn, p.extendedEvalCtx.Codec)
 		if err != nil {
 			return err
 		}
 		const allowAdding = true
-		return forEachTableDescWithTableLookupInternalFromDescriptors(
+		if err := forEachTableDescWithTableLookupInternalFromDescriptors(
 			ctx, p, dbContext, hideVirtual, allowAdding, descs, func(
 				dbDesc *dbdesc.Immutable, schema string, descriptor catalog.TableDescriptor, fn tableLookupFn,
 			) error {
@@ -3806,6 +3804,34 @@ CREATE TABLE crdb_internal.invalid_objects (
 				if dbDesc != nil {
 					dbName = dbDesc.GetName()
 				}
+				return addRow(
+					tree.NewDInt(tree.DInt(descriptor.GetID())),
+					tree.NewDString(dbName),
+					tree.NewDString(schema),
+					tree.NewDString(descriptor.GetName()),
+					tree.NewDString(err.Error()),
+				)
+			}); err != nil {
+			return err
+		}
+
+		// Validate type descriptors.
+		return forEachTypeDescWithTableLookupInternalFromDescriptors(
+			ctx, p, dbContext, allowAdding, descs, func(
+				dbDesc *dbdesc.Immutable, schema string, descriptor catalog.TypeDescriptor, fn tableLookupFn,
+			) error {
+				if descriptor == nil {
+					return nil
+				}
+				err = descriptor.Validate(ctx, fn)
+				if err == nil {
+					return nil
+				}
+				var dbName string
+				if dbDesc != nil {
+					dbName = dbDesc.GetName()
+				}
+
 				return addRow(
 					tree.NewDInt(tree.DInt(descriptor.GetID())),
 					tree.NewDString(dbName),

diff --git a/pkg/sql/information_schema.go b/pkg/sql/information_schema.go
@@ -2047,6 +2047,40 @@ func forEachTableDescWithTableLookupInternal(
 		ctx, p, dbContext, virtualOpts, allowAdding, descs, fn)
 }
 
+func forEachTypeDescWithTableLookupInternalFromDescriptors(
+	ctx context.Context,
+	p *planner,
+	dbContext *dbdesc.Immutable,
+	allowAdding bool,
+	descs []catalog.Descriptor,
+	fn func(*dbdesc.Immutable, string, catalog.TypeDescriptor, tableLookupFn) error,
+) error {
+	lCtx := newInternalLookupCtx(ctx, descs, dbContext,
+		catalogkv.NewOneLevelUncachedDescGetter(p.txn, p.execCfg.Codec))
+
+	for _, typID := range lCtx.typIDs {
+		typDesc := lCtx.typDescs[typID]
+		if typDesc.Dropped() || !userCanSeeDescriptor(ctx, p, typDesc, allowAdding) {
+			continue
+		}
+		var scName string
+		dbDesc, parentExists := lCtx.dbDescs[typDesc.GetParentID()]
+		if parentExists {
+			var ok bool
+			scName, ok = lCtx.schemaNames[typDesc.GetParentSchemaID()]
+			if !ok {
+				return errors.AssertionFailedf("schema id %d not found", typDesc.GetParentSchemaID())
+			}
+			if err := fn(dbDesc, scName, typDesc, lCtx); err != nil {
+				return err
+			}
+		} else {
+			return errors.AssertionFailedf("database id %d not found", typDesc.GetParentID())
+		}
+	}
+	return nil
+}
+
 func forEachTableDescWithTableLookupInternalFromDescriptors(
 	ctx context.Context,
 	p *planner,