Support for RENAME/DROP SCHEMA with sequences

Since commit for the support of ALTER SEQUENCE,
sequences use a global name in GTM based on:
db_name.schema_name.sequence_name

This commit permits to rename sequences on GTM
if their schema's name is modified.
This patch permits also to drop a sequence on GTM
in the case that its schema is being dropped in cascade.

7 files changed, 161 insertions, 35 deletions

diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 2932bffd1d..7fdef1c6ac 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -53,6 +53,10 @@
 #include "catalog/pg_user_mapping.h"
 #ifdef PGXC
 #include "catalog/pgxc_class.h"
+#include "pgxc/pgxc.h"
+#include "commands/sequence.h"
+#include "gtm/gtm_c.h"
+#include "access/gtm.h"
 #endif
 #include "commands/comment.h"
 #include "commands/dbcommands.h"
@@ -335,6 +339,89 @@ performMultipleDeletions(const ObjectAddresses *objects,
 	heap_close(depRel, RowExclusiveLock);
 }
 
+#ifdef PGXC
+/*
+ * Check type and class of the given object and rename it properly on GTM
+ */
+static void
+doRename(const ObjectAddress *object, const char *oldname, const char *newname)
+{
+	switch (getObjectClass(object))
+	{
+		case OCLASS_CLASS:
+		{
+			char        relKind = get_rel_relkind(object->objectId);
+
+			/*
+			 * If we are here, a schema is being renamed, a sequence depends on it.
+			 * as sequences' global name use the schema name, this sequence
+			 * has also to be renamed on GTM.
+			 */
+			if (relKind == RELKIND_SEQUENCE && IS_PGXC_COORDINATOR)
+			{
+				Relation relseq = relation_open(object->objectId, AccessShareLock);
+				char *seqname = GetGlobalSeqName(relseq, NULL, oldname);
+				char *newseqname = GetGlobalSeqName(relseq, NULL, newname);
+
+				/* We also need to rename this sequence on GTM, it has a global name ! */
+				if (RenameSequenceGTM(seqname, newseqname) < 0)
+					ereport(ERROR,
+							(errcode(ERRCODE_CONNECTION_FAILURE),
+							 errmsg("GTM error, could not rename sequence")));
+
+				pfree(seqname);
+				pfree(newseqname);
+
+				relation_close(relseq, AccessShareLock);
+			}
+		}
+		default:
+			/* Nothing to do, this object has not to be renamed, end of the story... */
+			break;
+	}
+}
+
+/*
+ * performRename: used to rename objects
+ * on GTM depending on another object(s)
+ */
+void
+performRename(const ObjectAddress *object, const char *oldname, const char *newname)
+{
+	Relation    depRel;
+	ObjectAddresses *targetObjects;
+	int i;
+
+	/*
+	 * Check the dependencies on this object
+	 * And rename object dependent if necessary
+	 */
+
+	depRel = heap_open(DependRelationId, RowExclusiveLock);
+
+	targetObjects = new_object_addresses();
+
+	findDependentObjects(object,
+						 DEPFLAG_ORIGINAL,
+						 NULL,      /* empty stack */
+						 targetObjects,
+						 NULL,
+						 depRel);
+
+	/* Check Objects one by one to see if some of them have to be renamed on GTM */
+	for (i = 0; i < targetObjects->numrefs; i++)
+	{
+		ObjectAddress *thisobj = targetObjects->refs + i;
+		doRename(thisobj, oldname, newname);
+	}
+
+	/* And clean up */
+	free_object_addresses(targetObjects);
+
+	heap_close(depRel, RowExclusiveLock);
+}
+#endif
+
 /*
  * deleteWhatDependsOn: attempt to drop everything that depends on the
  * specified object, though not the object itself.	Behavior is always
@@ -1043,6 +1130,33 @@ doDeletion(const ObjectAddress *object)
 					else
 						heap_drop_with_catalog(object->objectId);
 				}
+
+#ifdef PGXC
+				/* Drop the sequence on GTM */
+				if (relKind == RELKIND_SEQUENCE && IS_PGXC_COORDINATOR)
+				{
+					/*
+					 * The sequence has already been removed from coordinator,
+					 * finish the stuff on GTM too
+					 */
+					/* PGXCTODO: allow the ability to rollback or abort dropping sequences. */
+
+					Relation relseq;
+					char *seqname;
+					/*
+					 * A relation is opened to get the schema and database name as
+					 * such data is not available before when dropping a function.
+					 */
+					relseq = relation_open(object->objectId, AccessShareLock);
+					seqname = GetGlobalSeqName(relseq, NULL, NULL);
+
+					DropSequenceGTM(seqname);
+					pfree(seqname);
+
+					/* Then close the relation opened previously */
+					relation_close(relseq, AccessShareLock);
+				}
+#endif /* PGXC */
 				break;
 			}
 
