sql: fix bug causing sequenceIDs to be duplicated on column descriptors

Previously, there was a bug in the `ALTER SEQUENCE ... OWNED BY ...`
command that could cause sequence IDs to appear on more than one
column descriptor's 'OwnsSequenceIDs' field. This was problematic as
the dropping logic requires a sequence ID only be present on one column
descriptor's `OwnsSequenceIDs' field.

The field would get corrupted whenever ownership was altered between
columns of the same table. This was because we would read a table
descriptor before executing the removal code, but the removal would
modify the same table descriptor. This would make the table descriptor
read earlier stale, as the old owner's column descriptor still
referenced the sequence ID. This would mean that when a new owner was
added to a different column, the schema change that was written would
have the sequence ID on two different column descriptor's
`OwnsSequenceIDs` field.

This patch fixes that by forcing the `addSequenceOwner` to resolve the
most recent version of the table descriptor instead of being passed it.

Fixes #50711

Release note (bug fix): Previously, if there was a table
`t(a int, b int)`, and a sequence `seq` that was first owned by `t.a`
and then altered to be owned by `t.b`, it would make the table `t`
impossible to drop. This is now fixed.

pkg/sql/logictest/testdata/logic_test/sequences

@@ -1182,3 +1182,49 @@ DROP DATABASE db_50712 CASCADE
 
 statement ok
 DROP TABLE t_50712
+
+# previously, changing ownership of a sequence between columns of the same table
+# was causing the sequenceID to appear in multiple column's ownedSequences list.
+# This makes it impossible to drop the affected columns/table once the sequence
+# has been dropped. This tests for these scenarios and ensures the table/columns
+# remain drop-able.
+subtest regression_50711
+
+statement ok
+CREATE TABLE t_50711(a int, b int)
+
+statement ok
+CREATE SEQUENCE seq_50711 owned by t_50711.a
+
+statement ok
+ALTER SEQUENCE seq_50711 owned by t_50711.b
+
+statement ok
+ALTER SEQUENCE seq_50711 owned by t_50711.a
+
+statement ok
+DROP SEQUENCE seq_50711
+
+statement ok
+DROP TABLE t_50711
+
+statement ok
+CREATE TABLE t_50711(a int, b int)
+
+statement ok
+CREATE SEQUENCE seq_50711 owned by t_50711.a
+
+statement ok
+ALTER SEQUENCE seq_50711 owned by t_50711.b
+
+statement ok
+ALTER SEQUENCE seq_50711 owned by t_50711.a
+
+statement ok
+DROP SEQUENCE seq_50711
+
+statement ok
+ALTER TABLE t_50711 DROP COLUMN a
+
+statement ok
+ALTER TABLE t_50711 DROP COLUMN b

pkg/sql/sequence.go

