diff options
| -rw-r--r-- | contrib/postgres_fdw/connection.c | 60 | ||||
| -rw-r--r-- | contrib/postgres_fdw/deparse.c | 6 | ||||
| -rw-r--r-- | contrib/postgres_fdw/expected/postgres_fdw.out | 12 | ||||
| -rw-r--r-- | contrib/postgres_fdw/postgres_fdw.c | 65 | ||||
| -rw-r--r-- | contrib/postgres_fdw/postgres_fdw.h | 4 | ||||
| -rw-r--r-- | contrib/postgres_fdw/sql/postgres_fdw.sql | 4 |
6 files changed, 122 insertions, 29 deletions
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 22ac50e6f9f..0a723f0dfa1 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -67,6 +67,7 @@ static bool xact_got_connection = false; static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user); static void check_conn_params(const char **keywords, const char **values); static void configure_remote_session(PGconn *conn); +static void do_sql_command(PGconn *conn, const char *sql); static void begin_remote_xact(ConnCacheEntry *entry); static void pgfdw_xact_callback(XactEvent event, void *arg); static void pgfdw_subxact_callback(SubXactEvent event, @@ -314,11 +315,43 @@ check_conn_params(const char **keywords, const char **values) static void configure_remote_session(PGconn *conn) { - const char *sql; - PGresult *res; + int remoteversion = PQserverVersion(conn); /* Force the search path to contain only pg_catalog (see deparse.c) */ - sql = "SET search_path = pg_catalog"; + do_sql_command(conn, "SET search_path = pg_catalog"); + + /* + * Set remote timezone; this is basically just cosmetic, since all + * transmitted and returned timestamptzs should specify a zone explicitly + * anyway. However it makes the regression test outputs more predictable. + * + * We don't risk setting remote zone equal to ours, since the remote + * server might use a different timezone database. + */ + do_sql_command(conn, "SET timezone = UTC"); + + /* + * Set values needed to ensure unambiguous data output from remote. (This + * logic should match what pg_dump does. See also set_transmission_modes + * in postgres_fdw.c.) + */ + do_sql_command(conn, "SET datestyle = ISO"); + if (remoteversion >= 80400) + do_sql_command(conn, "SET intervalstyle = postgres"); + if (remoteversion >= 90000) + do_sql_command(conn, "SET extra_float_digits = 3"); + else + do_sql_command(conn, "SET extra_float_digits = 2"); +} + +/* + * Convenience subroutine to issue a non-data-returning SQL command to remote + */ +static void +do_sql_command(PGconn *conn, const char *sql) +{ + PGresult *res; + res = PQexec(conn, sql); if (PQresultStatus(res) != PGRES_COMMAND_OK) pgfdw_report_error(ERROR, res, true, sql); @@ -339,7 +372,6 @@ 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) @@ -353,10 +385,7 @@ begin_remote_xact(ConnCacheEntry *entry) 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); + do_sql_command(entry->conn, sql); entry->xact_depth = 1; } @@ -370,10 +399,7 @@ begin_remote_xact(ConnCacheEntry *entry) 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); + do_sql_command(entry->conn, sql); entry->xact_depth++; } } @@ -509,10 +535,7 @@ pgfdw_xact_callback(XactEvent event, void *arg) { 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); + do_sql_command(entry->conn, "COMMIT TRANSACTION"); /* * If there were any errors in subtransactions, and we made @@ -647,10 +670,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid, { /* 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); + do_sql_command(entry->conn, sql); } else { diff --git a/contrib/postgres_fdw/deparse.c b/contrib/postgres_fdw/deparse.c index f5d723cc38a..8ed915744af 100644 --- a/contrib/postgres_fdw/deparse.c +++ b/contrib/postgres_fdw/deparse.c @@ -469,8 +469,12 @@ appendWhereClause(StringInfo buf, List *exprs, bool is_first) { + int nestlevel; ListCell *lc; + /* Make sure any constants in the exprs are printed portably */ + nestlevel = set_transmission_modes(); + foreach(lc, exprs) { RestrictInfo *ri = (RestrictInfo *) lfirst(lc); @@ -487,6 +491,8 @@ appendWhereClause(StringInfo buf, is_first = false; } + + reset_transmission_modes(nestlevel); } /* diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index 50b1f239209..9b7ca313605 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -1983,10 +1983,10 @@ INSERT INTO ft1(c1, c2) VALUES(1111, -2); -- c2positive ERROR: new row for relation "T 1" violates check constraint "c2positive" DETAIL: Failing row contains (1111, -2, null, null, null, (^-^;), null, null). CONTEXT: Remote SQL command: INSERT INTO "S 1"."T 1"("C 1", c2) VALUES ($1, $2) -UPDATE ft1 SET c2 = -c2, c4 = null WHERE c1 = 1; -- c2positive +UPDATE ft1 SET c2 = -c2 WHERE c1 = 1; -- c2positive ERROR: new row for relation "T 1" violates check constraint "c2positive" -DETAIL: Failing row contains (1, -1, 00001_trig_update, null, 1970-01-02 00:00:00, 1, 1 , foo). -CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2, c4 = $3 WHERE ctid = $1 +DETAIL: Failing row contains (1, -1, 00001_trig_update, 1970-01-02 08:00:00+00, 1970-01-02 00:00:00, 1, 1 , foo). +CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2 WHERE ctid = $1 -- Test savepoint/rollback behavior select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count @@ -2142,10 +2142,10 @@ select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; (13 rows) savepoint s3; -update ft2 set c2 = -2, c4 = null where c2 = 42; -- fail on remote side +update ft2 set c2 = -2 where c2 = 42; -- fail on remote side ERROR: new row for relation "T 1" violates check constraint "c2positive" -DETAIL: Failing row contains (10, -2, 00010_trig_update_trig_update, null, 1970-01-11 00:00:00, 0, 0 , foo). -CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2, c4 = $3 WHERE ctid = $1 +DETAIL: Failing row contains (10, -2, 00010_trig_update_trig_update, 1970-01-11 08:00:00+00, 1970-01-11 00:00:00, 0, 0 , foo). +CONTEXT: Remote SQL command: UPDATE "S 1"."T 1" SET c2 = $2 WHERE ctid = $1 rollback to savepoint s3; select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; c2 | count diff --git a/contrib/postgres_fdw/postgres_fdw.c b/contrib/postgres_fdw/postgres_fdw.c index a6db061d603..95505c8a1c7 100644 --- a/contrib/postgres_fdw/postgres_fdw.c +++ b/contrib/postgres_fdw/postgres_fdw.c @@ -30,6 +30,7 @@ #include "optimizer/var.h" #include "parser/parsetree.h" #include "utils/builtins.h" +#include "utils/guc.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -1554,9 +1555,12 @@ create_cursor(ForeignScanState *node) if (numParams > 0 && !fsstate->extparams_done) { ParamListInfo params = node->ss.ps.state->es_param_list_info; + int nestlevel; List *param_numbers; ListCell *lc; + nestlevel = set_transmission_modes(); + param_numbers = (List *) list_nth(fsstate->fdw_private, FdwScanPrivateExternParamIds); foreach(lc, param_numbers) @@ -1598,6 +1602,9 @@ create_cursor(ForeignScanState *node) prm->value); } } + + reset_transmission_modes(nestlevel); + fsstate->extparams_done = true; } @@ -1706,6 +1713,56 @@ fetch_more_data(ForeignScanState *node) } /* + * Force assorted GUC parameters to settings that ensure that we'll output + * data values in a form that is unambiguous to the remote server. + * + * This is rather expensive and annoying to do once per row, but there's + * little choice if we want to be sure values are transmitted accurately; + * we can't leave the settings in place between rows for fear of affecting + * user-visible computations. + * + * We use the equivalent of a function SET option to allow the settings to + * persist only until the caller calls reset_transmission_modes(). If an + * error is thrown in between, guc.c will take care of undoing the settings. + * + * The return value is the nestlevel that must be passed to + * reset_transmission_modes() to undo things. + */ +int +set_transmission_modes(void) +{ + int nestlevel = NewGUCNestLevel(); + + /* + * The values set here should match what pg_dump does. See also + * configure_remote_session in connection.c. + */ + if (DateStyle != USE_ISO_DATES) + (void) set_config_option("datestyle", "ISO", + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0); + if (IntervalStyle != INTSTYLE_POSTGRES) + (void) set_config_option("intervalstyle", "postgres", + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0); + if (extra_float_digits < 3) + (void) set_config_option("extra_float_digits", "3", + PGC_USERSET, PGC_S_SESSION, + GUC_ACTION_SAVE, true, 0); + + return nestlevel; +} + +/* + * Undo the effects of set_transmission_modes(). + */ +void +reset_transmission_modes(int nestlevel) +{ + AtEOXact_GUC(true, nestlevel); +} + +/* * Utility routine to close a cursor. */ static void @@ -1791,16 +1848,20 @@ convert_prep_stmt_params(PgFdwModifyState *fmstate, /* 1st parameter should be ctid, if it's in use */ if (tupleid != NULL) { + /* don't need set_transmission_modes for TID output */ p_values[pindex] = OutputFunctionCall(&fmstate->p_flinfo[pindex], PointerGetDatum(tupleid)); pindex++; } /* get following parameters from slot */ - if (slot != NULL) + if (slot != NULL && fmstate->target_attrs != NIL) { + int nestlevel; ListCell *lc; + nestlevel = set_transmission_modes(); + foreach(lc, fmstate->target_attrs) { int attnum = lfirst_int(lc); @@ -1815,6 +1876,8 @@ convert_prep_stmt_params(PgFdwModifyState *fmstate, value); pindex++; } + + reset_transmission_modes(nestlevel); } Assert(pindex == fmstate->p_nums); diff --git a/contrib/postgres_fdw/postgres_fdw.h b/contrib/postgres_fdw/postgres_fdw.h index 236a60b4898..9149aa186f7 100644 --- a/contrib/postgres_fdw/postgres_fdw.h +++ b/contrib/postgres_fdw/postgres_fdw.h @@ -20,6 +20,10 @@ #include "libpq-fe.h" +/* in postgres_fdw.c */ +extern int set_transmission_modes(void); +extern void reset_transmission_modes(int nestlevel); + /* in connection.c */ extern PGconn *GetConnection(ForeignServer *server, UserMapping *user, bool will_prep_stmt); diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index 2e9e17f5f93..007109c7c76 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -316,7 +316,7 @@ ALTER TABLE "S 1"."T 1" ADD CONSTRAINT c2positive CHECK (c2 >= 0); INSERT INTO ft1(c1, c2) VALUES(11, 12); -- duplicate key INSERT INTO ft1(c1, c2) VALUES(1111, -2); -- c2positive -UPDATE ft1 SET c2 = -c2, c4 = null WHERE c1 = 1; -- c2positive +UPDATE ft1 SET c2 = -c2 WHERE c1 = 1; -- c2positive -- Test savepoint/rollback behavior select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; @@ -337,7 +337,7 @@ select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; release savepoint s2; select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; savepoint s3; -update ft2 set c2 = -2, c4 = null where c2 = 42; -- fail on remote side +update ft2 set c2 = -2 where c2 = 42; -- fail on remote side rollback to savepoint s3; select c2, count(*) from ft2 where c2 < 500 group by 1 order by 1; release savepoint s3; |