Fix all known problems with pg_dump's handling of serial sequences

by abandoning the idea that it should say SERIAL in the dump.  Instead,
dump serial sequences and column defaults just like regular ones.
Add a new backend command ALTER SEQUENCE OWNED BY to let pg_dump recreate
the sequence-to-column dependency that was formerly created "behind the
scenes" by SERIAL.  This restores SERIAL to being truly "just a macro"
consisting of component operations that can be stated explicitly in SQL.
Furthermore, the new command allows sequence ownership to be reassigned,
so that old mistakes can be cleaned up.

Also, downgrade the OWNED-BY dependency from INTERNAL to AUTO, since there
is no longer any very compelling argument why the sequence couldn't be
dropped while keeping the column.  (This forces initdb, to be sure the
right kinds of dependencies are in there.)

Along the way, add checks to prevent ALTER OWNER or SET SCHEMA on an
owned sequence; you can now only do this indirectly by changing the
owning table's owner or schema.  This is an oversight in previous
releases, but probably not worth back-patching.

2 files changed, 120 insertions, 100 deletions

diff --git a/src/backend/catalog/pg_depend.c b/src/backend/catalog/pg_depend.c
index 5a297d4a27..99cdf5e7e6 100644
--- a/src/backend/catalog/pg_depend.c
+++ b/src/backend/catalog/pg_depend.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/pg_depend.c,v 1.21 2006/07/11 17:26:58 momjian Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/pg_depend.c,v 1.22 2006/08/21 00:57:24 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -163,58 +163,6 @@ deleteDependencyRecordsFor(Oid classId, Oid objectId)
 }
 
 /*
- * objectIsInternalDependency -- return whether the specified object
- * is listed as an internal dependency for some other object.
- *
- * This is used to implement DROP/REASSIGN OWNED.  We cannot invoke
- * performDeletion blindly, because it may try to drop or modify an internal-
- * dependent object before the "main" object, so we need to skip the first
- * object and expect it to be automatically dropped when the main object is
- * dropped.
- */
-bool
-objectIsInternalDependency(Oid classId, Oid objectId)
-{
-	Relation	depRel;
-	ScanKeyData key[2];
-	SysScanDesc scan;
-	HeapTuple	tup;
-	bool		isdep = false;
-
-	depRel = heap_open(DependRelationId, AccessShareLock);
-
-	ScanKeyInit(&key[0],
-				Anum_pg_depend_classid,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(classId));
-	ScanKeyInit(&key[1],
-				Anum_pg_depend_objid,
-				BTEqualStrategyNumber, F_OIDEQ,
-				ObjectIdGetDatum(objectId));
-
-	scan = systable_beginscan(depRel, DependDependerIndexId, true,
-							  SnapshotNow, 2, key);
-
-	while (HeapTupleIsValid(tup = systable_getnext(scan)))
-	{
-		Form_pg_depend depForm = (Form_pg_depend) GETSTRUCT(tup);
-
-		if (depForm->deptype == DEPENDENCY_INTERNAL)
-		{
-			/* No need to keep scanning */
-			isdep = true;
-			break;
-		}
-	}
-
-	systable_endscan(scan);
-
-	heap_close(depRel, AccessShareLock);
-
-	return isdep;
-}
-
-/*
  * Adjust dependency record(s) to point to a different object of the same type
  *
  * classId/objectId specify the referencing object.
@@ -313,6 +261,105 @@ changeDependencyFor(Oid classId, Oid objectId,
 }
 
 /*
+ * Detect whether a sequence is marked as "owned" by a column
+ *
+ * An ownership marker is an AUTO dependency from the sequence to the
+ * column.  If we find one, store the identity of the owning column
+ * into *tableId and *colId and return TRUE; else return FALSE.
+ *
+ * Note: if there's more than one such pg_depend entry then you get
+ * a random one of them returned into the out parameters.  This should
+ * not happen, though.
+ */
+bool
+sequenceIsOwned(Oid seqId, Oid *tableId, int32 *colId)
+{
+	bool		ret = false;
+	Relation	depRel;
+	ScanKeyData key[2];
+	SysScanDesc scan;
+	HeapTuple	tup;
+
+	depRel = heap_open(DependRelationId, AccessShareLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_classid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationRelationId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_objid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(seqId));
+
+	scan = systable_beginscan(depRel, DependDependerIndexId, true,
+							  SnapshotNow, 2, key);
+
+	while (HeapTupleIsValid((tup = systable_getnext(scan))))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+
+		if (depform->refclassid == RelationRelationId &&
+			depform->deptype == DEPENDENCY_AUTO)
+		{
+			*tableId = depform->refobjid;
+			*colId = depform->refobjsubid;
+			ret = true;
+			break;				/* no need to keep scanning */
+		}
+	}
+
+	systable_endscan(scan);
+
+	heap_close(depRel, AccessShareLock);
+
+	return ret;
+}
+
+/*
+ * Remove any existing "owned" markers for the specified sequence.
+ *
+ * Note: we don't provide a special function to install an "owned"
+ * marker; just use recordDependencyOn().
+ */
+void
+markSequenceUnowned(Oid seqId)
+{
+	Relation	depRel;
+	ScanKeyData key[2];
+	SysScanDesc scan;
+	HeapTuple	tup;
+
+	depRel = heap_open(DependRelationId, RowExclusiveLock);
+
+	ScanKeyInit(&key[0],
+				Anum_pg_depend_classid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationRelationId));
+	ScanKeyInit(&key[1],
+				Anum_pg_depend_objid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(seqId));
+
+	scan = systable_beginscan(depRel, DependDependerIndexId, true,
+							  SnapshotNow, 2, key);
+
+	while (HeapTupleIsValid((tup = systable_getnext(scan))))
+	{
+		Form_pg_depend depform = (Form_pg_depend) GETSTRUCT(tup);
+
+		if (depform->refclassid == RelationRelationId &&
+			depform->deptype == DEPENDENCY_AUTO)
+		{
+			simple_heap_delete(depRel, &tup->t_self);
+		}
+	}
+
+	systable_endscan(scan);
+
+	heap_close(depRel, RowExclusiveLock);
+}
+
+/*
  * isObjectPinned()
  *
  * Test if an object is required for basic database functionality.
diff --git a/src/backend/catalog/pg_shdepend.c b/src/backend/catalog/pg_shdepend.c
index 31f1f654de..85e3d968d4 100644
--- a/src/backend/catalog/pg_shdepend.c
+++ b/src/backend/catalog/pg_shdepend.c
@@ -8,7 +8,7 @@
  *
  *
  * IDENTIFICATION
- *	  $PostgreSQL: pgsql/src/backend/catalog/pg_shdepend.c,v 1.13 2006/08/20 21:56:16 alvherre Exp $
+ *	  $PostgreSQL: pgsql/src/backend/catalog/pg_shdepend.c,v 1.14 2006/08/21 00:57:24 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -17,6 +17,7 @@
 #include "access/genam.h"
 #include "access/heapam.h"
 #include "access/xact.h"
+#include "catalog/catalog.h"
 #include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/pg_authid.h"
@@ -869,30 +870,17 @@ shdepDropDependency(Relation sdepRel, Oid classId, Oid objectId,
  * Get the database Id that should be used in pg_shdepend, given the OID
  * of the catalog containing the object.  For shared objects, it's 0
  * (InvalidOid); for all other objects, it's the current database Id.
- *
- * XXX it's awfully tempting to hard-wire this instead of doing a syscache
- * lookup ... but resist the temptation, unless you can prove it's a
- * bottleneck.
  */
 static Oid
 classIdGetDbId(Oid classId)
 {
 	Oid			dbId;
-	HeapTuple	tup;
-
-	tup = SearchSysCache(RELOID,
-						 ObjectIdGetDatum(classId),
-						 0, 0, 0);
-	if (!HeapTupleIsValid(tup))
-		elog(ERROR, "cache lookup failed for relation %u", classId);
 
-	if (((Form_pg_class) GETSTRUCT(tup))->relisshared)
+	if (IsSharedRelation(classId))
 		dbId = InvalidOid;
 	else
 		dbId = MyDatabaseId;
 
-	ReleaseSysCache(tup);
-
 	return dbId;
 }
 
