Message-ID: <002901c0e733$5e683590$0705a8c0@jecw2k1>
From: "Joe Conway" <joe.conway@mail.com>
To: <pgsql-interfaces@postgresql.org>
References: <Pine.LNX.4.30.0105271914200.14985-100000@commandprompt.com>
Subject: remote database queries
Date: Sun, 27 May 2001 22:02:17 -0700
MIME-Version: 1.0
Content-Type: multipart/mixed;
	boundary="----=_NextPart_000_0026_01C0E6F8.B18D9DE0"

This is a multi-part message in MIME format.

------=_NextPart_000_0026_01C0E6F8.B18D9DE0
Content-Type: text/plain; charset="iso-8859-1"
Content-Transfer-Encoding: 7bit

Hello all,

Inspired by others who have recently gotten PostgreSQL functions to return
sets, I set out to create my own. I have on more than one occasion wished I
could run a query across databases or servers (similar to a dblink in Oracle
or a linked db in MSSQL). Attached is a C file which is my attempt. It
exports two functions:

dblink(text connect_string, text sql, text field_separator)
dblink_tok(text delimited_text, text field_separator, int ord_position)

The functions are used as shown in the following example:

select
  dblink_tok(t1.f1,'~',0)::int as vl_id
 ,dblink_tok(t1.f1,'~',1)::text as vl_guid
 ,dblink_tok(t1.f1,'~',2)::text as vl_pri_email
 ,dblink_tok(t1.f1,'~',3)::text as vl_acct_pass_phrase
 ,dblink_tok(t1.f1,'~',4)::text as vl_email_relname
 ,dblink_tok(t1.f1,'~',5)::text as vl_hwsn_relname
 ,dblink_tok(t1.f1,'~',6)::timestamp as vl_mod_dt
 ,dblink_tok(t1.f1,'~',7)::int as vl_status
