sql/schemachanger: clean up SequenceOwner elements during restore

Previously, when restoring a backup taken in middle of a DROP COLUMN,
where a column had a sequence owner assigned, it was possible for the
backup to be unrestorable. This would happen because the sequence
reference would have been dropped in the plan, but the seqeunce owner
element was still within the state. To address this, this test updates
the rewrite logic to clean up any SequenceOwner elements which have the
referenced sequence already removed.

Fixes: cockroachdb#130778

Release note (bug fix): Addressed a rare bug that could prevent backups
taken during a DROP COLUMN operation with a sequence owner from
restoring with the error: "rewriting descriptor ids: missing rewrite for
<id> in SequenceOwner..."

pkg/sql/catalog/rewrite/rewrite.go

@@ -767,6 +767,16 @@ func rewriteSchemaChangerState(
 			data.SchemaID = rewrite.ParentSchemaID
 			continue
 		}
+
+		// removeElementAtCurrentIdx deletes the element at the current index.
+		removeElementAtCurrentIdx := func() {
+			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--
+		}
+
+		missingID := descpb.InvalidID
 		if err := screl.WalkDescIDs(t.Element(), func(id *descpb.ID) error {
 			if *id == descpb.InvalidID {
 				// Some descriptor ID fields in elements may be deliberately unset.
@@ -775,6 +785,7 @@ func rewriteSchemaChangerState(
 			}
 			rewrite, ok := descriptorRewrites[*id]
 			if !ok {
+				missingID = *id
 				return errors.Errorf("missing rewrite for id %d in %s", *id, screl.ElementString(t.Element()))
 			}
 			*id = rewrite.ID
@@ -785,29 +796,27 @@ func rewriteSchemaChangerState(
 				// We'll permit this in the special case of a schema parent element.
 				_, 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--
+					removeElementAtCurrentIdx()
 					continue
 				}
 			case *scpb.CheckConstraint:
 				// IF there is any dependency missing for check constraint, we just drop
 				// the target.
-				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--
+				removeElementAtCurrentIdx()
 				droppedConstraints.Add(el.ConstraintID)
 				continue
 			case *scpb.ColumnDefaultExpression:
 				// IF there is any dependency missing for column default expression, we
 				// just drop the target.
-				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--
+				removeElementAtCurrentIdx()
 				continue
+			case *scpb.SequenceOwner:
+				// If a sequence owner is missing the sequence, then the sequence
+				// was already dropped and this element can be safely removed.
+				if el.SequenceID == missingID {
+					removeElementAtCurrentIdx()
+					continue
+				}
 			}
 			return errors.Wrap(err, "rewriting descriptor ids")
 		}

pkg/sql/schemachanger/testdata/end_to_end/drop_column_sequence_owner/drop_column_sequence_owner.definition

@@ -0,0 +1,49 @@
+setup
+CREATE TABLE t (i INT PRIMARY KEY, j INT, k int);
+CREATE SEQUENCE sq1 OWNED BY t.j;
+COMMENT ON TABLE t IS 't has a comment';
+COMMENT ON COLUMN t.j IS 'j has a comment';
+INSERT INTO t VALUES(-1);
+INSERT INTO t VALUES(-2);
+INSERT INTO t VALUES(-3);
+----
+
+stage-exec phase=PostCommitPhase stage=:
+INSERT INTO t VALUES($stageKey);
+INSERT INTO t VALUES($stageKey + 1);
+UPDATE t SET k=$stageKey;
+UPDATE t SET k=i;
+DELETE FROM t WHERE i=-1;
+DELETE FROM t WHERE i=$stageKey;
+INSERT INTO t VALUES($stageKey);
+INSERT INTO t VALUES(-1);
+----
+
+# Each insert will be injected twice per stage, plus 3 injected
+# at the start.
+stage-query phase=PostCommitPhase stage=:
+SELECT count(*)=($successfulStageCount*2)+3 FROM t;
+----
+true
+
+stage-exec phase=PostCommitNonRevertiblePhase stage=:
+INSERT INTO t VALUES($stageKey);
+INSERT INTO t VALUES($stageKey + 1);
+UPDATE t SET k=$stageKey;
+UPDATE t SET k=i;
+DELETE FROM t WHERE i=-1;
+DELETE FROM t WHERE i=$stageKey;
+INSERT INTO t VALUES($stageKey);
+INSERT INTO t VALUES(-1);
+----
+
+# Each insert will be injected twice per stage, plus 3 injected
+# at the start.
+stage-query phase=PostCommitNonRevertiblePhase stage=:
+SELECT count(*)=($successfulStageCount*2)+3 FROM t;
+----
+true
+
+test
+ALTER TABLE t DROP COLUMN j
+----