Message-ID: <4A22DCCC.8030807@joeconway.com>
Date: Sun, 31 May 2009 12:38:52 -0700
From: Joe Conway <mail@joeconway.com>
User-Agent: Mozilla/5.0 (X11; U; Linux x86_64; en-US;
	rv:1.8.0.12) Gecko/20071019 Fedora/1.5.0.12-3.fc6 pango-text
	Thunderbird/1.5.0.12 Mnenhy/0.7.5.666
MIME-Version: 1.0
To: Tom Lane <tgl@sss.pgh.pa.us>
CC: "Hackers (PostgreSQL)" <pgsql-hackers@postgresql.org>
Subject: Re: dblink patches for comment
References: <4A1C9C8C.6030405@joeconway.com> <20613.1243466644@sss.pgh.pa.us>
In-Reply-To: <20613.1243466644@sss.pgh.pa.us>
Content-Type: multipart/mixed; boundary="------------090506060107090701050106"

This is a multi-part message in MIME format.
--------------090506060107090701050106
Content-Type: text/plain; charset=ISO-8859-1; format=flowed
Content-Transfer-Encoding: 7bit

Tom Lane wrote:
> It's hard to review it without any docs that say what it's supposed to do.
> (And you'd need to patch the docs anyway, eh?)

Fair enough :-)

Probably better if I break this up in logical chunks too. This patch 
only addresses the refactoring you requested here:
http://archives.postgresql.org/message-id/28719.1230996378@sss.pgh.pa.us

I'll follow up later today with a SQL/MED only patch which includes docs.

Joe

--------------090506060107090701050106
Content-Type: text/x-patch;
 name="dblink.2009.05.31.01-async_refactor.diff"
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
	filename="dblink.2009.05.31.01-async_refactor.diff"

Index: dblink.c
===================================================================
RCS file: /opt/src/cvs/pgsql/contrib/dblink/dblink.c,v
retrieving revision 1.77
diff -c -r1.77 dblink.c
*** dblink.c	1 Jan 2009 17:23:31 -0000	1.77
--- dblink.c	25 May 2009 22:57:22 -0000
***************
*** 77,83 ****
  /*
   * Internal declarations
   */
! static Datum dblink_record_internal(FunctionCallInfo fcinfo, bool is_async, bool do_get);
  static remoteConn *getConnectionByName(const char *name);
  static HTAB *createConnHash(void);
  static void createNewConnection(const char *name, remoteConn * rconn);
--- 78,84 ----
  /*
   * Internal declarations
   */
! static Datum dblink_record_internal(FunctionCallInfo fcinfo, bool is_async);
  static remoteConn *getConnectionByName(const char *name);
  static HTAB *createConnHash(void);
  static void createNewConnection(const char *name, remoteConn * rconn);
***************
*** 689,713 ****
  Datum
  dblink_record(PG_FUNCTION_ARGS)
  {
! 	return dblink_record_internal(fcinfo, false, false);
  }
  
  PG_FUNCTION_INFO_V1(dblink_send_query);
  Datum
  dblink_send_query(PG_FUNCTION_ARGS)
  {
! 	return dblink_record_internal(fcinfo, true, false);
  }
  
  PG_FUNCTION_INFO_V1(dblink_get_result);
  Datum
  dblink_get_result(PG_FUNCTION_ARGS)
  {
! 	return dblink_record_internal(fcinfo, true, true);
  }
  
  static Datum
