restore: delete comments after failed or cancelled cluster restore

This patch ensures that after a failed or canceled cluster restore, comments
from dropped descriptors are removed.

Informs #149453

Release note: none

Author: msbutler

pkg/backup/backup_test.go

@@ -11183,3 +11183,60 @@ CREATE TABLE child_pk (k INT8 PRIMARY KEY REFERENCES parent);
 		sqlDB.Exec(t, `DROP DATABASE test`)
 	}
 }
+
+func TestRestoreFailureDeletesComments(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+	defer log.Scope(t).Close(t)
+
+	_, sqlDB, cleanupFn := backupRestoreTestSetupEmpty(t, singleNode, "", InitManualReplication, base.TestClusterArgs{})
+	defer cleanupFn()
+
+	// Set pause point for after the system tables have been published.
+	sqlDB.Exec(t, `SET CLUSTER SETTING jobs.debug.pausepoints = 'restore.after_cleanup_temp_system_tables'`)
+
+	commentCountQuery := `SELECT count(*) FROM system.comments`
+
+	var count int
+	sqlDB.QueryRow(t, commentCountQuery).Scan(&count)
+	require.Equal(t, 0, count)
+
+	// Create a database with tables, types, and schemas that have comments
+	sqlDB.Exec(t, `CREATE DATABASE test_db`)
+	sqlDB.Exec(t, `USE test_db`)
+
+	sqlDB.Exec(t, `CREATE TYPE custom_type AS ENUM ('val1', 'val2')`)
+	sqlDB.Exec(t, `COMMENT ON TYPE custom_type IS 'This is a custom type comment'`)
+
+	sqlDB.Exec(t, `CREATE SCHEMA test_schema`)
+	sqlDB.Exec(t, `COMMENT ON SCHEMA test_schema IS 'This is a schema comment'`)
+
+	sqlDB.Exec(t, `CREATE TABLE test_schema.test_table (id INT PRIMARY KEY, name STRING)`)
+	sqlDB.Exec(t, `COMMENT ON TABLE test_schema.test_table IS 'This is a table comment'`)
+
+	sqlDB.Exec(t, `COMMENT ON DATABASE test_db IS 'This is a database comment'`)
+
+	sqlDB.QueryRow(t, commentCountQuery).Scan(&count)
+	require.Equal(t, 4, count)
+
+	sqlDB.Exec(t, `BACKUP INTO 'nodelocal://1/test_backup'`)
+
+	sqlDB.Exec(t, `USE system`)
+
+	sqlDB.Exec(t, `DROP DATABASE test_db CASCADE`)
+	sqlDB.QueryRow(t, commentCountQuery).Scan(&count)
+	require.Equal(t, 0, count)
+
+	var jobID jobspb.JobID
+	sqlDB.QueryRow(t, `RESTORE FROM LATEST IN 'nodelocal://1/test_backup' WITH detached`).Scan(&jobID)
+	jobutils.WaitForJobToPause(t, sqlDB, jobID)
+
+	sqlDB.QueryRow(t, commentCountQuery).Scan(&count)
+	require.Equal(t, 4, count)
+
+	// Cancel the restore job
+	sqlDB.Exec(t, `CANCEL JOB $1`, jobID)
+	jobutils.WaitForJobToCancel(t, sqlDB, jobID)
+
+	sqlDB.QueryRow(t, commentCountQuery).Scan(&count)
+	require.Equal(t, 0, count)
+}

pkg/backup/restore_job.go

@@ -38,6 +38,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
+	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catalogkeys"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/dbdesc"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descidgen"
@@ -2809,6 +2810,10 @@ func (r *restoreResumer) dropDescriptors(
 		tablesToGC = append(tablesToGC, tableToDrop.ID)
 		tableToDrop.SetDropped()
 
+		if err := descsCol.DeleteTableComments(ctx, kvTrace, b, tableToDrop.ID); err != nil {
+			return err
+		}
+
 		// Drop any schedules we may have implicitly created.
 		if tableToDrop.HasRowLevelTTL() {
 			scheduleID := tableToDrop.RowLevelTTL.ScheduleID
@@ -2864,6 +2869,14 @@ func (r *restoreResumer) dropDescriptors(
 		}
 		mutType.SetDropped()
 
+		if err := descsCol.DeleteCommentInBatch(
+			ctx,
+			kvTrace,
+			b,
+			catalogkeys.MakeCommentKey(uint32(typDesc.ID), 0, catalogkeys.TypeCommentType)); err != nil {
+			return err
+		}
+
 		if err := descsCol.DeleteNamespaceEntryToBatch(ctx, kvTrace, typDesc, b); err != nil {
 			return err
 		}
@@ -2959,6 +2972,14 @@ func (r *restoreResumer) dropDescriptors(
 			entry.db = mutParent.(*dbdesc.Mutable)
 		}
 
+		if err := descsCol.DeleteCommentInBatch(
+			ctx,
+			kvTrace,
+			b,
+			catalogkeys.MakeCommentKey(uint32(schemaDesc.GetID()), 0, catalogkeys.SchemaCommentType)); err != nil {
+			return err
+		}
+
 		// Delete schema entries in descriptor and namespace system tables.
 		if err := descsCol.DeleteNamespaceEntryToBatch(ctx, kvTrace, mutSchema, b); err != nil {
 			return err
@@ -3042,6 +3063,14 @@ func (r *restoreResumer) dropDescriptors(
 				return err
 			}
 		}
+		if err := descsCol.DeleteCommentInBatch(
+			ctx,
+			kvTrace,
+			b,
+			catalogkeys.MakeCommentKey(uint32(dbDesc.GetID()), 0, catalogkeys.DatabaseCommentType)); err != nil {
+			return err
+		}
+
 		if err := descsCol.DeleteNamespaceEntryToBatch(ctx, kvTrace, db, b); err != nil {
 			return err
 		}