Index: dblink.c
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/dblink.c,v
retrieving revision 1.47
diff -c -r1.47 dblink.c
*** dblink.c	15 Oct 2005 02:49:04 -0000	1.47
--- dblink.c	16 Oct 2005 02:04:13 -0000
***************
*** 60,68 ****
  
  typedef struct remoteConn
  {
! 	PGconn	   *conn;			/* Hold the remote connection */
! 	int			autoXactCursors;/* Indicates the number of open cursors,
! 								 * non-zero means we opened the xact ourselves */
  }	remoteConn;
  
  /*
--- 60,68 ----
  
  typedef struct remoteConn
  {
! 	PGconn	   *conn;				/* Hold the remote connection */
! 	int			openCursorCount;	/* The number of open cursors */
! 	bool		newXactForCursor;	/* Opened a transaction for a cursor */
  }	remoteConn;
  
  /*
***************
*** 84,93 ****
  static char *generate_relation_name(Oid relid);
  
  /* Global */
! List	   *res_id = NIL;
! int			res_id_index = 0;
! PGconn	   *persistent_conn = NULL;
! static HTAB *remoteConnHash = NULL;
  
  /*
   *	Following is list that holds multiple remote connections.
--- 84,91 ----
  static char *generate_relation_name(Oid relid);
  
  /* Global */
! static remoteConn	   *pconn = NULL;
! static HTAB			   *remoteConnHash = NULL;
  
  /*
   *	Following is list that holds multiple remote connections.
***************
*** 184,189 ****
--- 182,197 ----
  			} \
  	} while (0)
  
+ #define DBLINK_INIT \
+ 	do { \
+ 			if (!pconn) \
+ 			{ \
+ 				pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn)); \
+ 				pconn->conn = NULL; \
+ 				pconn->openCursorCount = 0; \
+ 				pconn->newXactForCursor = FALSE; \
+ 			} \
+ 	} while (0)
  
  /*
   * Create a persistent connection to another database
***************
*** 199,204 ****
--- 207,214 ----
  	PGconn	   *conn = NULL;
  	remoteConn *rconn = NULL;
  
+ 	DBLINK_INIT;
+ 
  	if (PG_NARGS() == 2)
  	{
  		connstr = GET_STR(PG_GETARG_TEXT_P(1));
***************
*** 234,240 ****
  		createNewConnection(connname, rconn);
  	}
  	else
! 		persistent_conn = conn;
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
--- 244,250 ----
  		createNewConnection(connname, rconn);
  	}
  	else
! 		pconn->conn = conn;
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
***************
*** 250,255 ****
--- 260,267 ----
  	remoteConn *rconn = NULL;
  	PGconn	   *conn = NULL;
  
+ 	DBLINK_INIT;
+ 
  	if (PG_NARGS() == 1)
  	{
  		conname = GET_STR(PG_GETARG_TEXT_P(0));
***************
*** 258,264 ****
  			conn = rconn->conn;
  	}
  	else
! 		conn = persistent_conn;
  
  	if (!conn)
  		DBLINK_CONN_NOT_AVAIL;
--- 270,276 ----
  			conn = rconn->conn;
  	}
  	else
! 		conn = pconn->conn;
  
  	if (!conn)
  		DBLINK_CONN_NOT_AVAIL;
***************
*** 270,276 ****
  		pfree(rconn);
  	}
  	else
! 		persistent_conn = NULL;
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
--- 282,288 ----
  		pfree(rconn);
  	}
  	else
! 		pconn->conn = NULL;
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
***************
*** 285,290 ****
--- 297,304 ----
  	char	   *msg;
  	PGresult   *res = NULL;
  	PGconn	   *conn = NULL;
+ 	int		   *openCursorCount = NULL;
+ 	bool	   *newXactForCursor = NULL;
  	char	   *curname = NULL;
  	char	   *sql = NULL;
  	char	   *conname = NULL;
***************
*** 292,303 ****
  	remoteConn *rconn = NULL;
  	bool		fail = true;	/* default to backward compatible behavior */
  
  	if (PG_NARGS() == 2)
  	{
  		/* text,text */
  		curname = GET_STR(PG_GETARG_TEXT_P(0));
  		sql = GET_STR(PG_GETARG_TEXT_P(1));
! 		conn = persistent_conn;
  	}
  	else if (PG_NARGS() == 3)
  	{
--- 306,321 ----
  	remoteConn *rconn = NULL;
  	bool		fail = true;	/* default to backward compatible behavior */
  
+ 	DBLINK_INIT;
+ 
  	if (PG_NARGS() == 2)
  	{
  		/* text,text */
  		curname = GET_STR(PG_GETARG_TEXT_P(0));
  		sql = GET_STR(PG_GETARG_TEXT_P(1));
! 		conn = pconn->conn;
! 		openCursorCount = &pconn->openCursorCount;
! 		newXactForCursor = &pconn->newXactForCursor;
  	}
  	else if (PG_NARGS() == 3)
  	{
***************
*** 307,313 ****
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			sql = GET_STR(PG_GETARG_TEXT_P(1));
  			fail = PG_GETARG_BOOL(2);
! 			conn = persistent_conn;
  		}
  		else
  		{
--- 325,333 ----
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			sql = GET_STR(PG_GETARG_TEXT_P(1));
  			fail = PG_GETARG_BOOL(2);
! 			conn = pconn->conn;
! 			openCursorCount = &pconn->openCursorCount;
! 			newXactForCursor = &pconn->newXactForCursor;
  		}
  		else
  		{
***************
*** 316,322 ****
--- 336,346 ----
  			sql = GET_STR(PG_GETARG_TEXT_P(2));
  			rconn = getConnectionByName(conname);
  			if (rconn)
+ 			{
  				conn = rconn->conn;
+ 				openCursorCount = &rconn->openCursorCount;
+ 				newXactForCursor = &rconn->newXactForCursor;
+ 			}
  		}
  	}
  	else if (PG_NARGS() == 4)
