diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c index 116be7d..9492f14 100644 --- a/contrib/postgres_fdw/connection.c +++ b/contrib/postgres_fdw/connection.c @@ -42,20 +42,21 @@ typedef struct ConnCacheKey } 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 */ bool have_prep_stmt; /* have we prepared any stmts in this xact? */ bool have_error; /* have any subxacts aborted in this xact? */ + char *prep_xact_name; /* Name of prepared transaction on this connection */ } ConnCacheEntry; /* * Connection cache (initialized on first use) */ static HTAB *ConnectionHash = NULL; /* for assigning cursor numbers and prepared statement numbers */ static unsigned int cursor_number = 0; static unsigned int prep_stmt_number = 0; @@ -135,20 +136,21 @@ GetConnection(ForeignServer *server, UserMapping *user, * 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; entry->have_prep_stmt = false; entry->have_error = false; + entry->prep_xact_name = NULL; } /* * 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 @@ -156,20 +158,21 @@ GetConnection(ForeignServer *server, UserMapping *user, * will be left in a valid empty state.) */ if (entry->conn == NULL) { entry->xact_depth = 0; /* just to be sure */ entry->have_prep_stmt = false; entry->have_error = false; entry->conn = connect_pg_server(server, user); elog(DEBUG3, "new postgres_fdw connection %p for server \"%s\"", entry->conn, server->servername); + entry->prep_xact_name = NULL; } /* * Start a new transaction or subtransaction if needed. */ begin_remote_xact(entry); /* Remember if caller will prepare statements */ entry->have_prep_stmt |= will_prep_stmt; @@ -507,20 +510,55 @@ pgfdw_report_error(int elevel, PGresult *res, PGconn *conn, if (clear) PQclear(res); PG_RE_THROW(); } PG_END_TRY(); if (clear) PQclear(res); } /* + * pgfdw_get_prep_xact_id + * The function crafts prepared transaction identifier. PostgreSQL documentation + * mentions two restrictions on the name + * 1. String literal, less than 200 bytes long. + * 2. Should not be same as any other concurrent prepared transaction id. + * + * The name should give enough hints to the DBA/user, in case a manual + * intervention is required as per the comment in its caller. + * + * To make the prepared transaction id, we should ideally use something like + * UUID, which gives unique ids with high probability, but that may be expensive + * here and UUID extension which provides the function to generate UUID is part + * of the extension not the core. + * + * In order to make it easy for the DBA/user to identify the originating server + * (so that he can verify the status of the originating transaction), we should + * let the DBA/user configure a prefix to be used for prepared transaction ids, + * but that requires changes to the core, so left out of the work. Instead we + * use the serverid and userid, which help in creating unique ids as well give + * hints (albeit weak) to the originating transaction. + */ +static char * +pgfdw_get_prep_xact_id(ConnCacheEntry *entry) +{ +/* Maximum length of the prepared transaction id, borrowed from twophase.c */ +#define PREP_XACT_ID_MAX_LEN 200 +#define RANDOM_LARGE_MULTIPLIER 1000 + char *prep_xact_name = (char *)palloc(PREP_XACT_ID_MAX_LEN * sizeof(char)); + + snprintf(prep_xact_name, PREP_XACT_ID_MAX_LEN, "%s_%4d_%d_%d", + "px", abs(random() * RANDOM_LARGE_MULTIPLIER), + entry->key.serverid, entry->key.userid); + return prep_xact_name; +} +/* * 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) @@ -540,24 +578,48 @@ pgfdw_xact_callback(XactEvent event, void *arg) continue; /* If it has an open remote transaction, try to close it */ if (entry->xact_depth > 0) { elog(DEBUG3, "closing remote transaction on connection %p", entry->conn); switch (event) { + char *prep_xact_name; + StringInfo command; case XACT_EVENT_PRE_COMMIT: - /* Commit all remote transactions during pre-commit */ - do_sql_command(entry->conn, "COMMIT TRANSACTION"); + /* + * Prepare the transaction on the remote nodes, to check the + * subsequent COMMIT would go through or not. + * TODO: + * We also don't need prepared transactions involving a + * single foreign server. How do we know that the current + * transaction involved only a single foreign server? + */ + prep_xact_name = pgfdw_get_prep_xact_id(entry); + command = makeStringInfo(); + appendStringInfo(command, "PREPARE TRANSACTION '%s'", prep_xact_name); + do_sql_command(entry->conn, command->data); + /* The transaction got prepared, register this fact */ + entry->prep_xact_name = prep_xact_name; + /* + * XXX: + * After this, if the server crashes or looses connection to + * the foreign server, before COMMIT/ABORT message is sent + * to the foreign server, the transaction prepared above would + * remain in that state till DBA/user manually commits or + * rolls it back. The automation of this step would require + * changes in the core transaction manager and left out for + * now. + */ /* * If there were any errors in subtransactions, and we * made prepared statements, do a DEALLOCATE ALL to make * sure we get rid of all prepared statements. This is * annoying and not terribly bulletproof, but it's * probably not worth trying harder. * * DEALLOCATE ALL only exists in 8.3 and later, so this * constrains how old a server postgres_fdw can * communicate with. We intentionally ignore errors in @@ -568,86 +630,115 @@ pgfdw_xact_callback(XactEvent event, void *arg) */ if (entry->have_prep_stmt && entry->have_error) { res = PQexec(entry->conn, "DEALLOCATE ALL"); PQclear(res); } entry->have_prep_stmt = false; entry->have_error = false; break; case XACT_EVENT_PRE_PREPARE: - /* * We disallow remote transactions that modified anything, * since it's not very 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: + /* + * If we prepared a transaction on a foreign server, commit + * it. + */ + if (entry->prep_xact_name) + { + command = makeStringInfo(); + appendStringInfo(command, "COMMIT PREPARED '%s'", + entry->prep_xact_name); + do_sql_command(entry->conn, command->data); + entry->prep_xact_name = NULL; + } + else + /* Pre-commit should have closed the open transaction */ + elog(ERROR, "missed cleaning up connection during pre-commit"); + + break; case XACT_EVENT_PREPARE: /* Pre-commit should have closed the open transaction */ elog(ERROR, "missed cleaning up connection during pre-commit"); break; case XACT_EVENT_ABORT: /* Assume we might have lost track of prepared statements */ entry->have_error = true; /* If we're aborting, abort all remote transactions too */ - res = PQexec(entry->conn, "ABORT TRANSACTION"); + if (entry->prep_xact_name) + { + command = makeStringInfo(); + appendStringInfo(command, "ROLLBACK PREPARED '%s'", entry->prep_xact_name); + res = PQexec(entry->conn, command->data); + pfree(entry->prep_xact_name); + entry->prep_xact_name = NULL; + } + else + 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, entry->conn, true, "ABORT TRANSACTION"); else { PQclear(res); /* As above, make sure to clear any prepared stmts */ if (entry->have_prep_stmt && entry->have_error) { res = PQexec(entry->conn, "DEALLOCATE ALL"); PQclear(res); } entry->have_prep_stmt = false; entry->have_error = false; } break; } } - /* Reset state to show we're out of a transaction */ - entry->xact_depth = 0; + /* + * If we have aborted or committed the transaction, reset state to show + * we're out of a transaction. + */ + if (event != XACT_EVENT_PRE_COMMIT) + 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.) + * If we have aborted or committed the transaction, we can now mark ourselves + * as out of the transaction. */ - xact_got_connection = false; + if (event != XACT_EVENT_PRE_COMMIT) + 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, diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out index f7e11ed..e7e9bf7 100644 --- a/contrib/postgres_fdw/expected/postgres_fdw.out +++ b/contrib/postgres_fdw/expected/postgres_fdw.out @@ -3063,10 +3063,122 @@ ERROR: type "public.Colors" does not exist LINE 4: "Col" public."Colors" OPTIONS (column_name 'Col') ^ QUERY: CREATE FOREIGN TABLE t5 ( c1 integer OPTIONS (column_name 'c1'), c2 text OPTIONS (column_name 'c2') COLLATE pg_catalog."C", "Col" public."Colors" OPTIONS (column_name 'Col') ) SERVER loopback OPTIONS (schema_name 'import_source', table_name 't5'); CONTEXT: importing foreign table "t5" ROLLBACK; +-- This will suppress the context of errors, which contains prepared transaction +-- IDs. Those come out to be different each time. +\set VERBOSITY terse +-- Test transactional consistency for multiple server case +-- create two loopback servers for testing consistency on two connections +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER loopback1 FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER loopback2 FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; +CREATE USER MAPPING FOR CURRENT_USER SERVER loopback1; +CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2; +-- create a local table to refer to as foreign table. Add a row. The table has +-- constraints which are deferred till end of transaction. This allows commit +-- time errors occur by inserting data which violates constraints. +CREATE TABLE lt(val int UNIQUE DEFERRABLE INITIALLY DEFERRED); +INSERT INTO lt VALUES (1); +INSERT INTO lt VALUES (3); +-- create two foreign tables each on separate server referring to the local table. +CREATE FOREIGN TABLE ft1_lt (val int) SERVER loopback1 OPTIONS (table_name 'lt'); +CREATE FOREIGN TABLE ft2_lt (val int) SERVER loopback2 OPTIONS (table_name 'lt'); +-- In a transaction insert two rows one each to the two foreign tables. One of +-- the rows violates the constraint and other not. At the time of commit +-- constraints on one of the server will rollback transaction on that server. +BEGIN TRANSACTION; + INSERT INTO ft1_lt VALUES (1); -- Violates constraint + INSERT INTO ft2_lt VALUES (2); +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +BEGIN TRANSACTION; + INSERT INTO ft1_lt VALUES (4); + INSERT INTO ft2_lt VALUES (3); -- Violates constraint +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +SELECT * FROM lt; + val +----- + 1 + 3 +(2 rows) + +-- Transaction involving local changes and remote changes, one of them or both +-- violating the constraints +BEGIN TRANSACTION; + INSERT INTO lt VALUES (1); -- violates constraints + INSERT INTO ft1_lt VALUES (2); +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +SELECT * FROM lt; + val +----- + 1 + 3 +(2 rows) + +BEGIN TRANSACTION; + INSERT INTO lt VALUES (2); + INSERT INTO ft1_lt VALUES (1); -- violates constraints +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +SELECT * FROM lt; + val +----- + 1 + 3 +(2 rows) + +BEGIN TRANSACTION; + INSERT INTO lt VALUES (1); -- violates constraints + INSERT INTO ft1_lt VALUES (3); -- violates constraints +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +SELECT * FROM lt; + val +----- + 1 + 3 +(2 rows) + +-- Next test shows local transaction fails if "any" of the remote transactions +-- fail to commit. But any COMMITted transaction on the remote servers remains +-- COMMITTED. +BEGIN TRANSACTION; + INSERT INTO lt VALUES (2); + INSERT INTO ft1_lt VALUES (4); + INSERT INTO ft2_lt VALUES (1); -- violates constraints +COMMIT TRANSACTION; +ERROR: duplicate key value violates unique constraint "lt_val_key" +SELECT * FROM lt; + val +----- + 1 + 3 +(2 rows) + +DROP SERVER loopback1 CASCADE; +NOTICE: drop cascades to 2 other objects +DROP SERVER loopback2 CASCADE; +NOTICE: drop cascades to 2 other objects +DROP TABLE lt; +\set VERBOSITY default diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql index ae96684..f15d302 100644 --- a/contrib/postgres_fdw/sql/postgres_fdw.sql +++ b/contrib/postgres_fdw/sql/postgres_fdw.sql @@ -665,10 +665,95 @@ IMPORT FOREIGN SCHEMA nonesuch FROM SERVER nowhere INTO notthere; -- We can fake this by dropping the type locally in our transaction. CREATE TYPE "Colors" AS ENUM ('red', 'green', 'blue'); CREATE TABLE import_source.t5 (c1 int, c2 text collate "C", "Col" "Colors"); CREATE SCHEMA import_dest5; BEGIN; DROP TYPE "Colors" CASCADE; IMPORT FOREIGN SCHEMA import_source LIMIT TO (t5) FROM SERVER loopback INTO import_dest5; -- ERROR ROLLBACK; + +-- This will suppress the context of errors, which contains prepared transaction +-- IDs. Those come out to be different each time. +\set VERBOSITY terse +-- Test transactional consistency for multiple server case +-- create two loopback servers for testing consistency on two connections +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER loopback1 FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; + +DO $d$ + BEGIN + EXECUTE $$CREATE SERVER loopback2 FOREIGN DATA WRAPPER postgres_fdw + OPTIONS (dbname '$$||current_database()||$$', + port '$$||current_setting('port')||$$' + )$$; + END; +$d$; + +CREATE USER MAPPING FOR CURRENT_USER SERVER loopback1; +CREATE USER MAPPING FOR CURRENT_USER SERVER loopback2; + +-- create a local table to refer to as foreign table. Add a row. The table has +-- constraints which are deferred till end of transaction. This allows commit +-- time errors occur by inserting data which violates constraints. +CREATE TABLE lt(val int UNIQUE DEFERRABLE INITIALLY DEFERRED); +INSERT INTO lt VALUES (1); +INSERT INTO lt VALUES (3); + +-- create two foreign tables each on separate server referring to the local table. +CREATE FOREIGN TABLE ft1_lt (val int) SERVER loopback1 OPTIONS (table_name 'lt'); +CREATE FOREIGN TABLE ft2_lt (val int) SERVER loopback2 OPTIONS (table_name 'lt'); + +-- In a transaction insert two rows one each to the two foreign tables. One of +-- the rows violates the constraint and other not. At the time of commit +-- constraints on one of the server will rollback transaction on that server. +BEGIN TRANSACTION; + INSERT INTO ft1_lt VALUES (1); -- Violates constraint + INSERT INTO ft2_lt VALUES (2); +COMMIT TRANSACTION; + +BEGIN TRANSACTION; + INSERT INTO ft1_lt VALUES (4); + INSERT INTO ft2_lt VALUES (3); -- Violates constraint +COMMIT TRANSACTION; +SELECT * FROM lt; + +-- Transaction involving local changes and remote changes, one of them or both +-- violating the constraints +BEGIN TRANSACTION; + INSERT INTO lt VALUES (1); -- violates constraints + INSERT INTO ft1_lt VALUES (2); +COMMIT TRANSACTION; +SELECT * FROM lt; + +BEGIN TRANSACTION; + INSERT INTO lt VALUES (2); + INSERT INTO ft1_lt VALUES (1); -- violates constraints +COMMIT TRANSACTION; +SELECT * FROM lt; + +BEGIN TRANSACTION; + INSERT INTO lt VALUES (1); -- violates constraints + INSERT INTO ft1_lt VALUES (3); -- violates constraints +COMMIT TRANSACTION; +SELECT * FROM lt; +-- Next test shows local transaction fails if "any" of the remote transactions +-- fail to commit. But any COMMITted transaction on the remote servers remains +-- COMMITTED. +BEGIN TRANSACTION; + INSERT INTO lt VALUES (2); + INSERT INTO ft1_lt VALUES (4); + INSERT INTO ft2_lt VALUES (1); -- violates constraints +COMMIT TRANSACTION; +SELECT * FROM lt; + +DROP SERVER loopback1 CASCADE; +DROP SERVER loopback2 CASCADE; +DROP TABLE lt; +\set VERBOSITY default