@@ -289,7 +289,7 @@ func assignSequenceOptions(
 					if err := removeSequenceOwnerIfExists(params.ctx, params.p, sequenceID, opts); err != nil {
 						return err
 					}
-					err := addSequenceOwner(params.ctx, params.p, tableDesc, col, sequenceID, opts)
+					err := addSequenceOwner(params.ctx, params.p, option.ColumnItemVal, sequenceID, opts)
 					if err != nil {
 						return err
 					}
@@ -390,11 +390,15 @@ func resolveColumnItemToDescriptors(
 func addSequenceOwner(
 	ctx context.Context,
 	p *planner,
-	tableDesc *MutableTableDescriptor,
-	col *sqlbase.ColumnDescriptor,
+	columnItemVal *tree.ColumnItem,
 	sequenceID sqlbase.ID,
 	opts *sqlbase.TableDescriptor_SequenceOpts,
 ) error {
+	tableDesc, col, err := resolveColumnItemToDescriptors(ctx, p, columnItemVal)
+	if err != nil {
+		return err
+	}
+
 	col.OwnsSequenceIds = append(col.OwnsSequenceIds, sequenceID)
 
 	opts.SequenceOwner.OwnerColumnID = col.ID

pkg/sql/sequence_test.go

@@ -15,7 +15,11 @@ import (
 	"testing"
 
 	"github.com/cockroachdb/cockroach/pkg/base"
+	"github.com/cockroachdb/cockroach/pkg/keys"
+	"github.com/cockroachdb/cockroach/pkg/kv"
+	"github.com/cockroachdb/cockroach/pkg/sql/sqlbase"
 	"github.com/cockroachdb/cockroach/pkg/testutils/serverutils"
+	"github.com/cockroachdb/cockroach/pkg/util/leaktest"
 )
 
 func BenchmarkSequenceIncrement(b *testing.B) {
@@ -70,3 +74,118 @@ func BenchmarkUniqueRowID(b *testing.B) {
 	}
 	b.StopTimer()
 }
+
+// Regression test for #50711. The root cause of #50711 was the fact that a
+// sequenceID popped up in multiple columns' column descriptor. This test
+// inspects the table descriptor to ascertain that sequence ownership integrity
+// is preserved in various scenarios.
+// Scenarios tested:
+// - ownership swaps between table columns
+// - two sequences being owned simultaneously
+// - sequence drops
+// - ownership removal
+func TestSequenceOwnershipDependencies(t *testing.T) {
+	defer leaktest.AfterTest(t)()
+
+	ctx := context.Background()
+	params := base.TestServerArgs{}
+	s, sqlConn, kvDB := serverutils.StartServer(t, params)
+	defer s.Stopper().Stop(ctx)
+
+	if _, err := sqlConn.Exec(`
+CREATE DATABASE t;
+CREATE TABLE t.test(a INT PRIMARY KEY, b INT)`); err != nil {
+		t.Fatal(err)
+	}
+
+	// Switch ownership between columns of the same table.
+	if _, err := sqlConn.Exec("CREATE SEQUENCE t.seq1 OWNED BY t.test.a"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq1"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+
+	if _, err := sqlConn.Exec("ALTER SEQUENCE t.seq1 OWNED BY t.test.b"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, nil /* seqNames */)
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, []string{"seq1"})
+
+	if _, err := sqlConn.Exec("ALTER SEQUENCE t.seq1 OWNED BY t.test.a"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq1"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+
+	// Add a second sequence in the mix and switch its ownership.
+	if _, err := sqlConn.Exec("CREATE SEQUENCE t.seq2 OWNED BY t.test.a"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq1", "seq2"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+
+	if _, err := sqlConn.Exec("ALTER SEQUENCE t.seq2 OWNED BY t.test.b"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq1"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, []string{"seq2"})
+
+	if _, err := sqlConn.Exec("ALTER SEQUENCE t.seq2 OWNED BY t.test.a"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq1", "seq2"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+
+	// Ensure dropping sequences removes the ownership dependencies.
+	if _, err := sqlConn.Exec("DROP SEQUENCE t.seq1"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, []string{"seq2"})
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+
+	// Ensure removing an owner removes the ownership dependency.
+	if _, err := sqlConn.Exec("ALTER SEQUENCE t.seq2 OWNED BY NONE"); err != nil {
+		t.Fatal(err)
+	}
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 0 /* colIdx */, nil /* seqNames */)
+	assertColumnOwnsSequences(t, kvDB, "t", "test", 1 /* colIdx */, nil /* seqNames */)
+}
+
+// assertColumnOwnsSequences ensures that the column at (DbName, tbName, colIdx)
+// owns all the sequences passed to it (in order) by looking up descriptors in
+// kvDB.
+func assertColumnOwnsSequences(
+	t *testing.T, kvDB *kv.DB, dbName string, tbName string, colIdx int, seqNames []string,
+) {
+	tableDesc := sqlbase.TestingGetTableDescriptor(kvDB, keys.SystemSQLCodec, dbName, tbName)
+	col := tableDesc.GetColumns()[colIdx]
+	var seqDescs []*SequenceDescriptor
+	for _, seqName := range seqNames {
+		seqDescs = append(
+			seqDescs,
+			sqlbase.TestingGetTableDescriptor(kvDB, keys.SystemSQLCodec, dbName, seqName),
+		)
+	}
+
+	if len(col.OwnsSequenceIds) != len(seqDescs) {
+		t.Fatalf(
+			"unexpected number of sequence ownership dependencies. expected: %d, got:%d",
+			len(seqDescs), len(col.OwnsSequenceIds),
+		)
+	}
+
+	for i, seqID := range col.OwnsSequenceIds {
+		if seqID != seqDescs[i].GetID() {
+			t.Fatalf("unexpected sequence id. expected %d got %d", seqDescs[i].GetID(), seqID)
+		}
+
+		ownerTableID := seqDescs[i].SequenceOpts.SequenceOwner.OwnerTableID
+		ownerColID := seqDescs[i].SequenceOpts.SequenceOwner.OwnerColumnID
+		if ownerTableID != tableDesc.GetID() || ownerColID != col.ID {
+			t.Fatalf(
+				"unexpected sequence owner. expected table id %d, got: %d; expected column id %d, got :%d",
+				tableDesc.GetID(), ownerTableID, col.ID, ownerColID,
+			)
+		}
+	}
+}