Support for CLEAN CONNECTION

Utility to clean up Postgres-XC Pooler connections.
This utility is launched to all the Coordinators of the cluster

Use of CLEAN CONNECTION is limited to a super user.
It is advised to clean connections before dropping a Database.

SQL query synopsis is as follows:
CLEAN CONNECTION TO
             (COORDINATOR num | DATANODE num | ALL {FORCE})
              FOR DATABASE dbname
Connection cleaning has to be made on a chosen database called dbname.

It is also possible to clean connections of several Coordinators or Datanodes
Ex:  CLEAN CONNECTION TO DATANODE 1,5,7 FOR DATABASE template1
             CLEAN CONNECTION TO COORDINATOR 2,4,6 FOR DATABASE template1

Or even to all Coordinators/Datanodes at the same time
Ex:  CLEAN CONNECTION TO DATANODE * FOR DATABASE template1
             CLEAN CONNECTION TO COORDINATOR * FOR DATABASE template1

When FORCE is used, all the transactions using pooler connections are aborted,
and pooler connections are cleaned up.
Ex:  CLEAN CONNECTION TO ALL FORCE FOR DATABASE template1;

FORCE can only be used with TO ALL, as it takes a lock on pooler to stop requests
asking for connections, aborts all the connections in the cluster, and cleans up
pool connections

6 files changed, 58 insertions, 0 deletions

diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 2d60d918d5..10b259ad85 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -355,6 +355,7 @@ typedef enum NodeTag
 	T_AlterUserMappingStmt,
 	T_DropUserMappingStmt,
 	T_ExecDirectStmt,
+	T_CleanConnStmt,
 
 	/*
 	 * TAGS FOR PARSE TREE NODES (parsenodes.h)
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 355c44847e..79b5fac560 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2416,6 +2416,18 @@ typedef struct ExecDirectStmt
 	List	   *nodes;
 	char	   *query;
 } ExecDirectStmt;
+
+/*
+ * CLEAN CONNECTION statement
+ */
+typedef struct CleanConnStmt
+{
+	NodeTag		type;
+	List	   *nodes;		/* list of nodes dropped */
+	char	   *dbname;		/* name of database to drop connections */
+	bool		is_coord;	/* type of connections dropped */
+	bool		is_force;	/* option force  */
+} CleanConnStmt;
 /* PGXC_END */
 
 #endif   /* PARSENODES_H */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index aec7b6b3d9..62c7c0dcb9 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -75,6 +75,9 @@ PG_KEYWORD("characteristics", CHARACTERISTICS, UNRESERVED_KEYWORD)
 PG_KEYWORD("check", CHECK, RESERVED_KEYWORD)
 PG_KEYWORD("checkpoint", CHECKPOINT, UNRESERVED_KEYWORD)
 PG_KEYWORD("class", CLASS, UNRESERVED_KEYWORD)
+#ifdef PGXC
+PG_KEYWORD("clean", CLEAN, UNRESERVED_KEYWORD)
+#endif
 PG_KEYWORD("close", CLOSE, UNRESERVED_KEYWORD)
 PG_KEYWORD("cluster", CLUSTER, UNRESERVED_KEYWORD)
 PG_KEYWORD("coalesce", COALESCE, COL_NAME_KEYWORD)
diff --git a/src/include/pgxc/poolcomm.h b/src/include/pgxc/poolcomm.h
index 9e286ab292..a10af3da9e 100644
--- a/src/include/pgxc/poolcomm.h
+++ b/src/include/pgxc/poolcomm.h
@@ -45,5 +45,9 @@ extern int	pool_putbytes(PoolPort *port, const char *s, size_t len);
 extern int	pool_flush(PoolPort *port);
 extern int	pool_sendfds(PoolPort *port, int *fds, int count);
 extern int	pool_recvfds(PoolPort *port, int *fds, int count);
+extern int	pool_sendres(PoolPort *port, int res);
+extern int	pool_recvres(PoolPort *port);
+extern int	pool_sendpids(PoolPort *port, int *pids, int count);
+extern int	pool_recvpids(PoolPort *port, int **pids);
 
 #endif   /* POOLCOMM_H */
diff --git a/src/include/pgxc/poolmgr.h b/src/include/pgxc/poolmgr.h
index 3a615cc0f3..7b6e4d4758 100644
--- a/src/include/pgxc/poolmgr.h
+++ b/src/include/pgxc/poolmgr.h
@@ -63,6 +63,8 @@ typedef struct databasepool
  */
 typedef struct
 {
+	/* Process ID of postmaster child process associated to pool agent */
+	int			pid;
 	/* communication channel */
 	PoolPort	port;
 	DatabasePool *pool;
@@ -132,6 +134,12 @@ extern void PoolManagerConnect(PoolHandle *handle, const char *database);
 /* Get pooled connections */
 extern int *PoolManagerGetConnections(List *datanodelist, List *coordlist);
 
+/* Clean pool connections */
+extern void PoolManagerCleanConnection(List *datanodelist, List *coordlist, char *dbname);
+
+/* Send Abort signal to transactions being run */
+extern int	PoolManagerAbortTransactions(char *dbname, int **proc_pids);
+
 /* Return connections back to the pool, for both Coordinator and Datanode connections */
 extern void PoolManagerReleaseConnections(int dn_ndisc, int* dn_discard, int co_ndisc, int* co_discard);
 
diff --git a/src/include/pgxc/poolutils.h b/src/include/pgxc/poolutils.h
new file mode 100644
index 0000000000..0f34561253
--- /dev/null
+++ b/src/include/pgxc/poolutils.h
@@ -0,0 +1,30 @@
+/*-------------------------------------------------------------------------
+ *
+ * poolutils.h
+ *
+ *		Utilities for Postgres-XC Pooler
+ *
+ * Portions Copyright (c) 1996-2009, PostgreSQL Global Development Group
+ * Portions Copyright (c) 2010 Nippon Telegraph and Telephone Corporation
+ *
+ * IDENTIFICATION
+ *	$$
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef POOLUTILS_H
+#define POOLUTILS_H
+
+#include "nodes/parsenodes.h"
+
+#define TIMEOUT_CLEAN_LOOP 10 /* Wait 10s for all the transactions to shutdown */
+
+/* Error codes for connection cleaning */
+#define CLEAN_CONNECTION_COMPLETED			0
+#define CLEAN_CONNECTION_NOT_COMPLETED		1
+#define CLEAN_CONNECTION_TX_REMAIN			2
+#define	CLEAN_CONNECTION_EOF				-1
+
+void CleanConnection(CleanConnStmt *stmt);
+#endif