From 6bbd2a3aa32ca15f6d0e67f7cec7c9893cba4593 Mon Sep 17 00:00:00 2001
From: Marius Posta <marius@cockroachlabs.com>
Date: Tue, 12 Jan 2021 13:37:58 -0500
Subject: [PATCH] sql: add doctor check for parent schema in parent db

With this patch, doctor now also checks that the parent id of an object
matches the parent id of the parent schema of the object.

Fixes #55716.

Release note: None
---
 pkg/sql/doctor/doctor.go      | 16 +++++++++++-----
 pkg/sql/doctor/doctor_test.go | 29 +++++++++++++++++++++++++++++
 2 files changed, 40 insertions(+), 5 deletions(-)

diff --git a/pkg/sql/doctor/doctor.go b/pkg/sql/doctor/doctor.go
index 2dcfc761397b..0ed6448d8017 100644
--- a/pkg/sql/doctor/doctor.go
+++ b/pkg/sql/doctor/doctor.go
@@ -141,7 +141,7 @@ func ExamineDescriptors(
 		}
 
 		_, parentExists := descGetter[desc.GetParentID()]
-		_, parentSchemaExists := descGetter[desc.GetParentSchemaID()]
+		parentSchema, parentSchemaExists := descGetter[desc.GetParentSchemaID()]
 		switch d := desc.(type) {
 		case catalog.TableDescriptor:
 			if err := d.Validate(ctx, descGetter); err != nil {
@@ -161,15 +161,21 @@ func ExamineDescriptors(
 			// parent schema id is always 0.
 			parentSchemaExists = true
 		}
+		var invalidParentID bool
 		if desc.GetParentID() != descpb.InvalidID && !parentExists {
 			problemsFound = true
+			invalidParentID = true
 			fmt.Fprint(stdout, reportMsg(desc, "invalid parent id %d", desc.GetParentID()))
 		}
 		if desc.GetParentSchemaID() != descpb.InvalidID &&
-			desc.GetParentSchemaID() != keys.PublicSchemaID &&
-			!parentSchemaExists {
-			problemsFound = true
-			fmt.Fprint(stdout, reportMsg(desc, "invalid parent schema id %d", desc.GetParentSchemaID()))
+			desc.GetParentSchemaID() != keys.PublicSchemaID {
+			if !parentSchemaExists {
+				problemsFound = true
+				fmt.Fprint(stdout, reportMsg(desc, "invalid parent schema id %d", desc.GetParentSchemaID()))
+			} else if !invalidParentID && parentSchema.GetParentID() != desc.GetParentID() {
+				problemsFound = true
+				fmt.Fprint(stdout, reportMsg(desc, "invalid parent schema parent id %d", parentSchema.GetParentID()))
+			}
 		}
 
 		// Process namespace entries pointing to this descriptor.
diff --git a/pkg/sql/doctor/doctor_test.go b/pkg/sql/doctor/doctor_test.go
index 9a57ae0b2986..8d0b1863df25 100644
--- a/pkg/sql/doctor/doctor_test.go
+++ b/pkg/sql/doctor/doctor_test.go
@@ -74,6 +74,10 @@ func TestExamineDescriptors(t *testing.T) {
 	descpb.TableFromDescriptor(droppedValidTableDesc, hlc.Timestamp{WallTime: 1}).
 		State = descpb.DescriptorState_DROP
 
+	inSchemaValidTableDesc := protoutil.Clone(validTableDesc).(*descpb.Descriptor)
+	descpb.TableFromDescriptor(inSchemaValidTableDesc, hlc.Timestamp{WallTime: 1}).
+		UnexposedParentSchemaID = 3
+
 	tests := []struct {
 		descTable      doctor.DescriptorTable
 		namespaceTable doctor.NamespaceTable
@@ -194,6 +198,31 @@ Database   1: ParentID   0, ParentSchemaID  0, Name 'db': not being dropped but
 			},
 			expected: `Examining 1 descriptors and 1 namespace entries...
     Type   1: ParentID   0, ParentSchemaID  0, Name 'type': invalid parentID 0
+`,
+		},
+		{
+			descTable: doctor.DescriptorTable{
+				{ID: 1, DescBytes: toBytes(t, inSchemaValidTableDesc)},
+				{
+					ID: 2,
+					DescBytes: toBytes(t, &descpb.Descriptor{Union: &descpb.Descriptor_Database{
+						Database: &descpb.DatabaseDescriptor{Name: "db", ID: 2},
+					}}),
+				},
+				{
+					ID: 3,
+					DescBytes: toBytes(t, &descpb.Descriptor{Union: &descpb.Descriptor_Schema{
+						Schema: &descpb.SchemaDescriptor{Name: "schema", ID: 3, ParentID: 0},
+					}}),
+				},
+			},
+			namespaceTable: doctor.NamespaceTable{
+				{NameInfo: descpb.NameInfo{ParentID: 2, ParentSchemaID: 3, Name: "t"}, ID: 1},
+				{NameInfo: descpb.NameInfo{Name: "db"}, ID: 2},
+				{NameInfo: descpb.NameInfo{ParentID: 0, Name: "schema"}, ID: 3},
+			},
+			expected: `Examining 3 descriptors and 3 namespace entries...
+   Table   1: ParentID   2, ParentSchemaID  3, Name 't': invalid parent schema parent id 0
 `,
 		},
 		{