***************
*** 328,344 ****
  		fail = PG_GETARG_BOOL(3);
  		rconn = getConnectionByName(conname);
  		if (rconn)
  			conn = rconn->conn;
  	}
  
  	if (!conn)
  		DBLINK_CONN_NOT_AVAIL;
  
! 	res = PQexec(conn, "BEGIN");
! 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
! 		DBLINK_RES_INTERNALERROR("begin error");
  
! 	PQclear(res);
  
  	appendStringInfo(str, "DECLARE %s CURSOR FOR %s", curname, sql);
  	res = PQexec(conn, str->data);
--- 352,380 ----
  		fail = PG_GETARG_BOOL(3);
  		rconn = getConnectionByName(conname);
  		if (rconn)
+ 		{
  			conn = rconn->conn;
+ 			openCursorCount = &rconn->openCursorCount;
+ 			newXactForCursor = &rconn->newXactForCursor;
+ 		}
  	}
  
  	if (!conn)
  		DBLINK_CONN_NOT_AVAIL;
  
! 	/*	If we are not in a transaction, start one */
! 	if (PQtransactionStatus(conn) == PQTRANS_IDLE)
! 	{
! 		res = PQexec(conn, "BEGIN");
! 		if (PQresultStatus(res) != PGRES_COMMAND_OK)
! 			DBLINK_RES_INTERNALERROR("begin error");
! 		PQclear(res);
! 		*newXactForCursor = TRUE;
! 	}
  
! 	/* if we started a transaction, increment cursor count */
! 	if (*newXactForCursor)
! 		(*openCursorCount)++;
  
  	appendStringInfo(str, "DECLARE %s CURSOR FOR %s", curname, sql);
  	res = PQexec(conn, str->data);
