release-20.1: combined fixes for ownership related bugs

pkg/sql/alter_table.go

@@ -418,7 +418,7 @@ func (n *alterTableNode) startExec(params runParams) error {
 				return err
 			}
 
-			if err := params.p.dropSequencesOwnedByCol(params.ctx, colToDrop); err != nil {
+			if err := params.p.dropSequencesOwnedByCol(params.ctx, colToDrop, true /* queueJob */); err != nil {
 				return err
 			}
 

pkg/sql/drop_table.go

@@ -288,7 +288,7 @@ func (p *planner) dropTableImpl(
 
 	// Drop sequences that the columns of the table own
 	for _, col := range tableDesc.Columns {
-		if err := p.dropSequencesOwnedByCol(ctx, &col); err != nil {
+		if err := p.dropSequencesOwnedByCol(ctx, &col, queueJob); err != nil {
 			return droppedViews, err
 		}
 	}
@@ -336,7 +336,7 @@ func (p *planner) initiateDropTable(
 	drainName bool,
 ) error {
 	if tableDesc.Dropped() {
-		return fmt.Errorf("table %q is being dropped", tableDesc.Name)
+		return errors.Errorf("table %q is already being dropped", tableDesc.Name)
 	}
 
 	// If the table is not interleaved , use the delayed GC mechanism to

pkg/sql/logictest/testdata/logic_test/sequences

@@ -1170,6 +1170,7 @@ DROP DATABASE db_50712 CASCADE
 
 statement ok
 DROP TABLE t_50712
+<<<<<<< HEAD
 
 # previously, changing ownership of a sequence between columns of the same table
 # was causing the sequenceID to appear in multiple column's ownedSequences list.

pkg/sql/sequence.go

@@ -336,6 +336,11 @@ func removeSequenceOwnerIfExists(
 	if err != nil {
 		return err
 	}
+	// If the table descriptor has already been dropped, there is no need to
+	// remove the reference.
+	if tableDesc.Dropped() {
+		return nil
+	}
 	col, err := tableDesc.FindColumnByID(opts.SequenceOwner.OwnerColumnID)
 	if err != nil {
 		return err
@@ -471,17 +476,21 @@ func maybeAddSequenceDependencies(
 // dropSequencesOwnedByCol drops all the sequences from col.OwnsSequenceIDs.
 // Called when the respective column (or the whole table) is being dropped.
 func (p *planner) dropSequencesOwnedByCol(
-	ctx context.Context, col *sqlbase.ColumnDescriptor,
+	ctx context.Context, col *sqlbase.ColumnDescriptor, queueJob bool,
 ) error {
 	for _, sequenceID := range col.OwnsSequenceIds {
 		seqDesc, err := p.Tables().getMutableTableVersionByID(ctx, sequenceID, p.txn)
 		if err != nil {
 			return err
 		}
+		// This sequence is already getting dropped. Don't do it twice.
+		if seqDesc.Dropped() {
+			continue
+		}
 		jobDesc := fmt.Sprintf("removing sequence %q dependent on column %q which is being dropped",
 			seqDesc.Name, col.ColName())
 		if err := p.dropSequenceImpl(
-			ctx, seqDesc, true /* queueJob */, jobDesc, tree.DropRestrict,
+			ctx, seqDesc, queueJob, jobDesc, tree.DropRestrict,
 		); err != nil {
 			return err
 		}

pkg/sql/table.go

@@ -1047,7 +1047,8 @@ func (p *planner) writeSchemaChange(
 	}
 	if tableDesc.Dropped() {
 		// We don't allow schema changes on a dropped table.
-		return fmt.Errorf("table %q is being dropped", tableDesc.Name)
+		return errors.Errorf("no schema changes allowed on table %q as it is being dropped",
+			tableDesc.Name)
 	}
 	if err := p.createOrUpdateSchemaChangeJob(ctx, tableDesc, jobDesc, mutationID); err != nil {
 		return err
@@ -1063,7 +1064,8 @@ func (p *planner) writeSchemaChangeToBatch(
 	}
 	if tableDesc.Dropped() {
 		// We don't allow schema changes on a dropped table.
-		return fmt.Errorf("table %q is being dropped", tableDesc.Name)
+		return errors.Errorf("no schema changes allowed on table %q as it is being dropped",
+			tableDesc.Name)
 	}
 	return p.writeTableDescToBatch(ctx, tableDesc, b)
 }