! dblink_record_internal(FunctionCallInfo fcinfo, bool is_async, bool do_get)
  {
  	FuncCallContext *funcctx;
  	TupleDesc	tupdesc = NULL;
--- 707,753 ----
  Datum
  dblink_record(PG_FUNCTION_ARGS)
  {
! 	return dblink_record_internal(fcinfo, false);
  }
  
  PG_FUNCTION_INFO_V1(dblink_send_query);
  Datum
  dblink_send_query(PG_FUNCTION_ARGS)
  {
! 	PGconn	   *conn = NULL;
! 	char	   *connstr = NULL;
! 	char	   *sql = NULL;
! 	remoteConn *rconn = NULL;
! 	char	   *msg;
! 	bool		freeconn = false;
! 	int			retval;
! 
! 	if (PG_NARGS() == 2)
! 	{
! 		DBLINK_GET_CONN;
! 		sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
! 	}
! 	else
! 		/* shouldn't happen */
! 		elog(ERROR, "wrong number of arguments");
! 
! 	/* async query send */
! 	retval = PQsendQuery(conn, sql);
! 	if (retval != 1)
! 		elog(NOTICE, "%s", PQerrorMessage(conn));
! 
! 	PG_RETURN_INT32(retval);
  }
  
  PG_FUNCTION_INFO_V1(dblink_get_result);
  Datum
  dblink_get_result(PG_FUNCTION_ARGS)
  {
! 	return dblink_record_internal(fcinfo, true);
  }
  
  static Datum
! dblink_record_internal(FunctionCallInfo fcinfo, bool is_async)
  {
  	FuncCallContext *funcctx;
  	TupleDesc	tupdesc = NULL;
***************
*** 775,788 ****
  				/* shouldn't happen */
  				elog(ERROR, "wrong number of arguments");
  		}
! 		else if (is_async && do_get)
  		{
  			/* get async result */
  			if (PG_NARGS() == 2)
  			{
  				/* text,bool */
  				DBLINK_GET_CONN;
! 				fail = PG_GETARG_BOOL(2);
  			}
  			else if (PG_NARGS() == 1)
  			{
--- 815,828 ----
  				/* shouldn't happen */
  				elog(ERROR, "wrong number of arguments");
  		}
! 		else /* is_async */
  		{
  			/* get async result */
  			if (PG_NARGS() == 2)
  			{
  				/* text,bool */
  				DBLINK_GET_CONN;
! 				fail = PG_GETARG_BOOL(1);
  			}
  			else if (PG_NARGS() == 1)
  			{
***************
*** 793,929 ****
  				/* shouldn't happen */
  				elog(ERROR, "wrong number of arguments");
  		}
- 		else
- 		{
- 			/* send async query */
- 			if (PG_NARGS() == 2)
- 			{
- 				DBLINK_GET_CONN;
- 				sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
- 			}
- 			else
- 				/* shouldn't happen */
- 				elog(ERROR, "wrong number of arguments");
- 		}
  
  		if (!conn)
  			DBLINK_CONN_NOT_AVAIL;
  
! 		if (!is_async || (is_async && do_get))
  		{
! 			/* synchronous query, or async result retrieval */
! 			if (!is_async)
! 				res = PQexec(conn, sql);
! 			else
  			{
- 				res = PQgetResult(conn);
- 				/* NULL means we're all done with the async results */
- 				if (!res)
- 				{
- 					MemoryContextSwitchTo(oldcontext);
- 					SRF_RETURN_DONE(funcctx);
- 				}
- 			}
- 
- 			if (!res ||
- 				(PQresultStatus(res) != PGRES_COMMAND_OK &&
- 				 PQresultStatus(res) != PGRES_TUPLES_OK))
- 			{
- 				dblink_res_error(conname, res, "could not execute query", fail);
- 				if (freeconn)
- 					PQfinish(conn);
  				MemoryContextSwitchTo(oldcontext);
  				SRF_RETURN_DONE(funcctx);
  			}
  
! 			if (PQresultStatus(res) == PGRES_COMMAND_OK)
! 			{
! 				is_sql_cmd = true;
! 
! 				/* need a tuple descriptor representing one TEXT column */
! 				tupdesc = CreateTemplateTupleDesc(1, false);
! 				TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
! 								   TEXTOID, -1, 0);
! 
! 				/*
! 				 * and save a copy of the command status string to return as
! 				 * our result tuple
! 				 */
! 				sql_cmd_status = PQcmdStatus(res);
! 				funcctx->max_calls = 1;
! 			}
! 			else
! 				funcctx->max_calls = PQntuples(res);
! 
! 			/* got results, keep track of them */
! 			funcctx->user_fctx = res;
! 
! 			/* if needed, close the connection to the database and cleanup */
  			if (freeconn)
  				PQfinish(conn);
  
! 			if (!is_sql_cmd)
! 			{
! 				/* get a tuple descriptor for our result type */
! 				switch (get_call_result_type(fcinfo, NULL, &tupdesc))
! 				{
! 					case TYPEFUNC_COMPOSITE:
! 						/* success */
! 						break;
! 					case TYPEFUNC_RECORD:
! 						/* failed to determine actual type of RECORD */
! 						ereport(ERROR,
! 								(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 						errmsg("function returning record called in context "
! 							   "that cannot accept type record")));
! 						break;
! 					default:
! 						/* result type isn't composite */
! 						elog(ERROR, "return type must be a row type");
! 						break;
! 				}
  
! 				/* make sure we have a persistent copy of the tupdesc */
! 				tupdesc = CreateTupleDescCopy(tupdesc);
! 			}
  
  			/*
! 			 * check result and tuple descriptor have the same number of
! 			 * columns
  			 */
! 			if (PQnfields(res) != tupdesc->natts)
! 				ereport(ERROR,
! 						(errcode(ERRCODE_DATATYPE_MISMATCH),
! 						 errmsg("remote query result rowtype does not match "
! 								"the specified FROM clause rowtype")));
  
! 			/* fast track when no results */
! 			if (funcctx->max_calls < 1)
  			{
! 				if (res)
! 					PQclear(res);
! 				MemoryContextSwitchTo(oldcontext);
! 				SRF_RETURN_DONE(funcctx);
  			}
  
! 			/* store needed metadata for subsequent calls */
! 			attinmeta = TupleDescGetAttInMetadata(tupdesc);
! 			funcctx->attinmeta = attinmeta;
! 
! 			MemoryContextSwitchTo(oldcontext);
  		}
! 		else
  		{
! 			/* async query send */
  			MemoryContextSwitchTo(oldcontext);
! 			PG_RETURN_INT32(PQsendQuery(conn, sql));
  		}
- 	}
  
! 	if (is_async && !do_get)
! 	{
! 		/* async query send -- should not happen */
! 		elog(ERROR, "async query send called more than once");
  
  	}
  
--- 833,942 ----
  				/* shouldn't happen */
  				elog(ERROR, "wrong number of arguments");
  		}
  
  		if (!conn)
  			DBLINK_CONN_NOT_AVAIL;
  
