mirror of https://github.com/postgres/postgres
There's still a lot of room for improvement, but it basically works, and we need this to be present before we can do anything much with the writable-foreign-tables patch. So let's commit it and get on with testing. Shigeru Hanada, reviewed by KaiGai Kohei and Tom Lanepull/3/head
Tom Lane
13 years ago
parent
f435cd1d38
commit
d0d75c4022
@ -0,0 +1,4 @@ |
||||
# Generated subdirectories |
||||
/log/ |
||||
/results/ |
||||
/tmp_check/ |
||||
@ -0,0 +1,27 @@ |
||||
# contrib/postgres_fdw/Makefile
|
||||
|
||||
MODULE_big = postgres_fdw
|
||||
OBJS = postgres_fdw.o option.o deparse.o connection.o
|
||||
|
||||
PG_CPPFLAGS = -I$(libpq_srcdir)
|
||||
SHLIB_LINK = $(libpq)
|
||||
SHLIB_PREREQS = submake-libpq
|
||||
|
||||
EXTENSION = postgres_fdw
|
||||
DATA = postgres_fdw--1.0.sql
|
||||
|
||||
REGRESS = postgres_fdw
|
||||
|
||||
# the db name is hard-coded in the tests
|
||||
override USE_MODULE_DB =
|
||||
|
||||
ifdef USE_PGXS |
||||
PG_CONFIG = pg_config
|
||||
PGXS := $(shell $(PG_CONFIG) --pgxs)
|
||||
include $(PGXS) |
||||
else |
||||
subdir = contrib/postgres_fdw
|
||||
top_builddir = ../..
|
||||
include $(top_builddir)/src/Makefile.global |
||||
include $(top_srcdir)/contrib/contrib-global.mk |
||||
endif |
||||
@ -0,0 +1,581 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* connection.c |
||||
* Connection management functions for postgres_fdw |
||||
* |
||||
* Portions Copyright (c) 2012-2013, PostgreSQL Global Development Group |
||||
* |
||||
* IDENTIFICATION |
||||
* contrib/postgres_fdw/connection.c |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#include "postgres.h" |
||||
|
||||
#include "postgres_fdw.h" |
||||
|
||||
#include "access/xact.h" |
||||
#include "mb/pg_wchar.h" |
||||
#include "miscadmin.h" |
||||
#include "utils/hsearch.h" |
||||
#include "utils/memutils.h" |
||||
|
||||
|
||||
/*
|
||||
* Connection cache hash table entry |
||||
* |
||||
* The lookup key in this hash table is the foreign server OID plus the user |
||||
* mapping OID. (We use just one connection per user per foreign server, |
||||
* so that we can ensure all scans use the same snapshot during a query.) |
||||
* |
||||
* The "conn" pointer can be NULL if we don't currently have a live connection. |
||||
* When we do have a connection, xact_depth tracks the current depth of |
||||
* transactions and subtransactions open on the remote side. We need to issue |
||||
* commands at the same nesting depth on the remote as we're executing at |
||||
* ourselves, so that rolling back a subtransaction will kill the right |
||||
* queries and not the wrong ones. |
||||
*/ |
||||
typedef struct ConnCacheKey |
||||
{ |
||||
Oid serverid; /* OID of foreign server */ |
||||
Oid userid; /* OID of local user whose mapping we use */ |
||||
} ConnCacheKey; |
||||
|
||||
typedef struct ConnCacheEntry |
||||
{ |
||||
ConnCacheKey key; /* hash key (must be first) */ |
||||
PGconn *conn; /* connection to foreign server, or NULL */ |
||||
int xact_depth; /* 0 = no xact open, 1 = main xact open, 2 =
|
||||
* one level of subxact open, etc */ |
||||
} ConnCacheEntry; |
||||
|
||||
/*
|
||||
* Connection cache (initialized on first use) |
||||
*/ |
||||
static HTAB *ConnectionHash = NULL; |
||||
|
||||
/* for assigning cursor numbers */ |
||||
static unsigned int cursor_number = 0; |
||||
|
||||
/* tracks whether any work is needed in callback functions */ |
||||
static bool xact_got_connection = false; |
||||
|
||||
/* prototypes of private functions */ |
||||
static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user); |
||||
static void check_conn_params(const char **keywords, const char **values); |
||||
static void begin_remote_xact(ConnCacheEntry *entry); |
||||
static void pgfdw_xact_callback(XactEvent event, void *arg); |
||||
static void pgfdw_subxact_callback(SubXactEvent event, |
||||
SubTransactionId mySubid, |
||||
SubTransactionId parentSubid, |
||||
void *arg); |
||||
|
||||
|
||||
/*
|
||||
* Get a PGconn which can be used to execute queries on the remote PostgreSQL |
||||
* server with the user's authorization. A new connection is established |
||||
* if we don't already have a suitable one, and a transaction is opened at |
||||
* the right subtransaction nesting depth if we didn't do that already. |
||||
* |
||||
* XXX Note that caching connections theoretically requires a mechanism to |
||||
* detect change of FDW objects to invalidate already established connections. |
||||
* We could manage that by watching for invalidation events on the relevant |
||||
* syscaches. For the moment, though, it's not clear that this would really |
||||
* be useful and not mere pedantry. We could not flush any active connections |
||||
* mid-transaction anyway. |
||||
*/ |
||||
PGconn * |
||||
GetConnection(ForeignServer *server, UserMapping *user) |
||||
{ |
||||
bool found; |
||||
ConnCacheEntry *entry; |
||||
ConnCacheKey key; |
||||
|
||||
/* First time through, initialize connection cache hashtable */ |
||||
if (ConnectionHash == NULL) |
||||
{ |
||||
HASHCTL ctl; |
||||
|
||||
MemSet(&ctl, 0, sizeof(ctl)); |
||||
ctl.keysize = sizeof(ConnCacheKey); |
||||
ctl.entrysize = sizeof(ConnCacheEntry); |
||||
ctl.hash = tag_hash; |
||||
/* allocate ConnectionHash in the cache context */ |
||||
ctl.hcxt = CacheMemoryContext; |
||||
ConnectionHash = hash_create("postgres_fdw connections", 8, |
||||
&ctl, |
||||
HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT); |
||||
|
||||
/*
|
||||
* Register some callback functions that manage connection cleanup. |
||||
* This should be done just once in each backend. |
||||
*/ |
||||
RegisterXactCallback(pgfdw_xact_callback, NULL); |
||||
RegisterSubXactCallback(pgfdw_subxact_callback, NULL); |
||||
} |
||||
|
||||
/* Set flag that we did GetConnection during the current transaction */ |
||||
xact_got_connection = true; |
||||
|
||||
/* Create hash key for the entry. Assume no pad bytes in key struct */ |
||||
key.serverid = server->serverid; |
||||
key.userid = user->userid; |
||||
|
||||
/*
|
||||
* Find or create cached entry for requested connection. |
||||
*/ |
||||
entry = hash_search(ConnectionHash, &key, HASH_ENTER, &found); |
||||
if (!found) |
||||
{ |
||||
/* initialize new hashtable entry (key is already filled in) */ |
||||
entry->conn = NULL; |
||||
entry->xact_depth = 0; |
||||
} |
||||
|
||||
/*
|
||||
* We don't check the health of cached connection here, because it would |
||||
* require some overhead. Broken connection will be detected when the |
||||
* connection is actually used. |
||||
*/ |
||||
|
||||
/*
|
||||
* If cache entry doesn't have a connection, we have to establish a new |
||||
* connection. (If connect_pg_server throws an error, the cache entry |
||||
* will be left in a valid empty state.) |
||||
*/ |
||||
if (entry->conn == NULL) |
||||
{ |
||||
entry->xact_depth = 0; /* just to be sure */ |
||||
entry->conn = connect_pg_server(server, user); |
||||
elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\"", |
||||
entry->conn, server->servername); |
||||
} |
||||
|
||||
/*
|
||||
* Start a new transaction or subtransaction if needed. |
||||
*/ |
||||
begin_remote_xact(entry); |
||||
|
||||
return entry->conn; |
||||
} |
||||
|
||||
/*
|
||||
* Connect to remote server using specified server and user mapping properties. |
||||
*/ |
||||
static PGconn * |
||||
connect_pg_server(ForeignServer *server, UserMapping *user) |
||||
{ |
||||
PGconn *volatile conn = NULL; |
||||
|
||||
/*
|
||||
* Use PG_TRY block to ensure closing connection on error. |
||||
*/ |
||||
PG_TRY(); |
||||
{ |
||||
const char **keywords; |
||||
const char **values; |
||||
int n; |
||||
|
||||
/*
|
||||
* Construct connection params from generic options of ForeignServer |
||||
* and UserMapping. (Some of them might not be libpq options, in |
||||
* which case we'll just waste a few array slots.) Add 3 extra slots |
||||
* for fallback_application_name, client_encoding, end marker. |
||||
*/ |
||||
n = list_length(server->options) + list_length(user->options) + 3; |
||||
keywords = (const char **) palloc(n * sizeof(char *)); |
||||
values = (const char **) palloc(n * sizeof(char *)); |
||||
|
||||
n = 0; |
||||
n += ExtractConnectionOptions(server->options, |
||||
keywords + n, values + n); |
||||
n += ExtractConnectionOptions(user->options, |
||||
keywords + n, values + n); |
||||
|
||||
/* Use "postgres_fdw" as fallback_application_name. */ |
||||
keywords[n] = "fallback_application_name"; |
||||
values[n] = "postgres_fdw"; |
||||
n++; |
||||
|
||||
/* Set client_encoding so that libpq can convert encoding properly. */ |
||||
keywords[n] = "client_encoding"; |
||||
values[n] = GetDatabaseEncodingName(); |
||||
n++; |
||||
|
||||
keywords[n] = values[n] = NULL; |
||||
|
||||
/* verify connection parameters and make connection */ |
||||
check_conn_params(keywords, values); |
||||
|
||||
conn = PQconnectdbParams(keywords, values, false); |
||||
if (!conn || PQstatus(conn) != CONNECTION_OK) |
||||
{ |
||||
char *connmessage; |
||||
int msglen; |
||||
|
||||
/* libpq typically appends a newline, strip that */ |
||||
connmessage = pstrdup(PQerrorMessage(conn)); |
||||
msglen = strlen(connmessage); |
||||
if (msglen > 0 && connmessage[msglen - 1] == '\n') |
||||
connmessage[msglen - 1] = '\0'; |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION), |
||||
errmsg("could not connect to server \"%s\"", |
||||
server->servername), |
||||
errdetail_internal("%s", connmessage))); |
||||
} |
||||
|
||||
/*
|
||||
* Check that non-superuser has used password to establish connection; |
||||
* otherwise, he's piggybacking on the postgres server's user |
||||
* identity. See also dblink_security_check() in contrib/dblink. |
||||
*/ |
||||
if (!superuser() && !PQconnectionUsedPassword(conn)) |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED), |
||||
errmsg("password is required"), |
||||
errdetail("Non-superuser cannot connect if the server does not request a password."), |
||||
errhint("Target server's authentication method must be changed."))); |
||||
|
||||
pfree(keywords); |
||||
pfree(values); |
||||
} |
||||
PG_CATCH(); |
||||
{ |
||||
/* Release PGconn data structure if we managed to create one */ |
||||
if (conn) |
||||
PQfinish(conn); |
||||
PG_RE_THROW(); |
||||
} |
||||
PG_END_TRY(); |
||||
|
||||
return conn; |
||||
} |
||||
|
||||
/*
|
||||
* For non-superusers, insist that the connstr specify a password. This |
||||
* prevents a password from being picked up from .pgpass, a service file, |
||||
* the environment, etc. We don't want the postgres user's passwords |
||||
* to be accessible to non-superusers. (See also dblink_connstr_check in |
||||
* contrib/dblink.) |
||||
*/ |
||||
static void |
||||
check_conn_params(const char **keywords, const char **values) |
||||
{ |
||||
int i; |
||||
|
||||
/* no check required if superuser */ |
||||
if (superuser()) |
||||
return; |
||||
|
||||
/* ok if params contain a non-empty password */ |
||||
for (i = 0; keywords[i] != NULL; i++) |
||||
{ |
||||
if (strcmp(keywords[i], "password") == 0 && values[i][0] != '\0') |
||||
return; |
||||
} |
||||
|
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED), |
||||
errmsg("password is required"), |
||||
errdetail("Non-superusers must provide a password in the user mapping."))); |
||||
} |
||||
|
||||
/*
|
||||
* Start remote transaction or subtransaction, if needed. |
||||
* |
||||
* Note that we always use at least REPEATABLE READ in the remote session. |
||||
* This is so that, if a query initiates multiple scans of the same or |
||||
* different foreign tables, we will get snapshot-consistent results from |
||||
* those scans. A disadvantage is that we can't provide sane emulation of |
||||
* READ COMMITTED behavior --- it would be nice if we had some other way to |
||||
* control which remote queries share a snapshot. |
||||
*/ |
||||
static void |
||||
begin_remote_xact(ConnCacheEntry *entry) |
||||
{ |
||||
int curlevel = GetCurrentTransactionNestLevel(); |
||||
PGresult *res; |
||||
|
||||
/* Start main transaction if we haven't yet */ |
||||
if (entry->xact_depth <= 0) |
||||
{ |
||||
const char *sql; |
||||
|
||||
elog(DEBUG3, "starting remote transaction on connection %p", |
||||
entry->conn); |
||||
|
||||
if (XactIsoLevel == XACT_SERIALIZABLE) |
||||
sql = "START TRANSACTION ISOLATION LEVEL SERIALIZABLE"; |
||||
else |
||||
sql = "START TRANSACTION ISOLATION LEVEL REPEATABLE READ"; |
||||
res = PQexec(entry->conn, sql); |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(ERROR, res, true, sql); |
||||
PQclear(res); |
||||
entry->xact_depth = 1; |
||||
} |
||||
|
||||
/*
|
||||
* If we're in a subtransaction, stack up savepoints to match our level. |
||||
* This ensures we can rollback just the desired effects when a |
||||
* subtransaction aborts. |
||||
*/ |
||||
while (entry->xact_depth < curlevel) |
||||
{ |
||||
char sql[64]; |
||||
|
||||
snprintf(sql, sizeof(sql), "SAVEPOINT s%d", entry->xact_depth + 1); |
||||
res = PQexec(entry->conn, sql); |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(ERROR, res, true, sql); |
||||
PQclear(res); |
||||
entry->xact_depth++; |
||||
} |
||||
} |
||||
|
||||
/*
|
||||
* Release connection reference count created by calling GetConnection. |
||||
*/ |
||||
void |
||||
ReleaseConnection(PGconn *conn) |
||||
{ |
||||
/*
|
||||
* Currently, we don't actually track connection references because all |
||||
* cleanup is managed on a transaction or subtransaction basis instead. So |
||||
* there's nothing to do here. |
||||
*/ |
||||
} |
||||
|
||||
/*
|
||||
* Assign a "unique" number for a cursor. |
||||
* |
||||
* These really only need to be unique per connection within a transaction. |
||||
* For the moment we ignore the per-connection point and assign them across |
||||
* all connections in the transaction, but we ask for the connection to be |
||||
* supplied in case we want to refine that. |
||||
* |
||||
* Note that even if wraparound happens in a very long transaction, actual |
||||
* collisions are highly improbable; just be sure to use %u not %d to print. |
||||
*/ |
||||
unsigned int |
||||
GetCursorNumber(PGconn *conn) |
||||
{ |
||||
return ++cursor_number; |
||||
} |
||||
|
||||
/*
|
||||
* Report an error we got from the remote server. |
||||
* |
||||
* elevel: error level to use (typically ERROR, but might be less) |
||||
* res: PGresult containing the error |
||||
* clear: if true, PQclear the result (otherwise caller will handle it) |
||||
* sql: NULL, or text of remote command we tried to execute |
||||
*/ |
||||
void |
||||
pgfdw_report_error(int elevel, PGresult *res, bool clear, const char *sql) |
||||
{ |
||||
/* If requested, PGresult must be released before leaving this function. */ |
||||
PG_TRY(); |
||||
{ |
||||
char *diag_sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); |
||||
char *message_primary = PQresultErrorField(res, PG_DIAG_MESSAGE_PRIMARY); |
||||
char *message_detail = PQresultErrorField(res, PG_DIAG_MESSAGE_DETAIL); |
||||
char *message_hint = PQresultErrorField(res, PG_DIAG_MESSAGE_HINT); |
||||
char *message_context = PQresultErrorField(res, PG_DIAG_CONTEXT); |
||||
int sqlstate; |
||||
|
||||
if (diag_sqlstate) |
||||
sqlstate = MAKE_SQLSTATE(diag_sqlstate[0], |
||||
diag_sqlstate[1], |
||||
diag_sqlstate[2], |
||||
diag_sqlstate[3], |
||||
diag_sqlstate[4]); |
||||
else |
||||
sqlstate = ERRCODE_CONNECTION_FAILURE; |
||||
|
||||
ereport(elevel, |
||||
(errcode(sqlstate), |
||||
message_primary ? errmsg_internal("%s", message_primary) : |
||||
errmsg("unknown error"), |
||||
message_detail ? errdetail_internal("%s", message_detail) : 0, |
||||
message_hint ? errhint("%s", message_hint) : 0, |
||||
message_context ? errcontext("%s", message_context) : 0, |
||||
sql ? errcontext("Remote SQL command: %s", sql) : 0)); |
||||
} |
||||
PG_CATCH(); |
||||
{ |
||||
if (clear) |
||||
PQclear(res); |
||||
PG_RE_THROW(); |
||||
} |
||||
PG_END_TRY(); |
||||
if (clear) |
||||
PQclear(res); |
||||
} |
||||
|
||||
/*
|
||||
* pgfdw_xact_callback --- cleanup at main-transaction end. |
||||
*/ |
||||
static void |
||||
pgfdw_xact_callback(XactEvent event, void *arg) |
||||
{ |
||||
HASH_SEQ_STATUS scan; |
||||
ConnCacheEntry *entry; |
||||
|
||||
/* Quick exit if no connections were touched in this transaction. */ |
||||
if (!xact_got_connection) |
||||
return; |
||||
|
||||
/*
|
||||
* Scan all connection cache entries to find open remote transactions, and |
||||
* close them. |
||||
*/ |
||||
hash_seq_init(&scan, ConnectionHash); |
||||
while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) |
||||
{ |
||||
PGresult *res; |
||||
|
||||
/* We only care about connections with open remote transactions */ |
||||
if (entry->conn == NULL || entry->xact_depth == 0) |
||||
continue; |
||||
|
||||
elog(DEBUG3, "closing remote transaction on connection %p", |
||||
entry->conn); |
||||
|
||||
switch (event) |
||||
{ |
||||
case XACT_EVENT_PRE_COMMIT: |
||||
/* Commit all remote transactions during pre-commit */ |
||||
res = PQexec(entry->conn, "COMMIT TRANSACTION"); |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(ERROR, res, true, "COMMIT TRANSACTION"); |
||||
PQclear(res); |
||||
break; |
||||
case XACT_EVENT_PRE_PREPARE: |
||||
|
||||
/*
|
||||
* We disallow remote transactions that modified anything, |
||||
* since it's not really reasonable to hold them open until |
||||
* the prepared transaction is committed. For the moment, |
||||
* throw error unconditionally; later we might allow read-only |
||||
* cases. Note that the error will cause us to come right |
||||
* back here with event == XACT_EVENT_ABORT, so we'll clean up |
||||
* the connection state at that point. |
||||
*/ |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED), |
||||
errmsg("cannot prepare a transaction that modified remote tables"))); |
||||
break; |
||||
case XACT_EVENT_COMMIT: |
||||
case XACT_EVENT_PREPARE: |
||||
/* Should not get here -- pre-commit should have handled it */ |
||||
elog(ERROR, "missed cleaning up connection during pre-commit"); |
||||
break; |
||||
case XACT_EVENT_ABORT: |
||||
/* If we're aborting, abort all remote transactions too */ |
||||
res = PQexec(entry->conn, "ABORT TRANSACTION"); |
||||
/* Note: can't throw ERROR, it would be infinite loop */ |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(WARNING, res, true, |
||||
"ABORT TRANSACTION"); |
||||
else |
||||
PQclear(res); |
||||
break; |
||||
} |
||||
|
||||
/* Reset state to show we're out of a transaction */ |
||||
entry->xact_depth = 0; |
||||
|
||||
/*
|
||||
* If the connection isn't in a good idle state, discard it to |
||||
* recover. Next GetConnection will open a new connection. |
||||
*/ |
||||
if (PQstatus(entry->conn) != CONNECTION_OK || |
||||
PQtransactionStatus(entry->conn) != PQTRANS_IDLE) |
||||
{ |
||||
elog(DEBUG3, "discarding connection %p", entry->conn); |
||||
PQfinish(entry->conn); |
||||
entry->conn = NULL; |
||||
} |
||||
} |
||||
|
||||
/*
|
||||
* Regardless of the event type, we can now mark ourselves as out of the |
||||
* transaction. (Note: if we are here during PRE_COMMIT or PRE_PREPARE, |
||||
* this saves a useless scan of the hashtable during COMMIT or PREPARE.) |
||||
*/ |
||||
xact_got_connection = false; |
||||
|
||||
/* Also reset cursor numbering for next transaction */ |
||||
cursor_number = 0; |
||||
} |
||||
|
||||
/*
|
||||
* pgfdw_subxact_callback --- cleanup at subtransaction end. |
||||
*/ |
||||
static void |
||||
pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, |
||||
SubTransactionId parentSubid, void *arg) |
||||
{ |
||||
HASH_SEQ_STATUS scan; |
||||
ConnCacheEntry *entry; |
||||
int curlevel; |
||||
|
||||
/* Nothing to do at subxact start, nor after commit. */ |
||||
if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB || |
||||
event == SUBXACT_EVENT_ABORT_SUB)) |
||||
return; |
||||
|
||||
/* Quick exit if no connections were touched in this transaction. */ |
||||
if (!xact_got_connection) |
||||
return; |
||||
|
||||
/*
|
||||
* Scan all connection cache entries to find open remote subtransactions |
||||
* of the current level, and close them. |
||||
*/ |
||||
curlevel = GetCurrentTransactionNestLevel(); |
||||
hash_seq_init(&scan, ConnectionHash); |
||||
while ((entry = (ConnCacheEntry *) hash_seq_search(&scan))) |
||||
{ |
||||
PGresult *res; |
||||
char sql[100]; |
||||
|
||||
/*
|
||||
* We only care about connections with open remote subtransactions of |
||||
* the current level. |
||||
*/ |
||||
if (entry->conn == NULL || entry->xact_depth < curlevel) |
||||
continue; |
||||
|
||||
if (entry->xact_depth > curlevel) |
||||
elog(ERROR, "missed cleaning up remote subtransaction at level %d", |
||||
entry->xact_depth); |
||||
|
||||
if (event == SUBXACT_EVENT_PRE_COMMIT_SUB) |
||||
{ |
||||
/* Commit all remote subtransactions during pre-commit */ |
||||
snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel); |
||||
res = PQexec(entry->conn, sql); |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(ERROR, res, true, sql); |
||||
PQclear(res); |
||||
} |
||||
else |
||||
{ |
||||
/* Rollback all remote subtransactions during abort */ |
||||
snprintf(sql, sizeof(sql), |
||||
"ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", |
||||
curlevel, curlevel); |
||||
res = PQexec(entry->conn, sql); |
||||
if (PQresultStatus(res) != PGRES_COMMAND_OK) |
||||
pgfdw_report_error(WARNING, res, true, sql); |
||||
else |
||||
PQclear(res); |
||||
} |
||||
|
||||
/* OK, we're outta that level of subtransaction */ |
||||
entry->xact_depth--; |
||||
} |
||||
} |
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,704 @@ |
||||
-- =================================================================== |
||||
-- create FDW objects |
||||
-- =================================================================== |
||||
CREATE EXTENSION postgres_fdw; |
||||
CREATE SERVER testserver1 FOREIGN DATA WRAPPER postgres_fdw; |
||||
CREATE SERVER loopback FOREIGN DATA WRAPPER postgres_fdw |
||||
OPTIONS (dbname 'contrib_regression'); |
||||
CREATE USER MAPPING FOR public SERVER testserver1 |
||||
OPTIONS (user 'value', password 'value'); |
||||
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback; |
||||
-- =================================================================== |
||||
-- create objects used through FDW loopback server |
||||
-- =================================================================== |
||||
CREATE TYPE user_enum AS ENUM ('foo', 'bar', 'buz'); |
||||
CREATE SCHEMA "S 1"; |
||||
CREATE TABLE "S 1"."T 1" ( |
||||
"C 1" int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum, |
||||
CONSTRAINT t1_pkey PRIMARY KEY ("C 1") |
||||
); |
||||
CREATE TABLE "S 1"."T 2" ( |
||||
c1 int NOT NULL, |
||||
c2 text, |
||||
CONSTRAINT t2_pkey PRIMARY KEY (c1) |
||||
); |
||||
INSERT INTO "S 1"."T 1" |
||||
SELECT id, |
||||
id % 10, |
||||
to_char(id, 'FM00000'), |
||||
'1970-01-01'::timestamptz + ((id % 100) || ' days')::interval, |
||||
'1970-01-01'::timestamp + ((id % 100) || ' days')::interval, |
||||
id % 10, |
||||
id % 10, |
||||
'foo'::user_enum |
||||
FROM generate_series(1, 1000) id; |
||||
INSERT INTO "S 1"."T 2" |
||||
SELECT id, |
||||
'AAA' || to_char(id, 'FM000') |
||||
FROM generate_series(1, 100) id; |
||||
ANALYZE "S 1"."T 1"; |
||||
ANALYZE "S 1"."T 2"; |
||||
-- =================================================================== |
||||
-- create foreign tables |
||||
-- =================================================================== |
||||
CREATE FOREIGN TABLE ft1 ( |
||||
c0 int, |
||||
c1 int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum |
||||
) SERVER loopback; |
||||
ALTER FOREIGN TABLE ft1 DROP COLUMN c0; |
||||
CREATE FOREIGN TABLE ft2 ( |
||||
c0 int, |
||||
c1 int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum |
||||
) SERVER loopback; |
||||
ALTER FOREIGN TABLE ft2 DROP COLUMN c0; |
||||
-- =================================================================== |
||||
-- tests for validator |
||||
-- =================================================================== |
||||
-- requiressl, krbsrvname and gsslib are omitted because they depend on |
||||
-- configure options |
||||
ALTER SERVER testserver1 OPTIONS ( |
||||
use_remote_explain 'false', |
||||
fdw_startup_cost '123.456', |
||||
fdw_tuple_cost '0.123', |
||||
service 'value', |
||||
connect_timeout 'value', |
||||
dbname 'value', |
||||
host 'value', |
||||
hostaddr 'value', |
||||
port 'value', |
||||
--client_encoding 'value', |
||||
application_name 'value', |
||||
--fallback_application_name 'value', |
||||
keepalives 'value', |
||||
keepalives_idle 'value', |
||||
keepalives_interval 'value', |
||||
-- requiressl 'value', |
||||
sslcompression 'value', |
||||
sslmode 'value', |
||||
sslcert 'value', |
||||
sslkey 'value', |
||||
sslrootcert 'value', |
||||
sslcrl 'value' |
||||
--requirepeer 'value', |
||||
-- krbsrvname 'value', |
||||
-- gsslib 'value', |
||||
--replication 'value' |
||||
); |
||||
ALTER USER MAPPING FOR public SERVER testserver1 |
||||
OPTIONS (DROP user, DROP password); |
||||
ALTER FOREIGN TABLE ft1 OPTIONS (schema_name 'S 1', table_name 'T 1'); |
||||
ALTER FOREIGN TABLE ft2 OPTIONS (schema_name 'S 1', table_name 'T 1'); |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c1 OPTIONS (column_name 'C 1'); |
||||
ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1'); |
||||
\det+ |
||||
List of foreign tables |
||||
Schema | Table | Server | FDW Options | Description |
||||
--------+-------+----------+---------------------------------------+------------- |
||||
public | ft1 | loopback | (schema_name 'S 1', table_name 'T 1') | |
||||
public | ft2 | loopback | (schema_name 'S 1', table_name 'T 1') | |
||||
(2 rows) |
||||
|
||||
-- Now we should be able to run ANALYZE. |
||||
-- To exercise multiple code paths, we use local stats on ft1 |
||||
-- and remote_explain mode on ft2. |
||||
ANALYZE ft1; |
||||
ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_explain 'true'); |
||||
-- =================================================================== |
||||
-- simple queries |
||||
-- =================================================================== |
||||
-- single table, with/without alias |
||||
EXPLAIN (COSTS false) SELECT * FROM ft1 ORDER BY c3, c1 OFFSET 100 LIMIT 10; |
||||
QUERY PLAN |
||||
--------------------------------- |
||||
Limit |
||||
-> Sort |
||||
Sort Key: c3, c1 |
||||
-> Foreign Scan on ft1 |
||||
(4 rows) |
||||
|
||||
SELECT * FROM ft1 ORDER BY c3, c1 OFFSET 100 LIMIT 10; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
-----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
101 | 1 | 00101 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
102 | 2 | 00102 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo |
||||
103 | 3 | 00103 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo |
||||
104 | 4 | 00104 | Mon Jan 05 00:00:00 1970 PST | Mon Jan 05 00:00:00 1970 | 4 | 4 | foo |
||||
105 | 5 | 00105 | Tue Jan 06 00:00:00 1970 PST | Tue Jan 06 00:00:00 1970 | 5 | 5 | foo |
||||
106 | 6 | 00106 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo |
||||
107 | 7 | 00107 | Thu Jan 08 00:00:00 1970 PST | Thu Jan 08 00:00:00 1970 | 7 | 7 | foo |
||||
108 | 8 | 00108 | Fri Jan 09 00:00:00 1970 PST | Fri Jan 09 00:00:00 1970 | 8 | 8 | foo |
||||
109 | 9 | 00109 | Sat Jan 10 00:00:00 1970 PST | Sat Jan 10 00:00:00 1970 | 9 | 9 | foo |
||||
110 | 0 | 00110 | Sun Jan 11 00:00:00 1970 PST | Sun Jan 11 00:00:00 1970 | 0 | 0 | foo |
||||
(10 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------- |
||||
Limit |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
-> Sort |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Sort Key: t1.c3, t1.c1 |
||||
-> Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" |
||||
(8 rows) |
||||
|
||||
SELECT * FROM ft1 t1 ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
-----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
101 | 1 | 00101 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
102 | 2 | 00102 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo |
||||
103 | 3 | 00103 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo |
||||
104 | 4 | 00104 | Mon Jan 05 00:00:00 1970 PST | Mon Jan 05 00:00:00 1970 | 4 | 4 | foo |
||||
105 | 5 | 00105 | Tue Jan 06 00:00:00 1970 PST | Tue Jan 06 00:00:00 1970 | 5 | 5 | foo |
||||
106 | 6 | 00106 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo |
||||
107 | 7 | 00107 | Thu Jan 08 00:00:00 1970 PST | Thu Jan 08 00:00:00 1970 | 7 | 7 | foo |
||||
108 | 8 | 00108 | Fri Jan 09 00:00:00 1970 PST | Fri Jan 09 00:00:00 1970 | 8 | 8 | foo |
||||
109 | 9 | 00109 | Sat Jan 10 00:00:00 1970 PST | Sat Jan 10 00:00:00 1970 | 9 | 9 | foo |
||||
110 | 0 | 00110 | Sun Jan 11 00:00:00 1970 PST | Sun Jan 11 00:00:00 1970 | 0 | 0 | foo |
||||
(10 rows) |
||||
|
||||
-- empty result |
||||
SELECT * FROM ft1 WHERE false; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+----+----+----+----+----+---- |
||||
(0 rows) |
||||
|
||||
-- with WHERE clause |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 101 AND t1.c6 = '1' AND t1.c7 >= '1'; |
||||
QUERY PLAN |
||||
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((c7 OPERATOR(pg_catalog.>=) '1'::bpchar)) AND (("C 1" OPERATOR(pg_catalog.=) 101)) AND ((c6::text OPERATOR(pg_catalog.=) '1'::text)) |
||||
(3 rows) |
||||
|
||||
SELECT * FROM ft1 t1 WHERE t1.c1 = 101 AND t1.c6 = '1' AND t1.c7 >= '1'; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
-----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
101 | 1 | 00101 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
(1 row) |
||||
|
||||
-- aggregate |
||||
SELECT COUNT(*) FROM ft1 t1; |
||||
count |
||||
------- |
||||
1000 |
||||
(1 row) |
||||
|
||||
-- join two tables |
||||
SELECT t1.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
c1 |
||||
----- |
||||
101 |
||||
102 |
||||
103 |
||||
104 |
||||
105 |
||||
106 |
||||
107 |
||||
108 |
||||
109 |
||||
110 |
||||
(10 rows) |
||||
|
||||
-- subquery |
||||
SELECT * FROM ft1 t1 WHERE t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 <= 10) ORDER BY c1; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo |
||||
3 | 3 | 00003 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo |
||||
4 | 4 | 00004 | Mon Jan 05 00:00:00 1970 PST | Mon Jan 05 00:00:00 1970 | 4 | 4 | foo |
||||
5 | 5 | 00005 | Tue Jan 06 00:00:00 1970 PST | Tue Jan 06 00:00:00 1970 | 5 | 5 | foo |
||||
6 | 6 | 00006 | Wed Jan 07 00:00:00 1970 PST | Wed Jan 07 00:00:00 1970 | 6 | 6 | foo |
||||
7 | 7 | 00007 | Thu Jan 08 00:00:00 1970 PST | Thu Jan 08 00:00:00 1970 | 7 | 7 | foo |
||||
8 | 8 | 00008 | Fri Jan 09 00:00:00 1970 PST | Fri Jan 09 00:00:00 1970 | 8 | 8 | foo |
||||
9 | 9 | 00009 | Sat Jan 10 00:00:00 1970 PST | Sat Jan 10 00:00:00 1970 | 9 | 9 | foo |
||||
10 | 0 | 00010 | Sun Jan 11 00:00:00 1970 PST | Sun Jan 11 00:00:00 1970 | 0 | 0 | foo |
||||
(10 rows) |
||||
|
||||
-- subquery+MAX |
||||
SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
------+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
1000 | 0 | 01000 | Thu Jan 01 00:00:00 1970 PST | Thu Jan 01 00:00:00 1970 | 0 | 0 | foo |
||||
(1 row) |
||||
|
||||
-- used in CTE |
||||
WITH t1 AS (SELECT * FROM ft1 WHERE c1 <= 10) SELECT t2.c1, t2.c2, t2.c3, t2.c4 FROM t1, ft2 t2 WHERE t1.c1 = t2.c1 ORDER BY t1.c1; |
||||
c1 | c2 | c3 | c4 |
||||
----+----+-------+------------------------------ |
||||
1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST |
||||
2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST |
||||
3 | 3 | 00003 | Sun Jan 04 00:00:00 1970 PST |
||||
4 | 4 | 00004 | Mon Jan 05 00:00:00 1970 PST |
||||
5 | 5 | 00005 | Tue Jan 06 00:00:00 1970 PST |
||||
6 | 6 | 00006 | Wed Jan 07 00:00:00 1970 PST |
||||
7 | 7 | 00007 | Thu Jan 08 00:00:00 1970 PST |
||||
8 | 8 | 00008 | Fri Jan 09 00:00:00 1970 PST |
||||
9 | 9 | 00009 | Sat Jan 10 00:00:00 1970 PST |
||||
10 | 0 | 00010 | Sun Jan 11 00:00:00 1970 PST |
||||
(10 rows) |
||||
|
||||
-- fixed values |
||||
SELECT 'fixed', NULL FROM ft1 t1 WHERE c1 = 1; |
||||
?column? | ?column? |
||||
----------+---------- |
||||
fixed | |
||||
(1 row) |
||||
|
||||
-- user-defined operator/function |
||||
CREATE FUNCTION postgres_fdw_abs(int) RETURNS int AS $$ |
||||
BEGIN |
||||
RETURN abs($1); |
||||
END |
||||
$$ LANGUAGE plpgsql IMMUTABLE; |
||||
CREATE OPERATOR === ( |
||||
LEFTARG = int, |
||||
RIGHTARG = int, |
||||
PROCEDURE = int4eq, |
||||
COMMUTATOR = ===, |
||||
NEGATOR = !== |
||||
); |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = postgres_fdw_abs(t1.c2); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c1 = postgres_fdw_abs(t1.c2)) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 === t1.c2; |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c1 === t1.c2) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = abs(t1.c2); |
||||
QUERY PLAN |
||||
----------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) pg_catalog.abs(c2))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = t1.c2; |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) c2)) |
||||
(3 rows) |
||||
|
||||
-- =================================================================== |
||||
-- WHERE with remotely-executable conditions |
||||
-- =================================================================== |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 1; -- Var, OpExpr(b), Const |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 100 AND t1.c2 = 0; -- BoolExpr |
||||
QUERY PLAN |
||||
-------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 100)) AND ((c2 OPERATOR(pg_catalog.=) 0)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 IS NULL; -- NullTest |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" IS NULL)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 IS NOT NULL; -- NullTest |
||||
QUERY PLAN |
||||
----------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" IS NOT NULL)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE round(abs(c1), 0) = 1; -- FuncExpr |
||||
QUERY PLAN |
||||
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((pg_catalog.round(pg_catalog."numeric"(pg_catalog.abs("C 1")), 0) OPERATOR(pg_catalog.=) 1::numeric)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = -c1; -- OpExpr(l) |
||||
QUERY PLAN |
||||
----------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) (OPERATOR(pg_catalog.-) "C 1"))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE 1 = c1!; -- OpExpr(r) |
||||
QUERY PLAN |
||||
--------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((1::numeric OPERATOR(pg_catalog.=) (pg_catalog.int8("C 1") OPERATOR(pg_catalog.!)))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE (c1 IS NOT NULL) IS DISTINCT FROM (c1 IS NOT NULL); -- DistinctExpr |
||||
QUERY PLAN |
||||
-------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((("C 1" IS NOT NULL) IS DISTINCT FROM ("C 1" IS NOT NULL))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = ANY(ARRAY[c2, 1, c1 + 0]); -- ScalarArrayOpExpr |
||||
QUERY PLAN |
||||
--------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) ANY (ARRAY[c2, 1, ("C 1" OPERATOR(pg_catalog.+) 0)]))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = (ARRAY[c1,c2,3])[1]; -- ArrayRef |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) ((ARRAY["C 1", c2, 3])[1]))) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c6 = E'foo''s\\bar'; -- check special chars |
||||
QUERY PLAN |
||||
---------------------------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE ((c6::text OPERATOR(pg_catalog.=) E'foo''s\\bar'::text)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c8 = 'foo'; -- can't be sent to remote |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" |
||||
(4 rows) |
||||
|
||||
-- =================================================================== |
||||
-- parameterized queries |
||||
-- =================================================================== |
||||
-- simple join |
||||
PREPARE st1(int, int) AS SELECT t1.c3, t2.c3 FROM ft1 t1, ft2 t2 WHERE t1.c1 = $1 AND t2.c1 = $2; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st1(1, 2); |
||||
QUERY PLAN |
||||
----------------------------------------------------------------------------------------------------------------------------------- |
||||
Nested Loop |
||||
Output: t1.c3, t2.c3 |
||||
-> Foreign Scan on public.ft1 t1 |
||||
Output: t1.c3 |
||||
Remote SQL: SELECT NULL, NULL, c3, NULL, NULL, NULL, NULL, NULL FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
-> Foreign Scan on public.ft2 t2 |
||||
Output: t2.c3 |
||||
Remote SQL: SELECT NULL, NULL, c3, NULL, NULL, NULL, NULL, NULL FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 2)) |
||||
(8 rows) |
||||
|
||||
EXECUTE st1(1, 1); |
||||
c3 | c3 |
||||
-------+------- |
||||
00001 | 00001 |
||||
(1 row) |
||||
|
||||
EXECUTE st1(101, 101); |
||||
c3 | c3 |
||||
-------+------- |
||||
00101 | 00101 |
||||
(1 row) |
||||
|
||||
-- subquery using stable function (can't be sent to remote) |
||||
PREPARE st2(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 < $2 AND t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 > $1 AND EXTRACT(dow FROM c4) = 6) ORDER BY c1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st2(10, 20); |
||||
QUERY PLAN |
||||
---------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Sort |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Sort Key: t1.c1 |
||||
-> Nested Loop Semi Join |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Join Filter: (t1.c3 = t2.c3) |
||||
-> Foreign Scan on public.ft1 t1 |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.<) 20)) |
||||
-> Materialize |
||||
Output: t2.c3 |
||||
-> Foreign Scan on public.ft2 t2 |
||||
Output: t2.c3 |
||||
Filter: (date_part('dow'::text, t2.c4) = 6::double precision) |
||||
Remote SQL: SELECT NULL, NULL, c3, c4, NULL, NULL, NULL, NULL FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.>) 10)) |
||||
(15 rows) |
||||
|
||||
EXECUTE st2(10, 20); |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
16 | 6 | 00016 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo |
||||
(1 row) |
||||
|
||||
EXECUTE st1(101, 101); |
||||
c3 | c3 |
||||
-------+------- |
||||
00101 | 00101 |
||||
(1 row) |
||||
|
||||
-- subquery using immutable function (can be sent to remote) |
||||
PREPARE st3(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 < $2 AND t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 > $1 AND EXTRACT(dow FROM c5) = 6) ORDER BY c1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st3(10, 20); |
||||
QUERY PLAN |
||||
----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
||||
Sort |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Sort Key: t1.c1 |
||||
-> Nested Loop Semi Join |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Join Filter: (t1.c3 = t2.c3) |
||||
-> Foreign Scan on public.ft1 t1 |
||||
Output: t1.c1, t1.c2, t1.c3, t1.c4, t1.c5, t1.c6, t1.c7, t1.c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.<) 20)) |
||||
-> Materialize |
||||
Output: t2.c3 |
||||
-> Foreign Scan on public.ft2 t2 |
||||
Output: t2.c3 |
||||
Remote SQL: SELECT NULL, NULL, c3, NULL, NULL, NULL, NULL, NULL FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.>) 10)) AND ((pg_catalog.date_part('dow'::text, c5) OPERATOR(pg_catalog.=) 6::double precision)) |
||||
(14 rows) |
||||
|
||||
EXECUTE st3(10, 20); |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
16 | 6 | 00016 | Sat Jan 17 00:00:00 1970 PST | Sat Jan 17 00:00:00 1970 | 6 | 6 | foo |
||||
(1 row) |
||||
|
||||
EXECUTE st3(20, 30); |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
23 | 3 | 00023 | Sat Jan 24 00:00:00 1970 PST | Sat Jan 24 00:00:00 1970 | 3 | 3 | foo |
||||
(1 row) |
||||
|
||||
-- custom plan should be chosen initially |
||||
PREPARE st4(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 = $1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(3 rows) |
||||
|
||||
-- once we try it enough times, should switch to generic plan |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) $1)) |
||||
(3 rows) |
||||
|
||||
-- value of $1 should not be sent to remote |
||||
PREPARE st5(user_enum,int) AS SELECT * FROM ft1 t1 WHERE c8 = $1 and c1 = $2; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------ |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = 'foo'::user_enum) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) 1)) |
||||
(4 rows) |
||||
|
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
QUERY PLAN |
||||
------------------------------------------------------------------------------------------------------------------- |
||||
Foreign Scan on public.ft1 t1 |
||||
Output: c1, c2, c3, c4, c5, c6, c7, c8 |
||||
Filter: (t1.c8 = $1) |
||||
Remote SQL: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (("C 1" OPERATOR(pg_catalog.=) $2)) |
||||
(4 rows) |
||||
|
||||
EXECUTE st5('foo', 1); |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
(1 row) |
||||
|
||||
-- cleanup |
||||
DEALLOCATE st1; |
||||
DEALLOCATE st2; |
||||
DEALLOCATE st3; |
||||
DEALLOCATE st4; |
||||
DEALLOCATE st5; |
||||
-- =================================================================== |
||||
-- used in pl/pgsql function |
||||
-- =================================================================== |
||||
CREATE OR REPLACE FUNCTION f_test(p_c1 int) RETURNS int AS $$ |
||||
DECLARE |
||||
v_c1 int; |
||||
BEGIN |
||||
SELECT c1 INTO v_c1 FROM ft1 WHERE c1 = p_c1 LIMIT 1; |
||||
PERFORM c1 FROM ft1 WHERE c1 = p_c1 AND p_c1 = v_c1 LIMIT 1; |
||||
RETURN v_c1; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
SELECT f_test(100); |
||||
f_test |
||||
-------- |
||||
100 |
||||
(1 row) |
||||
|
||||
DROP FUNCTION f_test(int); |
||||
-- =================================================================== |
||||
-- conversion error |
||||
-- =================================================================== |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE int; |
||||
SELECT * FROM ft1 WHERE c1 = 1; -- ERROR |
||||
ERROR: invalid input syntax for integer: "foo" |
||||
CONTEXT: column "c8" of foreign table "ft1" |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; |
||||
-- =================================================================== |
||||
-- subtransaction |
||||
-- + local/remote error doesn't break cursor |
||||
-- =================================================================== |
||||
BEGIN; |
||||
DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; |
||||
FETCH c; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
(1 row) |
||||
|
||||
SAVEPOINT s; |
||||
ERROR OUT; -- ERROR |
||||
ERROR: syntax error at or near "ERROR" |
||||
LINE 1: ERROR OUT; |
||||
^ |
||||
ROLLBACK TO s; |
||||
FETCH c; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
2 | 2 | 00002 | Sat Jan 03 00:00:00 1970 PST | Sat Jan 03 00:00:00 1970 | 2 | 2 | foo |
||||
(1 row) |
||||
|
||||
SAVEPOINT s; |
||||
SELECT * FROM ft1 WHERE 1 / (c1 - 1) > 0; -- ERROR |
||||
ERROR: division by zero |
||||
CONTEXT: Remote SQL command: SELECT "C 1", c2, c3, c4, c5, c6, c7, c8 FROM "S 1"."T 1" WHERE (((1 OPERATOR(pg_catalog./) ("C 1" OPERATOR(pg_catalog.-) 1)) OPERATOR(pg_catalog.>) 0)) |
||||
ROLLBACK TO s; |
||||
FETCH c; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
3 | 3 | 00003 | Sun Jan 04 00:00:00 1970 PST | Sun Jan 04 00:00:00 1970 | 3 | 3 | foo |
||||
(1 row) |
||||
|
||||
SELECT * FROM ft1 ORDER BY c1 LIMIT 1; |
||||
c1 | c2 | c3 | c4 | c5 | c6 | c7 | c8 |
||||
----+----+-------+------------------------------+--------------------------+----+------------+----- |
||||
1 | 1 | 00001 | Fri Jan 02 00:00:00 1970 PST | Fri Jan 02 00:00:00 1970 | 1 | 1 | foo |
||||
(1 row) |
||||
|
||||
COMMIT; |
||||
@ -0,0 +1,293 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* option.c |
||||
* FDW option handling for postgres_fdw |
||||
* |
||||
* Portions Copyright (c) 2012-2013, PostgreSQL Global Development Group |
||||
* |
||||
* IDENTIFICATION |
||||
* contrib/postgres_fdw/option.c |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#include "postgres.h" |
||||
|
||||
#include "postgres_fdw.h" |
||||
|
||||
#include "access/reloptions.h" |
||||
#include "catalog/pg_foreign_server.h" |
||||
#include "catalog/pg_foreign_table.h" |
||||
#include "catalog/pg_user_mapping.h" |
||||
#include "commands/defrem.h" |
||||
|
||||
|
||||
/*
|
||||
* Describes the valid options for objects that this wrapper uses. |
||||
*/ |
||||
typedef struct PgFdwOption |
||||
{ |
||||
const char *keyword; |
||||
Oid optcontext; /* OID of catalog in which option may appear */ |
||||
bool is_libpq_opt; /* true if it's used in libpq */ |
||||
} PgFdwOption; |
||||
|
||||
/*
|
||||
* Valid options for postgres_fdw. |
||||
* Allocated and filled in InitPgFdwOptions. |
||||
*/ |
||||
static PgFdwOption *postgres_fdw_options; |
||||
|
||||
/*
|
||||
* Valid options for libpq. |
||||
* Allocated and filled in InitPgFdwOptions. |
||||
*/ |
||||
static PQconninfoOption *libpq_options; |
||||
|
||||
/*
|
||||
* Helper functions |
||||
*/ |
||||
static void InitPgFdwOptions(void); |
||||
static bool is_valid_option(const char *keyword, Oid context); |
||||
static bool is_libpq_option(const char *keyword); |
||||
|
||||
|
||||
/*
|
||||
* Validate the generic options given to a FOREIGN DATA WRAPPER, SERVER, |
||||
* USER MAPPING or FOREIGN TABLE that uses postgres_fdw. |
||||
* |
||||
* Raise an ERROR if the option or its value is considered invalid. |
||||
*/ |
||||
extern Datum postgres_fdw_validator(PG_FUNCTION_ARGS); |
||||
|
||||
PG_FUNCTION_INFO_V1(postgres_fdw_validator); |
||||
|
||||
Datum |
||||
postgres_fdw_validator(PG_FUNCTION_ARGS) |
||||
{ |
||||
List *options_list = untransformRelOptions(PG_GETARG_DATUM(0)); |
||||
Oid catalog = PG_GETARG_OID(1); |
||||
ListCell *cell; |
||||
|
||||
/* Build our options lists if we didn't yet. */ |
||||
InitPgFdwOptions(); |
||||
|
||||
/*
|
||||
* Check that only options supported by postgres_fdw, and allowed for the |
||||
* current object type, are given. |
||||
*/ |
||||
foreach(cell, options_list) |
||||
{ |
||||
DefElem *def = (DefElem *) lfirst(cell); |
||||
|
||||
if (!is_valid_option(def->defname, catalog)) |
||||
{ |
||||
/*
|
||||
* Unknown option specified, complain about it. Provide a hint |
||||
* with list of valid options for the object. |
||||
*/ |
||||
PgFdwOption *opt; |
||||
StringInfoData buf; |
||||
|
||||
initStringInfo(&buf); |
||||
for (opt = postgres_fdw_options; opt->keyword; opt++) |
||||
{ |
||||
if (catalog == opt->optcontext) |
||||
appendStringInfo(&buf, "%s%s", (buf.len > 0) ? ", " : "", |
||||
opt->keyword); |
||||
} |
||||
|
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_FDW_INVALID_OPTION_NAME), |
||||
errmsg("invalid option \"%s\"", def->defname), |
||||
errhint("Valid options in this context are: %s", |
||||
buf.data))); |
||||
} |
||||
|
||||
/*
|
||||
* Validate option value, when we can do so without any context. |
||||
*/ |
||||
if (strcmp(def->defname, "use_remote_explain") == 0) |
||||
{ |
||||
/* use_remote_explain accepts only boolean values */ |
||||
(void) defGetBoolean(def); |
||||
} |
||||
else if (strcmp(def->defname, "fdw_startup_cost") == 0 || |
||||
strcmp(def->defname, "fdw_tuple_cost") == 0) |
||||
{ |
||||
/* these must have a non-negative numeric value */ |
||||
double val; |
||||
char *endp; |
||||
|
||||
val = strtod(defGetString(def), &endp); |
||||
if (*endp || val < 0) |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_SYNTAX_ERROR), |
||||
errmsg("%s requires a non-negative numeric value", |
||||
def->defname))); |
||||
} |
||||
} |
||||
|
||||
PG_RETURN_VOID(); |
||||
} |
||||
|
||||
/*
|
||||
* Initialize option lists. |
||||
*/ |
||||
static void |
||||
InitPgFdwOptions(void) |
||||
{ |
||||
int num_libpq_opts; |
||||
PQconninfoOption *lopt; |
||||
PgFdwOption *popt; |
||||
|
||||
/* non-libpq FDW-specific FDW options */ |
||||
static const PgFdwOption non_libpq_options[] = { |
||||
{"schema_name", ForeignTableRelationId, false}, |
||||
{"table_name", ForeignTableRelationId, false}, |
||||
{"column_name", AttributeRelationId, false}, |
||||
/* use_remote_explain is available on both server and table */ |
||||
{"use_remote_explain", ForeignServerRelationId, false}, |
||||
{"use_remote_explain", ForeignTableRelationId, false}, |
||||
/* cost factors */ |
||||
{"fdw_startup_cost", ForeignServerRelationId, false}, |
||||
{"fdw_tuple_cost", ForeignServerRelationId, false}, |
||||
{NULL, InvalidOid, false} |
||||
}; |
||||
|
||||
/* Prevent redundant initialization. */ |
||||
if (postgres_fdw_options) |
||||
return; |
||||
|
||||
/*
|
||||
* Get list of valid libpq options. |
||||
* |
||||
* To avoid unnecessary work, we get the list once and use it throughout |
||||
* the lifetime of this backend process. We don't need to care about |
||||
* memory context issues, because PQconndefaults allocates with malloc. |
||||
*/ |
||||
libpq_options = PQconndefaults(); |
||||
if (!libpq_options) /* assume reason for failure is OOM */ |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_FDW_OUT_OF_MEMORY), |
||||
errmsg("out of memory"), |
||||
errdetail("could not get libpq's default connection options"))); |
||||
|
||||
/* Count how many libpq options are available. */ |
||||
num_libpq_opts = 0; |
||||
for (lopt = libpq_options; lopt->keyword; lopt++) |
||||
num_libpq_opts++; |
||||
|
||||
/*
|
||||
* Construct an array which consists of all valid options for |
||||
* postgres_fdw, by appending FDW-specific options to libpq options. |
||||
* |
||||
* We use plain malloc here to allocate postgres_fdw_options because it |
||||
* lives as long as the backend process does. Besides, keeping |
||||
* libpq_options in memory allows us to avoid copying every keyword |
||||
* string. |
||||
*/ |
||||
postgres_fdw_options = (PgFdwOption *) |
||||
malloc(sizeof(PgFdwOption) * num_libpq_opts + |
||||
sizeof(non_libpq_options)); |
||||
if (postgres_fdw_options == NULL) |
||||
ereport(ERROR, |
||||
(errcode(ERRCODE_FDW_OUT_OF_MEMORY), |
||||
errmsg("out of memory"))); |
||||
|
||||
popt = postgres_fdw_options; |
||||
for (lopt = libpq_options; lopt->keyword; lopt++) |
||||
{ |
||||
/* Hide debug options, as well as settings we override internally. */ |
||||
if (strchr(lopt->dispchar, 'D') || |
||||
strcmp(lopt->keyword, "fallback_application_name") == 0 || |
||||
strcmp(lopt->keyword, "client_encoding") == 0) |
||||
continue; |
||||
|
||||
/* We don't have to copy keyword string, as described above. */ |
||||
popt->keyword = lopt->keyword; |
||||
|
||||
/*
|
||||
* "user" and any secret options are allowed only on user mappings. |
||||
* Everything else is a server option. |
||||
*/ |
||||
if (strcmp(lopt->keyword, "user") == 0 || strchr(lopt->dispchar, '*')) |
||||
popt->optcontext = UserMappingRelationId; |
||||
else |
||||
popt->optcontext = ForeignServerRelationId; |
||||
popt->is_libpq_opt = true; |
||||
|
||||
popt++; |
||||
} |
||||
|
||||
/* Append FDW-specific options and dummy terminator. */ |
||||
memcpy(popt, non_libpq_options, sizeof(non_libpq_options)); |
||||
} |
||||
|
||||
/*
|
||||
* Check whether the given option is one of the valid postgres_fdw options. |
||||
* context is the Oid of the catalog holding the object the option is for. |
||||
*/ |
||||
static bool |
||||
is_valid_option(const char *keyword, Oid context) |
||||
{ |
||||
PgFdwOption *opt; |
||||
|
||||
Assert(postgres_fdw_options); /* must be initialized already */ |
||||
|
||||
for (opt = postgres_fdw_options; opt->keyword; opt++) |
||||
{ |
||||
if (context == opt->optcontext && strcmp(opt->keyword, keyword) == 0) |
||||
return true; |
||||
} |
||||
|
||||
return false; |
||||
} |
||||
|
||||
/*
|
||||
* Check whether the given option is one of the valid libpq options. |
||||
*/ |
||||
static bool |
||||
is_libpq_option(const char *keyword) |
||||
{ |
||||
PgFdwOption *opt; |
||||
|
||||
Assert(postgres_fdw_options); /* must be initialized already */ |
||||
|
||||
for (opt = postgres_fdw_options; opt->keyword; opt++) |
||||
{ |
||||
if (opt->is_libpq_opt && strcmp(opt->keyword, keyword) == 0) |
||||
return true; |
||||
} |
||||
|
||||
return false; |
||||
} |
||||
|
||||
/*
|
||||
* Generate key-value arrays which include only libpq options from the |
||||
* given list (which can contain any kind of options). Caller must have |
||||
* allocated large-enough arrays. Returns number of options found. |
||||
*/ |
||||
int |
||||
ExtractConnectionOptions(List *defelems, const char **keywords, |
||||
const char **values) |
||||
{ |
||||
ListCell *lc; |
||||
int i; |
||||
|
||||
/* Build our options lists if we didn't yet. */ |
||||
InitPgFdwOptions(); |
||||
|
||||
i = 0; |
||||
foreach(lc, defelems) |
||||
{ |
||||
DefElem *d = (DefElem *) lfirst(lc); |
||||
|
||||
if (is_libpq_option(d->defname)) |
||||
{ |
||||
keywords[i] = d->defname; |
||||
values[i] = defGetString(d); |
||||
i++; |
||||
} |
||||
} |
||||
return i; |
||||
} |
||||
@ -0,0 +1,18 @@ |
||||
/* contrib/postgres_fdw/postgres_fdw--1.0.sql */ |
||||
|
||||
-- complain if script is sourced in psql, rather than via CREATE EXTENSION |
||||
\echo Use "CREATE EXTENSION postgres_fdw" to load this file. \quit |
||||
|
||||
CREATE FUNCTION postgres_fdw_handler() |
||||
RETURNS fdw_handler |
||||
AS 'MODULE_PATHNAME' |
||||
LANGUAGE C STRICT; |
||||
|
||||
CREATE FUNCTION postgres_fdw_validator(text[], oid) |
||||
RETURNS void |
||||
AS 'MODULE_PATHNAME' |
||||
LANGUAGE C STRICT; |
||||
|
||||
CREATE FOREIGN DATA WRAPPER postgres_fdw |
||||
HANDLER postgres_fdw_handler |
||||
VALIDATOR postgres_fdw_validator; |
||||
File diff suppressed because it is too large
Load Diff
@ -0,0 +1,5 @@ |
||||
# postgres_fdw extension |
||||
comment = 'foreign-data wrapper for remote PostgreSQL servers' |
||||
default_version = '1.0' |
||||
module_pathname = '$libdir/postgres_fdw' |
||||
relocatable = true |
||||
@ -0,0 +1,52 @@ |
||||
/*-------------------------------------------------------------------------
|
||||
* |
||||
* postgres_fdw.h |
||||
* Foreign-data wrapper for remote PostgreSQL servers |
||||
* |
||||
* Portions Copyright (c) 2012-2013, PostgreSQL Global Development Group |
||||
* |
||||
* IDENTIFICATION |
||||
* contrib/postgres_fdw/postgres_fdw.h |
||||
* |
||||
*------------------------------------------------------------------------- |
||||
*/ |
||||
#ifndef POSTGRES_FDW_H |
||||
#define POSTGRES_FDW_H |
||||
|
||||
#include "foreign/foreign.h" |
||||
#include "lib/stringinfo.h" |
||||
#include "nodes/relation.h" |
||||
#include "utils/rel.h" |
||||
|
||||
#include "libpq-fe.h" |
||||
|
||||
/* in connection.c */ |
||||
extern PGconn *GetConnection(ForeignServer *server, UserMapping *user); |
||||
extern void ReleaseConnection(PGconn *conn); |
||||
extern unsigned int GetCursorNumber(PGconn *conn); |
||||
extern void pgfdw_report_error(int elevel, PGresult *res, bool clear, |
||||
const char *sql); |
||||
|
||||
/* in option.c */ |
||||
extern int ExtractConnectionOptions(List *defelems, |
||||
const char **keywords, |
||||
const char **values); |
||||
|
||||
/* in deparse.c */ |
||||
extern void classifyConditions(PlannerInfo *root, |
||||
RelOptInfo *baserel, |
||||
List **remote_conds, |
||||
List **param_conds, |
||||
List **local_conds, |
||||
List **param_numbers); |
||||
extern void deparseSimpleSql(StringInfo buf, |
||||
PlannerInfo *root, |
||||
RelOptInfo *baserel, |
||||
List *local_conds); |
||||
extern void appendWhereClause(StringInfo buf, |
||||
bool has_where, |
||||
List *exprs, |
||||
PlannerInfo *root); |
||||
extern void deparseAnalyzeSql(StringInfo buf, Relation rel); |
||||
|
||||
#endif /* POSTGRES_FDW_H */ |
||||
@ -0,0 +1,272 @@ |
||||
-- =================================================================== |
||||
-- create FDW objects |
||||
-- =================================================================== |
||||
|
||||
CREATE EXTENSION postgres_fdw; |
||||
|
||||
CREATE SERVER testserver1 FOREIGN DATA WRAPPER postgres_fdw; |
||||
CREATE SERVER loopback FOREIGN DATA WRAPPER postgres_fdw |
||||
OPTIONS (dbname 'contrib_regression'); |
||||
|
||||
CREATE USER MAPPING FOR public SERVER testserver1 |
||||
OPTIONS (user 'value', password 'value'); |
||||
CREATE USER MAPPING FOR CURRENT_USER SERVER loopback; |
||||
|
||||
-- =================================================================== |
||||
-- create objects used through FDW loopback server |
||||
-- =================================================================== |
||||
CREATE TYPE user_enum AS ENUM ('foo', 'bar', 'buz'); |
||||
CREATE SCHEMA "S 1"; |
||||
CREATE TABLE "S 1"."T 1" ( |
||||
"C 1" int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum, |
||||
CONSTRAINT t1_pkey PRIMARY KEY ("C 1") |
||||
); |
||||
CREATE TABLE "S 1"."T 2" ( |
||||
c1 int NOT NULL, |
||||
c2 text, |
||||
CONSTRAINT t2_pkey PRIMARY KEY (c1) |
||||
); |
||||
|
||||
INSERT INTO "S 1"."T 1" |
||||
SELECT id, |
||||
id % 10, |
||||
to_char(id, 'FM00000'), |
||||
'1970-01-01'::timestamptz + ((id % 100) || ' days')::interval, |
||||
'1970-01-01'::timestamp + ((id % 100) || ' days')::interval, |
||||
id % 10, |
||||
id % 10, |
||||
'foo'::user_enum |
||||
FROM generate_series(1, 1000) id; |
||||
INSERT INTO "S 1"."T 2" |
||||
SELECT id, |
||||
'AAA' || to_char(id, 'FM000') |
||||
FROM generate_series(1, 100) id; |
||||
|
||||
ANALYZE "S 1"."T 1"; |
||||
ANALYZE "S 1"."T 2"; |
||||
|
||||
-- =================================================================== |
||||
-- create foreign tables |
||||
-- =================================================================== |
||||
CREATE FOREIGN TABLE ft1 ( |
||||
c0 int, |
||||
c1 int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum |
||||
) SERVER loopback; |
||||
ALTER FOREIGN TABLE ft1 DROP COLUMN c0; |
||||
|
||||
CREATE FOREIGN TABLE ft2 ( |
||||
c0 int, |
||||
c1 int NOT NULL, |
||||
c2 int NOT NULL, |
||||
c3 text, |
||||
c4 timestamptz, |
||||
c5 timestamp, |
||||
c6 varchar(10), |
||||
c7 char(10), |
||||
c8 user_enum |
||||
) SERVER loopback; |
||||
ALTER FOREIGN TABLE ft2 DROP COLUMN c0; |
||||
|
||||
-- =================================================================== |
||||
-- tests for validator |
||||
-- =================================================================== |
||||
-- requiressl, krbsrvname and gsslib are omitted because they depend on |
||||
-- configure options |
||||
ALTER SERVER testserver1 OPTIONS ( |
||||
use_remote_explain 'false', |
||||
fdw_startup_cost '123.456', |
||||
fdw_tuple_cost '0.123', |
||||
service 'value', |
||||
connect_timeout 'value', |
||||
dbname 'value', |
||||
host 'value', |
||||
hostaddr 'value', |
||||
port 'value', |
||||
--client_encoding 'value', |
||||
application_name 'value', |
||||
--fallback_application_name 'value', |
||||
keepalives 'value', |
||||
keepalives_idle 'value', |
||||
keepalives_interval 'value', |
||||
-- requiressl 'value', |
||||
sslcompression 'value', |
||||
sslmode 'value', |
||||
sslcert 'value', |
||||
sslkey 'value', |
||||
sslrootcert 'value', |
||||
sslcrl 'value' |
||||
--requirepeer 'value', |
||||
-- krbsrvname 'value', |
||||
-- gsslib 'value', |
||||
--replication 'value' |
||||
); |
||||
ALTER USER MAPPING FOR public SERVER testserver1 |
||||
OPTIONS (DROP user, DROP password); |
||||
ALTER FOREIGN TABLE ft1 OPTIONS (schema_name 'S 1', table_name 'T 1'); |
||||
ALTER FOREIGN TABLE ft2 OPTIONS (schema_name 'S 1', table_name 'T 1'); |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c1 OPTIONS (column_name 'C 1'); |
||||
ALTER FOREIGN TABLE ft2 ALTER COLUMN c1 OPTIONS (column_name 'C 1'); |
||||
\det+ |
||||
|
||||
-- Now we should be able to run ANALYZE. |
||||
-- To exercise multiple code paths, we use local stats on ft1 |
||||
-- and remote_explain mode on ft2. |
||||
ANALYZE ft1; |
||||
ALTER FOREIGN TABLE ft2 OPTIONS (use_remote_explain 'true'); |
||||
|
||||
-- =================================================================== |
||||
-- simple queries |
||||
-- =================================================================== |
||||
-- single table, with/without alias |
||||
EXPLAIN (COSTS false) SELECT * FROM ft1 ORDER BY c3, c1 OFFSET 100 LIMIT 10; |
||||
SELECT * FROM ft1 ORDER BY c3, c1 OFFSET 100 LIMIT 10; |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
SELECT * FROM ft1 t1 ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
-- empty result |
||||
SELECT * FROM ft1 WHERE false; |
||||
-- with WHERE clause |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 101 AND t1.c6 = '1' AND t1.c7 >= '1'; |
||||
SELECT * FROM ft1 t1 WHERE t1.c1 = 101 AND t1.c6 = '1' AND t1.c7 >= '1'; |
||||
-- aggregate |
||||
SELECT COUNT(*) FROM ft1 t1; |
||||
-- join two tables |
||||
SELECT t1.c1 FROM ft1 t1 JOIN ft2 t2 ON (t1.c1 = t2.c1) ORDER BY t1.c3, t1.c1 OFFSET 100 LIMIT 10; |
||||
-- subquery |
||||
SELECT * FROM ft1 t1 WHERE t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 <= 10) ORDER BY c1; |
||||
-- subquery+MAX |
||||
SELECT * FROM ft1 t1 WHERE t1.c3 = (SELECT MAX(c3) FROM ft2 t2) ORDER BY c1; |
||||
-- used in CTE |
||||
WITH t1 AS (SELECT * FROM ft1 WHERE c1 <= 10) SELECT t2.c1, t2.c2, t2.c3, t2.c4 FROM t1, ft2 t2 WHERE t1.c1 = t2.c1 ORDER BY t1.c1; |
||||
-- fixed values |
||||
SELECT 'fixed', NULL FROM ft1 t1 WHERE c1 = 1; |
||||
-- user-defined operator/function |
||||
CREATE FUNCTION postgres_fdw_abs(int) RETURNS int AS $$ |
||||
BEGIN |
||||
RETURN abs($1); |
||||
END |
||||
$$ LANGUAGE plpgsql IMMUTABLE; |
||||
CREATE OPERATOR === ( |
||||
LEFTARG = int, |
||||
RIGHTARG = int, |
||||
PROCEDURE = int4eq, |
||||
COMMUTATOR = ===, |
||||
NEGATOR = !== |
||||
); |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = postgres_fdw_abs(t1.c2); |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 === t1.c2; |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = abs(t1.c2); |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = t1.c2; |
||||
|
||||
-- =================================================================== |
||||
-- WHERE with remotely-executable conditions |
||||
-- =================================================================== |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 1; -- Var, OpExpr(b), Const |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE t1.c1 = 100 AND t1.c2 = 0; -- BoolExpr |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 IS NULL; -- NullTest |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 IS NOT NULL; -- NullTest |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE round(abs(c1), 0) = 1; -- FuncExpr |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = -c1; -- OpExpr(l) |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE 1 = c1!; -- OpExpr(r) |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE (c1 IS NOT NULL) IS DISTINCT FROM (c1 IS NOT NULL); -- DistinctExpr |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = ANY(ARRAY[c2, 1, c1 + 0]); -- ScalarArrayOpExpr |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c1 = (ARRAY[c1,c2,3])[1]; -- ArrayRef |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c6 = E'foo''s\\bar'; -- check special chars |
||||
EXPLAIN (VERBOSE, COSTS false) SELECT * FROM ft1 t1 WHERE c8 = 'foo'; -- can't be sent to remote |
||||
|
||||
-- =================================================================== |
||||
-- parameterized queries |
||||
-- =================================================================== |
||||
-- simple join |
||||
PREPARE st1(int, int) AS SELECT t1.c3, t2.c3 FROM ft1 t1, ft2 t2 WHERE t1.c1 = $1 AND t2.c1 = $2; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st1(1, 2); |
||||
EXECUTE st1(1, 1); |
||||
EXECUTE st1(101, 101); |
||||
-- subquery using stable function (can't be sent to remote) |
||||
PREPARE st2(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 < $2 AND t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 > $1 AND EXTRACT(dow FROM c4) = 6) ORDER BY c1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st2(10, 20); |
||||
EXECUTE st2(10, 20); |
||||
EXECUTE st1(101, 101); |
||||
-- subquery using immutable function (can be sent to remote) |
||||
PREPARE st3(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 < $2 AND t1.c3 IN (SELECT c3 FROM ft2 t2 WHERE c1 > $1 AND EXTRACT(dow FROM c5) = 6) ORDER BY c1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st3(10, 20); |
||||
EXECUTE st3(10, 20); |
||||
EXECUTE st3(20, 30); |
||||
-- custom plan should be chosen initially |
||||
PREPARE st4(int) AS SELECT * FROM ft1 t1 WHERE t1.c1 = $1; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
-- once we try it enough times, should switch to generic plan |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st4(1); |
||||
-- value of $1 should not be sent to remote |
||||
PREPARE st5(user_enum,int) AS SELECT * FROM ft1 t1 WHERE c8 = $1 and c1 = $2; |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXPLAIN (VERBOSE, COSTS false) EXECUTE st5('foo', 1); |
||||
EXECUTE st5('foo', 1); |
||||
|
||||
-- cleanup |
||||
DEALLOCATE st1; |
||||
DEALLOCATE st2; |
||||
DEALLOCATE st3; |
||||
DEALLOCATE st4; |
||||
DEALLOCATE st5; |
||||
|
||||
-- =================================================================== |
||||
-- used in pl/pgsql function |
||||
-- =================================================================== |
||||
CREATE OR REPLACE FUNCTION f_test(p_c1 int) RETURNS int AS $$ |
||||
DECLARE |
||||
v_c1 int; |
||||
BEGIN |
||||
SELECT c1 INTO v_c1 FROM ft1 WHERE c1 = p_c1 LIMIT 1; |
||||
PERFORM c1 FROM ft1 WHERE c1 = p_c1 AND p_c1 = v_c1 LIMIT 1; |
||||
RETURN v_c1; |
||||
END; |
||||
$$ LANGUAGE plpgsql; |
||||
SELECT f_test(100); |
||||
DROP FUNCTION f_test(int); |
||||
|
||||
-- =================================================================== |
||||
-- conversion error |
||||
-- =================================================================== |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE int; |
||||
SELECT * FROM ft1 WHERE c1 = 1; -- ERROR |
||||
ALTER FOREIGN TABLE ft1 ALTER COLUMN c8 TYPE user_enum; |
||||
|
||||
-- =================================================================== |
||||
-- subtransaction |
||||
-- + local/remote error doesn't break cursor |
||||
-- =================================================================== |
||||
BEGIN; |
||||
DECLARE c CURSOR FOR SELECT * FROM ft1 ORDER BY c1; |
||||
FETCH c; |
||||
SAVEPOINT s; |
||||
ERROR OUT; -- ERROR |
||||
ROLLBACK TO s; |
||||
FETCH c; |
||||
SAVEPOINT s; |
||||
SELECT * FROM ft1 WHERE 1 / (c1 - 1) > 0; -- ERROR |
||||
ROLLBACK TO s; |
||||
FETCH c; |
||||
SELECT * FROM ft1 ORDER BY c1 LIMIT 1; |
||||
COMMIT; |
||||
@ -0,0 +1,325 @@ |
||||
<!-- doc/src/sgml/postgres-fdw.sgml --> |
||||
|
||||
<sect1 id="postgres-fdw" xreflabel="postgres_fdw"> |
||||
<title>postgres_fdw</title> |
||||
|
||||
<indexterm zone="postgres-fdw"> |
||||
<primary>postgres_fdw</primary> |
||||
</indexterm> |
||||
|
||||
<para> |
||||
The <filename>postgres_fdw</> module provides the foreign-data wrapper |
||||
<literal>postgres_fdw</literal>, which can be used to access data |
||||
stored in external <productname>PostgreSQL</productname> servers. |
||||
</para> |
||||
|
||||
<para> |
||||
The functionality provided by this module overlaps substantially |
||||
with the functionality of the older <xref linkend="dblink"> module. |
||||
But <filename>postgres_fdw</> provides more transparent and |
||||
standards-compliant syntax for accessing remote tables, and can give |
||||
better performance in many cases. |
||||
</para> |
||||
|
||||
<para> |
||||
To prepare for remote access using <filename>postgres_fdw</>: |
||||
<orderedlist spacing="compact"> |
||||
<listitem> |
||||
<para> |
||||
Install the <filename>postgres_fdw</> extension using <xref |
||||
linkend="sql-createextension">. |
||||
</para> |
||||
</listitem> |
||||
<listitem> |
||||
<para> |
||||
Create a foreign server object, using <xref linkend="sql-createserver">, |
||||
to represent each remote database you want to connect to. |
||||
Specify connection information, except <literal>user</literal> and |
||||
<literal>password</literal>, as options of the server object. |
||||
</para> |
||||
</listitem> |
||||
<listitem> |
||||
<para> |
||||
Create a user mapping, using <xref linkend="sql-createusermapping">, for |
||||
each database user you want to allow to access each foreign server. |
||||
Specify the remote user name and password to use as |
||||
<literal>user</literal> and <literal>password</literal> options of the |
||||
user mapping. |
||||
</para> |
||||
</listitem> |
||||
<listitem> |
||||
<para> |
||||
Create a foreign table, using <xref linkend="sql-createforeigntable">, |
||||
for each remote table you want to access. The columns of the foreign |
||||
table must match the referenced remote table. You can, however, use |
||||
table and/or column names different from the remote table's, if you |
||||
specify the correct remote names as options of the foreign table object. |
||||
</para> |
||||
</listitem> |
||||
</orderedlist> |
||||
</para> |
||||
|
||||
<para> |
||||
Now you need only <command>SELECT</> from a foreign table to access |
||||
the data stored in its underlying remote table. |
||||
</para> |
||||
|
||||
<para> |
||||
It is generally recommended that the columns of a foreign table be declared |
||||
with exactly the same data types, and collations if applicable, as the |
||||
referenced columns of the remote table. Although <filename>postgres_fdw</> |
||||
is currently rather forgiving about performing data type conversions at |
||||
need, surprising semantic anomalies may arise when types or collations do |
||||
not match, due to the remote server interpreting <literal>WHERE</> clauses |
||||
slightly differently from the local server. |
||||
</para> |
||||
|
||||
<para> |
||||
Note that a foreign table can be declared with fewer columns, or with a |
||||
different column order, than its underlying remote table has. Matching |
||||
of columns to the remote table is by name, not position. |
||||
</para> |
||||
|
||||
<sect2> |
||||
<title>FDW Options of postgres_fdw</title> |
||||
|
||||
<sect3> |
||||
<title>Connection Options</title> |
||||
|
||||
<para> |
||||
A foreign server using the <filename>postgres_fdw</> foreign data wrapper |
||||
can have the same options that <application>libpq</> accepts in |
||||
connection strings, as described in <xref linkend="libpq-paramkeywords">, |
||||
except that these options are not allowed: |
||||
|
||||
<itemizedlist spacing="compact"> |
||||
<listitem> |
||||
<para> |
||||
<literal>user</literal> and <literal>password</literal> (specify these |
||||
for a user mapping, instead) |
||||
</para> |
||||
</listitem> |
||||
<listitem> |
||||
<para> |
||||
<literal>client_encoding</> (this is automatically set from the local |
||||
server encoding) |
||||
</para> |
||||
</listitem> |
||||
<listitem> |
||||
<para> |
||||
<literal>fallback_application_name</> (always set to |
||||
<literal>postgres_fdw</>) |
||||
</para> |
||||
</listitem> |
||||
</itemizedlist> |
||||
</para> |
||||
|
||||
<para> |
||||
Only superusers may connect to foreign servers without password |
||||
authentication, so always specify the <literal>password</literal> option |
||||
for user mappings belonging to non-superusers. |
||||
</para> |
||||
</sect3> |
||||
|
||||
<sect3> |
||||
<title>Object Name Options</title> |
||||
|
||||
<para> |
||||
These options can be used to control the names used in SQL statements |
||||
sent to the remote <productname>PostgreSQL</productname> server. These |
||||
options are needed when a foreign table is created with names different |
||||
from the underlying remote table's names. |
||||
</para> |
||||
|
||||
<variablelist> |
||||
|
||||
<varlistentry> |
||||
<term><literal>schema_name</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a foreign table, gives the |
||||
schema name to use for the foreign table on the remote server. If this |
||||
option is omitted, the name of the foreign table's schema is used. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
<varlistentry> |
||||
<term><literal>table_name</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a foreign table, gives the |
||||
table name to use for the foreign table on the remote server. If this |
||||
option is omitted, the foreign table's name is used. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
<varlistentry> |
||||
<term><literal>column_name</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a column of a foreign table, |
||||
gives the column name to use for the column on the remote server. |
||||
If this option is omitted, the column's name is used. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
</variablelist> |
||||
|
||||
</sect3> |
||||
|
||||
<sect3> |
||||
<title>Cost Estimation Options</title> |
||||
|
||||
<para> |
||||
<filename>postgres_fdw</> retrieves remote data by executing queries |
||||
against remote servers, so ideally the estimated cost of scanning a |
||||
foreign table should be whatever it costs to be done on the remote |
||||
server, plus some overhead for communication. The most reliable way to |
||||
get such an estimate is to ask the remote server and then add something |
||||
for overhead — but for simple queries, it may not be worth the cost |
||||
of an additional remote query to get a cost estimate. |
||||
So <filename>postgres_fdw</> provides the following options to control |
||||
how cost estimation is done: |
||||
</para> |
||||
|
||||
<variablelist> |
||||
|
||||
<varlistentry> |
||||
<term><literal>use_remote_estimate</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a foreign table or a foreign |
||||
server, controls whether <filename>postgres_fdw</> issues remote |
||||
<command>EXPLAIN</command> commands to obtain cost estimates. |
||||
A setting for a foreign table overrides any setting for its server, |
||||
but only for that table. |
||||
The default is <literal>false</literal>. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
<varlistentry> |
||||
<term><literal>fdw_startup_cost</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a foreign server, is a numeric |
||||
value that is added to the estimated startup cost of any foreign-table |
||||
scan on that server. This represents the additional overhead of |
||||
establishing a connection, parsing and planning the query on the |
||||
remote side, etc. |
||||
The default value is <literal>100</literal>. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
<varlistentry> |
||||
<term><literal>fdw_tuple_cost</literal></term> |
||||
<listitem> |
||||
<para> |
||||
This option, which can be specified for a foreign server, is a numeric |
||||
value that is used as extra cost per-tuple for foreign-table |
||||
scans on that server. This represents the additional overhead of |
||||
data transfer between servers. You might increase or decrease this |
||||
number to reflect higher or lower network delay to the remote server. |
||||
The default value is <literal>0.01</literal>. |
||||
</para> |
||||
</listitem> |
||||
</varlistentry> |
||||
|
||||
</variablelist> |
||||
|
||||
<para> |
||||
When <literal>use_remote_estimate</literal> is true, |
||||
<filename>postgres_fdw</> obtains rowcount and cost estimates from the |
||||
remote server and then adds <literal>fdw_startup_cost</literal> and |
||||
<literal>fdw_tuple_cost</literal> to the cost estimates. When |
||||
<literal>use_remote_estimate</literal> is false, |
||||
<filename>postgres_fdw</> performs local rowcount and cost estimation |
||||
and then adds <literal>fdw_startup_cost</literal> and |
||||
<literal>fdw_tuple_cost</literal> to the cost estimates. This local |
||||
estimation is unlikely to be very accurate unless local copies of the |
||||
remote table's statistics are available. Running |
||||
<xref linkend="sql-analyze"> on the foreign table is the way to update |
||||
the local statistics; this will perform a scan of the remote table and |
||||
then calculate and store statistics just as though the table were local. |
||||
Keeping local statistics can be a useful way to reduce per-query planning |
||||
overhead for a remote table — but if the remote table is |
||||
frequently updated, the local statistics will soon be obsolete. |
||||
</para> |
||||
|
||||
</sect3> |
||||
</sect2> |
||||
|
||||
<sect2> |
||||
<title>Connection Management</title> |
||||
|
||||
<para> |
||||
<filename>postgres_fdw</filename> establishes a connection to a |
||||
foreign server during the first query that uses a foreign table |
||||
associated with the foreign server. This connection is kept and |
||||
re-used for subsequent queries in the same session. However, if |
||||
multiple user identities (user mappings) are used to access the foreign |
||||
server, a connection is established for each user mapping. |
||||
</para> |
||||
</sect2> |
||||
|
||||
<sect2> |
||||
<title>Transaction Management</title> |
||||
|
||||
<para> |
||||
During a query that references any remote tables on a foreign server, |
||||
<filename>postgres_fdw</filename> opens a transaction on the |
||||
remote server if one is not already open corresponding to the current |
||||
local transaction. The remote transaction is committed or aborted when |
||||
the local transaction commits or aborts. Savepoints are similarly |
||||
managed by creating corresponding remote savepoints. |
||||
</para> |
||||
|
||||
<para> |
||||
The remote transaction uses <literal>SERIALIZABLE</> |
||||
isolation level when the local transaction has <literal>SERIALIZABLE</> |
||||
isolation level; otherwise it uses <literal>REPEATABLE READ</> |
||||
isolation level. This choice ensures that if a query performs multiple |
||||
table scans on the remote server, it will get snapshot-consistent results |
||||
for all the scans. A consequence is that successive queries within a |
||||
single transaction will see the same data from the remote server, even if |
||||
concurrent updates are occurring on the remote server due to other |
||||
activities. That behavior would be expected anyway if the local |
||||
transaction uses <literal>SERIALIZABLE</> or <literal>REPEATABLE READ</> |
||||
isolation level, but it might be surprising for a <literal>READ |
||||
COMMITTED</> local transaction. A future |
||||
<productname>PostgreSQL</productname> release might modify these rules. |
||||
</para> |
||||
</sect2> |
||||
|
||||
<sect2> |
||||
<title>Remote Query Optimization</title> |
||||
|
||||
<para> |
||||
<filename>postgres_fdw</> attempts to optimize remote queries to reduce |
||||
the amount of data transferred from foreign servers. This is done by |
||||
sending query <literal>WHERE</> clauses to the remote server for |
||||
execution, and by not retrieving table columns that are not needed for |
||||
the current query. To reduce the risk of misexecution of queries, |
||||
<literal>WHERE</> clauses are not sent to the remote server unless they use |
||||
only built-in data types, operators, and functions. Operators and |
||||
functions in the clauses must be <literal>IMMUTABLE</> as well. |
||||
</para> |
||||
|
||||
<para> |
||||
The query that is actually sent to the remote server for execution can |
||||
be examined using <command>EXPLAIN VERBOSE</>. |
||||
</para> |
||||
</sect2> |
||||
|
||||
<sect2> |
||||
<title>Author</title> |
||||
<para> |
||||
Shigeru Hanada <email>shigeru.hanada@gmail.com</email> |
||||
</para> |
||||
</sect2> |
||||
|
||||
</sect1> |
||||
Loading…
Reference in new issue