@@ -1055,6 +1043,11 @@ isSharedObjectPinned(Oid classId, Oid objectId, Relation sdepRel)
  * Drop the objects owned by any one of the given RoleIds.	If a role has
  * access to an object, the grant will be removed as well (but the object
  * will not, of course.)
+ *
+ * We can revoke grants immediately while doing the scan, but drops are
+ * saved up and done all at once with performMultipleDeletions.  This
+ * is necessary so that we don't get failures from trying to delete
+ * interdependent objects in the wrong order.
  */
 void
 shdepDropOwned(List *roleids, DropBehavior behavior)
@@ -1113,7 +1106,7 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 			InternalGrant	istmt;
 			Form_pg_shdepend sdepForm = (Form_pg_shdepend) GETSTRUCT(tuple);
 
-			/* We only operate on objects on the current database */
+			/* We only operate on objects in the current database */
 			if (sdepForm->dbid != MyDatabaseId)
 				continue;
 
@@ -1128,24 +1121,8 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 					switch (sdepForm->classid)
 					{
 						case RelationRelationId:
-							{
-								/* is it a sequence or non-sequence? */
-								Form_pg_class pg_class_tuple;
-								HeapTuple	tuple;
-
-								tuple = SearchSysCache(RELOID,
-													   ObjectIdGetDatum(sdepForm->objid),
-													   0, 0, 0);
-								if (!HeapTupleIsValid(tuple))
-									elog(ERROR, "cache lookup failed for relation %u",
-										 sdepForm->objid);
-								pg_class_tuple = (Form_pg_class) GETSTRUCT(tuple);
-								if (pg_class_tuple->relkind == RELKIND_SEQUENCE)
-									istmt.objtype = ACL_OBJECT_SEQUENCE;
-								else
-									istmt.objtype = ACL_OBJECT_RELATION;
-								ReleaseSysCache(tuple);
-							}
+							/* it's OK to use RELATION for a sequence */
+							istmt.objtype = ACL_OBJECT_RELATION;
 							break;
 						case DatabaseRelationId:
 							istmt.objtype = ACL_OBJECT_DATABASE;
@@ -1180,11 +1157,10 @@ shdepDropOwned(List *roleids, DropBehavior behavior)
 					ExecGrantStmt_oids(&istmt);
 					break;
 				case SHARED_DEPENDENCY_OWNER:
-					/* Save it for later deleting it */
+					/* Save it for deletion below */
 					obj.classId = sdepForm->classid;
 					obj.objectId = sdepForm->objid;
 					obj.objectSubId = 0;
-
 					add_exact_object_address(&obj, deleteobjs);
 					break;
 			}
