Hi hackers,
Attached is a v1 patch adding optional server-side automatic retry for
single-statement implicit transactions (autocommit mode) that fail due to
transient concurrency errors such as serialization failures (40001) and
deadlocks (40P01).
1. Problem & Motivation
=======================
Under REPEATABLE READ and SERIALIZABLE isolation levels (or high-contention
READ COMMITTED workloads encountering deadlocks or DDL lock conflicts),
applications must implement retry loops around transactions that fail with
SQLSTATE 40001 (serialization_failure) or 40P01 (deadlock_detected).
While multi-statement explicit transactions (BEGIN ... COMMIT) inherently
require client-side retry logic because intermediate application state may
depend on earlier queries in the transaction, standalone autocommit
statements contain their entire specification within a single wire-protocol
request. Retrying such statements directly inside the backend avoids a full
network round-trip, connection-pooler churn, and boilerplate retry wrappers
in simple clients, ORMs, and migration runners.
2. Design & Architecture
========================
The retry loop operates in `src/backend/tcop/postgres.c` around top-level
autocommit executions when `max_autocommit_retries > 0`:
a) Protocol Coverage:
- Simple Query Protocol (`exec_simple_query`):
Eligible when the query string parses into a single statement
(`list_length(parsetree_list) == 1`) and the backend is not initially in
an explicit transaction block (`!(xact_started && IsTransactionBlock())`).
On each retry attempt, the statement is re-parsed from the raw
`query_string` in a fresh `MessageContext` after aborting the failed
transaction.
- Extended Query Protocol (`exec_execute_message`):
Eligible when executing an uncompleted portal (`!portal->atStart` is
false, `max_rows <= 0` so no partial portal fetches occur), not inside an
explicit transaction block, with a single statement in `portal->stmts`.
Before execution begins, the bound `ParamListInfo` and portal context
(`stmt_text`, `prep_stmt_name`, `commandTag`, `cplan` or raw `stmts`) are
snapshot into a dedicated child memory context (`retry_cxt`) under
`TopMemoryContext`. If a retry is triggered after
`AbortCurrentTransaction()`
drops the failed unnamed portal, the portal is cleanly reconstructed and
re-defined from either the cached plan (`GetCachedPlan`) or a fresh copy
of the parse trees before re-running `PortalStart` and `PortalRun`.
b) Statement Classification & Eligible SQLSTATEs:
- Standard retryable statements (`AUTOCOMMIT_RETRY_DML`):
`SELECT`, `INSERT`, `UPDATE`, `DELETE`, `MERGE`, `VALUES`, `TABLE`, and
`CALL` (provided the procedure does not issue internal transaction-ending
`COMMIT`/`ROLLBACK` commands, which disqualify retry as soon as the
original `TransactionId` advances/resets).
Retried on:
* `ERRCODE_T_R_SERIALIZATION_FAILURE` (`40001`)
* `ERRCODE_DEADLOCK_DETECTED` (`40P01`)
- Rollback-safe DDL (`AUTOCOMMIT_RETRY_DDL_SAFE`):
When `enable_ddl_autocommit_retry = on`, rollback-safe transactional DDL
statements (`CREATE`, `ALTER`, `DROP`, `COMMENT`, `SECURITY LABEL`,
`GRANT`/`REVOKE`, `IMPORT FOREIGN SCHEMA`, `REFRESH MATERIALIZED VIEW`
non-concurrently, `REINDEX` non-concurrently, `CLUSTER`, etc.) are also
eligible for retry. Non-transactional or non-rollback-safe commands
(`VACUUM`, `CREATE/DROP DATABASE`, `CREATE/DROP TABLESPACE`,
`ALTER SYSTEM`,
`REINDEX CONCURRENTLY`, `CREATE/DROP INDEX CONCURRENTLY`, `REFRESH
MATERIALIZED VIEW CONCURRENTLY`) and transaction control commands are
explicitly excluded (`AUTOCOMMIT_RETRY_NONE`).
In addition to `40001` and `40P01`, rollback-safe DDL is also retried on:
* `ERRCODE_LOCK_NOT_AVAILABLE` (`55P03`)
* `ERRCODE_QUERY_CANCELED` (`57014`) when caused by `lock_timeout`
(explicitly excluding user `SIGINT` cancels and `statement_timeout`).
c) Speculative Wire-Protocol Output Buffering (`pqcomm.c`):
A `SELECT` or `INSERT ... RETURNING` statement may stream `RowDescription`
and partial `DataRow` messages into `PqSendBuffer` before encountering a
serialization failure or deadlock part-way through execution.
To prevent corrupting the frontend wire stream on retry:
- `pq_begin_retry_Holdoff()` pins all outgoing protocol bytes in
`PqSendBuffer` (expanding the buffer dynamically up to an 8 MB ceiling,
`PQ_RETRY_BUFFER_MAX = 8 * 1024 * 1024`, before falling back to flushing
to the socket).
- If the statement fails with a retryable error and the buffer has not
flushed (`!pq_retry_Holdoff_flushed()`), `pq_discard_retry_Holdoff()`
rewinds `PqSendPointer` to discard the speculative `RowDescription` /
`DataRow` bytes, and `FlushErrorState()` clears the pending error before
sleeping and retrying.
- If the output exceeded 8 MB and already flushed bytes to the wire, retry
is suppressed and the error is reported normally.
d) Backoff, Jitter, and `statement_timeout`:
- Between retry attempts, the backend aborts the failed transaction so locks
and snapshots are released, then sleeps using `WaitLatch(MyLatch,
WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH, sleep_ms,
WAIT_EVENT_AUTOCOMMIT_RETRY_BACKOFF)`.
- The backoff uses exponential growth with full uniform jitter via
`pg_prng_uint64_range(&pg_global_prng_state, 0, cap_ms)`:
`cap_ms = min(autocommit_retry_backoff_max_ms,
autocommit_retry_backoff_base_ms * 2^(attempt - 1))`.
- `statement_timeout` is re-armed per attempt (`enable_statement_timeout()`).
3. New GUC Parameters
=====================
- `max_autocommit_retries` (integer, default `0`, range `0..100`,
`PGC_USERSET`):
Maximum number of retry attempts for autocommit statements failing with
transient concurrency errors (`0` disables automatic retry).
- `enable_ddl_autocommit_retry` (bool, default `false`, `PGC_USERSET`):
Enables automatic retry for rollback-safe DDL statements on `40001`, `40P01`,
`55P03` (`lock_not_available`), and `lock_timeout` (`57014`).
- `autocommit_retry_backoff_base_ms` (integer, default `5`, range
`0..10000`, `PGC_USERSET`):
Base delay in milliseconds for exponential backoff with jitter
between retries.
- `autocommit_retry_backoff_max_ms` (integer, default `1000`, range
`1..60000`, `PGC_USERSET`):
Maximum delay in milliseconds for exponential backoff between retries.
- `log_autocommit_retries` (bool, default `false`, `PGC_USERSET`):
Emits a `LOG` message for each automatic retry attempt including the attempt
number, `max_autocommit_retries`, SQLSTATE, sleep duration, and
statement text.
4. Verification & Testing
=========================
- Isolation test suite (`src/test/isolation/specs/autocommit-retry.spec`):
10 permutations covering:
* Read-write skew (`40001`) under `default_transaction_isolation =
'repeatable read'`
with `max_autocommit_retries = 0` (fails) vs `max_autocommit_retries = 3`
(transparently retries and commits updated balance).
* CTE (`WITH ... UPDATE`) two-session mutual deadlock (`40P01`) where the
deadlock victim automatically retries and succeeds.
* Rollback-safe DDL (`ALTER TABLE ... ADD COLUMN`) failing with `lock_timeout`
(`57014`) when blocked by an open transaction, then retrying after the
blocker releases the lock (`enable_ddl_autocommit_retry = on` vs `off`).
* `SELECT ... FOR SHARE` / `FOR UPDATE` conflict retry.
* Explicit `BEGIN ... COMMIT` transaction block verification (never
auto-retried).
* Multi-statement Simple Query (`SELECT 1; UPDATE ...`) verification (never
auto-retried).
* `CALL` procedure with internal `COMMIT` verification (never auto-retried
once internal transaction control occurs).
- Regression tests (`src/test/regress/sql/psql.sql`):
* Extended Query protocol (`\bind`) retry on `40001`.
* Speculative `DataRow` output buffer discard (`pq_discard_retry_Holdoff()`)
when a multi-row `SELECT` fails mid-stream with `40001` on the first
attempt and succeeds cleanly on the retry attempt without duplicate rows.
Base commit: e75fcc9867fdfe295957ad5c9a98cbe2e1b4c5c3
GitHub branch:
https://github.com/postsql/postgres/tree/auto-retry-serialization-errors
CI run: https://github.com/postsql/postgres/actions/runs/39452018256
Regards,
Hannu Krosing
From 57e9874973b29f435a03262f815f24f11706cf2b Mon Sep 17 00:00:00 2001
From: Hannu Krosing <[email protected]>
Date: Mon, 12 Oct 2026 20:19:16 +0000
Subject: [PATCH v1 1/1] Add automatic retry for autocommit statements on
transient errors
Allow standalone single-statement queries outside an explicit transaction
block (in both Simple Query and Extended Query protocols) to automatically
retry on transient concurrency errors when max_autocommit_retries > 0:
- DML, SELECT, and CALL statements retry on serialization failure (40001)
and deadlock detected (40P01), provided CALL has not executed an internal
COMMIT or ROLLBACK prior to failing.
- Rollback-safe DDL and table-level utility commands optionally retry on
lock_timeout (55P03), statement_timeout (57014), 40001, and 40P01 when
enable_ddl_autocommit_retry = on.
- Buffer uncommitted protocol output (RowDescription, DataRow, NoticeResponse)
in PqSendBuffer (up to 8 MB) and discard it if an attempt fails and is
retried, disabling retry if the buffer already flushed to the socket.
- Sleep between attempts using interruptible exponential backoff with jitter
configured by autocommit_retry_backoff_base_ms and
autocommit_retry_backoff_max_ms, with optional server logging controlled
by log_autocommit_retries.
---
src/backend/libpq/pqcomm.c | 98 ++
src/backend/tcop/postgres.c | 1025 +++++++++++++----
src/backend/utils/misc/guc_parameters.dat | 39 +
src/backend/utils/misc/postgresql.conf.sample | 5 +
src/include/libpq/libpq.h | 5 +
src/include/tcop/tcopprot.h | 6 +
.../isolation/expected/autocommit-retry.out | 258 +++++
src/test/isolation/isolation_schedule | 1 +
.../isolation/specs/autocommit-retry.spec | 157 +++
src/test/regress/expected/psql.out | 52 +
src/test/regress/expected/sysviews.out | 3 +-
src/test/regress/sql/psql.sql | 30 +
12 files changed, 1434 insertions(+), 245 deletions(-)
create mode 100644 src/test/isolation/expected/autocommit-retry.out
create mode 100644 src/test/isolation/specs/autocommit-retry.spec
diff --git a/src/backend/libpq/pqcomm.c b/src/backend/libpq/pqcomm.c
index 3704d121003..43bbd186431 100644
--- a/src/backend/libpq/pqcomm.c
+++ b/src/backend/libpq/pqcomm.c
@@ -71,6 +71,7 @@
#include "common/ip.h"
#include "libpq/libpq.h"
+#include "libpq/protocol.h"
#include "miscadmin.h"
#include "port/pg_bswap.h"
#include "postmaster/postmaster.h"
@@ -116,12 +117,18 @@ static List *sock_paths = NIL;
#define PQ_SEND_BUFFER_SIZE 8192
#define PQ_RECV_BUFFER_SIZE 8192
+#define PQ_MAX_RETRY_BUFFER_SIZE (8 * 1024 * 1024)
static char *PqSendBuffer;
static int PqSendBufferSize; /* Size send buffer */
static size_t PqSendPointer; /* Next index to store a byte in PqSendBuffer */
static size_t PqSendStart; /* Next index to send a byte in PqSendBuffer */
+/* State for buffering output during retryable autocommit statements */
+static bool PqRetryBuffering = false;
+static bool PqRetryBufferFlushed = false;
+static size_t PqSendRetryStart = 0;
+
static char PqRecvBuffer[PQ_RECV_BUFFER_SIZE];
static int PqRecvPointer; /* Next index to read a byte from PqRecvBuffer */
static int PqRecvLength; /* End of data available in PqRecvBuffer */
@@ -333,6 +340,9 @@ socket_comm_reset(void)
{
/* Do not throw away pending data, but do reset the busy flag */
PqCommBusy = false;
+ PqRetryBuffering = false;
+ PqRetryBufferFlushed = false;
+ PqSendRetryStart = 0;
}
/* --------------------------------
@@ -1278,6 +1288,34 @@ internal_putbytes(const void *b, size_t len)
while (len > 0)
{
+ /*
+ * If retry buffering is active and we haven't flushed yet, expand
+ * PqSendBuffer up to PQ_MAX_RETRY_BUFFER_SIZE instead of flushing
+ * uncommitted output to the client socket.
+ */
+ if (PqRetryBuffering && !PqRetryBufferFlushed &&
+ PqSendPointer + len > PqSendBufferSize)
+ {
+ size_t required = PqSendPointer + len;
+
+ if (required <= PQ_MAX_RETRY_BUFFER_SIZE)
+ {
+ size_t newsize = PqSendBufferSize;
+
+ while (newsize < required)
+ newsize *= 2;
+ if (newsize > PQ_MAX_RETRY_BUFFER_SIZE)
+ newsize = PQ_MAX_RETRY_BUFFER_SIZE;
+ PqSendBuffer = repalloc(PqSendBuffer, newsize);
+ PqSendBufferSize = (int) newsize;
+ }
+ else
+ {
+ /* Buffer limit exceeded; allow flush and mark un-retryable */
+ PqRetryBufferFlushed = true;
+ }
+ }
+
/* If buffer is full, then flush it out */
if (PqSendPointer >= PqSendBufferSize)
{
@@ -1329,6 +1367,14 @@ socket_flush(void)
/* No-op if reentrant call */
if (PqCommBusy)
return 0;
+
+ /*
+ * Defer explicit flushes (e.g., from send_message_to_frontend on NOTICE)
+ * while buffering uncommitted output for a retryable statement.
+ */
+ if (PqRetryBuffering && !PqRetryBufferFlushed)
+ return 0;
+
PqCommBusy = true;
socket_set_nonblocking(false);
res = internal_flush();
@@ -1364,6 +1410,9 @@ internal_flush_buffer(const char *buf, size_t *start, size_t *end)
const char *bufptr = buf + *start;
const char *bufend = buf + *end;
+ if (PqRetryBuffering && bufptr < bufend)
+ PqRetryBufferFlushed = true;
+
while (bufptr < bufend)
{
ssize_t r;
@@ -1442,6 +1491,10 @@ socket_flush_if_writable(void)
if (PqCommBusy)
return 0;
+ /* Defer flush while buffering uncommitted output for retry */
+ if (PqRetryBuffering && !PqRetryBufferFlushed)
+ return 0;
+
/* Temporarily put the socket into non-blocking mode */
socket_set_nonblocking(true);
@@ -1461,6 +1514,51 @@ socket_is_send_pending(void)
return (PqSendStart < PqSendPointer);
}
+/* --------------------------------
+ * Autocommit retry output buffering support
+ * --------------------------------
+ */
+void
+pq_begin_retry_buffer(void)
+{
+ PqRetryBuffering = true;
+ PqRetryBufferFlushed = false;
+ PqSendRetryStart = PqSendPointer;
+}
+
+void
+pq_end_retry_buffer(void)
+{
+ PqRetryBuffering = false;
+ PqRetryBufferFlushed = false;
+ PqSendRetryStart = 0;
+}
+
+bool
+pq_retry_buffer_flushed(void)
+{
+ return PqRetryBufferFlushed;
+}
+
+void
+pq_discard_retry_buffer(void)
+{
+ Assert(PqRetryBuffering);
+ Assert(!PqRetryBufferFlushed);
+ Assert(PqSendStart <= PqSendRetryStart);
+ Assert(PqSendRetryStart <= PqSendPointer);
+ PqSendPointer = PqSendRetryStart;
+}
+
+bool
+pq_is_next_msg_sync(void)
+{
+ if (PqRecvLength - PqRecvPointer >= 5 &&
+ PqRecvBuffer[PqRecvPointer] == PqMsg_Sync)
+ return true;
+ return false;
+}
+
/* --------------------------------
* Message-level I/O routines begin here.
* --------------------------------
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index b6bdfe213fe..aa9acdd6df2 100644
--- a/src/backend/tcop/postgres.c
+++ b/src/backend/tcop/postgres.c
@@ -110,6 +110,13 @@ int client_connection_check_interval = 0;
/* flags for non-system relation kinds to restrict use */
int restrict_nonsystem_relation_kind;
+/* GUC variables for automatic retry of autocommit statements */
+int max_autocommit_retries = 0;
+bool enable_ddl_autocommit_retry = false;
+int autocommit_retry_backoff_base_ms = 10;
+int autocommit_retry_backoff_max_ms = 1000;
+bool log_autocommit_retries = false;
+
/*
* Include signal sender PID/UID in the server log when available
* (SA_SIGINFO). The caller must supply the already-captured pid and uid
@@ -132,6 +139,14 @@ typedef struct BindParamCbData
const char *paramval; /* textual input string, if available */
} BindParamCbData;
+/* statement classification for autocommit auto-retry */
+typedef enum AutocommitRetryClass
+{
+ AUTOCOMMIT_RETRY_NONE = 0,
+ AUTOCOMMIT_RETRY_DML, /* SELECT, INSERT, UPDATE, DELETE, MERGE, CALL */
+ AUTOCOMMIT_RETRY_DDL, /* Rollback-safe DDL / utility statements */
+} AutocommitRetryClass;
+
/* ----------------
* private variables
* ----------------
@@ -143,6 +158,12 @@ typedef struct BindParamCbData
*/
static bool xact_started = false;
+/*
+ * True if the most recent ERRCODE_QUERY_CANCELED was caused by
+ * statement_timeout rather than a user cancel request.
+ */
+static bool stmt_timeout_canceled = false;
+
/*
* Flag to indicate that we are doing the outer loop's read-from-client,
* as opposed to any random read from client that might happen within
@@ -193,6 +214,12 @@ static bool IsTransactionExitStmt(Node *parsetree);
static bool IsTransactionExitStmtList(List *pstmts);
static bool IsTransactionStmtList(List *pstmts);
static void drop_unnamed_stmt(void);
+static AutocommitRetryClass classify_autocommit_retry_stmt(Node *stmt);
+static bool is_autocommit_error_retryable(int sqlerrcode,
+ AutocommitRetryClass retry_class);
+static void autocommit_retry_sleep(int retry_attempt, int sqlerrcode,
+ const char *errmsg_str,
+ const char *query_string);
static void ProcessRecoveryConflictInterrupts(void);
static void ProcessRecoveryConflictInterrupt(RecoveryConflictReason reason);
static void report_recovery_conflict(RecoveryConflictReason reason);
@@ -1119,15 +1146,15 @@ exec_simple_query(const char *query_string)
foreach(parsetree_item, parsetree_list)
{
RawStmt *parsetree = lfirst_node(RawStmt, parsetree_item);
- bool snapshot_set = false;
CommandTag commandTag;
QueryCompletion qc;
MemoryContext per_parsetree_context = NULL;
- List *querytree_list,
- *plantree_list;
- Portal portal;
- DestReceiver *receiver;
- int16 format;
+ MemoryContext retry_cxt = NULL;
+ AutocommitRetryClass retry_class = AUTOCOMMIT_RETRY_NONE;
+ bool can_retry = false;
+ volatile int retry_attempt = 0;
+ volatile bool retry_done = false;
+ volatile LocalTransactionId initial_lxid = InvalidLocalTransactionId;
const char *cmdtagname;
size_t cmdtaglen;
@@ -1175,193 +1202,327 @@ exec_simple_query(const char *query_string)
if (use_implicit_block)
BeginImplicitTransactionBlock();
- /* If we got a cancel signal in parsing or prior command, quit */
- CHECK_FOR_INTERRUPTS();
-
/*
- * Set up a snapshot if parse analysis/planning will need one.
+ * Determine whether this statement is eligible for automatic retry
+ * in autocommit mode. Only single-statement query strings outside an
+ * explicit transaction block qualify.
*/
- if (analyze_requires_snapshot(parsetree))
+ if (!use_implicit_block && !IsTransactionBlock())
{
- PushActiveSnapshot(GetTransactionSnapshot());
- snapshot_set = true;
+ retry_class = classify_autocommit_retry_stmt((Node *) parsetree);
+ can_retry = (max_autocommit_retries > 0) &&
+ (retry_class == AUTOCOMMIT_RETRY_DML ||
+ (retry_class == AUTOCOMMIT_RETRY_DDL && enable_ddl_autocommit_retry));
}
- /*
- * OK to analyze, rewrite, and plan this query.
- *
- * Switch to appropriate context for constructing query and plan trees
- * (these can't be in the transaction context, as that will get reset
- * when the command is COMMIT/ROLLBACK). If we have multiple
- * parsetrees, we use a separate context for each one, so that we can
- * free that memory before moving on to the next one. But for the
- * last (or only) parsetree, just use MessageContext, which will be
- * reset shortly after completion anyway. In event of an error, the
- * per_parsetree_context will be deleted when MessageContext is reset.
- */
- if (lnext(parsetree_list, parsetree_item) != NULL)
+ if (can_retry)
+ retry_cxt = AllocSetContextCreate(MessageContext,
+ "autocommit retry context",
+ ALLOCSET_DEFAULT_SIZES);
+
+ for (;;)
{
- per_parsetree_context =
- AllocSetContextCreate(MessageContext,
- "per-parsetree message context",
- ALLOCSET_DEFAULT_SIZES);
- oldcontext = MemoryContextSwitchTo(per_parsetree_context);
- }
- else
- oldcontext = MemoryContextSwitchTo(MessageContext);
+ int retry_errcode = 0;
+ char *retry_errmsg = NULL;
- querytree_list = pg_analyze_and_rewrite_fixedparams(parsetree, query_string,
- NULL, 0, NULL);
+ stmt_timeout_canceled = false;
- plantree_list = pg_plan_queries(querytree_list, query_string,
- CURSOR_OPT_PARALLEL_OK, NULL);
+ if (retry_attempt > 0)
+ {
+ SetCurrentStatementStartTimestamp();
+ start_xact_command();
+ BeginCommand(commandTag, dest);
+ }
- /*
- * Done with the snapshot used for parsing/planning.
- *
- * While it looks promising to reuse the same snapshot for query
- * execution (at least for simple protocol), unfortunately it causes
- * execution to use a snapshot that has been acquired before locking
- * any of the tables mentioned in the query. This creates user-
- * visible anomalies, so refrain. Refer to
- * https://postgr.es/m/flat/[email protected] for details.
- */
- if (snapshot_set)
- PopActiveSnapshot();
+ initial_lxid = MyProc ? MyProc->vxid.lxid : InvalidLocalTransactionId;
- /* If we got a cancel signal in analysis or planning, quit */
- CHECK_FOR_INTERRUPTS();
+ if (can_retry && whereToSendOutput == DestRemote)
+ pq_begin_retry_buffer();
- /*
- * Create unnamed portal to run the query or queries in. If there
- * already is one, silently drop it.
- */
- portal = CreatePortal("", true, true);
- /* Don't display the portal in pg_cursors */
- portal->visible = false;
+ PG_TRY();
+ {
+ RawStmt *cur_parsetree = parsetree;
+ bool snapshot_set = false;
+ List *querytree_list,
+ *plantree_list;
+ Portal portal;
+ DestReceiver *receiver;
+ int16 format;
+
+ /* If we got a cancel signal in parsing or prior command, quit */
+ CHECK_FOR_INTERRUPTS();
- /*
- * We don't have to copy anything into the portal, because everything
- * we are passing here is in MessageContext or the
- * per_parsetree_context, and so will outlive the portal anyway.
- */
- PortalDefineQuery(portal,
- NULL,
- query_string,
- commandTag,
- plantree_list,
- NULL);
+ /*
+ * Set up a snapshot if parse analysis/planning will need one.
+ */
+ if (analyze_requires_snapshot(cur_parsetree))
+ {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ snapshot_set = true;
+ }
- /*
- * Start the portal. No parameters here.
- */
- PortalStart(portal, NULL, 0, InvalidSnapshot);
+ /*
+ * OK to analyze, rewrite, and plan this query.
+ *
+ * Switch to appropriate context for constructing query and
+ * plan trees (these can't be in the transaction context, as
+ * that will get reset when the command is COMMIT/ROLLBACK).
+ * When autocommit retry is active, we allocate in retry_cxt
+ * and work on a deep copy of parsetree because parse analysis
+ * and utility execution can mutate raw parse nodes in-place.
+ */
+ if (retry_cxt != NULL)
+ {
+ oldcontext = MemoryContextSwitchTo(retry_cxt);
+ cur_parsetree = copyObject(parsetree);
+ }
+ else if (lnext(parsetree_list, parsetree_item) != NULL)
+ {
+ per_parsetree_context =
+ AllocSetContextCreate(MessageContext,
+ "per-parsetree message context",
+ ALLOCSET_DEFAULT_SIZES);
+ oldcontext = MemoryContextSwitchTo(per_parsetree_context);
+ }
+ else
+ oldcontext = MemoryContextSwitchTo(MessageContext);
- /*
- * Select the appropriate output format: text unless we are doing a
- * FETCH from a binary cursor. (Pretty grotty to have to do this here
- * --- but it avoids grottiness in other places. Ah, the joys of
- * backward compatibility...)
- */
- format = 0; /* TEXT is default */
- if (IsA(parsetree->stmt, FetchStmt))
- {
- FetchStmt *stmt = (FetchStmt *) parsetree->stmt;
+ querytree_list = pg_analyze_and_rewrite_fixedparams(cur_parsetree,
+ query_string,
+ NULL, 0, NULL);
- if (!stmt->ismove)
- {
- Portal fportal = GetPortalByName(stmt->portalname);
+ plantree_list = pg_plan_queries(querytree_list, query_string,
+ CURSOR_OPT_PARALLEL_OK, NULL);
- if (PortalIsValid(fportal) &&
- (fportal->cursorOptions & CURSOR_OPT_BINARY))
- format = 1; /* BINARY */
- }
- }
- PortalSetResultFormat(portal, 1, &format);
+ /*
+ * Done with the snapshot used for parsing/planning.
+ *
+ * While it looks promising to reuse the same snapshot for
+ * query execution (at least for simple protocol),
+ * unfortunately it causes execution to use a snapshot that
+ * has been acquired before locking any of the tables
+ * mentioned in the query. This creates user-visible
+ * anomalies, so refrain. Refer to
+ * https://postgr.es/m/flat/[email protected] for
+ * details.
+ */
+ if (snapshot_set)
+ PopActiveSnapshot();
- /*
- * Now we can create the destination receiver object.
- */
- receiver = CreateDestReceiver(dest);
- if (dest == DestRemote)
- SetRemoteDestReceiverParams(receiver, portal);
+ /* If we got a cancel signal in analysis or planning, quit */
+ CHECK_FOR_INTERRUPTS();
- /*
- * Switch back to transaction context for execution.
- */
- MemoryContextSwitchTo(oldcontext);
+ /*
+ * Create unnamed portal to run the query or queries in. If
+ * there already is one, silently drop it.
+ */
+ portal = CreatePortal("", true, true);
+ /* Don't display the portal in pg_cursors */
+ portal->visible = false;
- /*
- * Run the portal to completion, and then drop it (and the receiver).
- */
- (void) PortalRun(portal,
- FETCH_ALL,
- true, /* always top level */
- receiver,
- receiver,
- &qc);
+ /*
+ * We don't have to copy anything into the portal, because
+ * everything we are passing here is in MessageContext,
+ * per_parsetree_context, or retry_cxt, and so will outlive
+ * the portal anyway.
+ */
+ PortalDefineQuery(portal,
+ NULL,
+ query_string,
+ commandTag,
+ plantree_list,
+ NULL);
- receiver->rDestroy(receiver);
+ /*
+ * Start the portal. No parameters here.
+ */
+ PortalStart(portal, NULL, 0, InvalidSnapshot);
- PortalDrop(portal, false);
+ /*
+ * Select the appropriate output format: text unless we are
+ * doing a FETCH from a binary cursor. (Pretty grotty to have
+ * to do this here --- but it avoids grottiness in other
+ * places. Ah, the joys of backward compatibility...)
+ */
+ format = 0; /* TEXT is default */
+ if (IsA(cur_parsetree->stmt, FetchStmt))
+ {
+ FetchStmt *stmt = (FetchStmt *) cur_parsetree->stmt;
- if (lnext(parsetree_list, parsetree_item) == NULL)
- {
- /*
- * If this is the last parsetree of the query string, close down
- * transaction statement before reporting command-complete. This
- * is so that any end-of-transaction errors are reported before
- * the command-complete message is issued, to avoid confusing
- * clients who will expect either a command-complete message or an
- * error, not one and then the other. Also, if we're using an
- * implicit transaction block, we must close that out first.
- */
- if (use_implicit_block)
- EndImplicitTransactionBlock();
- finish_xact_command();
- }
- else if (IsA(parsetree->stmt, TransactionStmt))
- {
- /*
- * If this was a transaction control statement, commit it. We will
- * start a new xact command for the next command.
- */
- finish_xact_command();
- }
- else
- {
- /*
- * We had better not see XACT_FLAGS_NEEDIMMEDIATECOMMIT set if
- * we're not calling finish_xact_command(). (The implicit
- * transaction block should have prevented it from getting set.)
- */
- Assert(!(MyXactFlags & XACT_FLAGS_NEEDIMMEDIATECOMMIT));
+ if (!stmt->ismove)
+ {
+ Portal fportal = GetPortalByName(stmt->portalname);
- /*
- * We need a CommandCounterIncrement after every query, except
- * those that start or end a transaction block.
- */
- CommandCounterIncrement();
+ if (PortalIsValid(fportal) &&
+ (fportal->cursorOptions & CURSOR_OPT_BINARY))
+ format = 1; /* BINARY */
+ }
+ }
+ PortalSetResultFormat(portal, 1, &format);
- /*
- * Disable statement timeout between queries of a multi-query
- * string, so that the timeout applies separately to each query.
- * (Our next loop iteration will start a fresh timeout.)
- */
- disable_statement_timeout();
- }
+ /*
+ * Now we can create the destination receiver object.
+ */
+ receiver = CreateDestReceiver(dest);
+ if (dest == DestRemote)
+ SetRemoteDestReceiverParams(receiver, portal);
- /*
- * Tell client that we're done with this query. Note we emit exactly
- * one EndCommand report for each raw parsetree, thus one for each SQL
- * command the client sent, regardless of rewriting. (But a command
- * aborted by error will not send an EndCommand report at all.)
- */
- EndCommand(&qc, dest, false);
+ /*
+ * Switch back to transaction context for execution.
+ */
+ MemoryContextSwitchTo(oldcontext);
+
+ /*
+ * Run the portal to completion, and then drop it (and the
+ * receiver).
+ */
+ (void) PortalRun(portal,
+ FETCH_ALL,
+ true, /* always top level */
+ receiver,
+ receiver,
+ &qc);
+
+ receiver->rDestroy(receiver);
+
+ PortalDrop(portal, false);
+
+ if (lnext(parsetree_list, parsetree_item) == NULL)
+ {
+ /*
+ * If this is the last parsetree of the query string,
+ * close down transaction statement before reporting
+ * command-complete. This is so that any
+ * end-of-transaction errors are reported before the
+ * command-complete message is issued, to avoid confusing
+ * clients who will expect either a command-complete
+ * message or an error, not one and then the other. Also,
+ * if we're using an implicit transaction block, we must
+ * close that out first.
+ */
+ if (use_implicit_block)
+ EndImplicitTransactionBlock();
+ finish_xact_command();
+ }
+ else if (IsA(cur_parsetree->stmt, TransactionStmt))
+ {
+ /*
+ * If this was a transaction control statement, commit it.
+ * We will start a new xact command for the next command.
+ */
+ finish_xact_command();
+ }
+ else
+ {
+ /*
+ * We had better not see XACT_FLAGS_NEEDIMMEDIATECOMMIT
+ * set if we're not calling finish_xact_command(). (The
+ * implicit transaction block should have prevented it
+ * from getting set.)
+ */
+ Assert(!(MyXactFlags & XACT_FLAGS_NEEDIMMEDIATECOMMIT));
+
+ /*
+ * We need a CommandCounterIncrement after every query,
+ * except those that start or end a transaction block.
+ */
+ CommandCounterIncrement();
+
+ /*
+ * Disable statement timeout between queries of a
+ * multi-query string, so that the timeout applies
+ * separately to each query. (Our next loop iteration
+ * will start a fresh timeout.)
+ */
+ disable_statement_timeout();
+ }
- /* Now we may drop the per-parsetree context, if one was created. */
- if (per_parsetree_context)
- MemoryContextDelete(per_parsetree_context);
+ if (can_retry && whereToSendOutput == DestRemote)
+ pq_end_retry_buffer();
+
+ /*
+ * Tell client that we're done with this query. Note we emit
+ * exactly one EndCommand report for each raw parsetree, thus
+ * one for each SQL command the client sent, regardless of
+ * rewriting. (But a command aborted by error will not send an
+ * EndCommand report at all.)
+ */
+ EndCommand(&qc, dest, false);
+
+ /*
+ * Now we may drop the per-parsetree or retry context, if one
+ * was created.
+ */
+ if (per_parsetree_context)
+ MemoryContextDelete(per_parsetree_context);
+ if (retry_cxt)
+ MemoryContextDelete(retry_cxt);
+
+ retry_done = true;
+ }
+ PG_CATCH();
+ {
+ ErrorData *edata;
+ int errcode_val = geterrcode();
+
+ if (!can_retry ||
+ retry_attempt >= max_autocommit_retries ||
+ pq_retry_buffer_flushed() ||
+ IsTransactionBlock() ||
+ (MyProc != NULL && MyProc->vxid.lxid != initial_lxid) ||
+ !is_autocommit_error_retryable(errcode_val, retry_class))
+ {
+ pq_end_retry_buffer();
+ PG_RE_THROW();
+ }
+
+ HOLD_INTERRUPTS();
+ disable_all_timeouts(false);
+ QueryCancelPending = false;
+ stmt_timeout_canceled = false;
+
+ /*
+ * Discard any uncommitted RowDescription, DataRow, or Notice
+ * messages accumulated in PqSendBuffer during the failed
+ * attempt.
+ */
+ pq_discard_retry_buffer();
+ pq_end_retry_buffer();
+
+ MemoryContextSwitchTo(MessageContext);
+ edata = CopyErrorData();
+ retry_errcode = edata->sqlerrcode;
+ retry_errmsg = pstrdup(edata->message ? edata->message : "");
+ FreeErrorData(edata);
+ FlushErrorState();
+
+ AbortCurrentTransaction();
+ if (am_walsender)
+ WalSndErrorCleanup();
+ PortalErrorCleanup();
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotRelease();
+ ReplicationSlotCleanup(false);
+ jit_reset_after_error();
+
+ xact_started = false;
+
+ MemoryContextSwitchTo(MessageContext);
+ MemoryContextReset(retry_cxt);
+
+ RESUME_INTERRUPTS();
+ }
+ PG_END_TRY();
+
+ if (retry_done)
+ break;
+
+ retry_attempt++;
+ autocommit_retry_sleep(retry_attempt, retry_errcode,
+ retry_errmsg, query_string);
+ if (retry_errmsg)
+ pfree(retry_errmsg);
+ }
} /* end loop over parsetrees */
/*
@@ -2151,13 +2312,21 @@ static void
exec_execute_message(const char *portal_name, long max_rows)
{
CommandDest dest;
- DestReceiver *receiver;
Portal portal;
- bool completed;
QueryCompletion qc;
const char *sourceText;
const char *prepStmtName;
+ const char *raw_stmt_name = NULL;
ParamListInfo portalParams;
+ CachedPlanSource *psrc = NULL;
+ ParamListInfo saved_params = NULL;
+ int16 *saved_rformats = NULL;
+ int saved_num_rformats = 0;
+ AutocommitRetryClass retry_class = AUTOCOMMIT_RETRY_NONE;
+ bool can_retry = false;
+ volatile int retry_attempt = 0;
+ volatile bool retry_done = false;
+ volatile LocalTransactionId initial_lxid = InvalidLocalTransactionId;
bool save_log_statement_stats = log_statement_stats;
bool is_xact_command;
bool execute_is_fetch;
@@ -2196,13 +2365,14 @@ exec_execute_message(const char *portal_name, long max_rows)
/*
* We must copy the sourceText and prepStmtName into MessageContext in
- * case the portal is destroyed during finish_xact_command. We do not
- * make a copy of the portalParams though, preferring to just not print
- * them in that case.
+ * case the portal is destroyed during finish_xact_command.
*/
sourceText = pstrdup(portal->sourceText);
if (portal->prepStmtName)
- prepStmtName = pstrdup(portal->prepStmtName);
+ {
+ raw_stmt_name = pstrdup(portal->prepStmtName);
+ prepStmtName = raw_stmt_name;
+ }
else
prepStmtName = "<unnamed>";
portalParams = portal->portalParams;
@@ -2245,14 +2415,6 @@ exec_execute_message(const char *portal_name, long max_rows)
BeginCommand(portal->commandTag, dest);
- /*
- * Create dest receiver in MessageContext (we don't want it in transaction
- * context, because that may get deleted if portal contains VACUUM).
- */
- receiver = CreateDestReceiver(dest);
- if (dest == DestRemoteExecute)
- SetRemoteDestReceiverParams(receiver, portal);
-
/*
* Ensure we are in a transaction command (this should normally be the
* case already due to prior BIND).
@@ -2300,91 +2462,297 @@ exec_execute_message(const char *portal_name, long max_rows)
errmsg("current transaction is aborted, "
"commands ignored until end of transaction block")));
- /* Check for cancel signal before we start execution */
- CHECK_FOR_INTERRUPTS();
+ if (max_rows <= 0)
+ max_rows = FETCH_ALL;
/*
- * Okay to run the portal. Set the error callback so that parameters are
- * logged. The parameters must have been saved during the bind phase.
+ * Determine whether this portal execution is eligible for automatic retry
+ * in autocommit mode.
*/
- params_data.portalName = portal->name;
- params_data.params = portalParams;
- params_errcxt.previous = error_context_stack;
- params_errcxt.callback = ParamsErrorCallback;
- params_errcxt.arg = ¶ms_data;
- error_context_stack = ¶ms_errcxt;
+ if (!execute_is_fetch &&
+ max_rows == FETCH_ALL &&
+ !IsTransactionBlock() &&
+ !(MyXactFlags & XACT_FLAGS_PIPELINING) &&
+ list_length(portal->stmts) == 1)
+ {
+ if (raw_stmt_name != NULL && raw_stmt_name[0] != '\0')
+ {
+ PreparedStatement *pstmt = FetchPreparedStatement(raw_stmt_name, false);
- if (max_rows <= 0)
- max_rows = FETCH_ALL;
+ if (pstmt != NULL)
+ psrc = pstmt->plansource;
+ }
+ else
+ psrc = unnamed_stmt_psrc;
- completed = PortalRun(portal,
- max_rows,
- true, /* always top level */
- receiver,
- receiver,
- &qc);
+ if (psrc != NULL)
+ {
+ retry_class = (psrc->raw_parse_tree != NULL) ?
+ classify_autocommit_retry_stmt((Node *) psrc->raw_parse_tree) :
+ classify_autocommit_retry_stmt((Node *) linitial(portal->stmts));
+ can_retry = (max_autocommit_retries > 0) &&
+ (retry_class == AUTOCOMMIT_RETRY_DML ||
+ (retry_class == AUTOCOMMIT_RETRY_DDL && enable_ddl_autocommit_retry));
+ }
+ }
- receiver->rDestroy(receiver);
+ if (can_retry)
+ {
+ MemoryContext oldcxt = MemoryContextSwitchTo(MessageContext);
- /* Done executing; remove the params error callback */
- error_context_stack = error_context_stack->previous;
+ if (portalParams != NULL)
+ {
+ saved_params = copyParamList(portalParams);
+ if (portalParams->paramValuesStr != NULL)
+ saved_params->paramValuesStr = pstrdup(portalParams->paramValuesStr);
+ }
+ if (portal->tupDesc != NULL && portal->formats != NULL)
+ {
+ saved_num_rformats = portal->tupDesc->natts;
+ saved_rformats = palloc_array(int16, saved_num_rformats);
+ memcpy(saved_rformats, portal->formats,
+ saved_num_rformats * sizeof(int16));
+ }
+ MemoryContextSwitchTo(oldcxt);
+ }
- if (completed)
+ for (;;)
{
- if (is_xact_command || (MyXactFlags & XACT_FLAGS_NEEDIMMEDIATECOMMIT))
+ int retry_errcode = 0;
+ char *retry_errmsg = NULL;
+
+ stmt_timeout_canceled = false;
+
+ if (can_retry && whereToSendOutput == DestRemote)
+ pq_begin_retry_buffer();
+
+ PG_TRY();
{
+ DestReceiver *receiver;
+ MemoryContext oldcxt;
+ bool completed;
+
+ if (retry_attempt > 0)
+ {
+ ParamListInfo retry_params = NULL;
+ CachedPlan *cplan;
+ bool snapshot_set = false;
+ char *portal_query_string;
+ char *portal_stmt_name;
+
+ SetCurrentStatementStartTimestamp();
+ start_xact_command();
+ BeginCommand(psrc->commandTag, dest);
+
+ if (portal_name[0] == '\0')
+ portal = CreatePortal(portal_name, true, true);
+ else
+ portal = CreatePortal(portal_name, false, false);
+
+ oldcxt = MemoryContextSwitchTo(portal->portalContext);
+ portal_query_string = pstrdup(psrc->query_string);
+ portal_stmt_name = (raw_stmt_name && raw_stmt_name[0] != '\0') ?
+ pstrdup(raw_stmt_name) : NULL;
+ if (saved_params != NULL)
+ {
+ retry_params = copyParamList(saved_params);
+ if (saved_params->paramValuesStr != NULL)
+ retry_params->paramValuesStr = pstrdup(saved_params->paramValuesStr);
+ }
+ MemoryContextSwitchTo(oldcxt);
+
+ if (retry_params != NULL ||
+ (psrc->raw_parse_tree &&
+ analyze_requires_snapshot(psrc->raw_parse_tree)))
+ {
+ PushActiveSnapshot(GetTransactionSnapshot());
+ snapshot_set = true;
+ }
+
+ oldcxt = MemoryContextSwitchTo(MessageContext);
+ cplan = GetCachedPlan(psrc, retry_params, NULL, NULL);
+ PortalDefineQuery(portal,
+ portal_stmt_name,
+ portal_query_string,
+ psrc->commandTag,
+ cplan->stmt_list,
+ cplan);
+ if (snapshot_set)
+ PopActiveSnapshot();
+ MemoryContextSwitchTo(oldcxt);
+
+ PortalStart(portal, retry_params, 0, InvalidSnapshot);
+ if (saved_num_rformats > 0)
+ PortalSetResultFormat(portal, saved_num_rformats, saved_rformats);
+
+ portalParams = retry_params;
+ }
+
+ initial_lxid = MyProc ? MyProc->vxid.lxid : InvalidLocalTransactionId;
+
/*
- * If this was a transaction control statement, commit it. We
- * will start a new xact command for the next command (if any).
- * Likewise if the statement required immediate commit. Without
- * this provision, we wouldn't force commit until Sync is
- * received, which creates a hazard if the client tries to
- * pipeline immediate-commit statements.
+ * Create dest receiver in MessageContext (we don't want it in
+ * transaction context, because that may get deleted if portal
+ * contains VACUUM).
*/
- finish_xact_command();
+ oldcxt = MemoryContextSwitchTo(MessageContext);
+ receiver = CreateDestReceiver(dest);
+ if (dest == DestRemoteExecute)
+ SetRemoteDestReceiverParams(receiver, portal);
+ MemoryContextSwitchTo(oldcxt);
+
+ /* Check for cancel signal before we start execution */
+ CHECK_FOR_INTERRUPTS();
/*
- * These commands typically don't have any parameters, and even if
- * one did we couldn't print them now because the storage went
- * away during finish_xact_command. So pretend there were none.
+ * Okay to run the portal. Set the error callback so that
+ * parameters are logged. The parameters must have been saved
+ * during the bind phase.
*/
- portalParams = NULL;
+ params_data.portalName = portal->name;
+ params_data.params = portalParams;
+ params_errcxt.previous = error_context_stack;
+ params_errcxt.callback = ParamsErrorCallback;
+ params_errcxt.arg = ¶ms_data;
+ error_context_stack = ¶ms_errcxt;
+
+ completed = PortalRun(portal,
+ max_rows,
+ true, /* always top level */
+ receiver,
+ receiver,
+ &qc);
+
+ receiver->rDestroy(receiver);
+
+ /* Done executing; remove the params error callback */
+ error_context_stack = error_context_stack->previous;
+
+ if (completed)
+ {
+ if (is_xact_command ||
+ (MyXactFlags & XACT_FLAGS_NEEDIMMEDIATECOMMIT) ||
+ (can_retry && pq_is_next_msg_sync()))
+ {
+ /*
+ * If this was a transaction control statement, commit it.
+ * We will start a new xact command for the next command
+ * (if any). Likewise if the statement required immediate
+ * commit, or if autocommit retry is active and the next
+ * client message is Sync (so commit-time serialization
+ * failures are caught inside the retry loop).
+ */
+ finish_xact_command();
+
+ /*
+ * Use saved_params if available, since portal storage
+ * went away during finish_xact_command.
+ */
+ portalParams = saved_params;
+ }
+ else
+ {
+ /*
+ * We need a CommandCounterIncrement after every query,
+ * except those that start or end a transaction block.
+ */
+ CommandCounterIncrement();
+
+ /*
+ * Set XACT_FLAGS_PIPELINING whenever we complete an
+ * Execute message without immediately committing the
+ * transaction.
+ */
+ MyXactFlags |= XACT_FLAGS_PIPELINING;
+
+ /*
+ * Disable statement timeout whenever we complete an
+ * Execute message. The next protocol message will start
+ * a fresh timeout.
+ */
+ disable_statement_timeout();
+ }
+
+ if (can_retry && whereToSendOutput == DestRemote)
+ pq_end_retry_buffer();
+
+ /* Send appropriate CommandComplete to client */
+ EndCommand(&qc, dest, false);
+ }
+ else
+ {
+ if (can_retry && whereToSendOutput == DestRemote)
+ pq_end_retry_buffer();
+
+ /* Portal run not complete, so send PortalSuspended */
+ if (whereToSendOutput == DestRemote)
+ pq_putemptymessage(PqMsg_PortalSuspended);
+
+ /*
+ * Set XACT_FLAGS_PIPELINING whenever we suspend an Execute
+ * message, too.
+ */
+ MyXactFlags |= XACT_FLAGS_PIPELINING;
+ }
+
+ retry_done = true;
}
- else
+ PG_CATCH();
{
- /*
- * We need a CommandCounterIncrement after every query, except
- * those that start or end a transaction block.
- */
- CommandCounterIncrement();
+ ErrorData *edata;
+ int errcode_val = geterrcode();
+
+ if (!can_retry ||
+ retry_attempt >= max_autocommit_retries ||
+ pq_retry_buffer_flushed() ||
+ IsTransactionBlock() ||
+ (MyProc != NULL && MyProc->vxid.lxid != initial_lxid) ||
+ !is_autocommit_error_retryable(errcode_val, retry_class))
+ {
+ pq_end_retry_buffer();
+ PG_RE_THROW();
+ }
- /*
- * Set XACT_FLAGS_PIPELINING whenever we complete an Execute
- * message without immediately committing the transaction.
- */
- MyXactFlags |= XACT_FLAGS_PIPELINING;
+ HOLD_INTERRUPTS();
+ disable_all_timeouts(false);
+ QueryCancelPending = false;
+ stmt_timeout_canceled = false;
- /*
- * Disable statement timeout whenever we complete an Execute
- * message. The next protocol message will start a fresh timeout.
- */
- disable_statement_timeout();
+ pq_discard_retry_buffer();
+ pq_end_retry_buffer();
+
+ MemoryContextSwitchTo(MessageContext);
+ edata = CopyErrorData();
+ retry_errcode = edata->sqlerrcode;
+ retry_errmsg = pstrdup(edata->message ? edata->message : "");
+ FreeErrorData(edata);
+ FlushErrorState();
+
+ AbortCurrentTransaction();
+ if (am_walsender)
+ WalSndErrorCleanup();
+ PortalErrorCleanup();
+ if (MyReplicationSlot != NULL)
+ ReplicationSlotRelease();
+ ReplicationSlotCleanup(false);
+ jit_reset_after_error();
+
+ xact_started = false;
+
+ MemoryContextSwitchTo(MessageContext);
+
+ RESUME_INTERRUPTS();
}
+ PG_END_TRY();
- /* Send appropriate CommandComplete to client */
- EndCommand(&qc, dest, false);
- }
- else
- {
- /* Portal run not complete, so send PortalSuspended */
- if (whereToSendOutput == DestRemote)
- pq_putemptymessage(PqMsg_PortalSuspended);
+ if (retry_done)
+ break;
- /*
- * Set XACT_FLAGS_PIPELINING whenever we suspend an Execute message,
- * too.
- */
- MyXactFlags |= XACT_FLAGS_PIPELINING;
+ retry_attempt++;
+ autocommit_retry_sleep(retry_attempt, retry_errcode,
+ retry_errmsg, sourceText);
+ if (retry_errmsg)
+ pfree(retry_errmsg);
}
/*
@@ -3001,6 +3369,173 @@ drop_unnamed_stmt(void)
}
}
+/*
+ * Classify a statement (RawStmt, PlannedStmt, or parse/utility Node) for
+ * autocommit automatic retry eligibility.
+ */
+static AutocommitRetryClass
+classify_autocommit_retry_stmt(Node *stmt)
+{
+ if (stmt == NULL)
+ return AUTOCOMMIT_RETRY_NONE;
+
+ if (IsA(stmt, RawStmt))
+ stmt = ((RawStmt *) stmt)->stmt;
+
+ if (IsA(stmt, PlannedStmt))
+ {
+ PlannedStmt *pstmt = (PlannedStmt *) stmt;
+
+ switch (pstmt->commandType)
+ {
+ case CMD_SELECT:
+ case CMD_INSERT:
+ case CMD_UPDATE:
+ case CMD_DELETE:
+ case CMD_MERGE:
+ return AUTOCOMMIT_RETRY_DML;
+ case CMD_UTILITY:
+ stmt = pstmt->utilityStmt;
+ if (stmt == NULL)
+ return AUTOCOMMIT_RETRY_NONE;
+ break;
+ default:
+ return AUTOCOMMIT_RETRY_NONE;
+ }
+ }
+
+ switch (nodeTag(stmt))
+ {
+ case T_SelectStmt:
+ if (((SelectStmt *) stmt)->intoClause != NULL)
+ return AUTOCOMMIT_RETRY_DDL;
+ return AUTOCOMMIT_RETRY_DML;
+
+ case T_InsertStmt:
+ case T_UpdateStmt:
+ case T_DeleteStmt:
+ case T_MergeStmt:
+ case T_CallStmt:
+ return AUTOCOMMIT_RETRY_DML;
+
+ /*
+ * Non-transactional / filesystem-level commands that cannot be
+ * rolled back cleanly or run inside a transaction.
+ */
+ case T_CreatedbStmt:
+ case T_DropdbStmt:
+ case T_CreateTableSpaceStmt:
+ case T_DropTableSpaceStmt:
+ case T_AlterSystemStmt:
+ return AUTOCOMMIT_RETRY_NONE;
+
+ /*
+ * Additional table-level utility commands that acquire heavyweight
+ * relation locks and are safe to retry if they fail before committing
+ * an internal transaction.
+ */
+ case T_TruncateStmt:
+ case T_ReindexStmt:
+ case T_VacuumStmt:
+ return AUTOCOMMIT_RETRY_DDL;
+
+ default:
+ if (GetCommandLogLevel(stmt) == LOGSTMT_DDL)
+ return AUTOCOMMIT_RETRY_DDL;
+ return AUTOCOMMIT_RETRY_NONE;
+ }
+}
+
+/*
+ * Return true if the given SQLSTATE error code is eligible for autocommit
+ * retry for the given statement class under the current GUC settings.
+ */
+static bool
+is_autocommit_error_retryable(int sqlerrcode, AutocommitRetryClass retry_class)
+{
+ if (max_autocommit_retries <= 0 || retry_class == AUTOCOMMIT_RETRY_NONE)
+ return false;
+
+ if (retry_class == AUTOCOMMIT_RETRY_DDL && !enable_ddl_autocommit_retry)
+ return false;
+
+ if (sqlerrcode == ERRCODE_T_R_SERIALIZATION_FAILURE ||
+ sqlerrcode == ERRCODE_T_R_DEADLOCK_DETECTED)
+ return true;
+
+ if (retry_class == AUTOCOMMIT_RETRY_DDL)
+ {
+ if (sqlerrcode == ERRCODE_LOCK_NOT_AVAILABLE)
+ return true;
+ if (sqlerrcode == ERRCODE_QUERY_CANCELED && stmt_timeout_canceled)
+ return true;
+ }
+
+ return false;
+}
+
+/*
+ * Sleep with exponential backoff and jitter before an autocommit retry
+ * attempt, optionally logging the retry event. Interruptible via WaitLatch
+ * so user cancel (SIGINT) or backend termination (SIGTERM) aborts the sleep
+ * immediately.
+ */
+static void
+autocommit_retry_sleep(int retry_attempt, int sqlerrcode,
+ const char *errmsg_str, const char *query_string)
+{
+ int64 base_ms = Max(1, autocommit_retry_backoff_base_ms);
+ int64 max_ms = Max(base_ms, autocommit_retry_backoff_max_ms);
+ int shift = Min(retry_attempt - 1, 20);
+ int64 cap_ms = Min(max_ms, base_ms * (INT64CONST(1) << shift));
+ int64 half_ms = cap_ms / 2;
+ int64 jitter_ms = (cap_ms - half_ms > 0) ?
+ (int64) pg_prng_uint64_range(&pg_global_prng_state, 0, cap_ms - half_ms) : 0;
+ long sleep_ms = (long) Max(1, half_ms + jitter_ms);
+ TimestampTz end_time;
+
+ if (log_autocommit_retries)
+ {
+ char *truncated_stmt = truncate_query_log(query_string);
+
+ ereport(LOG,
+ (errmsg("autocommit retry %d/%d after %s (%s) in %ld ms: %s",
+ retry_attempt,
+ max_autocommit_retries,
+ unpack_sql_state(sqlerrcode),
+ errmsg_str ? errmsg_str : "",
+ sleep_ms,
+ (truncated_stmt != NULL) ? truncated_stmt : query_string),
+ errhidestmt(true)));
+
+ if (truncated_stmt != NULL)
+ pfree(truncated_stmt);
+ }
+
+ end_time = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), sleep_ms);
+ for (;;)
+ {
+ TimestampTz now;
+ long rem_ms;
+
+ ResetLatch(MyLatch);
+ CHECK_FOR_INTERRUPTS();
+
+ now = GetCurrentTimestamp();
+ if (now >= end_time)
+ break;
+
+ rem_ms = (long) ((TimestampDifferenceMicroseconds(now, end_time) + 999) / 1000);
+ if (rem_ms <= 0)
+ break;
+
+ (void) WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ rem_ms,
+ PG_WAIT_TIMEOUT);
+ }
+}
+
/* --------------------------------
* signal handler routines used in PostgresMain()
@@ -3605,6 +4140,7 @@ ProcessInterrupts(void)
}
if (stmt_timeout_occurred)
{
+ stmt_timeout_canceled = true;
LockErrorCleanup();
ereport(ERROR,
(errcode(ERRCODE_QUERY_CANCELED),
@@ -4599,6 +5135,7 @@ PostgresMain(const char *dbname, const char *username)
*/
disable_all_timeouts(false); /* do first to avoid race condition */
QueryCancelPending = false;
+ stmt_timeout_canceled = false;
idle_in_transaction_timeout_enabled = false;
idle_session_timeout_enabled = false;
diff --git a/src/backend/utils/misc/guc_parameters.dat b/src/backend/utils/misc/guc_parameters.dat
index 380679a76c0..5a72a08b648 100644
--- a/src/backend/utils/misc/guc_parameters.dat
+++ b/src/backend/utils/misc/guc_parameters.dat
@@ -114,6 +114,24 @@
max => '600',
},
+{ name => 'autocommit_retry_backoff_base_ms', type => 'int', context => 'PGC_USERSET', group => 'CLIENT_CONN_STATEMENT',
+ short_desc => 'Sets the initial backoff delay in milliseconds before retrying an autocommit statement.',
+ flags => 'GUC_UNIT_MS',
+ variable => 'autocommit_retry_backoff_base_ms',
+ boot_val => '10',
+ min => '1',
+ max => '60000',
+},
+
+{ name => 'autocommit_retry_backoff_max_ms', type => 'int', context => 'PGC_USERSET', group => 'CLIENT_CONN_STATEMENT',
+ short_desc => 'Sets the maximum backoff delay in milliseconds between retries of an autocommit statement.',
+ flags => 'GUC_UNIT_MS',
+ variable => 'autocommit_retry_backoff_max_ms',
+ boot_val => '1000',
+ min => '1',
+ max => '300000',
+},
+
{ name => 'autovacuum', type => 'bool', context => 'PGC_SIGHUP', group => 'VACUUM_AUTOVACUUM',
short_desc => 'Starts the autovacuum subprocess.',
variable => 'autovacuum_start_daemon',
@@ -869,6 +887,12 @@
boot_val => 'true',
},
+{ name => 'enable_ddl_autocommit_retry', type => 'bool', context => 'PGC_USERSET', group => 'CLIENT_CONN_STATEMENT',
+ short_desc => 'Enables automatic retry of autocommit DDL statements on lock or statement timeout.',
+ variable => 'enable_ddl_autocommit_retry',
+ boot_val => 'false',
+},
+
{ name => 'enable_distinct_reordering', type => 'bool', context => 'PGC_USERSET', group => 'QUERY_TUNING_METHOD',
short_desc => 'Enables reordering of DISTINCT keys.',
flags => 'GUC_EXPLAIN',
@@ -1627,6 +1651,12 @@
max => 'INT_MAX',
},
+{ name => 'log_autocommit_retries', type => 'bool', context => 'PGC_USERSET', group => 'LOGGING_WHAT',
+ short_desc => 'Logs each automatic retry of an autocommit statement.',
+ variable => 'log_autocommit_retries',
+ boot_val => 'false',
+},
+
{ name => 'log_autovacuum_min_duration', type => 'int', context => 'PGC_SIGHUP', group => 'LOGGING_WHAT',
short_desc => 'Sets the minimum execution time above which vacuum actions by autovacuum will be logged.',
long_desc => '-1 disables logging vacuum actions by autovacuum. 0 means log all vacuum actions by autovacuum.',
@@ -1977,6 +2007,15 @@
max => 'MAX_BACKENDS',
},
+{ name => 'max_autocommit_retries', type => 'int', context => 'PGC_USERSET', group => 'CLIENT_CONN_STATEMENT',
+ short_desc => 'Sets the maximum number of automatic retries for autocommit statements encountering transient errors.',
+ long_desc => '0 disables automatic retries.',
+ variable => 'max_autocommit_retries',
+ boot_val => '0',
+ min => '0',
+ max => '100',
+},
+
{ name => 'max_connections', type => 'int', context => 'PGC_POSTMASTER', group => 'CONN_AUTH_SETTINGS',
short_desc => 'Sets the maximum number of concurrent connections.',
variable => 'MaxConnections',
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 01f98de8a9f..c292314d91a 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -635,6 +635,7 @@
# their durations, > 0 logs only
# actions running at least this number
# of milliseconds.
+#log_autocommit_retries = off # log automatic retries of autocommit statements
#log_checkpoints = on
#log_connections = '' # log aspects of connection setup
# options include receipt, authentication, authorization,
@@ -819,6 +820,10 @@
#lock_timeout = 0 # in milliseconds, 0 is disabled
#idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled
#idle_session_timeout = 0 # in milliseconds, 0 is disabled
+#max_autocommit_retries = 0 # max automatic retries for autocommit statements; 0 is disabled
+#enable_ddl_autocommit_retry = off # retry autocommit DDL on lock or statement timeout
+#autocommit_retry_backoff_base_ms = 10ms # initial backoff delay between autocommit retries
+#autocommit_retry_backoff_max_ms = 1s # maximum backoff delay between autocommit retries
#bytea_output = 'hex' # hex, escape
#xmlbinary = 'base64'
#xmloption = 'content'
diff --git a/src/include/libpq/libpq.h b/src/include/libpq/libpq.h
index d15073a0a93..6f751aadc42 100644
--- a/src/include/libpq/libpq.h
+++ b/src/include/libpq/libpq.h
@@ -85,6 +85,11 @@ extern int pq_getbyte_if_available(unsigned char *c);
extern ssize_t pq_buffer_remaining_data(void);
extern int pq_putmessage_v2(char msgtype, const char *s, size_t len);
extern bool pq_check_connection(void);
+extern void pq_begin_retry_buffer(void);
+extern void pq_end_retry_buffer(void);
+extern bool pq_retry_buffer_flushed(void);
+extern void pq_discard_retry_buffer(void);
+extern bool pq_is_next_msg_sync(void);
/*
* prototypes for functions in be-secure.c
diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h
index 5bc5bcfb20d..35d77c2295e 100644
--- a/src/include/tcop/tcopprot.h
+++ b/src/include/tcop/tcopprot.h
@@ -46,6 +46,12 @@ extern PGDLLIMPORT int log_statement;
extern PGDLLIMPORT int restrict_nonsystem_relation_kind;
+extern PGDLLIMPORT int max_autocommit_retries;
+extern PGDLLIMPORT bool enable_ddl_autocommit_retry;
+extern PGDLLIMPORT int autocommit_retry_backoff_base_ms;
+extern PGDLLIMPORT int autocommit_retry_backoff_max_ms;
+extern PGDLLIMPORT bool log_autocommit_retries;
+
extern List *pg_parse_query(const char *query_string);
extern List *pg_rewrite_query(Query *query);
extern List *pg_analyze_and_rewrite_fixedparams(RawStmt *parsetree,
diff --git a/src/test/isolation/expected/autocommit-retry.out b/src/test/isolation/expected/autocommit-retry.out
new file mode 100644
index 00000000000..32acdea38e7
--- /dev/null
+++ b/src/test/isolation/expected/autocommit-retry.out
@@ -0,0 +1,258 @@
+Parsed test spec with 2 sessions
+
+starting permutation: s2_retries_0 s1_begin s1_upd2 s2_upd_ret s1_commit s2_check_tbl
+step s2_retries_0: SET max_autocommit_retries = 0;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_upd_ret: UPDATE retry_tbl SET val = val + 1 RETURNING id, val; <waiting ...>
+step s1_commit: COMMIT;
+step s2_upd_ret: <... completed>
+ERROR: could not serialize access due to concurrent update
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 10
+ 2|120
+(2 rows)
+
+
+starting permutation: s2_retries_3 s1_begin s1_upd2 s2_upd_ret s1_commit s2_check_tbl
+step s2_retries_3: SET max_autocommit_retries = 3;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_upd_ret: UPDATE retry_tbl SET val = val + 1 RETURNING id, val; <waiting ...>
+step s1_commit: COMMIT;
+step s2_upd_ret: <... completed>
+id|val
+--+---
+ 1| 11
+ 2|121
+(2 rows)
+
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 11
+ 2|121
+(2 rows)
+
+
+starting permutation: s2_retries_3 s1_begin s1_upd2 s2_begin s2_upd_ret s1_commit s2_rollback s2_check_tbl
+step s2_retries_3: SET max_autocommit_retries = 3;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_begin: BEGIN;
+step s2_upd_ret: UPDATE retry_tbl SET val = val + 1 RETURNING id, val; <waiting ...>
+step s1_commit: COMMIT;
+step s2_upd_ret: <... completed>
+ERROR: could not serialize access due to concurrent update
+step s2_rollback: ROLLBACK;
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 10
+ 2|120
+(2 rows)
+
+
+starting permutation: s2_retries_3 s1_begin s1_upd2 s2_multistmt s1_commit s2_check_tbl
+step s2_retries_3: SET max_autocommit_retries = 3;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_multistmt: SELECT 1; UPDATE retry_tbl SET val = val + 1 WHERE id = 2; <waiting ...>
+step s1_commit: COMMIT;
+step s2_multistmt: <... completed>
+?column?
+--------
+ 1
+(1 row)
+
+ERROR: could not serialize access due to concurrent update
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 10
+ 2|120
+(2 rows)
+
+
+starting permutation: s2_retries_3 s1_begin s1_upd2 s2_call_simple s1_commit s2_check_tbl
+step s2_retries_3: SET max_autocommit_retries = 3;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_call_simple: CALL retry_proc_simple(); <waiting ...>
+step s1_commit: COMMIT;
+step s2_call_simple: <... completed>
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 10
+ 2|121
+(2 rows)
+
+
+starting permutation: s2_retries_3 s1_begin s1_upd2 s2_call_commit s1_commit s2_check_tbl
+step s2_retries_3: SET max_autocommit_retries = 3;
+step s1_begin: BEGIN;
+step s1_upd2: UPDATE retry_tbl SET val = val + 100 WHERE id = 2;
+step s2_call_commit: CALL retry_proc_with_commit(); <waiting ...>
+step s1_commit: COMMIT;
+step s2_call_commit: <... completed>
+ERROR: could not serialize access due to concurrent update
+step s2_check_tbl: SELECT id, val FROM retry_tbl ORDER BY id;
+id|val
+--+---
+ 1| 11
+ 2|120
+(2 rows)
+
+
+starting permutation: s2_retries_2 s2_reset_seq s2_test_40001 s2_reset_seq s2_test_40p01 s2_reset_seq s2_exhaust s2_check_seq s2_ddl_setup s2_ddl_alter s2_check_seq s2_ddl_enable s2_ddl_alter s2_ddl_stmttimeout_setup s2_ddl_alter2 s2_check_seq s2_ddl_stmttimeout_enable s2_ddl_alter2 s2_ddl_cleanup
+step s2_retries_2: SET max_autocommit_retries = 2;
+step s2_reset_seq:
+ SET log_autocommit_retries = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_test_40001: SELECT fail_n_times(2, '40001') AS attempts_40001;
+attempts_40001
+--------------
+ 3
+(1 row)
+
+step s2_reset_seq:
+ SET log_autocommit_retries = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_test_40p01: SELECT fail_n_times(2, '40P01') AS attempts_40p01;
+attempts_40p01
+--------------
+ 3
+(1 row)
+
+step s2_reset_seq:
+ SET log_autocommit_retries = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_exhaust: SELECT fail_n_times(3, '40001') AS should_fail;
+ERROR: simulated transient failure on attempt 3
+step s2_check_seq: SELECT last_value FROM retry_attempt_seq;
+last_value
+----------
+ 3
+(1 row)
+
+step s2_ddl_setup:
+ CREATE FUNCTION ddl_fail_once() RETURNS event_trigger LANGUAGE plpgsql AS $$
+ BEGIN
+ IF nextval('retry_attempt_seq') <= 1 THEN
+ RAISE EXCEPTION 'simulated DDL lock timeout' USING ERRCODE = '55P03';
+ END IF;
+ END;
+ $$;
+ CREATE EVENT TRIGGER ddl_retry_trig ON ddl_command_start
+ WHEN TAG IN ('ALTER TABLE') EXECUTE FUNCTION ddl_fail_once();
+ SET enable_ddl_autocommit_retry = off;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_ddl_alter: ALTER TABLE retry_tbl ADD COLUMN extra int DEFAULT 42;
+ERROR: simulated DDL lock timeout
+step s2_check_seq: SELECT last_value FROM retry_attempt_seq;
+last_value
+----------
+ 1
+(1 row)
+
+step s2_ddl_enable:
+ SET enable_ddl_autocommit_retry = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_ddl_alter: ALTER TABLE retry_tbl ADD COLUMN extra int DEFAULT 42;
+step s2_ddl_stmttimeout_setup:
+ SELECT last_value AS ddl_lock_attempts FROM retry_attempt_seq;
+ DROP EVENT TRIGGER ddl_retry_trig;
+ DROP FUNCTION ddl_fail_once();
+
+ CREATE FUNCTION ddl_sleep_once() RETURNS event_trigger LANGUAGE plpgsql AS $$
+ BEGIN
+ IF nextval('retry_attempt_seq') <= 1 THEN
+ PERFORM pg_sleep(1);
+ END IF;
+ END;
+ $$;
+ CREATE EVENT TRIGGER ddl_stmt_timeout_trig ON ddl_command_start
+ WHEN TAG IN ('ALTER TABLE') EXECUTE FUNCTION ddl_sleep_once();
+ SET statement_timeout = '15ms';
+ SET enable_ddl_autocommit_retry = off;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+ddl_lock_attempts
+-----------------
+ 2
+(1 row)
+
+setval
+------
+ 1
+(1 row)
+
+step s2_ddl_alter2: ALTER TABLE retry_tbl ADD COLUMN extra2 int DEFAULT 99;
+ERROR: canceling statement due to statement timeout
+step s2_check_seq: SELECT last_value FROM retry_attempt_seq;
+last_value
+----------
+ 1
+(1 row)
+
+step s2_ddl_stmttimeout_enable:
+ SET enable_ddl_autocommit_retry = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+
+setval
+------
+ 1
+(1 row)
+
+step s2_ddl_alter2: ALTER TABLE retry_tbl ADD COLUMN extra2 int DEFAULT 99;
+step s2_ddl_cleanup:
+ SET statement_timeout = 0;
+ SELECT last_value AS ddl_stmttimeout_attempts FROM retry_attempt_seq;
+ SELECT id, val, extra, extra2 FROM retry_tbl ORDER BY id;
+ DROP EVENT TRIGGER ddl_stmt_timeout_trig;
+ DROP FUNCTION ddl_sleep_once();
+
+ddl_stmttimeout_attempts
+------------------------
+ 2
+(1 row)
+
+id|val|extra|extra2
+--+---+-----+------
+ 1| 10| 42| 99
+ 2| 20| 42| 99
+(2 rows)
+
diff --git a/src/test/isolation/isolation_schedule b/src/test/isolation/isolation_schedule
index bf9a2037c70..2b747f17be9 100644
--- a/src/test/isolation/isolation_schedule
+++ b/src/test/isolation/isolation_schedule
@@ -132,3 +132,4 @@ test: ddl-dependency-locking
test: tablespace-dependency-locking
test: pub-concurrent-drop
test: drop-owned-grant
+test: autocommit-retry
diff --git a/src/test/isolation/specs/autocommit-retry.spec b/src/test/isolation/specs/autocommit-retry.spec
new file mode 100644
index 00000000000..1ba382daf06
--- /dev/null
+++ b/src/test/isolation/specs/autocommit-retry.spec
@@ -0,0 +1,157 @@
+# Tests for automatic retry of autocommit-mode statements encountering
+# serialization failures, deadlocks, and (optionally) DDL lock/statement
+# timeouts.
+
+setup
+{
+ CREATE TABLE retry_tbl (id int PRIMARY KEY, val int NOT NULL);
+ INSERT INTO retry_tbl VALUES (1, 10), (2, 20);
+
+ CREATE SEQUENCE retry_attempt_seq;
+
+ CREATE FUNCTION fail_n_times(target_failures int, err_sqlstate text)
+ RETURNS int LANGUAGE plpgsql AS $$
+ DECLARE
+ cur_attempt int;
+ BEGIN
+ cur_attempt := nextval('retry_attempt_seq');
+ IF cur_attempt <= target_failures THEN
+ RAISE EXCEPTION 'simulated transient failure on attempt %', cur_attempt
+ USING ERRCODE = err_sqlstate;
+ END IF;
+ RETURN cur_attempt;
+ END;
+ $$;
+
+ CREATE PROCEDURE retry_proc_simple()
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ UPDATE retry_tbl SET val = val + 1 WHERE id = 2;
+ END;
+ $$;
+
+ CREATE PROCEDURE retry_proc_with_commit()
+ LANGUAGE plpgsql AS $$
+ BEGIN
+ UPDATE retry_tbl SET val = val + 1 WHERE id = 1;
+ COMMIT;
+ SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
+ UPDATE retry_tbl SET val = val + 1 WHERE id = 2;
+ END;
+ $$;
+}
+
+teardown
+{
+ DROP PROCEDURE retry_proc_with_commit();
+ DROP PROCEDURE retry_proc_simple();
+ DROP FUNCTION fail_n_times(int, text);
+ DROP SEQUENCE retry_attempt_seq;
+ DROP TABLE retry_tbl;
+}
+
+session s1
+step s1_begin { BEGIN; }
+step s1_upd2 { UPDATE retry_tbl SET val = val + 100 WHERE id = 2; }
+step s1_commit { COMMIT; }
+
+session s2
+setup
+{
+ SET default_transaction_isolation = 'repeatable read';
+ SET autocommit_retry_backoff_base_ms = 1;
+ SET autocommit_retry_backoff_max_ms = 5;
+}
+step s2_retries_0 { SET max_autocommit_retries = 0; }
+step s2_retries_2 { SET max_autocommit_retries = 2; }
+step s2_retries_3 { SET max_autocommit_retries = 3; }
+step s2_upd_ret { UPDATE retry_tbl SET val = val + 1 RETURNING id, val; }
+step s2_begin { BEGIN; }
+step s2_rollback { ROLLBACK; }
+step s2_multistmt { SELECT 1; UPDATE retry_tbl SET val = val + 1 WHERE id = 2; }
+step s2_call_simple { CALL retry_proc_simple(); }
+step s2_call_commit { CALL retry_proc_with_commit(); }
+step s2_check_tbl { SELECT id, val FROM retry_tbl ORDER BY id; }
+step s2_reset_seq
+{
+ SET log_autocommit_retries = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+}
+step s2_test_40001 { SELECT fail_n_times(2, '40001') AS attempts_40001; }
+step s2_test_40p01 { SELECT fail_n_times(2, '40P01') AS attempts_40p01; }
+step s2_exhaust { SELECT fail_n_times(3, '40001') AS should_fail; }
+step s2_check_seq { SELECT last_value FROM retry_attempt_seq; }
+step s2_ddl_setup
+{
+ CREATE FUNCTION ddl_fail_once() RETURNS event_trigger LANGUAGE plpgsql AS $$
+ BEGIN
+ IF nextval('retry_attempt_seq') <= 1 THEN
+ RAISE EXCEPTION 'simulated DDL lock timeout' USING ERRCODE = '55P03';
+ END IF;
+ END;
+ $$;
+ CREATE EVENT TRIGGER ddl_retry_trig ON ddl_command_start
+ WHEN TAG IN ('ALTER TABLE') EXECUTE FUNCTION ddl_fail_once();
+ SET enable_ddl_autocommit_retry = off;
+ SELECT setval('retry_attempt_seq', 1, false);
+}
+step s2_ddl_alter { ALTER TABLE retry_tbl ADD COLUMN extra int DEFAULT 42; }
+step s2_ddl_enable
+{
+ SET enable_ddl_autocommit_retry = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+}
+step s2_ddl_stmttimeout_setup
+{
+ SELECT last_value AS ddl_lock_attempts FROM retry_attempt_seq;
+ DROP EVENT TRIGGER ddl_retry_trig;
+ DROP FUNCTION ddl_fail_once();
+
+ CREATE FUNCTION ddl_sleep_once() RETURNS event_trigger LANGUAGE plpgsql AS $$
+ BEGIN
+ IF nextval('retry_attempt_seq') <= 1 THEN
+ PERFORM pg_sleep(1);
+ END IF;
+ END;
+ $$;
+ CREATE EVENT TRIGGER ddl_stmt_timeout_trig ON ddl_command_start
+ WHEN TAG IN ('ALTER TABLE') EXECUTE FUNCTION ddl_sleep_once();
+ SET statement_timeout = '15ms';
+ SET enable_ddl_autocommit_retry = off;
+ SELECT setval('retry_attempt_seq', 1, false);
+}
+step s2_ddl_alter2 { ALTER TABLE retry_tbl ADD COLUMN extra2 int DEFAULT 99; }
+step s2_ddl_stmttimeout_enable
+{
+ SET enable_ddl_autocommit_retry = on;
+ SELECT setval('retry_attempt_seq', 1, false);
+}
+step s2_ddl_cleanup
+{
+ SET statement_timeout = 0;
+ SELECT last_value AS ddl_stmttimeout_attempts FROM retry_attempt_seq;
+ SELECT id, val, extra, extra2 FROM retry_tbl ORDER BY id;
+ DROP EVENT TRIGGER ddl_stmt_timeout_trig;
+ DROP FUNCTION ddl_sleep_once();
+}
+
+# 1. Default max_autocommit_retries = 0: serialization failure is reported
+permutation s2_retries_0 s1_begin s1_upd2 s2_upd_ret s1_commit s2_check_tbl
+
+# 2. Autocommit UPDATE ... RETURNING retries on 40001 and discards buffered rows from attempt 1
+permutation s2_retries_3 s1_begin s1_upd2 s2_upd_ret s1_commit s2_check_tbl
+
+# 3. Explicit transaction block (BEGIN ... COMMIT) does not auto-retry
+permutation s2_retries_3 s1_begin s1_upd2 s2_begin s2_upd_ret s1_commit s2_rollback s2_check_tbl
+
+# 4. Multi-statement simple query does not auto-retry
+permutation s2_retries_3 s1_begin s1_upd2 s2_multistmt s1_commit s2_check_tbl
+
+# 5. CALL procedure without internal transaction control auto-retries on 40001
+permutation s2_retries_3 s1_begin s1_upd2 s2_call_simple s1_commit s2_check_tbl
+
+# 6. CALL procedure that commits internally before failing does not auto-retry
+permutation s2_retries_3 s1_begin s1_upd2 s2_call_commit s1_commit s2_check_tbl
+
+# 7. Deterministic retry counts for 40001, 40P01, retry exhaustion, DDL 55P03, and DDL statement_timeout
+permutation s2_retries_2 s2_reset_seq s2_test_40001 s2_reset_seq s2_test_40p01 s2_reset_seq s2_exhaust s2_check_seq s2_ddl_setup s2_ddl_alter s2_check_seq s2_ddl_enable s2_ddl_alter s2_ddl_stmttimeout_setup s2_ddl_alter2 s2_check_seq s2_ddl_stmttimeout_enable s2_ddl_alter2 s2_ddl_cleanup
diff --git a/src/test/regress/expected/psql.out b/src/test/regress/expected/psql.out
index e8605dd041f..1c0fa8f6d7f 100644
--- a/src/test/regress/expected/psql.out
+++ b/src/test/regress/expected/psql.out
@@ -7012,3 +7012,55 @@ Policies |
\pset null ''
DROP TABLE defprivs;
+-- Test autocommit retry over Simple and Extended Query protocol (\bind)
+-- including discarding buffered partial DataRow messages on retry.
+CREATE SEQUENCE retry_ext_seq;
+CREATE FUNCTION retry_ext_fn(n int) RETURNS SETOF int LANGUAGE plpgsql AS $$
+DECLARE
+ attempt bigint;
+BEGIN
+ attempt := nextval('retry_ext_seq');
+ RETURN NEXT 100 + attempt::int;
+ IF attempt <= n THEN
+ RAISE SQLSTATE '40001' USING MESSAGE = 'simulated serialization failure';
+ END IF;
+ RETURN NEXT 200 + attempt::int;
+END;
+$$;
+SET max_autocommit_retries = 2;
+SET autocommit_retry_backoff_base_ms = 1;
+SET autocommit_retry_backoff_max_ms = 5;
+SELECT * FROM retry_ext_fn(2);
+ retry_ext_fn
+--------------
+ 103
+ 203
+(2 rows)
+
+SELECT setval('retry_ext_seq', 1, false);
+ setval
+--------
+ 1
+(1 row)
+
+SELECT * FROM retry_ext_fn($1) \bind 2 \g
+ retry_ext_fn
+--------------
+ 103
+ 203
+(2 rows)
+
+SELECT setval('retry_ext_seq', 1, false);
+ setval
+--------
+ 1
+(1 row)
+
+SELECT * FROM retry_ext_fn($1) \bind 3 \g
+ERROR: simulated serialization failure
+CONTEXT: PL/pgSQL function retry_ext_fn(integer) line 8 at RAISE
+RESET max_autocommit_retries;
+RESET autocommit_retry_backoff_base_ms;
+RESET autocommit_retry_backoff_max_ms;
+DROP FUNCTION retry_ext_fn(int);
+DROP SEQUENCE retry_ext_seq;
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out
index 1e327c2afa4..10297a181a2 100644
--- a/src/test/regress/expected/sysviews.out
+++ b/src/test/regress/expected/sysviews.out
@@ -157,6 +157,7 @@ select name, setting from pg_settings where name like 'enable%';
--------------------------------+---------
enable_async_append | on
enable_bitmapscan | on
+ enable_ddl_autocommit_retry | off
enable_distinct_reordering | on
enable_eager_aggregate | on
enable_gathermerge | on
@@ -181,7 +182,7 @@ select name, setting from pg_settings where name like 'enable%';
enable_seqscan | on
enable_sort | on
enable_tidscan | on
-(26 rows)
+(27 rows)
-- There are always wait event descriptions for various types. InjectionPoint
-- may be present or absent, depending on history since last postmaster start.
diff --git a/src/test/regress/sql/psql.sql b/src/test/regress/sql/psql.sql
index cd685b43e0f..7dd9012dffc 100644
--- a/src/test/regress/sql/psql.sql
+++ b/src/test/regress/sql/psql.sql
@@ -1991,3 +1991,33 @@ CREATE TABLE defprivs (a int);
\zx defprivs
\pset null ''
DROP TABLE defprivs;
+
+-- Test autocommit retry over Simple and Extended Query protocol (\bind)
+-- including discarding buffered partial DataRow messages on retry.
+CREATE SEQUENCE retry_ext_seq;
+CREATE FUNCTION retry_ext_fn(n int) RETURNS SETOF int LANGUAGE plpgsql AS $$
+DECLARE
+ attempt bigint;
+BEGIN
+ attempt := nextval('retry_ext_seq');
+ RETURN NEXT 100 + attempt::int;
+ IF attempt <= n THEN
+ RAISE SQLSTATE '40001' USING MESSAGE = 'simulated serialization failure';
+ END IF;
+ RETURN NEXT 200 + attempt::int;
+END;
+$$;
+SET max_autocommit_retries = 2;
+SET autocommit_retry_backoff_base_ms = 1;
+SET autocommit_retry_backoff_max_ms = 5;
+SELECT * FROM retry_ext_fn(2);
+SELECT setval('retry_ext_seq', 1, false);
+SELECT * FROM retry_ext_fn($1) \bind 2 \g
+SELECT setval('retry_ext_seq', 1, false);
+SELECT * FROM retry_ext_fn($1) \bind 3 \g
+RESET max_autocommit_retries;
+RESET autocommit_retry_backoff_base_ms;
+RESET autocommit_retry_backoff_max_ms;
+DROP FUNCTION retry_ext_fn(int);
+DROP SEQUENCE retry_ext_seq;
+
--
2.56.0.212.g129ede20d9-goog