diff --git a/src/backend/commands/schemacmds.c b/src/backend/commands/schemacmds.c
index 5680da3657..adf5dec63e 100644
--- a/src/backend/commands/schemacmds.c
+++ b/src/backend/commands/schemacmds.c
@@ -31,6 +31,9 @@
 #include "utils/lsyscache.h"
 #include "utils/syscache.h"
 
+#ifdef PGXC
+#include "pgxc/pgxc.h"
+#endif
 
 static void AlterSchemaOwner_internal(HeapTuple tup, Relation rel, Oid newOwnerId);
 
@@ -297,6 +300,26 @@ RenameSchema(const char *oldname, const char *newname)
 	simple_heap_update(rel, &tup->t_self, tup);
 	CatalogUpdateIndexes(rel, tup);
 
+#ifdef PGXC
+	if (IS_PGXC_COORDINATOR)
+	{
+		ObjectAddress		object;
+		Oid					namespaceId;
+
+		/* Check object dependency and see if there is a sequence. If yes rename it */
+		namespaceId = GetSysCacheOid(NAMESPACENAME,
+									 CStringGetDatum(oldname),
+									 0, 0, 0);
+		/* Create the object that will be checked for the dependencies */
+		object.classId = NamespaceRelationId;
+		object.objectId = namespaceId;
+		object.objectSubId = 0;
+
+		/* Rename all the objects depending on this schema */
+		performRename(&object, oldname, newname);
+	}
+#endif
+
 	heap_close(rel, NoLock);
 	heap_freetuple(tup);
 }
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index f4470770ec..cd2fd97078 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -352,7 +352,7 @@ DefineSequence(CreateSeqStmt *seq)
 #ifdef PGXC  /* PGXC_COORD */
 	if (IS_PGXC_COORDINATOR)
 	{
-		char *seqname = GetGlobalSeqName(rel, NULL);
+		char *seqname = GetGlobalSeqName(rel, NULL, NULL);
 
 		/* We also need to create it on the GTM */
 		if (CreateSequenceGTM(seqname,
@@ -494,7 +494,7 @@ AlterSequenceInternal(Oid relid, List *options)
 #ifdef PGXC
 	if (IS_PGXC_COORDINATOR)
 	{
-		char *seqname = GetGlobalSeqName(seqrel, NULL);
+		char *seqname = GetGlobalSeqName(seqrel, NULL, NULL);
 
 		/* We also need to create it on the GTM */
 		if (AlterSequenceGTM(seqname,
@@ -587,7 +587,7 @@ nextval_internal(Oid relid)
 #ifdef PGXC  /* PGXC_COORD */
 	if (IS_PGXC_COORDINATOR)
 	{
-		char *seqname = GetGlobalSeqName(seqrel, NULL);
+		char *seqname = GetGlobalSeqName(seqrel, NULL, NULL);
 
 		/*
 		 * Above, we still use the page as a locking mechanism to handle
@@ -785,7 +785,7 @@ currval_oid(PG_FUNCTION_ARGS)
 #ifdef PGXC
 	if (IS_PGXC_COORDINATOR)
 	{
-		char *seqname = GetGlobalSeqName(seqrel, NULL);
+		char *seqname = GetGlobalSeqName(seqrel, NULL, NULL);
 
 		result = (int64) GetCurrentValGTM(seqname);
 		if (result < 0)
@@ -911,7 +911,7 @@ do_setval(Oid relid, int64 next, bool iscalled)
 #ifdef PGXC
 	if (IS_PGXC_COORDINATOR)
 	{
-		char *seqname = GetGlobalSeqName(seqrel, NULL);
+		char *seqname = GetGlobalSeqName(seqrel, NULL, NULL);
 
 		if (SetValGTM(seqname, next, iscalled) < 0)
 			ereport(ERROR,
@@ -1423,20 +1423,24 @@ init_params(List *options, bool isInit,
  */
 
 char *
-GetGlobalSeqName(Relation seqrel, const char *new_seqname)
+GetGlobalSeqName(Relation seqrel, const char *new_seqname, const char *new_schemaname)
 {
 	char *seqname, *dbname, *schemaname, *relname;
 	int charlen;
 
 	/* Get all the necessary relation names */
 	dbname = get_database_name(seqrel->rd_node.dbNode);
-	schemaname = get_namespace_name(RelationGetNamespace(seqrel));
 	
 	if (new_seqname)
-		relname = new_seqname;
+		relname = (char *) new_seqname;
 	else
 		relname = RelationGetRelationName(seqrel);
 
+	if (new_schemaname)
+		schemaname = (char *) new_schemaname;
+	else
+		schemaname = get_namespace_name(RelationGetNamespace(seqrel));
+
 	/* Calculate the global name size including the dots and \0 */
 	charlen = strlen(dbname) + strlen(schemaname) + strlen(relname) + 3;
 	seqname = (char *) palloc(charlen);
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index dd80748436..ff3a1748da 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -758,29 +758,6 @@ RemoveRelations(DropStmt *drop)
 
 		add_exact_object_address(&obj, objects);
 
-#ifdef PGXC  /* PGXC_COORD */
-		/* PGXCTODO: allow the ability to rollback dropping sequences. */
-
-		/* Drop the sequence */
-		if (IS_PGXC_COORDINATOR && classform->relkind == RELKIND_SEQUENCE)
-		{
-			Relation relseq;
-			char *seqname;
-
-			/*
-			 * A relation is opened to get the schema and database name as 
-			 * such data is not available before when dropping a function.
-			 */
-			relseq = relation_open(obj.objectId, AccessShareLock);
-			seqname = GetGlobalSeqName(relseq, NULL);
-
-			DropSequenceGTM(seqname);
-			pfree(seqname);
-
-			/* Then close the relation opened previously */
-			relation_close(relseq, AccessShareLock);
-		}
-#endif
 		ReleaseSysCache(tuple);
 	}
 
@@ -2110,14 +2087,17 @@ RenameRelation(Oid myrelid, const char *newrelname, ObjectType reltype)
 	if (IS_PGXC_COORDINATOR &&
 		(reltype == OBJECT_SEQUENCE || relkind == RELKIND_SEQUENCE)) /* It is possible to rename a sequence with ALTER TABLE */
 	{
-		char *seqname = GetGlobalSeqName(targetrelation, NULL);
-		char *newseqname = GetGlobalSeqName(targetrelation, newrelname);
+		char *seqname = GetGlobalSeqName(targetrelation, NULL, NULL);
+		char *newseqname = GetGlobalSeqName(targetrelation, newrelname, NULL);
 
 		/* We also need to rename it on the GTM */
 		if (RenameSequenceGTM(seqname, newseqname) < 0)
 			ereport(ERROR,
 					(errcode(ERRCODE_CONNECTION_FAILURE),
 					 errmsg("GTM error, could not rename sequence")));
+
+		pfree(seqname);
+		pfree(newseqname);
 	}
 #endif
 
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 61cfc8ece1..0bb4b4645f 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -581,7 +581,6 @@ ProcessUtility(Node *parsetree,
 			{
 				uint64		processed;
 #ifdef PGXC
-				bool		done;
 				processed = DoCopy((CopyStmt *) parsetree, queryString, true);
 #else
 				processed = DoCopy((CopyStmt *) parsetree, queryString):
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index b2af292585..1a7e3601ad 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -162,6 +162,12 @@ extern void performDeletion(const ObjectAddress *object,
 extern void performMultipleDeletions(const ObjectAddresses *objects,
 						 DropBehavior behavior);
 
+#ifdef PGXC
+extern void performRename(const ObjectAddress *object,
+						  const char *oldname,
+						  const char *newname);
+#endif
+
 extern void deleteWhatDependsOn(const ObjectAddress *object,
 					bool showNotices);
 
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 4f5271d309..a2a6455aa2 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -103,7 +103,7 @@ extern void seq_redo(XLogRecPtr lsn, XLogRecord *rptr);
 extern void seq_desc(StringInfo buf, uint8 xl_info, char *rec);
 
 #ifdef PGXC
-extern char *GetGlobalSeqName(Relation rel, const char *new_seqname);
+extern char *GetGlobalSeqName(Relation rel, const char *new_seqname, const char *new_schemaname);
 #endif
 
 #endif   /* SEQUENCE_H */