! 		/* synchronous query, or async result retrieval */
! 		if (!is_async)
! 			res = PQexec(conn, sql);
! 		else
  		{
! 			res = PQgetResult(conn);
! 			/* NULL means we're all done with the async results */
! 			if (!res)
  			{
  				MemoryContextSwitchTo(oldcontext);
  				SRF_RETURN_DONE(funcctx);
  			}
+ 		}
  
! 		if (!res ||
! 			(PQresultStatus(res) != PGRES_COMMAND_OK &&
! 			 PQresultStatus(res) != PGRES_TUPLES_OK))
! 		{
! 			dblink_res_error(conname, res, "could not execute query", fail);
  			if (freeconn)
  				PQfinish(conn);
+ 			MemoryContextSwitchTo(oldcontext);
+ 			SRF_RETURN_DONE(funcctx);
+ 		}
  
! 		if (PQresultStatus(res) == PGRES_COMMAND_OK)
! 		{
! 			is_sql_cmd = true;
  
! 			/* need a tuple descriptor representing one TEXT column */
! 			tupdesc = CreateTemplateTupleDesc(1, false);
! 			TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status",
! 							   TEXTOID, -1, 0);
  
  			/*
! 			 * and save a copy of the command status string to return as
! 			 * our result tuple
  			 */
! 			sql_cmd_status = PQcmdStatus(res);
! 			funcctx->max_calls = 1;
! 		}
! 		else
! 			funcctx->max_calls = PQntuples(res);
  
! 		/* got results, keep track of them */
! 		funcctx->user_fctx = res;
! 
! 		/* if needed, close the connection to the database and cleanup */
! 		if (freeconn)
! 			PQfinish(conn);
! 
! 		if (!is_sql_cmd)
! 		{
! 			/* get a tuple descriptor for our result type */
! 			switch (get_call_result_type(fcinfo, NULL, &tupdesc))
  			{
! 				case TYPEFUNC_COMPOSITE:
! 					/* success */
! 					break;
! 				case TYPEFUNC_RECORD:
! 					/* failed to determine actual type of RECORD */
! 					ereport(ERROR,
! 							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
! 					errmsg("function returning record called in context "
! 						   "that cannot accept type record")));
! 					break;
! 				default:
! 					/* result type isn't composite */
! 					elog(ERROR, "return type must be a row type");
! 					break;
  			}
  
! 			/* make sure we have a persistent copy of the tupdesc */
! 			tupdesc = CreateTupleDescCopy(tupdesc);
  		}
! 
! 		/*
! 		 * check result and tuple descriptor have the same number of
! 		 * columns
! 		 */
! 		if (PQnfields(res) != tupdesc->natts)
! 			ereport(ERROR,
! 					(errcode(ERRCODE_DATATYPE_MISMATCH),
! 					 errmsg("remote query result rowtype does not match "
! 							"the specified FROM clause rowtype")));
! 
! 		/* fast track when no results */
! 		if (funcctx->max_calls < 1)
  		{
! 			if (res)
! 				PQclear(res);
  			MemoryContextSwitchTo(oldcontext);
! 			SRF_RETURN_DONE(funcctx);
  		}
  
! 		/* store needed metadata for subsequent calls */
! 		attinmeta = TupleDescGetAttInMetadata(tupdesc);
! 		funcctx->attinmeta = attinmeta;
! 
! 		MemoryContextSwitchTo(oldcontext);
  
  	}
  

--------------090506060107090701050106--