***************
*** 365,370 ****
--- 401,408 ----
  dblink_close(PG_FUNCTION_ARGS)
  {
  	PGconn	   *conn = NULL;
+ 	int		   *openCursorCount = NULL;
+ 	bool	   *newXactForCursor = NULL;
  	PGresult   *res = NULL;
  	char	   *curname = NULL;
  	char	   *conname = NULL;
***************
*** 373,383 ****
  	remoteConn *rconn = NULL;
  	bool		fail = true;	/* default to backward compatible behavior */
  
  	if (PG_NARGS() == 1)
  	{
  		/* text */
  		curname = GET_STR(PG_GETARG_TEXT_P(0));
! 		conn = persistent_conn;
  	}
  	else if (PG_NARGS() == 2)
  	{
--- 411,425 ----
  	remoteConn *rconn = NULL;
  	bool		fail = true;	/* default to backward compatible behavior */
  
+ 	DBLINK_INIT;
+ 
  	if (PG_NARGS() == 1)
  	{
  		/* text */
  		curname = GET_STR(PG_GETARG_TEXT_P(0));
! 		conn = pconn->conn;
! 		openCursorCount = &pconn->openCursorCount;
! 		newXactForCursor = &pconn->newXactForCursor;
  	}
  	else if (PG_NARGS() == 2)
  	{
***************
*** 386,392 ****
  		{
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			fail = PG_GETARG_BOOL(1);
! 			conn = persistent_conn;
  		}
  		else
  		{
--- 428,436 ----
  		{
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			fail = PG_GETARG_BOOL(1);
! 			conn = pconn->conn;
! 			openCursorCount = &pconn->openCursorCount;
! 			newXactForCursor = &pconn->newXactForCursor;
  		}
  		else
  		{
***************
*** 394,400 ****
--- 438,448 ----
  			curname = GET_STR(PG_GETARG_TEXT_P(1));
  			rconn = getConnectionByName(conname);
  			if (rconn)
+ 			{
  				conn = rconn->conn;
+ 				openCursorCount = &rconn->openCursorCount;
+ 				newXactForCursor = &rconn->newXactForCursor;
+ 			}
  		}
  	}
  	if (PG_NARGS() == 3)
***************
*** 405,411 ****
--- 453,463 ----
  		fail = PG_GETARG_BOOL(2);
  		rconn = getConnectionByName(conname);
  		if (rconn)
+ 		{
  			conn = rconn->conn;
+ 			openCursorCount = &rconn->openCursorCount;
+ 			newXactForCursor = &rconn->newXactForCursor;
+ 		}
  	}
  
  	if (!conn)
***************
*** 428,439 ****
  
  	PQclear(res);
  
! 	/* commit the transaction */
! 	res = PQexec(conn, "COMMIT");
! 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
! 		DBLINK_RES_INTERNALERROR("commit error");
  
! 	PQclear(res);
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
--- 480,501 ----
  
  	PQclear(res);
  
! 	/* if we started a transaction, decrement cursor count */
! 	if (*newXactForCursor)
! 	{
! 		(*openCursorCount)--;
  
! 		/* if count is zero, commit the transaction */
! 		if (*openCursorCount == 0)
! 		{
! 			*newXactForCursor = FALSE;
! 
! 			res = PQexec(conn, "COMMIT");
! 			if (PQresultStatus(res) != PGRES_COMMAND_OK)
! 				DBLINK_RES_INTERNALERROR("commit error");
! 			PQclear(res);
! 		}
! 	}
  
  	PG_RETURN_TEXT_P(GET_TEXT("OK"));
  }
***************
*** 456,461 ****
--- 518,525 ----
  	char	   *conname = NULL;
  	remoteConn *rconn = NULL;
  
+ 	DBLINK_INIT;
+ 
  	/* stuff done only on the first call of the function */
  	if (SRF_IS_FIRSTCALL())
  	{
***************
*** 485,491 ****
  				curname = GET_STR(PG_GETARG_TEXT_P(0));
  				howmany = PG_GETARG_INT32(1);
  				fail = PG_GETARG_BOOL(2);
! 				conn = persistent_conn;
  			}
  			else
  			{
--- 549,555 ----
  				curname = GET_STR(PG_GETARG_TEXT_P(0));
  				howmany = PG_GETARG_INT32(1);
  				fail = PG_GETARG_BOOL(2);
! 				conn = pconn->conn;
  			}
  			else
  			{
***************
*** 503,509 ****
  			/* text,int */
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			howmany = PG_GETARG_INT32(1);
! 			conn = persistent_conn;
  		}
  
  		if (!conn)
--- 567,573 ----
  			/* text,int */
  			curname = GET_STR(PG_GETARG_TEXT_P(0));
  			howmany = PG_GETARG_INT32(1);
! 			conn = pconn->conn;
  		}
  
  		if (!conn)