@@ -1259,7 +1235,7 @@ shdepReassignOwned(List *roleids, Oid newrole)
 		{
 			Form_pg_shdepend sdepForm = (Form_pg_shdepend) GETSTRUCT(tuple);
 
-			/* We only operate on objects on the current database */
+			/* We only operate on objects in the current database */
 			if (sdepForm->dbid != MyDatabaseId)
 				continue;
 
@@ -1271,15 +1247,7 @@ shdepReassignOwned(List *roleids, Oid newrole)
 			if (sdepForm->deptype != SHARED_DEPENDENCY_OWNER)
 				continue;
 
-			/*
-			 * If there's a regular (non-shared) dependency on this object
-			 * marked with DEPENDENCY_INTERNAL, skip this object.  We will
-			 * alter the referencer object instead.
-			 */
-			if (objectIsInternalDependency(sdepForm->classid, sdepForm->objid))
-				continue;
-
-			/* Issue the appropiate ALTER OWNER call */
+			/* Issue the appropriate ALTER OWNER call */
 			switch (sdepForm->classid)
 			{
 				case ConversionRelationId:
@@ -1299,7 +1267,12 @@ shdepReassignOwned(List *roleids, Oid newrole)
 					break;
 
 				case RelationRelationId:
-					ATExecChangeOwner(sdepForm->objid, newrole, false);
+					/*
+					 * Pass recursing = true so that we don't fail on
+					 * indexes, owned sequences, etc when we happen
+					 * to visit them before their parent table.
+					 */
+					ATExecChangeOwner(sdepForm->objid, newrole, true);
 					break;
 
 				case ProcedureRelationId: