sql: disallow table/view/sequence rename from making cross DB references

Fixes: #55709

Previously, disallowed cross DB references could be created
using a ALTER TABLE/VIEW/SEQUENCE rename operations. This
was inadequate because users could accidentally start using
deprecated functionality. To address this, this patch detects
such scenarios are returns an appropriate error.

Release note (bug fix): ALTER TABLE/VIEW/SEQUENCE can no longer
be used to incorrectly create cross DB references.

Release justification: Low risk fix to avoid users from accidentally,
using deprecated functionality.

pkg/sql/logictest/testdata/logic_test/rename_table

@@ -381,3 +381,88 @@ DROP DATABASE olddb CASCADE
 
 statement ok
 DROP DATABASE newdb CASCADE
+
+# Sanity check for #55709 renaming tables and cross DB references
+
+statement ok
+CREATE DATABASE newdb;
+
+statement ok
+SET database = newdb;
+
+# First validate renames are blocked
+
+statement ok
+CREATE TABLE t1ref (i INT8 PRIMARY KEY);
+
+statement ok
+CREATE TABLE t2ref (
+	j INT8,
+	CONSTRAINT t1_reference FOREIGN KEY (j) REFERENCES t1ref (i)
+);
+
+statement error a foreign key constraint "t1_reference" will exist between databases after rename
+ALTER TABLE t2ref RENAME TO test2.public.t2;
+
+statement error a foreign key constraint "t1_reference" will exist between databases after rename
+ALTER TABLE t1ref RENAME TO test2.public.t1;
+
+statement ok
+CREATE TABLE t1vref (i INT8 PRIMARY KEY);
+
+statement ok
+CREATE VIEW t1view AS (SELECT i FROM t1vref);
+
+statement error cannot rename relation "t1vref" because view "t1view" depends on it
+ALTER TABLE t1vref RENAME TO test2.public.t1vref;
+
+statement error this view will reference a table "t1vref" in another databases after rename
+ALTER VIEW t1view RENAME TO test2.public.t1view;
+
+statement ok
+CREATE TABLE t1s1ref (i INT8 PRIMARY KEY);
+
+statement ok
+CREATE SEQUENCE s1 OWNED BY t1s1ref.i;
+
+statement error a sequence "s1" will be OWNED BY a table in a different database after rename
+ALTER TABLE t1s1ref RENAME TO test2.public.t1s1ref;
+
+statement error this sequence will be OWNED BY a table "t1s1ref" in a different database after rename
+ALTER SEQUENCE s1 RENAME TO test2.public.s1;
+
+# Enable cluster setting to allow foreign keys across DB's
+statement ok
+SET CLUSTER SETTING "sql.cross_db_fks.enabled" = true;
+
+statement ok
+ALTER TABLE t2ref RENAME TO test2.public.t2ref;
+
+statement ok
+ALTER TABLE test2.public.t2ref RENAME TO t2ref;
+
+statement ok
+ALTER TABLE t1ref RENAME TO test2.public.t1;
+
+#Enable cluster setting for views across DB's
+statement ok
+SET CLUSTER SETTING "sql.cross_db_views.enabled" = true;
+
+statement error cannot rename relation "t1vref" because view "t1view" depends on it
+ALTER TABLE t1vref RENAME TO test2.public.t1vref;
+
+statement ok
+ALTER VIEW t1view RENAME TO test2.public.t1view;
+
+#Enable cluster setting for sequence owned by across DB's
+statement ok
+SET CLUSTER SETTING "sql.cross_db_sequence_owners.enabled" = true;
+
+statement ok
+ALTER TABLE t1s1ref RENAME TO test2.public.t1s1ref;
+
+statement ok
+ALTER TABLE test2.public.t1s1ref RENAME TO t1s1ref;
+
+statement ok
+ALTER SEQUENCE s1 RENAME TO test2.public.s1;

pkg/sql/rename_table.go

@@ -188,6 +188,15 @@ func (n *renameTableNode) startExec(params runParams) error {
 		}
 	}
 
+	// Special checks for tables, view and sequences to determine if cross
+	// DB references would occur.
+	if oldTn.Catalog() != newTn.Catalog() {
+		err := n.checkForCrossDbReferences(ctx, p, targetDbDesc)
+		if err != nil {
+			return err
+		}
+	}
+
 	// oldTn and newTn are already normalized, so we can compare directly here.
 	if oldTn.Catalog() == newTn.Catalog() &&
 		oldTn.Schema() == newTn.Schema() &&
@@ -295,6 +304,175 @@ func (p *planner) dependentViewError(
 		"you can drop %s instead.", viewName)
 }
 
+// checkForCrossDbReferences validates if any cross DB references
+// will exist after any rename operation.
+func (n *renameTableNode) checkForCrossDbReferences(
+	ctx context.Context, p *planner, targetDbDesc catalog.DatabaseDescriptor,
+) error {
+	tableDesc := n.tableDesc
+
+	// Checks inbound / outbound foreign key references for cross DB references.
+	// The refTableID flag determines if the reference or origin field are checked.
+	checkFkForCrossDbDep := func(fk *descpb.ForeignKeyConstraint, refTableID bool) error {
+		tableID := fk.ReferencedTableID
+		if !refTableID {
+			tableID = fk.OriginTableID
+		}
+
+		referencedTable, err := p.Descriptors().GetImmutableTableByID(ctx, p.txn, tableID,
+			tree.ObjectLookupFlags{
+				CommonLookupFlags: tree.CommonLookupFlags{
+					Required:    true,
+					AvoidCached: true,
+				},
+			})
+		if err != nil {
+			return err
+		}
+		// No cross DB reference
+		if referencedTable.GetParentID() == targetDbDesc.GetID() {
+			return nil
+		}
+		return errors.WithHintf(
+			pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
+				"a foreign key constraint %q will exist between databases after rename "+
+					"(see the '%s' cluster setting)",
+				fk.Name,
+				allowCrossDatabaseFKsSetting),
+			crossDBReferenceDeprecationHint(),
+		)
+	}
+	// Validates if a given dependency on a relation will
+	// lead to a cross DB reference, and an appropriate
+	// error is generated.
+	checkDepForCrossDbRef := func(depID descpb.ID) error {
+		dependentObject, err := p.Descriptors().GetImmutableTableByID(ctx, p.txn, depID,
+			tree.ObjectLookupFlags{
+				CommonLookupFlags: tree.CommonLookupFlags{
+					Required:    true,
+					AvoidCached: true,
+				}})
+		if err != nil {
+			return err
+		}
+		// No cross DB reference detected
+		if dependentObject.GetParentID() == targetDbDesc.GetID() {
+			return nil
+		}
+		// For tables return an error based on if we are depending
+		// on a view or sequence.
+		if tableDesc.IsTable() {
+			if dependentObject.IsView() {
+				return errors.WithHintf(
+					pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
+						"a view %q reference to this table will refer to another databases after rename "+
+							"(see the '%s' cluster setting)",
+						dependentObject.GetName(),
+						allowCrossDatabaseViewsSetting),
+					crossDBReferenceDeprecationHint(),
+				)
+			} else if !allowCrossDatabaseSeqOwner.Get(&p.execCfg.Settings.SV) &&
+				dependentObject.IsSequence() {
+				return errors.WithHintf(
+					pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
+						"a sequence %q will be OWNED BY a table in a different database after rename "+
+							"(see the '%s' cluster setting)",
+						dependentObject.GetName(),
+						allowCrossDatabaseSeqOwnerSetting),
+					crossDBReferenceDeprecationHint(),
+				)
+			}
+		} else if tableDesc.IsView() {
+			// For views it can only be a relation.
+			return errors.WithHintf(
+				pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
+					"this view will reference a table %q in another databases after rename "+
+						"(see the '%s' cluster setting)",
+					dependentObject.GetName(),
+					allowCrossDatabaseViewsSetting),
+				crossDBReferenceDeprecationHint(),
+			)
+		} else if tableDesc.IsSequence() {
+			return errors.WithHintf(
+				pgerror.Newf(pgcode.ObjectNotInPrerequisiteState,
+					"this sequence will be OWNED BY a table %q in a different database after rename "+
+						"(see the '%s' cluster setting)",
+					dependentObject.GetName(),
+					allowCrossDatabaseSeqOwnerSetting),
+				crossDBReferenceDeprecationHint(),
+			)
+		}
+		return nil
+	}
+
+	// For tables check if any outbound or inbound foreign key references would
+	// be impacted.
+	if tableDesc.IsTable() {
+		if !allowCrossDatabaseFKs.Get(&p.execCfg.Settings.SV) {
+			err := tableDesc.ForeachOutboundFK(func(fk *descpb.ForeignKeyConstraint) error {
+				return checkFkForCrossDbDep(fk, true)
+			})
+			if err != nil {
+				return err
+			}
+
+			err = tableDesc.ForeachInboundFK(func(fk *descpb.ForeignKeyConstraint) error {
+				return checkFkForCrossDbDep(fk, false)
+			})
+			if err != nil {
+				return err
+			}
+		}
+
+		// If cross database sequence owners are not allowed, then
+		// check if any column owns a sequence.
+		if !allowCrossDatabaseSeqOwner.Get(&p.execCfg.Settings.SV) {
+			for _, columnDesc := range tableDesc.Columns {
+				for _, ownsSequenceID := range columnDesc.OwnsSequenceIds {
+					err := checkDepForCrossDbRef(ownsSequenceID)
+					if err != nil {
+						return err
+					}
+				}
+			}
+		}
+
+		// Check if any views depend on this table, while
+		// DependsOnBy contains sequences these are only
+		// once that are in use.
+		if !allowCrossDatabaseViews.Get(&p.execCfg.Settings.SV) {
+			err := tableDesc.ForeachDependedOnBy(func(dep *descpb.TableDescriptor_Reference) error {
+				return checkDepForCrossDbRef(dep.ID)
+			})
+			if err != nil {
+				return err
+			}
+		}
+	} else if tableDesc.IsView() &&
+		!allowCrossDatabaseViews.Get(&p.execCfg.Settings.SV) {
+		// For views check if we depend on tables in a different database.
+		dependsOn := tableDesc.GetDependsOn()
+		for _, dependency := range dependsOn {
+			err := checkDepForCrossDbRef(dependency)
+			if err != nil {
+				return err
+			}
+		}
+	} else if tableDesc.IsSequence() &&
+		!allowCrossDatabaseSeqOwner.Get(&p.execCfg.Settings.SV) {
+		// For sequences check if the sequence is owned by
+		// a different database.
+		sequenceOpts := tableDesc.GetSequenceOpts()
+		if sequenceOpts.SequenceOwner.OwnerTableID != descpb.InvalidID {
+			err := checkDepForCrossDbRef(sequenceOpts.SequenceOwner.OwnerTableID)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 // writeNameKey writes a name key to a batch and runs the batch.
 func (p *planner) writeNameKey(
 	ctx context.Context, key catalogkeys.DescriptorKey, ID descpb.ID,