from
 (select dblink('host=192.168.5.150 port=5432 dbname=vsreg_001 user=postgres
password=postgres','select * from vs_lkup','~') as f1) as t1

By doing "create view vs_lkup_rm as . . ." with the above query, from a
database on another server, I can then write:
"select * from vs_lkup" and get results just as if I were on 192.168.5.150
(sort of -- see problem below).

I have one question, and one problem regarding this.

First the question: is there any way to get the dblink function to return
setof composite -- i.e. return tuples instead of scalar values? The
documentation indicates that a function can return a composite type, but my
attempts all seemed to produce only pointer values (to the tuples?)

Now the problem: as I stated above, "select * from vs_lkup" returns results
just as if I were on 192.168.5.150 -- but if I try "select * from vs_lkup
WHERE vl_id = 1" or "select * from vs_lkup WHERE vl_pri_email in
('email1@foo.com')" I get the following error message: "ERROR:  Set-valued
function called in context that cannot accept a set". Any ideas how to work
around this?

Thanks,

Joe Conway


------=_NextPart_000_0026_01C0E6F8.B18D9DE0
Content-Type: application/octet-stream; name="dblink.h"
Content-Transfer-Encoding: quoted-printable
Content-Disposition: attachment; filename="dblink.h"

/*
 * dblink.h
 *
 * Functions returning info about current database
 *
 * Copyright (c) Joseph Conway <joe.conway@mail.com>, 2001;
 *=20
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without a written agreem=
ent
 * is hereby granted, provided that the above copyright notice and this
 * paragraph and the following two paragraphs appear in all copies.
 *=20
 * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDI=
NG
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF T=
HE
 * POSSIBILITY OF SUCH DAMAGE.
 *=20
 * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS =
TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 */

#ifndef DBLINK_H
#define DBLINK_H

#include <string.h>
#include "postgres.h"
#include "libpq-fe.h"
#include "libpq-int.h"
#include "fmgr.h"
#include "access/tupdesc.h"
#include "executor/executor.h"
#include "nodes/nodes.h"
#include "nodes/execnodes.h"
#include "utils/builtins.h"

/*
 * This struct holds the results of the remote query.
 * Use fn_extra to hold a pointer to it across calls
 */
typedef struct
{
	int			tup_num;		/* last tuple number accessed */
	char		*fldsep;		/* field separator */
	PGresult	*res;			/* the actual query results */
} dblink_results;


extern Datum dblink(PG_FUNCTION_ARGS);
extern Datum dblink_tok(PG_FUNCTION_ARGS);

dblink_results *init_dblink_results(MemoryContext fn_mcxt);

#define MAX_DBLINK_CALLS	16

#endif	 /* DBLINK_H */

------=_NextPart_000_0026_01C0E6F8.B18D9DE0
Content-Type: application/octet-stream; name="dblink.c"
Content-Transfer-Encoding: quoted-printable
Content-Disposition: attachment; filename="dblink.c"

/*
 * dblink.c
 *
 * Functions returning results from a remote database
 *
 * Copyright (c) Joseph Conway <joe.conway@mail.com>, 2001;
 *=20
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without a written agreem=
ent
 * is hereby granted, provided that the above copyright notice and this
 * paragraph and the following two paragraphs appear in all copies.
 *=20
 * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDI=
NG
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF T=
HE
 * POSSIBILITY OF SUCH DAMAGE.
 *=20
 * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS =
TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 */

#include "dblink.h"

PG_FUNCTION_INFO_V1(dblink);
Datum
dblink(PG_FUNCTION_ARGS)
{
	PGconn			*conn =3D NULL;
	PGresult		*res =3D NULL;
	dblink_results	*results;
	char			*fldsep;
	char			*optstr;
	char			*sqlstatement;
	char			*curstr =3D "DECLARE mycursor CURSOR FOR ";
	char			*execstatement;
	char			*msg;
	int				ntuples =3D 0;
	int				nfields =3D 0;
	ReturnSetInfo 	*rsi;
	char			*result;
	text			*result_text;
	int				j =3D 0;
	int				text_len =3D 0;

	if (PG_ARGISNULL(0) || PG_ARGISNULL(1) || PG_ARGISNULL(2)) {
		elog(ERROR, "dblink: NULL arguments are not permitted");
	}

	if (fcinfo->resultinfo =3D=3D NULL || ! IsA(fcinfo->resultinfo, ReturnSetI=
nfo)) {
		elog(ERROR, "dblink: function called in context that does not accept a se=
t result");
	}

	optstr =3D DatumGetCString(DirectFunctionCall1(textout, PointerGetDatum(PG=
_GETARG_TEXT_P(0))));
	sqlstatement =3D DatumGetCString(DirectFunctionCall1(textout, PointerGetDa=
tum(PG_GETARG_TEXT_P(1))));
	fldsep =3D DatumGetCString(DirectFunctionCall1(textout, PointerGetDatum(PG=
_GETARG_TEXT_P(2))));

	if (fcinfo->flinfo->fn_extra =3D=3D NULL) {

		conn =3D PQconnectdb(optstr);
		if (PQstatus(conn) =3D=3D CONNECTION_BAD)
		{
			msg =3D PQerrorMessage(conn);
			PQfinish(conn);
			elog(ERROR, "dblink: connection error: %s", msg);
		}

		res =3D PQexec(conn, "BEGIN");
		if (PQresultStatus(res) !=3D PGRES_COMMAND_OK)
		{
			msg =3D PQerrorMessage(conn);
			PQclear(res);
			PQfinish(conn);
			elog(ERROR, "dblink: begin error: %s", msg);
		}
		PQclear(res);


		execstatement =3D (char *) palloc(strlen(curstr) + strlen(sqlstatement) +=
 1);
		if (execstatement !=3D NULL) {
			strcpy(execstatement, curstr);
			strcat(execstatement, sqlstatement);
			strcat(execstatement, "\0");
		} else {
			elog(ERROR, "dblink: insufficient memory" );
		}

		res =3D PQexec(conn, execstatement);
		if (!res || (PQresultStatus(res) !=3D PGRES_COMMAND_OK && PQresultStatus(=
res) !=3D PGRES_TUPLES_OK))
		{
			msg =3D PQerrorMessage(conn);
			PQclear(res);
			PQfinish(conn);
			elog(ERROR, "dblink: sql error: %s", msg);
		} else {
			/*
			 * got results, start fetching them
			 */
		    PQclear(res);

		    res =3D PQexec(conn, "FETCH ALL in mycursor");
		    if (!res || PQresultStatus(res) !=3D PGRES_TUPLES_OK) {
				msg =3D PQerrorMessage(conn);
				PQclear(res);
				PQfinish(conn);
				elog(ERROR, "dblink: sql error: %s", msg);
		    }

			ntuples =3D PQntuples(res);

			if (ntuples > 0) {

				results =3D init_dblink_results(fcinfo->flinfo->fn_mcxt);

				results->fldsep =3D fldsep;
				results->tup_num =3D 0;
				results->res =3D res;
				res =3D NULL;

				(dblink_results *) fcinfo->flinfo->fn_extra =3D results;

				results =3D NULL;
				results =3D fcinfo->flinfo->fn_extra;

				nfields =3D PQnfields(results->res);

				for (j =3D 0; j < nfields; j++) {
					text_len +=3D PQgetlength(results->res, results->tup_num, j);
				}

				text_len +=3D (nfields - 1) * strlen(results->fldsep);
				result =3D (char *) palloc(text_len + 1);

				if (result !=3D NULL) {
					strcpy(result, PQgetvalue(results->res, results->tup_num, 0));
					for (j =3D 1; j < nfields; j++) {
						result =3D strcat(result, results->fldsep);
						result =3D strcat(result, PQgetvalue(results->res, results->tup_num, =
j));
					}
					strcat(result, "\0");
				} else {
					elog(ERROR, "dblink: insufficient memory" );
				}

				result_text =3D DatumGetTextP(DirectFunctionCall1(textin, CStringGetDat=
um(result)));

			    /* close the cursor */
			    res =3D PQexec(conn, "CLOSE mycursor");
			    PQclear(res);

			    /* commit the transaction */
			    res =3D PQexec(conn, "COMMIT");
			    PQclear(res);

			    /* close the connection to the database and cleanup */
			    PQfinish(conn);

				rsi =3D (ReturnSetInfo *)fcinfo->resultinfo;
				rsi->isDone =3D ExprMultipleResult;

				PG_RETURN_TEXT_P(result_text);

			} else {

			    PQclear(res);

			    /* close the cursor */
			    res =3D PQexec(conn, "CLOSE mycursor");
			    PQclear(res);

			    /* commit the transaction */
			    res =3D PQexec(conn, "COMMIT");
			    PQclear(res);

			    /* close the connection to the database and cleanup */
			    PQfinish(conn);

				rsi =3D (ReturnSetInfo *)fcinfo->resultinfo;
				rsi->isDone =3D ExprEndResult ;

				PG_RETURN_NULL();
			}
		}
	} else {
		/*
		 * check for more results
		 */

		results =3D fcinfo->flinfo->fn_extra;

		ntuples =3D PQntuples(results->res);
		results->tup_num++;

		if (results->tup_num < ntuples) {

			/*
			 * fetch them if available
			 */

			nfields =3D PQnfields(results->res);

			for (j =3D 0; j < nfields; j++) {
				text_len +=3D PQgetlength(results->res, results->tup_num, j);
			}

			text_len +=3D (nfields - 1) * strlen(results->fldsep);
			result =3D (char *) palloc(text_len + 1);

			if (result !=3D NULL) {
				strcpy(result, PQgetvalue(results->res, results->tup_num, 0));
				for (j =3D 1; j < nfields; j++) {
					result =3D strcat(result, results->fldsep);
					result =3D strcat(result, PQgetvalue(results->res, results->tup_num, j=
));
				}
				strcat(result, "\0");
			} else {
				elog(ERROR, "dblink: insufficient memory" );
			}

			result_text =3D DatumGetTextP(DirectFunctionCall1(textin, CStringGetDatu=
m(result)));

			rsi =3D (ReturnSetInfo *)fcinfo->resultinfo;
			rsi->isDone =3D ExprMultipleResult;

			PG_RETURN_TEXT_P(result_text);


		} else {

			/*
			 * or if no more, clean things up
			 */

			results =3D fcinfo->flinfo->fn_extra;

		    PQclear(results->res);

			rsi =3D (ReturnSetInfo *)fcinfo->resultinfo;
			rsi->isDone =3D ExprEndResult ;

			PG_RETURN_NULL();
		}
	}
	PG_RETURN_NULL();
}


/*
 * dblink_tok
 * parse dblink output string
 * return fldnum item (0 based)
 * based on provided field separator
 */

PG_FUNCTION_INFO_V1(dblink_tok);
Datum
dblink_tok(PG_FUNCTION_ARGS)
{
	char		*fldtext;
	char		*fldsep;
	int			fldnum;
	char		*buffer =3D "\0";
	int			j =3D 0;
	text		*result_text;


	if (PG_ARGISNULL(0) || PG_ARGISNULL(1)) {
		elog(ERROR, "dblink: NULL arguments are not permitted");
	}

	fldtext =3D DatumGetCString(DirectFunctionCall1(textout, PointerGetDatum(P=
G_GETARG_TEXT_P(0))));
	fldsep =3D DatumGetCString(DirectFunctionCall1(textout, PointerGetDatum(PG=
_GETARG_TEXT_P(1))));
	fldnum =3D PG_GETARG_INT32(2);

	if (fldnum < 0) {
		elog(ERROR, "dblink: field number < 0 not permitted");
	}

	buffer =3D strtok(fldtext, fldsep);
	for (j =3D 1; j < fldnum + 1; j++) {
		buffer =3D strtok(NULL, fldsep);
		if (buffer =3D=3D NULL) {
			elog(ERROR, "dblink: field number %d does not exist", fldnum);
		}
	}=20

	result_text =3D DatumGetTextP(DirectFunctionCall1(textin, CStringGetDatum(=
buffer)));
	PG_RETURN_TEXT_P(result_text);
}


/*
 * internal functions
 */


/*
 * init_dblink_results
 *	 - create an empty dblink_results data structure
 */
dblink_results *
init_dblink_results(MemoryContext fn_mcxt)
{
	MemoryContext	oldcontext;
	dblink_results	*retval;

	oldcontext =3D MemoryContextSwitchTo(fn_mcxt);

	retval =3D (dblink_results *) palloc(sizeof(dblink_results));
	MemSet(retval, 0, sizeof(dblink_results));

	retval->tup_num =3D -1;
	retval->fldsep =3D NULL;
	retval->res =3D NULL;

	MemoryContextSwitchTo(oldcontext);

	return retval;
}

------=_NextPart_000_0026_01C0E6F8.B18D9DE0--