public inbox for [email protected]
help / color / mirror / Atom feed[PATCH 03/10] Make sure published XIDs are persistent
4+ messages / 3 participants
[nested] [flat]
* [PATCH 03/10] Make sure published XIDs are persistent
@ 2021-03-08 06:43 Kyotaro Horiguchi <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Kyotaro Horiguchi @ 2021-03-08 06:43 UTC (permalink / raw)
pg_xact_status() premises that XIDs obtained by
pg_current_xact_id(_if_assigned)() are persistent beyond a crash. But
XIDs are not guaranteed to go beyond WAL buffers before commit and
thus XIDs may vanish if server crashes before commit. This patch
guarantees the XID shown by the functions to be flushed out to disk.
Copied from: https://www.postgresql.org/message-id/20210308.173242.463790587797836129.horikyota.ntt%40gmail.com
---
src/backend/access/transam/xact.c | 55 +++++++++++++++++++++++++------
src/backend/access/transam/xlog.c | 2 +-
src/backend/utils/adt/xid8funcs.c | 12 ++++++-
src/include/access/xact.h | 3 +-
4 files changed, 59 insertions(+), 13 deletions(-)
diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 6395a9b240..38e978d238 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -201,7 +201,7 @@ typedef struct TransactionStateData
int prevSecContext; /* previous SecurityRestrictionContext */
bool prevXactReadOnly; /* entry-time xact r/o state */
bool startedInRecovery; /* did we start in recovery? */
- bool didLogXid; /* has xid been included in WAL record? */
+ XLogRecPtr minLSN; /* LSN needed to reach to record the xid */
int parallelModeLevel; /* Enter/ExitParallelMode counter */
bool chain; /* start a new block after this one */
bool assigned; /* assigned to top-level XID */
@@ -520,14 +520,46 @@ GetCurrentFullTransactionIdIfAny(void)
* MarkCurrentTransactionIdLoggedIfAny
*
* Remember that the current xid - if it is assigned - now has been wal logged.
+ *
+ * upto is the LSN up to which we need to flush WAL to ensure the current xid
+ * to be persistent. See EnsureCurrentTransactionIdLogged().
*/
void
-MarkCurrentTransactionIdLoggedIfAny(void)
+MarkCurrentTransactionIdLoggedIfAny(XLogRecPtr upto)
{
- if (FullTransactionIdIsValid(CurrentTransactionState->fullTransactionId))
- CurrentTransactionState->didLogXid = true;
+ if (FullTransactionIdIsValid(CurrentTransactionState->fullTransactionId) &&
+ XLogRecPtrIsInvalid(CurrentTransactionState->minLSN))
+ CurrentTransactionState->minLSN = upto;
}
+/*
+ * EnsureCurrentTransactionIdLogged
+ *
+ * Make sure that the current top XID is WAL-logged.
+ */
+void
+EnsureTopTransactionIdLogged(void)
+{
+ /*
+ * We need at least one WAL record for the current top transaction to be
+ * flushed out. Write one if we don't have one yet.
+ */
+ if (XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
+ {
+ xl_xact_assignment xlrec;
+
+ xlrec.xtop = XidFromFullTransactionId(XactTopFullTransactionId);
+ Assert(TransactionIdIsValid(xlrec.xtop));
+ xlrec.nsubxacts = 0;
+
+ XLogBeginInsert();
+ XLogRegisterData((char *) &xlrec, MinSizeOfXactAssignment);
+ TopTransactionStateData.minLSN =
+ XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
+ }
+
+ XLogFlush(TopTransactionStateData.minLSN);
+}
/*
* GetStableLatestTransactionId
@@ -616,14 +648,14 @@ AssignTransactionId(TransactionState s)
* When wal_level=logical, guarantee that a subtransaction's xid can only
* be seen in the WAL stream if its toplevel xid has been logged before.
* If necessary we log an xact_assignment record with fewer than
- * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set
+ * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if minLSN isn't set
* for a transaction even though it appears in a WAL record, we just might
* superfluously log something. That can happen when an xid is included
* somewhere inside a wal record, but not in XLogRecord->xl_xid, like in
* xl_standby_locks.
*/
if (isSubXact && XLogLogicalInfoActive() &&
- !TopTransactionStateData.didLogXid)
+ XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
log_unknown_top = true;
/*
@@ -693,6 +725,7 @@ AssignTransactionId(TransactionState s)
log_unknown_top)
{
xl_xact_assignment xlrec;
+ XLogRecPtr endptr;
/*
* xtop is always set by now because we recurse up transaction
@@ -707,11 +740,13 @@ AssignTransactionId(TransactionState s)
XLogRegisterData((char *) unreportedXids,
nUnreportedXids * sizeof(TransactionId));
- (void) XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
+ endptr = XLogInsert(RM_XACT_ID, XLOG_XACT_ASSIGNMENT);
nUnreportedXids = 0;
- /* mark top, not current xact as having been logged */
- TopTransactionStateData.didLogXid = true;
+
+ /* set minLSN of top, not of current xact if not yet */
+ if (XLogRecPtrIsInvalid(TopTransactionStateData.minLSN))
+ TopTransactionStateData.minLSN = endptr;
}
}
}
@@ -2022,7 +2057,7 @@ StartTransaction(void)
* initialize reported xid accounting
*/
nUnreportedXids = 0;
- s->didLogXid = false;
+ s->minLSN = InvalidXLogRecPtr;
/*
* must initialize resource-management stuff first
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 15da91a8dd..6eb46ea8a7 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -1172,7 +1172,7 @@ XLogInsertRecord(XLogRecData *rdata,
*/
WALInsertLockRelease();
- MarkCurrentTransactionIdLoggedIfAny();
+ MarkCurrentTransactionIdLoggedIfAny(EndPos);
END_CRIT_SECTION();
diff --git a/src/backend/utils/adt/xid8funcs.c b/src/backend/utils/adt/xid8funcs.c
index cc2b4ac797..992482f8c8 100644
--- a/src/backend/utils/adt/xid8funcs.c
+++ b/src/backend/utils/adt/xid8funcs.c
@@ -357,6 +357,8 @@ bad_format:
Datum
pg_current_xact_id(PG_FUNCTION_ARGS)
{
+ FullTransactionId xid;
+
/*
* Must prevent during recovery because if an xid is not assigned we try
* to assign one, which would fail. Programs already rely on this function
@@ -365,7 +367,12 @@ pg_current_xact_id(PG_FUNCTION_ARGS)
*/
PreventCommandDuringRecovery("pg_current_xact_id()");
- PG_RETURN_FULLTRANSACTIONID(GetTopFullTransactionId());
+ xid = GetTopFullTransactionId();
+
+ /* the XID is going to be published, make sure it is psersistent */
+ EnsureTopTransactionIdLogged();
+
+ PG_RETURN_FULLTRANSACTIONID(xid);
}
/*
@@ -380,6 +387,9 @@ pg_current_xact_id_if_assigned(PG_FUNCTION_ARGS)
if (!FullTransactionIdIsValid(topfxid))
PG_RETURN_NULL();
+ /* the XID is going to be published, make sure it is psersistent */
+ EnsureTopTransactionIdLogged();
+
PG_RETURN_FULLTRANSACTIONID(topfxid);
}
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
index 34cfaf542c..a61e4d6da5 100644
--- a/src/include/access/xact.h
+++ b/src/include/access/xact.h
@@ -386,7 +386,8 @@ extern FullTransactionId GetTopFullTransactionId(void);
extern FullTransactionId GetTopFullTransactionIdIfAny(void);
extern FullTransactionId GetCurrentFullTransactionId(void);
extern FullTransactionId GetCurrentFullTransactionIdIfAny(void);
-extern void MarkCurrentTransactionIdLoggedIfAny(void);
+extern void MarkCurrentTransactionIdLoggedIfAny(XLogRecPtr upto);
+extern void EnsureTopTransactionIdLogged(void);
extern bool SubTransactionIsActive(SubTransactionId subxid);
extern CommandId GetCurrentCommandId(bool used);
extern void SetParallelStartTimestamps(TimestampTz xact_ts, TimestampTz stmt_ts);
--
2.17.0
--jozmn01XJZjDjM3N
Content-Type: text/x-diff; charset=us-ascii
Content-Disposition: attachment;
filename="0004-wal_compression_method-default-to-zlib.patch"
^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit
@ 2022-01-13 02:54 Fujii Masao <[email protected]>
0 siblings, 1 reply; 4+ messages in thread
From: Fujii Masao @ 2022-01-13 02:54 UTC (permalink / raw)
To: Etsuro Fujita <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
On 2022/01/06 17:29, Etsuro Fujita wrote:
> On Fri, Dec 3, 2021 at 6:07 PM Fujii Masao <[email protected]> wrote:
>> On 2021/11/16 18:55, Etsuro Fujita wrote:
>>> I changed my mind; I’ll update the patch to ignore the error as
>>> before, because 1) as far as I know, there are no reports from the
>>> field concerning that we ignore all kinds of errors in cleaning up the
>>> prepared statements, so maybe we don’t need to change that, and 2) we
>>> already committed at least one of the remote transactions, so it’s not
>>> good to abort the local transaction unless we really have to.
>
> Done.
>
>> Are you planning to update the patch? In addition to this change,
>> at least documentation about new parallel_commit parameter needs
>> to be included in the patch.
>
> Done. Attached is a new version.
>
> * 0001
> This is an updated version of the previous patch. In addition to the
> above, I expanded a comment in do_sql_command_end() a bit to explain
> why we do PQconsumeInput() before doing pgfdw_get_result(), to address
> your comment. Also, I moved the code to finish closing pending
> (sub)transactions in pgfdw_xact_callback()(pgfdw_subxact_callback())
> into separate functions. Also, I modified regression test cases a bit
> to access multiple foreign servers.
Thanks for updating the patch!
At first I'm reading the 0001 patch. Here are the comments for the patch.
0001 patch failed to be applied. Could you rebase the patch?
+ entry->changing_xact_state = true;
+ do_sql_command_begin(entry->conn, "DEALLOCATE ALL");
+ pending_deallocs = lappend(pending_deallocs, entry);
Originally entry->changing_xact_state is not set to true when executing DEALLOCATE ALL. But the patch do that. Why do we need this change?
The source comment explains that we intentionally ignore errors in the DEALLOCATE. But the patch changes DEALLOCATE ALL so that it's executed via do_sql_command_begin() that can cause an error. Is this OK?
+ if (ignore_errors)
+ pgfdw_report_error(WARNING, res, conn, true, sql);
When DEALLOCATE fails, originally even warning message is not logged. But the patch changes DEALLOCATE so that its result is received via do_sql_command_end() that can log warning message even when ignore_errors argument is enabled. Why do we need to change the behavior?
+ <para>
+ This option controls whether <filename>postgres_fdw</filename> commits
+ multiple remote (sub)transactions opened in a local (sub)transaction
+ in parallel when the local (sub)transaction commits.
Since parallel_commit is an option for foreign server, how the server with this option enabled is handled by postgres_fdw should be documented, instead?
+ <para>
+ Note that if many remote (sub)transactions are opened on a remote server
+ in a local (sub)transaction, this option might increase the remote
+ server’s load significantly when those remote (sub)transactions are
+ committed. So be careful when using this option.
+ </para>
This paragraph should be inside the listitem for parallel_commit, shouldn't it?
async_capable=true also may cause the similar issue? If so, this kind of note should be documented also in async_capable?
This explains that the remote server's load will be increased *significantly*. But "significantly" part is really true? I'd like to know how much parallel_commit=true actually can increase the load in a remote server.
Regards,
--
Fujii Masao
Advanced Computing Technology Center
Research and Development Headquarters
NTT DATA CORPORATION
^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit
@ 2022-02-07 05:35 Etsuro Fujita <[email protected]>
parent: Fujii Masao <[email protected]>
0 siblings, 1 reply; 4+ messages in thread
From: Etsuro Fujita @ 2022-02-07 05:35 UTC (permalink / raw)
To: Fujii Masao <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
On Thu, Jan 13, 2022 at 11:54 AM Fujii Masao
<[email protected]> wrote:
> At first I'm reading the 0001 patch. Here are the comments for the patch.
Thanks for reviewing!
> 0001 patch failed to be applied. Could you rebase the patch?
Done. Attached is an updated version of the patch set.
> + entry->changing_xact_state = true;
> + do_sql_command_begin(entry->conn, "DEALLOCATE ALL");
> + pending_deallocs = lappend(pending_deallocs, entry);
>
> Originally entry->changing_xact_state is not set to true when executing DEALLOCATE ALL. But the patch do that. Why do we need this change?
>
> The source comment explains that we intentionally ignore errors in the DEALLOCATE. But the patch changes DEALLOCATE ALL so that it's executed via do_sql_command_begin() that can cause an error. Is this OK?
>
> + if (ignore_errors)
> + pgfdw_report_error(WARNING, res, conn, true, sql);
>
> When DEALLOCATE fails, originally even warning message is not logged. But the patch changes DEALLOCATE so that its result is received via do_sql_command_end() that can log warning message even when ignore_errors argument is enabled. Why do we need to change the behavior?
Yeah, we don’t need to change the behavior as discussed before, so I
fixed these. I worked on the patch after a while, so I forgot about
that. :-(
> + <para>
> + This option controls whether <filename>postgres_fdw</filename> commits
> + multiple remote (sub)transactions opened in a local (sub)transaction
> + in parallel when the local (sub)transaction commits.
>
> Since parallel_commit is an option for foreign server, how the server with this option enabled is handled by postgres_fdw should be documented, instead?
Agreed. I rewrote this slightly like the attached. Does that make sense?
> + <para>
> + Note that if many remote (sub)transactions are opened on a remote server
> + in a local (sub)transaction, this option might increase the remote
> + server’s load significantly when those remote (sub)transactions are
> + committed. So be careful when using this option.
> + </para>
>
> This paragraph should be inside the listitem for parallel_commit, shouldn't it?
I put this note outside, because it’s rewritten to a note about both
the parallel_commit and parallel_abort options in the following patch.
But it would be good to make the parallel-commit patch independent, so
I moved it into the list.
> async_capable=true also may cause the similar issue? If so, this kind of note should be documented also in async_capable?
That’s right. I think it would be good to add a similar note about
that, but I’d like to leave that for another patch.
> This explains that the remote server's load will be increased *significantly*. But "significantly" part is really true?
I think that that would depend on how many transactions are committed
on the remote side at the same time. But the word “significantly”
might be too strong, so I dropped the word.
> I'd like to know how much parallel_commit=true actually can increase the load in a remote server.
Ok, I’ll do a load test.
About the #0002 patch:
This is in preparation for the parallel-abort patch (#0003), but I’d
like to propose a minor cleanup for commit 85c696112: 1) build an
abort command to be sent to the remote in pgfdw_abort_cleanup(), using
a macro, only when/if necessary, as before, and 2) add/modify comments
a little bit.
Sorry for the delay again.
Best regards,
Etsuro Fujita
Attachments:
[application/octet-stream] v3-0001-postgres-fdw-Add-support-for-parallel-commit.patch (18.1K, ../../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/2-v3-0001-postgres-fdw-Add-support-for-parallel-commit.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 29fcb6a76e..d59f91f14c 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -58,6 +58,7 @@ typedef struct ConnCacheEntry
bool have_prep_stmt; /* have we prepared any stmts in this xact? */
bool have_error; /* have any subxacts aborted in this xact? */
bool changing_xact_state; /* xact state change in process */
+ bool parallel_commit; /* do we commit (sub)xacts in parallel? */
bool invalidated; /* true if reconnect is pending */
bool keep_connections; /* setting value of keep_connections
* server option */
@@ -92,6 +93,8 @@ static PGconn *connect_pg_server(ForeignServer *server, UserMapping *user);
static void disconnect_pg_server(ConnCacheEntry *entry);
static void check_conn_params(const char **keywords, const char **values, UserMapping *user);
static void configure_remote_session(PGconn *conn);
+static void do_sql_command_begin(PGconn *conn, const char *sql);
+static void do_sql_command_end(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,
@@ -100,6 +103,7 @@ static void pgfdw_subxact_callback(SubXactEvent event,
void *arg);
static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue);
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry);
+static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel);
static bool pgfdw_cancel_query(PGconn *conn);
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
bool ignore_errors);
@@ -107,6 +111,8 @@ static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
PGresult **result, bool *timed_out);
static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql,
bool toplevel);
+static void pgfdw_finish_pre_commit_cleanup(List *pending_entries);
+static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
@@ -316,14 +322,20 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
* is changed will be closed and re-made later.
*
* By default, all the connections to any foreign servers are kept open.
+ *
+ * Also determine to commit (sub)transactions opened on the remote server
+ * in parallel at (sub)transaction end.
*/
entry->keep_connections = true;
+ entry->parallel_commit = false;
foreach(lc, server->options)
{
DefElem *def = (DefElem *) lfirst(lc);
if (strcmp(def->defname, "keep_connections") == 0)
entry->keep_connections = defGetBoolean(def);
+ if (strcmp(def->defname, "parallel_commit") == 0)
+ entry->parallel_commit = defGetBoolean(def);
}
/* Now try to make the connection */
@@ -633,6 +645,32 @@ do_sql_command(PGconn *conn, const char *sql)
PQclear(res);
}
+static void
+do_sql_command_begin(PGconn *conn, const char *sql)
+{
+ if (!PQsendQuery(conn, sql))
+ pgfdw_report_error(ERROR, NULL, conn, false, sql);
+}
+
+static void
+do_sql_command_end(PGconn *conn, const char *sql)
+{
+ PGresult *res;
+
+ /*
+ * Consume whatever data is available from the socket (Note that if all
+ * data is available, this allows us to call PQgetResult without forcing
+ * the overhead of WaitLatchOrSocket in pgfdw_get_result, which would be
+ * very large compared to the overhead of PQconsumeInput.)
+ */
+ if (!PQconsumeInput(conn))
+ pgfdw_report_error(ERROR, NULL, conn, false, sql);
+ res = pgfdw_get_result(conn, sql);
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ pgfdw_report_error(ERROR, res, conn, true, sql);
+ PQclear(res);
+}
+
/*
* Start remote transaction or subtransaction, if needed.
*
@@ -888,6 +926,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
{
HASH_SEQ_STATUS scan;
ConnCacheEntry *entry;
+ List *pending_entries = NIL;
/* Quick exit if no connections were touched in this transaction. */
if (!xact_got_connection)
@@ -925,6 +964,12 @@ pgfdw_xact_callback(XactEvent event, void *arg)
/* Commit all remote transactions during pre-commit */
entry->changing_xact_state = true;
+ if (entry->parallel_commit)
+ {
+ do_sql_command_begin(entry->conn, "COMMIT TRANSACTION");
+ pending_entries = lappend(pending_entries, entry);
+ continue;
+ }
do_sql_command(entry->conn, "COMMIT TRANSACTION");
entry->changing_xact_state = false;
@@ -981,23 +1026,15 @@ pgfdw_xact_callback(XactEvent event, void *arg)
}
/* Reset state to show we're out of a transaction */
- entry->xact_depth = 0;
+ pgfdw_reset_xact_state(entry, true);
+ }
- /*
- * If the connection isn't in a good idle state, it is marked as
- * invalid or keep_connections option of its server is disabled, then
- * discard it to recover. Next GetConnection will open a new
- * connection.
- */
- if (PQstatus(entry->conn) != CONNECTION_OK ||
- PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
- entry->changing_xact_state ||
- entry->invalidated ||
- !entry->keep_connections)
- {
- elog(DEBUG3, "discarding connection %p", entry->conn);
- disconnect_pg_server(entry);
- }
+ /* If there are any pending remote transactions, finish closing them */
+ if (pending_entries)
+ {
+ Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
+ event == XACT_EVENT_PRE_COMMIT);
+ pgfdw_finish_pre_commit_cleanup(pending_entries);
}
/*
@@ -1021,6 +1058,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
HASH_SEQ_STATUS scan;
ConnCacheEntry *entry;
int curlevel;
+ List *pending_entries = NIL;
/* Nothing to do at subxact start, nor after commit. */
if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
@@ -1063,6 +1101,12 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
/* Commit all remote subtransactions during pre-commit */
snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
entry->changing_xact_state = true;
+ if (entry->parallel_commit)
+ {
+ do_sql_command_begin(entry->conn, sql);
+ pending_entries = lappend(pending_entries, entry);
+ continue;
+ }
do_sql_command(entry->conn, sql);
entry->changing_xact_state = false;
}
@@ -1076,7 +1120,14 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
}
/* OK, we're outta that level of subtransaction */
- entry->xact_depth--;
+ pgfdw_reset_xact_state(entry, false);
+ }
+
+ /* If there are any pending remote subtransactions, finish closing them */
+ if (pending_entries)
+ {
+ Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB);
+ pgfdw_finish_pre_subcommit_cleanup(pending_entries);
}
}
@@ -1169,6 +1220,40 @@ pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry)
server->servername)));
}
+/*
+ * Reset state to show we're out of a (sub)transaction.
+ */
+static void
+pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel)
+{
+ if (toplevel)
+ {
+ /* Reset state to show we're out of a transaction */
+ entry->xact_depth = 0;
+
+ /*
+ * If the connection isn't in a good idle state, it is marked as
+ * invalid or keep_connections option of its server is disabled, then
+ * discard it to recover. Next GetConnection will open a new
+ * connection.
+ */
+ if (PQstatus(entry->conn) != CONNECTION_OK ||
+ PQtransactionStatus(entry->conn) != PQTRANS_IDLE ||
+ entry->changing_xact_state ||
+ entry->invalidated ||
+ !entry->keep_connections)
+ {
+ elog(DEBUG3, "discarding connection %p", entry->conn);
+ disconnect_pg_server(entry);
+ }
+ }
+ else
+ {
+ /* Reset state to show we're out of a subtransaction */
+ entry->xact_depth--;
+ }
+}
+
/*
* Cancel the currently-in-progress query (whose query text we do not have)
* and ignore the result. Returns true if we successfully cancel the query
@@ -1456,6 +1541,91 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel)
entry->changing_xact_state = false;
}
+static void
+pgfdw_finish_pre_commit_cleanup(List *pending_entries)
+{
+ ConnCacheEntry *entry;
+ List *pending_deallocs = NIL;
+ ListCell *lc;
+
+ Assert(pending_entries);
+
+ foreach(lc, pending_entries)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+
+ Assert(entry->changing_xact_state);
+ do_sql_command_end(entry->conn, "COMMIT TRANSACTION");
+ entry->changing_xact_state = false;
+
+ /* Do a DEALLOCATE ALL if needed */
+ if (entry->have_prep_stmt && entry->have_error)
+ {
+ /* Ignore errors in the DEALLOCATE (see note above) */
+ if (PQsendQuery(entry->conn, "DEALLOCATE ALL"))
+ {
+ pending_deallocs = lappend(pending_deallocs, entry);
+ continue;
+ }
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ pgfdw_reset_xact_state(entry, true);
+ }
+
+ if (pending_deallocs)
+ {
+ foreach(lc, pending_deallocs)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+ PGresult *res;
+
+ /* Ignore errors in the DEALLOCATE (see note above) */
+ if ((res = PQgetResult(entry->conn)) != NULL)
+ {
+ PQclear(res);
+ /*
+ * Stop if the connection is lost (else we'll loop infinitely)
+ */
+ if (PQstatus(entry->conn) == CONNECTION_BAD)
+ break;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ pgfdw_reset_xact_state(entry, true);
+ }
+ }
+}
+
+static void
+pgfdw_finish_pre_subcommit_cleanup(List *pending_entries)
+{
+ ConnCacheEntry *entry;
+ int curlevel;
+ char sql[100];
+ ListCell *lc;
+
+ Assert(pending_entries);
+
+ curlevel = GetCurrentTransactionNestLevel();
+ snprintf(sql, sizeof(sql), "RELEASE SAVEPOINT s%d", curlevel);
+
+ foreach(lc, pending_entries)
+ {
+ entry = (ConnCacheEntry *) lfirst(lc);
+
+ Assert(entry->changing_xact_state);
+ do_sql_command_end(entry->conn, sql);
+ entry->changing_xact_state = false;
+
+ pgfdw_reset_xact_state(entry, false);
+ }
+}
+
/*
* List active foreign server connections.
*
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index b2e02caefe..8043b207c5 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9509,7 +9509,7 @@ DO $d$
END;
$d$;
ERROR: invalid option "password"
-HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, keep_connections
+HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, keep_connections
CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')"
PL/pgSQL function inline_code_block line 3 at EXECUTE
-- If we add a password for our user mapping instead, we should get a different
@@ -10913,3 +10913,79 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
--Clean up
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
+-- ===================================================================
+-- test parallel commit
+-- ===================================================================
+ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+CREATE TABLE ploc1 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
+ SERVER loopback OPTIONS (table_name 'ploc1');
+CREATE TABLE ploc2 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem2 (f1 int, f2 text)
+ SERVER loopback2 OPTIONS (table_name 'ploc2');
+BEGIN;
+INSERT INTO prem1 VALUES (101, 'foo');
+INSERT INTO prem2 VALUES (201, 'bar');
+COMMIT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+-----
+ 101 | foo
+(1 row)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+-----
+ 201 | bar
+(1 row)
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (102, 'foofoo');
+INSERT INTO prem2 VALUES (202, 'barbar');
+RELEASE SAVEPOINT s;
+COMMIT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+(2 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+(2 rows)
+
+-- This tests executing DEALLOCATE ALL against foreign servers in parallel
+-- during pre-commit
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (103, 'baz');
+INSERT INTO prem2 VALUES (203, 'qux');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (104, 'bazbaz');
+INSERT INTO prem2 VALUES (204, 'quxqux');
+COMMIT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+ 104 | bazbaz
+(3 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+ 204 | quxqux
+(3 rows)
+
+ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index fc3ce6a53a..a09c0b6db7 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -121,6 +121,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
strcmp(def->defname, "updatable") == 0 ||
strcmp(def->defname, "truncatable") == 0 ||
strcmp(def->defname, "async_capable") == 0 ||
+ strcmp(def->defname, "parallel_commit") == 0 ||
strcmp(def->defname, "keep_connections") == 0)
{
/* these accept only boolean values */
@@ -249,6 +250,7 @@ InitPgFdwOptions(void)
/* async_capable is available on both server and table */
{"async_capable", ForeignServerRelationId, false},
{"async_capable", ForeignTableRelationId, false},
+ {"parallel_commit", ForeignServerRelationId, false},
{"keep_connections", ForeignServerRelationId, false},
{"password_required", UserMappingRelationId, false},
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index e050639b57..2dc6386b40 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -3504,3 +3504,49 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
--Clean up
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
+
+-- ===================================================================
+-- test parallel commit
+-- ===================================================================
+ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+
+CREATE TABLE ploc1 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
+ SERVER loopback OPTIONS (table_name 'ploc1');
+CREATE TABLE ploc2 (f1 int, f2 text);
+CREATE FOREIGN TABLE prem2 (f1 int, f2 text)
+ SERVER loopback2 OPTIONS (table_name 'ploc2');
+
+BEGIN;
+INSERT INTO prem1 VALUES (101, 'foo');
+INSERT INTO prem2 VALUES (201, 'bar');
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (102, 'foofoo');
+INSERT INTO prem2 VALUES (202, 'barbar');
+RELEASE SAVEPOINT s;
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+-- This tests executing DEALLOCATE ALL against foreign servers in parallel
+-- during pre-commit
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (103, 'baz');
+INSERT INTO prem2 VALUES (203, 'qux');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (104, 'bazbaz');
+INSERT INTO prem2 VALUES (204, 'quxqux');
+COMMIT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index 2bb31f1125..7886436aae 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -455,6 +455,50 @@ OPTIONS (ADD password_required 'false');
</variablelist>
</sect3>
+ <sect3>
+ <title>Transaction Management Options</title>
+
+ <para>
+ When multiple remote (sub)transactions are involved in a local
+ (sub)transaction, by default <filename>postgres_fdw</filename> commits
+ those remote (sub)transactions one by one when the local (sub)transaction
+ commits.
+ Performance can be improved with the following option:
+ </para>
+
+ <variablelist>
+
+ <varlistentry>
+ <term><literal>parallel_commit</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ This option controls whether <filename>postgres_fdw</filename> commits
+ remote (sub)transactions opened on a foreign server in a local
+ (sub)transaction in parallel when the local (sub)transaction commits.
+ This option can only be specified for foreign servers, not per-table.
+ The default is <literal>false</literal>.
+ </para>
+
+ <para>
+ If multiple foreign servers with this option enabled are involved in
+ a local (sub)transaction, remote (sub)transactions opened on those
+ foreign servers in the local (sub)transaction are committed in parallel
+ across those foreign servers.
+ </para>
+
+ <para>
+ For a foreign server with this option enabled, if many remote
+ (sub)transactions are opened on it in a local (sub)transaction, this
+ option might increase the remote server’s load when the local
+ (sub)transactions commits, so be careful when using this option.
+ </para>
+ </listitem>
+ </varlistentry>
+
+ </variablelist>
+
+ </sect3>
+
<sect3>
<title>Updatability Options</title>
[application/octet-stream] v3-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch (3.2K, ../../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/3-v3-0002-postgres_fdw-Minor-cleanup-for-pgfdw_abort_cleanup.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index d59f91f14c..22e72c99c7 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -80,6 +80,18 @@ static unsigned int prep_stmt_number = 0;
/* tracks whether any work is needed in callback functions */
static bool xact_got_connection = false;
+/* macro for constructing abort command to be sent */
+#define CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel) \
+ do { \
+ if (toplevel) \
+ snprintf((sql), sizeof(sql), \
+ "ABORT TRANSACTION"); \
+ else \
+ snprintf((sql), sizeof(sql), \
+ "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d", \
+ (entry)->xact_depth, (entry)->xact_depth); \
+ } while(0)
+
/*
* SQL functions
*/
@@ -109,8 +121,7 @@ static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
bool ignore_errors);
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
PGresult **result, bool *timed_out);
-static void pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql,
- bool toplevel);
+static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel);
static void pgfdw_finish_pre_commit_cleanup(List *pending_entries);
static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries);
static bool UserMappingPasswordRequired(UserMapping *user);
@@ -1019,8 +1030,8 @@ pgfdw_xact_callback(XactEvent event, void *arg)
break;
case XACT_EVENT_PARALLEL_ABORT:
case XACT_EVENT_ABORT:
-
- pgfdw_abort_cleanup(entry, "ABORT TRANSACTION", true);
+ /* Rollback all remote transactions during abort */
+ pgfdw_abort_cleanup(entry, true);
break;
}
}
@@ -1113,10 +1124,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
else
{
/* Rollback all remote subtransactions during abort */
- snprintf(sql, sizeof(sql),
- "ROLLBACK TO SAVEPOINT s%d; RELEASE SAVEPOINT s%d",
- curlevel, curlevel);
- pgfdw_abort_cleanup(entry, sql, false);
+ pgfdw_abort_cleanup(entry, false);
}
/* OK, we're outta that level of subtransaction */
@@ -1469,10 +1477,7 @@ exit: ;
}
/*
- * Abort remote transaction.
- *
- * The statement specified in "sql" is sent to the remote server,
- * in order to rollback the remote transaction.
+ * Abort remote transaction or subtransaction.
*
* "toplevel" should be set to true if toplevel (main) transaction is
* rollbacked, false otherwise.
@@ -1480,8 +1485,10 @@ exit: ;
* Set entry->changing_xact_state to false on success, true on failure.
*/
static void
-pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel)
+pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel)
{
+ char sql[100];
+
/*
* Don't try to clean up the connection if we're already in error
* recursion trouble.
@@ -1513,8 +1520,9 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, const char *sql, bool toplevel)
!pgfdw_cancel_query(entry->conn))
return; /* Unable to cancel running query */
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
if (!pgfdw_exec_cleanup_query(entry->conn, sql, false))
- return; /* Unable to abort remote transaction */
+ return; /* Unable to abort remote (sub)transaction */
if (toplevel)
{
[application/octet-stream] v3-0003-postgres-fdw-Add-support-for-parallel-abort.patch (23.0K, ../../CAPmGK158hrd=ZfXmgkmNFHivgh18e4oE2Gz151C2Q4OBDjZ08A@mail.gmail.com/4-v3-0003-postgres-fdw-Add-support-for-parallel-abort.patch)
download | inline diff:
diff --git a/contrib/postgres_fdw/connection.c b/contrib/postgres_fdw/connection.c
index 22e72c99c7..9b041361f3 100644
--- a/contrib/postgres_fdw/connection.c
+++ b/contrib/postgres_fdw/connection.c
@@ -59,6 +59,7 @@ typedef struct ConnCacheEntry
bool have_error; /* have any subxacts aborted in this xact? */
bool changing_xact_state; /* xact state change in process */
bool parallel_commit; /* do we commit (sub)xacts in parallel? */
+ bool parallel_abort; /* do we abort (sub)xacts in parallel? */
bool invalidated; /* true if reconnect is pending */
bool keep_connections; /* setting value of keep_connections
* server option */
@@ -117,13 +118,25 @@ static void pgfdw_inval_callback(Datum arg, int cacheid, uint32 hashvalue);
static void pgfdw_reject_incomplete_xact_state_change(ConnCacheEntry *entry);
static void pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel);
static bool pgfdw_cancel_query(PGconn *conn);
+static bool pgfdw_cancel_query_begin(PGconn *conn);
+static bool pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime);
static bool pgfdw_exec_cleanup_query(PGconn *conn, const char *query,
bool ignore_errors);
+static bool pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query);
+static bool pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query,
+ bool ignore_errors,
+ TimestampTz endtime);
static bool pgfdw_get_cleanup_result(PGconn *conn, TimestampTz endtime,
PGresult **result, bool *timed_out);
static void pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel);
+static bool pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel,
+ List **pending_entries,
+ List **cancel_requested);
static void pgfdw_finish_pre_commit_cleanup(List *pending_entries);
static void pgfdw_finish_pre_subcommit_cleanup(List *pending_entries);
+static void pgfdw_finish_abort_cleanup(List *pending_entries,
+ List *cancel_requested,
+ bool toplevel);
static bool UserMappingPasswordRequired(UserMapping *user);
static bool disconnect_cached_connections(Oid serverid);
@@ -334,11 +347,12 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
*
* By default, all the connections to any foreign servers are kept open.
*
- * Also determine to commit (sub)transactions opened on the remote server
- * in parallel at (sub)transaction end.
+ * Also determine to commit/abort (sub)transactions opened on the remote
+ * server in parallel at (sub)transaction end.
*/
entry->keep_connections = true;
entry->parallel_commit = false;
+ entry->parallel_abort = false;
foreach(lc, server->options)
{
DefElem *def = (DefElem *) lfirst(lc);
@@ -347,6 +361,8 @@ make_new_connection(ConnCacheEntry *entry, UserMapping *user)
entry->keep_connections = defGetBoolean(def);
if (strcmp(def->defname, "parallel_commit") == 0)
entry->parallel_commit = defGetBoolean(def);
+ if (strcmp(def->defname, "parallel_abort") == 0)
+ entry->parallel_abort = defGetBoolean(def);
}
/* Now try to make the connection */
@@ -938,6 +954,7 @@ pgfdw_xact_callback(XactEvent event, void *arg)
HASH_SEQ_STATUS scan;
ConnCacheEntry *entry;
List *pending_entries = NIL;
+ List *cancel_requested = NIL;
/* Quick exit if no connections were touched in this transaction. */
if (!xact_got_connection)
@@ -1031,7 +1048,15 @@ pgfdw_xact_callback(XactEvent event, void *arg)
case XACT_EVENT_PARALLEL_ABORT:
case XACT_EVENT_ABORT:
/* Rollback all remote transactions during abort */
- pgfdw_abort_cleanup(entry, true);
+ if (entry->parallel_abort)
+ {
+ if (pgfdw_abort_cleanup_begin(entry, true,
+ &pending_entries,
+ &cancel_requested))
+ continue;
+ }
+ else
+ pgfdw_abort_cleanup(entry, true);
break;
}
}
@@ -1041,11 +1066,21 @@ pgfdw_xact_callback(XactEvent event, void *arg)
}
/* If there are any pending remote transactions, finish closing them */
- if (pending_entries)
+ if (pending_entries || cancel_requested)
{
- Assert(event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
- event == XACT_EVENT_PRE_COMMIT);
- pgfdw_finish_pre_commit_cleanup(pending_entries);
+ if (event == XACT_EVENT_PARALLEL_PRE_COMMIT ||
+ event == XACT_EVENT_PRE_COMMIT)
+ {
+ Assert(cancel_requested == NIL);
+ pgfdw_finish_pre_commit_cleanup(pending_entries);
+ }
+ else
+ {
+ Assert(event == XACT_EVENT_PARALLEL_ABORT ||
+ event == XACT_EVENT_ABORT);
+ pgfdw_finish_abort_cleanup(pending_entries, cancel_requested,
+ true);
+ }
}
/*
@@ -1070,6 +1105,7 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
ConnCacheEntry *entry;
int curlevel;
List *pending_entries = NIL;
+ List *cancel_requested = NIL;
/* Nothing to do at subxact start, nor after commit. */
if (!(event == SUBXACT_EVENT_PRE_COMMIT_SUB ||
@@ -1124,7 +1160,15 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
else
{
/* Rollback all remote subtransactions during abort */
- pgfdw_abort_cleanup(entry, false);
+ if (entry->parallel_abort)
+ {
+ if (pgfdw_abort_cleanup_begin(entry, false,
+ &pending_entries,
+ &cancel_requested))
+ continue;
+ }
+ else
+ pgfdw_abort_cleanup(entry, false);
}
/* OK, we're outta that level of subtransaction */
@@ -1132,10 +1176,19 @@ pgfdw_subxact_callback(SubXactEvent event, SubTransactionId mySubid,
}
/* If there are any pending remote subtransactions, finish closing them */
- if (pending_entries)
+ if (pending_entries || cancel_requested)
{
- Assert(event == SUBXACT_EVENT_PRE_COMMIT_SUB);
- pgfdw_finish_pre_subcommit_cleanup(pending_entries);
+ if (event == SUBXACT_EVENT_PRE_COMMIT_SUB)
+ {
+ Assert(cancel_requested == NIL);
+ pgfdw_finish_pre_subcommit_cleanup(pending_entries);
+ }
+ else
+ {
+ Assert(event == SUBXACT_EVENT_ABORT_SUB);
+ pgfdw_finish_abort_cleanup(pending_entries, cancel_requested,
+ false);
+ }
}
}
@@ -1279,11 +1332,7 @@ pgfdw_reset_xact_state(ConnCacheEntry *entry, bool toplevel)
static bool
pgfdw_cancel_query(PGconn *conn)
{
- PGcancel *cancel;
- char errbuf[256];
- PGresult *result = NULL;
TimestampTz endtime;
- bool timed_out;
/*
* If it takes too long to cancel the query and discard the result, assume
@@ -1291,6 +1340,17 @@ pgfdw_cancel_query(PGconn *conn)
*/
endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+ if (!pgfdw_cancel_query_begin(conn))
+ return false;
+ return pgfdw_cancel_query_end(conn, endtime);
+}
+
+static bool
+pgfdw_cancel_query_begin(PGconn *conn)
+{
+ PGcancel *cancel;
+ char errbuf[256];
+
/*
* Issue cancel request. Unfortunately, there's no good way to limit the
* amount of time that we might block inside PQgetCancel().
@@ -1309,6 +1369,15 @@ pgfdw_cancel_query(PGconn *conn)
PQfreeCancel(cancel);
}
+ return true;
+}
+
+static bool
+pgfdw_cancel_query_end(PGconn *conn, TimestampTz endtime)
+{
+ PGresult *result = NULL;
+ bool timed_out;
+
/* Get and discard the result of the query. */
if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out))
{
@@ -1343,9 +1412,7 @@ pgfdw_cancel_query(PGconn *conn)
static bool
pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
{
- PGresult *result = NULL;
TimestampTz endtime;
- bool timed_out;
/*
* If it takes too long to execute a cleanup query, assume the connection
@@ -1355,6 +1422,14 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
*/
endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+ if (!pgfdw_exec_cleanup_query_begin(conn, query))
+ return false;
+ return pgfdw_exec_cleanup_query_end(conn, query, ignore_errors, endtime);
+}
+
+static bool
+pgfdw_exec_cleanup_query_begin(PGconn *conn, const char *query)
+{
/*
* Submit a query. Since we don't use non-blocking mode, this also can
* block. But its risk is relatively small, so we ignore that for now.
@@ -1365,6 +1440,16 @@ pgfdw_exec_cleanup_query(PGconn *conn, const char *query, bool ignore_errors)
return false;
}
+ return true;
+}
+
+static bool
+pgfdw_exec_cleanup_query_end(PGconn *conn, const char *query,
+ bool ignore_errors, TimestampTz endtime)
+{
+ PGresult *result = NULL;
+ bool timed_out;
+
/* Get the result of the query. */
if (pgfdw_get_cleanup_result(conn, endtime, &result, &timed_out))
{
@@ -1549,6 +1634,56 @@ pgfdw_abort_cleanup(ConnCacheEntry *entry, bool toplevel)
entry->changing_xact_state = false;
}
+static bool
+pgfdw_abort_cleanup_begin(ConnCacheEntry *entry, bool toplevel,
+ List **pending_entries, List **cancel_requested)
+{
+ /*
+ * Don't try to clean up the connection if we're already in error
+ * recursion trouble.
+ */
+ if (in_error_recursion_trouble())
+ entry->changing_xact_state = true;
+
+ /*
+ * If connection is already unsalvageable, don't touch it further.
+ */
+ if (entry->changing_xact_state)
+ return false;
+
+ /*
+ * Mark this connection as in the process of changing transaction state.
+ */
+ entry->changing_xact_state = true;
+
+ /* Assume we might have lost track of prepared statements */
+ entry->have_error = true;
+
+ /*
+ * If a command has been submitted to the remote server by using an
+ * asynchronous execution function, the command might not have yet
+ * completed. Check to see if a command is still being processed by the
+ * remote server, and if so, request cancellation of the command.
+ */
+ if (PQtransactionStatus(entry->conn) == PQTRANS_ACTIVE)
+ {
+ if (!pgfdw_cancel_query_begin(entry->conn))
+ return false; /* Unable to cancel running query */
+ *cancel_requested = lappend(*cancel_requested, entry);
+ }
+ else
+ {
+ char sql[100];
+
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
+ if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql))
+ return false; /* Unable to abort remote transaction */
+ *pending_entries = lappend(*pending_entries, entry);
+ }
+
+ return true;
+}
+
static void
pgfdw_finish_pre_commit_cleanup(List *pending_entries)
{
@@ -1634,6 +1769,155 @@ pgfdw_finish_pre_subcommit_cleanup(List *pending_entries)
}
}
+static void
+pgfdw_finish_abort_cleanup(List *pending_entries, List *cancel_requested,
+ bool toplevel)
+{
+ List *pending_deallocs = NIL;
+ ListCell *lc;
+
+ if (cancel_requested)
+ {
+ foreach(lc, cancel_requested)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+ char sql[100];
+
+ Assert(entry->changing_xact_state);
+
+ /*
+ * Set end time. You might think we should do so before issuing
+ * cancel request like in normal mode, but that is problematic,
+ * because if, for example, it took longer than 30 seconds to
+ * process the first few entries in the cancel_requested list, it
+ * would cause a timeout for each of the remaining entries in the
+ * list when processing it, leading to slamming the connection of
+ * it shut.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(),
+ 30000);
+
+ /* Get and discard the result of the query. */
+ if (!pgfdw_cancel_query_end(entry->conn, endtime))
+ {
+ /* Unable to cancel running query */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
+ if (!pgfdw_exec_cleanup_query_begin(entry->conn, sql))
+ {
+ /* Unable to abort remote (sub)transaction */
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+ else
+ pending_entries = lappend(pending_entries, entry);
+ }
+ }
+
+ if (!pending_entries)
+ return;
+
+ foreach(lc, pending_entries)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+ char sql[100];
+
+ Assert(entry->changing_xact_state);
+
+ /*
+ * Set end time. We do this now, not before issuing the command like
+ * in normal mode, for the same reason as for the cancel_requested
+ * entries.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+
+ /* Get the result of the command. */
+ CONSTRUCT_ABORT_COMMAND(sql, entry, toplevel);
+ if (!pgfdw_exec_cleanup_query_end(entry->conn, sql, false, endtime))
+ {
+ /* Unable to abort remote (sub)transaction */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ /*
+ * If called for cleanup at main-transaction end, do a DEALLOCATE ALL
+ * if needed.
+ */
+ if (toplevel)
+ {
+ if (entry->have_prep_stmt && entry->have_error)
+ {
+ if (!pgfdw_exec_cleanup_query_begin(entry->conn,
+ "DEALLOCATE ALL"))
+ {
+ /* Trouble clearing prepared statements */
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+ else
+ pending_deallocs = lappend(pending_deallocs, entry);
+ continue;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+ }
+
+ /* Reset the per-connection state if needed */
+ if (entry->state.pendingAreq)
+ memset(&entry->state, 0, sizeof(entry->state));
+
+ /* We're done with this entry; unset the changing_xact_state flag */
+ entry->changing_xact_state = false;
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+
+ if (!pending_deallocs)
+ return;
+ Assert(toplevel);
+
+ foreach(lc, pending_deallocs)
+ {
+ ConnCacheEntry *entry = (ConnCacheEntry *) lfirst(lc);
+ TimestampTz endtime;
+
+ Assert(entry->changing_xact_state);
+ Assert(entry->have_prep_stmt);
+ Assert(entry->have_error);
+
+ /*
+ * Set end time. We do this now, not before issuing the command like
+ * in normal mode, for the same reason as for the cancel_requested
+ * entries.
+ */
+ endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 30000);
+
+ /* Get the result of the command. */
+ if (!pgfdw_exec_cleanup_query_end(entry->conn, "DEALLOCATE ALL",
+ true, endtime))
+ {
+ /* Trouble clearing prepared statements */
+ pgfdw_reset_xact_state(entry, toplevel);
+ continue;
+ }
+
+ entry->have_prep_stmt = false;
+ entry->have_error = false;
+
+ /* Reset the per-connection state if needed */
+ if (entry->state.pendingAreq)
+ memset(&entry->state, 0, sizeof(entry->state));
+
+ /* We're done with this entry; unset the changing_xact_state flag */
+ entry->changing_xact_state = false;
+ pgfdw_reset_xact_state(entry, toplevel);
+ }
+}
+
/*
* List active foreign server connections.
*
diff --git a/contrib/postgres_fdw/expected/postgres_fdw.out b/contrib/postgres_fdw/expected/postgres_fdw.out
index 8043b207c5..2adb0ac125 100644
--- a/contrib/postgres_fdw/expected/postgres_fdw.out
+++ b/contrib/postgres_fdw/expected/postgres_fdw.out
@@ -9509,7 +9509,7 @@ DO $d$
END;
$d$;
ERROR: invalid option "password"
-HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, keep_connections
+HINT: Valid options in this context are: service, passfile, channel_binding, connect_timeout, dbname, host, hostaddr, port, options, application_name, keepalives, keepalives_idle, keepalives_interval, keepalives_count, tcp_user_timeout, sslmode, sslcompression, sslcert, sslkey, sslrootcert, sslcrl, sslcrldir, sslsni, requirepeer, ssl_min_protocol_version, ssl_max_protocol_version, gssencmode, krbsrvname, gsslib, target_session_attrs, use_remote_estimate, fdw_startup_cost, fdw_tuple_cost, extensions, updatable, truncatable, fetch_size, batch_size, async_capable, parallel_commit, parallel_abort, keep_connections
CONTEXT: SQL statement "ALTER SERVER loopback_nopw OPTIONS (ADD password 'dummypw')"
PL/pgSQL function inline_code_block line 3 at EXECUTE
-- If we add a password for our user mapping instead, we should get a different
@@ -10914,10 +10914,12 @@ SELECT pg_terminate_backend(pid, 180000) FROM pg_stat_activity
RESET postgres_fdw.application_name;
RESET debug_discard_caches;
-- ===================================================================
--- test parallel commit
+-- test parallel commit and parallel abort
-- ===================================================================
ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true');
ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true');
CREATE TABLE ploc1 (f1 int, f2 text);
CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
SERVER loopback OPTIONS (table_name 'ploc1');
@@ -10987,5 +10989,52 @@ SELECT * FROM prem2;
204 | quxqux
(3 rows)
+BEGIN;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+ 104 | bazbaz
+(3 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+ 204 | quxqux
+(3 rows)
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+ f1 | f2
+-----+--------
+ 101 | foo
+ 102 | foofoo
+ 104 | bazbaz
+(3 rows)
+
+SELECT * FROM prem2;
+ f1 | f2
+-----+--------
+ 201 | bar
+ 202 | barbar
+ 204 | quxqux
+(3 rows)
+
ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback OPTIONS (DROP parallel_abort);
ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_abort);
diff --git a/contrib/postgres_fdw/option.c b/contrib/postgres_fdw/option.c
index a09c0b6db7..e8c8b4ab36 100644
--- a/contrib/postgres_fdw/option.c
+++ b/contrib/postgres_fdw/option.c
@@ -122,6 +122,7 @@ postgres_fdw_validator(PG_FUNCTION_ARGS)
strcmp(def->defname, "truncatable") == 0 ||
strcmp(def->defname, "async_capable") == 0 ||
strcmp(def->defname, "parallel_commit") == 0 ||
+ strcmp(def->defname, "parallel_abort") == 0 ||
strcmp(def->defname, "keep_connections") == 0)
{
/* these accept only boolean values */
@@ -251,6 +252,7 @@ InitPgFdwOptions(void)
{"async_capable", ForeignServerRelationId, false},
{"async_capable", ForeignTableRelationId, false},
{"parallel_commit", ForeignServerRelationId, false},
+ {"parallel_abort", ForeignServerRelationId, false},
{"keep_connections", ForeignServerRelationId, false},
{"password_required", UserMappingRelationId, false},
diff --git a/contrib/postgres_fdw/sql/postgres_fdw.sql b/contrib/postgres_fdw/sql/postgres_fdw.sql
index 2dc6386b40..58ec80fad7 100644
--- a/contrib/postgres_fdw/sql/postgres_fdw.sql
+++ b/contrib/postgres_fdw/sql/postgres_fdw.sql
@@ -3506,10 +3506,12 @@ RESET postgres_fdw.application_name;
RESET debug_discard_caches;
-- ===================================================================
--- test parallel commit
+-- test parallel commit and parallel abort
-- ===================================================================
ALTER SERVER loopback OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback OPTIONS (ADD parallel_abort 'true');
ALTER SERVER loopback2 OPTIONS (ADD parallel_commit 'true');
+ALTER SERVER loopback2 OPTIONS (ADD parallel_abort 'true');
CREATE TABLE ploc1 (f1 int, f2 text);
CREATE FOREIGN TABLE prem1 (f1 int, f2 text)
@@ -3548,5 +3550,26 @@ COMMIT;
SELECT * FROM prem1;
SELECT * FROM prem2;
+BEGIN;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
+BEGIN;
+SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ROLLBACK TO SAVEPOINT s;
+RELEASE SAVEPOINT s;
+INSERT INTO prem1 VALUES (105, 'test1');
+INSERT INTO prem2 VALUES (205, 'test2');
+ABORT;
+SELECT * FROM prem1;
+SELECT * FROM prem2;
+
ALTER SERVER loopback OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback OPTIONS (DROP parallel_abort);
ALTER SERVER loopback2 OPTIONS (DROP parallel_commit);
+ALTER SERVER loopback2 OPTIONS (DROP parallel_abort);
diff --git a/doc/src/sgml/postgres-fdw.sgml b/doc/src/sgml/postgres-fdw.sgml
index 7886436aae..19e408ec9a 100644
--- a/doc/src/sgml/postgres-fdw.sgml
+++ b/doc/src/sgml/postgres-fdw.sgml
@@ -460,10 +460,10 @@ OPTIONS (ADD password_required 'false');
<para>
When multiple remote (sub)transactions are involved in a local
- (sub)transaction, by default <filename>postgres_fdw</filename> commits
- those remote (sub)transactions one by one when the local (sub)transaction
- commits.
- Performance can be improved with the following option:
+ (sub)transaction, by default <filename>postgres_fdw</filename> commits or
+ aborts those remote (sub)transactions one by one when the local
+ (sub)transaction commits or aborts.
+ Performance can be improved with the following options:
</para>
<variablelist>
@@ -478,25 +478,38 @@ OPTIONS (ADD password_required 'false');
This option can only be specified for foreign servers, not per-table.
The default is <literal>false</literal>.
</para>
+ </listitem>
+ </varlistentry>
+ <varlistentry>
+ <term><literal>parallel_abort</literal> (<type>boolean</type>)</term>
+ <listitem>
<para>
- If multiple foreign servers with this option enabled are involved in
- a local (sub)transaction, remote (sub)transactions opened on those
- foreign servers in the local (sub)transaction are committed in parallel
- across those foreign servers.
- </para>
-
- <para>
- For a foreign server with this option enabled, if many remote
- (sub)transactions are opened on it in a local (sub)transaction, this
- option might increase the remote server’s load when the local
- (sub)transactions commits, so be careful when using this option.
+ This option controls whether <filename>postgres_fdw</filename> aborts
+ remote (sub)transactions opened on a foreign server in a local
+ (sub)transaction in parallel when the local (sub)transaction aborts.
+ This option can only be specified for foreign servers, not per-table.
+ The default is <literal>false</literal>.
</para>
</listitem>
</varlistentry>
</variablelist>
+ <para>
+ If multiple foreign servers with this option enabled are involved in a
+ local (sub)transaction, remote (sub)transactions opened on those foreign
+ servers in the local (sub)transaction are committed or aborted in parallel
+ across those foreign servers.
+ </para>
+
+ <para>
+ For a foreign server with this option enabled, if many remote
+ (sub)transactions are opened on it in a local (sub)transaction, this
+ option might increase the remote server’s load when the local
+ (sub)transactions commits or aborts, so be careful when using this option.
+ </para>
+
</sect3>
<sect3>
^ permalink raw reply [nested|flat] 4+ messages in thread
* Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit
@ 2022-02-07 18:49 Fujii Masao <[email protected]>
parent: Etsuro Fujita <[email protected]>
0 siblings, 0 replies; 4+ messages in thread
From: Fujii Masao @ 2022-02-07 18:49 UTC (permalink / raw)
To: Etsuro Fujita <[email protected]>; +Cc: PostgreSQL Hackers <[email protected]>
On 2022/02/07 14:35, Etsuro Fujita wrote:
>> 0001 patch failed to be applied. Could you rebase the patch?
>
> Done. Attached is an updated version of the patch set.
Thanks for updating the patch! Here are the review comments for 0001 patch.
I got the following compiler warning.
[16:58:07.120] connection.c: In function ‘pgfdw_finish_pre_commit_cleanup’:
[16:58:07.120] connection.c:1726:4: error: ISO C90 forbids mixed declarations and code [-Werror=declaration-after-statement]
[16:58:07.120] 1726 | PGresult *res;
[16:58:07.120] | ^~~~~~~~
+ /* Ignore errors in the DEALLOCATE (see note above) */
+ if ((res = PQgetResult(entry->conn)) != NULL)
Doesn't PQgetResult() need to be called repeatedly until it returns NULL or the connection is lost because there can be more than one messages to receive?
+ if (pending_deallocs)
+ {
+ foreach(lc, pending_deallocs)
If pending_deallocs is NIL, we don't enter this foreach loop. So probably "if (pending_deallocs)" seems not necessary.
entry->keep_connections = defGetBoolean(def);
+ if (strcmp(def->defname, "parallel_commit") == 0)
+ entry->parallel_commit = defGetBoolean(def);
Isn't it better to use "else if" here, instead?
+static void do_sql_command_begin(PGconn *conn, const char *sql);
+static void do_sql_command_end(PGconn *conn, const char *sql);
To simplify the code more, I'm tempted to change do_sql_command() so that it just calls the above two functions, instead of calling PQsendQuery() and pgfw_get_result() directly. Thought? If we do this, probably we also need to change do_sql_command_end() so that it accepts boolean flag which specifies whether PQconsumeInput() is called or not, as follows.
do_sql_command_end(PGconn *conn, const char *sql, bool consumeInput)
{
/*
* If any data is expected to be available from the socket, consume it.
* ...
* When parallel_commit is enabled, since there can be a time window between
* sending query and receiving result, we can expect data is already available
* from the socket. In this case we try to consume it at first.... Otherwise..
*/
if (consumeInput && !PQconsumeInput(conn))
...
Regards,
--
Fujii Masao
Advanced Computing Technology Center
Research and Development Headquarters
NTT DATA CORPORATION
^ permalink raw reply [nested|flat] 4+ messages in thread
end of thread, other threads:[~2022-02-07 18:49 UTC | newest]
Thread overview: 4+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2021-03-08 06:43 [PATCH 03/10] Make sure published XIDs are persistent Kyotaro Horiguchi <[email protected]>
2022-01-13 02:54 Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit Fujii Masao <[email protected]>
2022-02-07 05:35 ` Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit Etsuro Fujita <[email protected]>
2022-02-07 18:49 ` Re: postgres_fdw: commit remote (sub)transactions in parallel during pre-commit Fujii Masao <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox