Re: dblink patches for comment - Mailing list pgsql-hackers

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
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);

      }