***************
*** 648,653 ****
--- 712,719 ----
  	MemoryContext oldcontext;
  	bool		freeconn = false;
  
+ 	DBLINK_INIT;
+ 
  	/* stuff done only on the first call of the function */
  	if (SRF_IS_FIRSTCALL())
  	{
***************
*** 678,684 ****
  			/* text,text or text,bool */
  			if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
  			{
! 				conn = persistent_conn;
  				sql = GET_STR(PG_GETARG_TEXT_P(0));
  				fail = PG_GETARG_BOOL(1);
  			}
--- 744,750 ----
  			/* text,text or text,bool */
  			if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
  			{
! 				conn = pconn->conn;
  				sql = GET_STR(PG_GETARG_TEXT_P(0));
  				fail = PG_GETARG_BOOL(1);
  			}
***************
*** 691,697 ****
  		else if (PG_NARGS() == 1)
  		{
  			/* text */
! 			conn = persistent_conn;
  			sql = GET_STR(PG_GETARG_TEXT_P(0));
  		}
  		else
--- 757,763 ----
  		else if (PG_NARGS() == 1)
  		{
  			/* text */
! 			conn = pconn->conn;
  			sql = GET_STR(PG_GETARG_TEXT_P(0));
  		}
  		else
***************
*** 857,862 ****
--- 923,930 ----
  	bool		freeconn = false;
  	bool		fail = true;	/* default to backward compatible behavior */
  
+ 	DBLINK_INIT;
+ 
  	if (PG_NARGS() == 3)
  	{
  		/* must be text,text,bool */
***************
*** 869,875 ****
  		/* might be text,text or text,bool */
  		if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
  		{
! 			conn = persistent_conn;
  			sql = GET_STR(PG_GETARG_TEXT_P(0));
  			fail = PG_GETARG_BOOL(1);
  		}
--- 937,943 ----
  		/* might be text,text or text,bool */
  		if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
  		{
! 			conn = pconn->conn;
  			sql = GET_STR(PG_GETARG_TEXT_P(0));
  			fail = PG_GETARG_BOOL(1);
  		}
***************
*** 882,888 ****
  	else if (PG_NARGS() == 1)
  	{
  		/* must be single text argument */
! 		conn = persistent_conn;
  		sql = GET_STR(PG_GETARG_TEXT_P(0));
  	}
  	else
--- 950,956 ----
  	else if (PG_NARGS() == 1)
  	{
  		/* must be single text argument */
! 		conn = pconn->conn;
  		sql = GET_STR(PG_GETARG_TEXT_P(0));
  	}
  	else
Index: expected/dblink.out
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/expected/dblink.out,v
retrieving revision 1.15
diff -c -r1.15 dblink.out
*** expected/dblink.out	8 Oct 2005 16:10:38 -0000	1.15
--- expected/dblink.out	16 Oct 2005 02:05:28 -0000
***************
*** 436,441 ****
--- 436,523 ----
   ROLLBACK
  (1 row)
  
+ -- test opening cursor in a transaction
+ SELECT dblink_exec('myconn','BEGIN');
+  dblink_exec 
+ -------------
+  BEGIN
+ (1 row)
+ 
+ -- an open transaction will prevent dblink_open() from opening its own
+ SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');
+  dblink_open 
+ -------------
+  OK
+ (1 row)
+ 
+ -- this should not commit the transaction because the client opened it
+ SELECT dblink_close('myconn','rmt_foo_cursor');
+  dblink_close 
+ --------------
+  OK
+ (1 row)
+ 
+ -- this should succeed because we have an open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+   dblink_exec   
+ ----------------
+  DECLARE CURSOR
+ (1 row)
+ 
+ -- commit remote transaction
+ SELECT dblink_exec('myconn','COMMIT');
+  dblink_exec 
+ -------------
+  COMMIT
+ (1 row)
+ 
+ -- test automatic transactions for multiple cursor opens
+ SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');
+  dblink_open 
+ -------------
+  OK
+ (1 row)
+ 
+ -- the second cursor
+ SELECT dblink_open('myconn','rmt_foo_cursor2','SELECT * FROM foo');
+  dblink_open 
+ -------------
+  OK
+ (1 row)
+ 
+ -- this should not commit the transaction
+ SELECT dblink_close('myconn','rmt_foo_cursor2');
+  dblink_close 
+ --------------
+  OK
+ (1 row)
+ 
+ -- this should succeed because we have an open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+   dblink_exec   
+ ----------------
+  DECLARE CURSOR
+ (1 row)
+ 
+ -- this should commit the transaction
+ SELECT dblink_close('myconn','rmt_foo_cursor');
+  dblink_close 
+ --------------
+  OK
+ (1 row)
+ 
+ -- this should fail because there is no open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+ ERROR:  sql error
+ DETAIL:  ERROR:  cursor "xact_test" already exists
+ 
+ -- reset remote transaction state
+ SELECT dblink_exec('myconn','ABORT');
+  dblink_exec 
+ -------------
+  ROLLBACK
+ (1 row)
+ 
  -- open a cursor
  SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');
   dblink_open 
Index: sql/dblink.sql
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/sql/dblink.sql,v
retrieving revision 1.14
diff -c -r1.14 dblink.sql
*** sql/dblink.sql	8 Oct 2005 16:10:38 -0000	1.14
--- sql/dblink.sql	16 Oct 2005 01:59:11 -0000
***************
*** 217,222 ****
--- 217,258 ----
  -- reset remote transaction state
  SELECT dblink_exec('myconn','ABORT');
  
+ -- test opening cursor in a transaction
+ SELECT dblink_exec('myconn','BEGIN');
+ 
+ -- an open transaction will prevent dblink_open() from opening its own
+ SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');
+ 
+ -- this should not commit the transaction because the client opened it
+ SELECT dblink_close('myconn','rmt_foo_cursor');
+ 
+ -- this should succeed because we have an open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+ 
+ -- commit remote transaction
+ SELECT dblink_exec('myconn','COMMIT');
+ 
+ -- test automatic transactions for multiple cursor opens
+ SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');
+ 
+ -- the second cursor
+ SELECT dblink_open('myconn','rmt_foo_cursor2','SELECT * FROM foo');
+ 
+ -- this should not commit the transaction
+ SELECT dblink_close('myconn','rmt_foo_cursor2');
+ 
+ -- this should succeed because we have an open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+ 
+ -- this should commit the transaction
+ SELECT dblink_close('myconn','rmt_foo_cursor');
+ 
+ -- this should fail because there is no open transaction
+ SELECT dblink_exec('myconn','DECLARE xact_test CURSOR FOR SELECT * FROM foo');
+ 
+ -- reset remote transaction state
+ SELECT dblink_exec('myconn','ABORT');
+ 
  -- open a cursor
  SELECT dblink_open('myconn','rmt_foo_cursor','SELECT * FROM foo');