Hi,
[Sending this message (first) to -hackers for discussion about the
extension and followed implementation.]
On Apr 01 09:39, Volkan YAZICI wrote:
> I've prepared a patch for the Describe <-> ParameterDescription
> messaging which is available via current extended query protocol.
Here's a written from scratch patch for the above mentioned extension.
It adds PQdescribePrepared() and PQdescribePortal() functions to the
libpq. New functions work as follows:
1. Issue a PQdescribePrepared() call.
2. First PQgetResult() will return a PGresult with input parameter
types of the prepared statement. (You can use PQftype() on this
PGresult to extract information.)
3. Second PQgetResult() will return another PGresult which holds the
column information for the will be returned tuples. (All PQf*()
functions can be used on this result.)
(A PQdescribePortal() call will just skip the 2nd step in the above
list.)
Patch passes regression tests and there're two examples attached for
PQdescribePrepared() and PQdescribePortal() usage.
To mention about the followed implementation, it needed some hack on
pqParseInput3() code to make it understand if a received message is
a reponse to a Describe ('D') query or to another tuple returning
query. To summarize problem, there're two possible forms of a 'D'
response:
1. Description of a prepared statement: t, T, Z
2. Description of a portal: T, Z
The problem is, AFAICS, it's not possible to distinguish between a tuple
returning query (T, ..., C, Z or T, E) and a description of a portal (T,
Z). Therefore, I've created a global flag (parsing_row_desc) which is
turned on when we receive a 'T' and turned off if we receive a 'C' or
'E'. It's a kind of ugly method but the only solution I could come up
with.
Regards.
Index: src/backend/tcop/postgres.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/backend/tcop/postgres.c,v
retrieving revision 1.483
diff -c -r1.483 postgres.c
*** src/backend/tcop/postgres.c 4 Apr 2006 19:35:35 -0000 1.483
--- src/backend/tcop/postgres.c 15 Apr 2006 07:39:49 -0000
***************
*** 1870,1875 ****
--- 1870,1876 ----
static void
exec_describe_statement_message(const char *stmt_name)
{
+ MemoryContext oldContext;
PreparedStatement *pstmt;
TupleDesc tupdesc;
ListCell *l;
***************
*** 1882,1888 ****
start_xact_command();
/* Switch back to message context */
! MemoryContextSwitchTo(MessageContext);
/* Find prepared statement */
if (stmt_name[0] != '\0')
--- 1883,1889 ----
start_xact_command();
/* Switch back to message context */
! oldContext = MemoryContextSwitchTo(MessageContext);
/* Find prepared statement */
if (stmt_name[0] != '\0')
***************
*** 1940,1946 ****
--- 1941,1950 ----
NULL);
else
pq_putemptymessage('n'); /* NoData */
+
+ MemoryContextSwitchTo(oldContext);
+ finish_xact_command();
}
/*
***************
*** 1951,1956 ****
--- 1955,1961 ----
static void
exec_describe_portal_message(const char *portal_name)
{
+ MemoryContext oldContext;
Portal portal;
/*
***************
*** 1960,1966 ****
start_xact_command();
/* Switch back to message context */
! MemoryContextSwitchTo(MessageContext);
portal = GetPortalByName(portal_name);
if (!PortalIsValid(portal))
--- 1965,1971 ----
start_xact_command();
/* Switch back to message context */
! oldContext = MemoryContextSwitchTo(MessageContext);
portal = GetPortalByName(portal_name);
if (!PortalIsValid(portal))
***************
*** 1992,1997 ****
--- 1997,2006 ----
portal->formats);
else
pq_putemptymessage('n'); /* NoData */
+
+ MemoryContextSwitchTo(oldContext);
+
+ finish_xact_command();
}
Index: src/interfaces/libpq/fe-exec.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/interfaces/libpq/fe-exec.c,v
retrieving revision 1.182
diff -c -r1.182 fe-exec.c
*** src/interfaces/libpq/fe-exec.c 14 Mar 2006 22:48:23 -0000 1.182
--- src/interfaces/libpq/fe-exec.c 15 Apr 2006 07:39:58 -0000
***************
*** 55,60 ****
--- 55,62 ----
static void parseInput(PGconn *conn);
static bool PQexecStart(PGconn *conn);
static PGresult *PQexecFinish(PGconn *conn);
+ static int pqDescribe(PGconn *conn, const char desc_type,
+ const char *desc_target);
/* ----------------
***************
*** 2281,2286 ****
--- 2283,2392 ----
return 0;
}
+
+ /*
+ * pqDescribe - Describe given prepared statement or portal.
+ *
+ * Available options for target_type are
+ * 'S' to describe a prepared statement; or
+ * 'P' to describe a portal.
+ * Returns 0 on success and 1 on failure.
+ *
+ * By issuing a PQgetResult(), response from the server will be placed
+ * in an empty PGresult which will be extractable via PQf*() function family.
+ */
+ static int
+ pqDescribe(PGconn *conn, const char desc_type, const char *desc_target)
+ {
+ int ret;
+
+ /* This isn't gonna work on a 2.0 server. */
+ if (PG_PROTOCOL_MAJOR(conn->pversion) < 3)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("function
requires at least protocol version 3.0\n"));
+ return 1;
+ }
+
+ if (!conn)
+ return 1;
+
+ /* Clear the connection error message. */
+ resetPQExpBuffer(&conn->errorMessage);
+
+ /* Don't try to send if we know there's no live connection. */
+ if (conn->status != CONNECTION_OK)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("no connection
to the server\n"));
+ return 1;
+ }
+
+ /* Can't send while already busy, either. */
+ if (conn->asyncStatus != PGASYNC_IDLE)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("another
command is already in progress\n"));
+ return 1;
+ }
+
+ /* Initialize async result-accumulation state. */
+ conn->result = NULL;
+ conn->curTuple = NULL;
+
+ if (desc_target)
+ ret = (pqPutMsgStart('D', false, conn) < 0 ||
+ pqPutc(desc_type, conn) < 0 ||
+ pqPuts(desc_target, conn) < 0 ||
+ pqPutMsgEnd(conn) < 0);
+
+ /* NULL desc_target values will be treated as an empty string. */
+ else
+ ret = (pqPutMsgStart('D', false, conn) < 0 ||
+ pqPutc(desc_type, conn) < 0 ||
+ pqPutc('\0', conn) < 0 ||
+ pqPutMsgEnd(conn) < 0);
+
+ if (ret)
+ goto SendFailure;
+
+ /* Extended protocol requires a Sync message. */
+ if (pqPutMsgStart('S', false, conn) < 0 ||
+ pqPutMsgEnd(conn) < 0)
+ goto SendFailure;
+
+ /* Remember we are using extended query protocol. */
+ conn->queryclass = PGQUERY_EXTENDED;
+
+ /* Free last query string. */
+ if (conn->last_query)
+ {
+ free(conn->last_query);
+ conn->last_query = NULL;
+ }
+
+ /* Describe request is sent. */
+ conn->asyncStatus = PGASYNC_BUSY;
+ return 0;
+
+ SendFailure:
+ pqHandleSendFailure(conn);
+ return 1;
+ }
+
+ int
+ PQdescribePrepared(PGconn *conn, const char *stmt)
+ {
+ return pqDescribe(conn, 'S', stmt);
+ }
+
+ int
+ PQdescribePortal(PGconn *conn, const char *portal)
+ {
+ return pqDescribe(conn, 'P', portal);
+ }
+
+
/* PQsetnonblocking:
* sets the PGconn's database connection non-blocking if the arg is TRUE
* or makes it non-blocking if the arg is FALSE, this will not protect
***************
*** 2346,2351 ****
--- 2452,2458 ----
free(ptr);
}
+
/*
* PQfreeNotify - free's the memory associated with a PGnotify
*
Index: src/interfaces/libpq/fe-protocol3.c
===================================================================
RCS file: /projects/cvsroot/pgsql/src/interfaces/libpq/fe-protocol3.c,v
retrieving revision 1.26
diff -c -r1.26 fe-protocol3.c
*** src/interfaces/libpq/fe-protocol3.c 14 Mar 2006 22:48:23 -0000 1.26
--- src/interfaces/libpq/fe-protocol3.c 15 Apr 2006 07:40:05 -0000
***************
*** 42,50 ****
--- 42,56 ----
((id) == 'T' || (id) == 'D' || (id) == 'd' || (id) == 'V' || \
(id) == 'E' || (id) == 'N' || (id) == 'A')
+ /*
+ * Flag to distinguish between an odd RowDescription and start of
+ * a tuple returning query.
+ */
+ static bool parsing_row_desc = false;
static void handleSyncLoss(PGconn *conn, char id, int msgLength);
static int getRowDescriptions(PGconn *conn);
+ static int getParamDescriptions(PGconn *conn);
static int getAnotherTuple(PGconn *conn, int msgLength);
static int getParameterStatus(PGconn *conn);
static int getNotify(PGconn *conn);
***************
*** 206,221 ****
--- 212,247 ----
}
strncpy(conn->result->cmdStatus,
conn->workBuffer.data,
CMDSTATUS_LEN);
+
+ /* This cannot be a Describe ('D')
response no more. */
+ parsing_row_desc = false;
+
conn->asyncStatus = PGASYNC_READY;
break;
case 'E': /* error return */
if (pqGetErrorNotice3(conn, true))
return;
conn->asyncStatus = PGASYNC_READY;
+
+ /* No further RowDescription parsing is
possible. */
+ parsing_row_desc = false;
break;
case 'Z': /* backend is ready for
new query */
if (getReadyForQuery(conn))
return;
+
+ if (parsing_row_desc)
+ {
+ /*
+ * This is probably received
from a Describe ('D')
+ * query's response.
+ */
+ parsing_row_desc = false;
+ conn->asyncStatus =
PGASYNC_READY;
+ return;
+ }
+
+ /* Backend is ready for a new query. */
conn->asyncStatus = PGASYNC_IDLE;
break;
case 'I': /* empty query */
***************
*** 268,273 ****
--- 294,302 ----
/* First 'T' in a query
sequence */
if (getRowDescriptions(conn))
return;
+
+ /* Turning "RowDesc is parsed"
flag to on. */
+ parsing_row_desc = true;
}
else
{
***************
*** 294,299 ****
--- 323,348 ----
conn->result =
PQmakeEmptyPGresult(conn,
PGRES_COMMAND_OK);
break;
+ case 't': /* Parameter
Description */
+ if (!conn->result)
+ {
+ /* A new PGresult will be
created to load parsed data into. */
+ if (getParamDescriptions(conn))
+ return;
+
+ /* Result is ready. */
+ conn->asyncStatus =
PGASYNC_READY;
+ }
+ else
+ {
+ /*
+ * We'll wait until the
application accepts the current
+ * existing result.
+ */
+ conn->asyncStatus =
PGASYNC_READY;
+ return;
+ }
+ break;
case 'D': /* Data Row */
if (conn->result != NULL &&
conn->result->resultStatus ==
PGRES_TUPLES_OK)
***************
*** 500,505 ****
--- 549,620 ----
}
/*
+ * parseInput subroutine to read a 't' (ParameterDescription) message.
+ * Returns 0 if parsing is completed, 1 if there isn't enough data yet
+ * and EOF if an error occurs.
+ *
+ * Parsed data will get loaded into the existing conn->result.
+ */
+ static int
+ getParamDescriptions(PGconn *conn)
+ {
+ PGresult *res;
+ int nattr;
+ int i;
+ Oid typid;
+
+ res = PQmakeEmptyPGresult(conn, PGRES_COMMAND_OK);
+ if (!res)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("out of
memory\n"));
+ EOF;
+ }
+
+ /* First extracting number of attributes from the message. */
+ if (pqGetInt(&nattr, 2, conn) < 0)
+ goto NotEnoughData;
+ res->numAttributes = nattr;
+
+ /* Allocate space for the attribute descriptors. */
+ if (nattr > 0)
+ {
+ int sz = nattr * sizeof(PGresAttDesc);
+
+ res->attDescs = (PGresAttDesc *) pqResultAlloc(res, sz, TRUE);
+ if (!res->attDescs)
+ {
+ printfPQExpBuffer(&conn->errorMessage,
+ libpq_gettext("out of
memory\n"));
+ goto Error;
+ }
+
+ /* Reset attribute values. */
+ MemSet(res->attDescs, 0, sz);
+ }
+
+ /* Loop through attribute's type OIDs. */
+ for (i = 0; i < nattr; i++)
+ {
+ if (pqGetInt(&typid, 4, conn) < 0)
+ goto NotEnoughData;
+ res->attDescs[i].typid = typid;
+ }
+
+ /* Success. */
+ conn->result = res;
+ return 0;
+
+ NotEnoughData:
+ PQclear(res);
+ return 1;
+
+ Error:
+ PQclear(res);
+ return EOF;
+ }
+
+ /*
* parseInput subroutine to read a 'D' (row data) message.
* We add another tuple to the existing PGresult structure.
* Returns: 0 if completed message, EOF if error or not enough data yet.
Index: src/interfaces/libpq/libpq-fe.h
===================================================================
RCS file: /projects/cvsroot/pgsql/src/interfaces/libpq/libpq-fe.h,v
retrieving revision 1.126
diff -c -r1.126 libpq-fe.h
*** src/interfaces/libpq/libpq-fe.h 20 Mar 2006 15:07:05 -0000 1.126
--- src/interfaces/libpq/libpq-fe.h 15 Apr 2006 07:40:08 -0000
***************
*** 414,419 ****
--- 414,423 ----
extern int PQgetlength(const PGresult *res, int tup_num, int field_num);
extern int PQgetisnull(const PGresult *res, int tup_num, int field_num);
+ /* Describe prepared statements and portals. */
+ extern int PQdescribePrepared(PGconn *conn, const char *stmt);
+ extern int PQdescribePortal(PGconn *conn, const char *portal);
+
/* Delete a PGresult */
extern void PQclear(PGresult *res);
#include <stdio.h>
#include "libpq-fe.h"
int
main(void)
{
PGconn *conn = PQconnectdb("dbname=template1");
PGresult *res;
int ret;
/*
* Make an erronous Describe ('D') request.
*/
printf("=== PQdescribePrepared>\n");
ret = PQdescribePrepared(conn, "not_exists");
if (ret)
{
printf("PQdescribePrepared() failed! [%s]\n",
PQerrorMessage(conn));
PQfinish(conn);
return 1;
}
/*
* Receive erronous describe response.
*/
printf("=== PQgetResult>\n");
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
printf("PQgetResult() failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
printf("(***Expected Error.***)\n");
}
PQclear(res);
/*
* Flush result set.
*/
printf("=== PQgetResult> (Flush result set)\n");
res = PQgetResult(conn);
if (res)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
/*
* PREPARE a statement to describe.
*/
printf("=== PREPARE>\n");
res = PQexec(conn,
"PREPARE my_plan (int4, text, float) AS"
" SELECT "
" (random()*$1)::float8 AS c1, "
" $2 AS c2, "
" (random()*$3)::int8 AS c3, "
" 'some text' AS c4;");
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
PQclear(res);
/*
* Make a Describe ('D') request.
*/
printf("=== PQdescribePrepared>\n");
ret = PQdescribePrepared(conn, "my_plan");
if (ret)
{
printf("PQdescribePrepared() failed! [%s]\n",
PQerrorMessage(conn));
PQfinish(conn);
return 1;
}
/*
* Receive Describe response (ParameterDescription).
*/
printf("=== PQgetResult>\n");
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
printf("PQgetResult() failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
else
{
int i, n = PQnfields(res);
printf("PQftype\n");
for (i = 0; i < n; i++)
printf(" %d:%d\n", i, PQftype(res, i));
}
PQclear(res);
printf("=== PQgetResult>\n");
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
printf("PQgetResult() failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
else
{
int i, n = PQnfields(res);
printf("PQfname\n");
for (i = 0; i < n; i++)
printf(" %d:%s\n", i, PQfname(res, i));
}
PQclear(res);
/*
* Flush result set.
*/
printf("=== PQgetResult> (Flush result set)\n");
res = PQgetResult(conn);
if (res)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
PQclear(res);
PQfinish(conn);
return 0;
}
#include <stdio.h>
#include "libpq-fe.h"
int
main(void)
{
PGconn *conn = PQconnectdb("dbname=template1");
PGresult *res;
int ret;
/*
* Make an erronous Describe ('D') request.
*/
printf("=== PQdescribePrepared>\n");
ret = PQdescribePrepared(conn, "not_exists");
if (ret)
{
printf("PQdescribePrepared() failed! [%s]\n",
PQerrorMessage(conn));
PQfinish(conn);
return 1;
}
/*
* Receive erronous describe response.
*/
printf("=== PQgetResult>\n");
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
printf("PQgetResult() failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
printf("(***Expected Error.***)\n");
}
PQclear(res);
/*
* Flush result set.
*/
printf("=== PQgetResult> (Flush result set)\n");
res = PQgetResult(conn);
if (res)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
/*
* DECLARE a statement to describe.
*/
printf("=== DECLARE>\n");
res = PQexec(conn,
"BEGIN; DECLARE curs CURSOR FOR"
" SELECT "
" (random()*100)::float8 AS c1, "
" 43 AS c2, "
" (random()*1000)::int8 AS c3, "
" 'some text' AS c4;");
if (PQresultStatus(res) != PGRES_COMMAND_OK)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
PQclear(res);
/*
* Make a Describe ('D') request.
*/
printf("=== PQdescribePortal>\n");
ret = PQdescribePortal(conn, "curs");
if (ret)
{
printf("PQdescribePortal() failed! [%s]\n",
PQerrorMessage(conn));
PQfinish(conn);
return 1;
}
/*
* Receive Describe response (ParameterDescription).
*/
printf("=== PQgetResult>\n");
res = PQgetResult(conn);
if (PQresultStatus(res) != PGRES_TUPLES_OK)
{
printf("PQgetResult() failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
else
{
int i, n = PQnfields(res);
printf("PQfname\n");
for (i = 0; i < n; i++)
printf(" %d:%s\n", i, PQfname(res, i));
}
PQclear(res);
/*
* Flush result set.
*/
printf("=== PQgetResult> (Flush result set)\n");
res = PQgetResult(conn);
if (res)
{
printf("Failed! %s [%s || Conn: %s]\n",
PQresStatus(PQresultStatus(res)),
PQresultErrorMessage(res),
PQerrorMessage(conn));
PQclear(res);
PQfinish(conn);
return 1;
}
PQclear(res);
PQfinish(conn);
return 0;
}
---------------------------(end of broadcast)---------------------------
TIP 2: Don't 'kill -9' the postmaster
