Hi, On Sat, Oct 4, 2025 at 9:35 AM Xuneng Zhou <[email protected]> wrote: > > Hi, > > On Sun, Sep 28, 2025 at 5:02 PM Xuneng Zhou <[email protected]> wrote: > > > > Hi, > > > > On Fri, Sep 26, 2025 at 7:22 PM Xuneng Zhou <[email protected]> wrote: > > > > > > Hi Álvaro, > > > > > > Thanks for your review. > > > > > > On Tue, Sep 16, 2025 at 4:24 AM Álvaro Herrera <[email protected]> > > > wrote: > > > > > > > > On 2025-Sep-15, Alexander Korotkov wrote: > > > > > > > > > > It's LGTM. The same pattern is observed in VACUUM, EXPLAIN, and > > > > > > CREATE > > > > > > PUBLICATION - all use minimal grammar rules that produce generic > > > > > > option lists, with the actual interpretation done in their > > > > > > respective > > > > > > implementation files. The moderate complexity in wait.c seems > > > > > > acceptable. > > > > > > > > Actually I find the code in ExecWaitStmt pretty unusual. We tend to use > > > > lists of DefElem (a name optionally followed by a value) instead of > > > > individual scattered elements that must later be matched up. Why not > > > > use utility_option_list instead and then loop on the list of DefElems? > > > > It'd be a lot simpler. > > > > > > I took a look at commands like VACUUM and EXPLAIN and they do follow > > > this pattern. v11 will make use of utility_option_list. > > > > > > > Also, we've found that failing to surround the options by parens leads > > > > to pain down the road, so maybe add that. Given that the LSN seems to > > > > be mandatory, maybe make it something like > > > > > > > > WAIT FOR LSN 'xy/zzy' [ WITH ( utility_option_list ) ] > > > > > > > > This requires that you make LSN a keyword, albeit unreserved. Or you > > > > could make it > > > > WAIT FOR Ident [the rest] > > > > and then ensure in C that the identifier matches the word LSN, such as > > > > we do for "permissive" and "restrictive" in > > > > RowSecurityDefaultPermissive. > > > > > > Shall make LSN an unreserved keyword as well. > > > > Here's the updated v11. Many thanks Jian for off-list discussions and > > review. > > v12 removed unused > +WaitStmt > +WaitStmtParam in pgindent/typedefs.list. >
Hi, I’ve split the patch into multiple patch sets for easier review, per Michael’s advice [1]. [1] https://www.postgresql.org/message-id/aOMsv9TszlB1n-W7%40paquier.xyz Best, Xuneng
From c3dd9972d8043c07247bb3e2b476026268ee1bad Mon Sep 17 00:00:00 2001 From: alterego655 <[email protected]> Date: Tue, 14 Oct 2025 20:50:04 +0800 Subject: [PATCH v13 3/3] Implement WAIT FOR command WAIT FOR is to be used on standby and specifies waiting for the specific WAL location to be replayed. This option is useful when the user makes some data changes on primary and needs a guarantee to see these changes are on standby. WAIT FOR needs to wait without any snapshot held. Otherwise, the snapshot could prevent the replay of WAL records, implying a kind of self-deadlock. This is why separate utility command seems appears to be the most robust way to implement this functionality. It's not possible to implement this as a function. Previous experience shows that stored procedures also have limitation in this aspect. Discussion: https://www.postgresql.org/message-id/flat/CAPpHfdsjtZLVzxjGT8rJHCYbM0D5dwkO+BBjcirozJ6nYbOW8Q@mail.gmail.com https://www.postgresql.org/message-id/flat/CABPTF7UNft368x-RgOXkfj475OwEbp%2BVVO-wEXz7StgjD_%3D6sw%40mail.gmail.com Author: Kartyshov Ivan <[email protected]> Author: Alexander Korotkov <[email protected]> Co-authored-by: Xuneng Zhou <[email protected]> Reviewed-by: Michael Paquier <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Dilip Kumar <[email protected]> Reviewed-by: Amit Kapila <[email protected]> Reviewed-by: Alexander Lakhin <[email protected]> Reviewed-by: Bharath Rupireddy <[email protected]> Reviewed-by: Euler Taveira <[email protected]> Reviewed-by: Heikki Linnakangas <[email protected]> Reviewed-by: Kyotaro Horiguchi <[email protected]> Reviewed-by: jian he <[email protected]> --- doc/src/sgml/high-availability.sgml | 54 ++++ doc/src/sgml/ref/allfiles.sgml | 1 + doc/src/sgml/ref/wait_for.sgml | 234 +++++++++++++++++ doc/src/sgml/reference.sgml | 1 + src/backend/access/transam/xact.c | 6 + src/backend/access/transam/xlog.c | 7 + src/backend/access/transam/xlogrecovery.c | 11 + src/backend/access/transam/xlogwait.c | 27 +- src/backend/commands/Makefile | 3 +- src/backend/commands/meson.build | 1 + src/backend/commands/wait.c | 212 ++++++++++++++++ src/backend/parser/gram.y | 33 ++- src/backend/storage/lmgr/proc.c | 5 + src/backend/tcop/pquery.c | 12 +- src/backend/tcop/utility.c | 22 ++ src/include/access/xlogwait.h | 3 +- src/include/commands/wait.h | 22 ++ src/include/nodes/parsenodes.h | 8 + src/include/parser/kwlist.h | 2 + src/include/tcop/cmdtaglist.h | 1 + src/test/recovery/meson.build | 3 +- src/test/recovery/t/049_wait_for_lsn.pl | 293 ++++++++++++++++++++++ src/tools/pgindent/typedefs.list | 3 + 23 files changed, 951 insertions(+), 13 deletions(-) create mode 100644 doc/src/sgml/ref/wait_for.sgml create mode 100644 src/backend/commands/wait.c create mode 100644 src/include/commands/wait.h create mode 100644 src/test/recovery/t/049_wait_for_lsn.pl diff --git a/doc/src/sgml/high-availability.sgml b/doc/src/sgml/high-availability.sgml index b47d8b4106e..b3fafb8b48c 100644 --- a/doc/src/sgml/high-availability.sgml +++ b/doc/src/sgml/high-availability.sgml @@ -1376,6 +1376,60 @@ synchronous_standby_names = 'ANY 2 (s1, s2, s3)' </sect3> </sect2> + <sect2 id="read-your-writes-consistency"> + <title>Read-Your-Writes Consistency</title> + + <para> + In asynchronous replication, there is always a short window where changes + on the primary may not yet be visible on the standby due to replication + lag. This can lead to inconsistencies when an application writes data on + the primary and then immediately issues a read query on the standby. + However, it is possible to address this without switching to synchronous + replication. + </para> + + <para> + To address this, PostgreSQL offers a mechanism for read-your-writes + consistency. The key idea is to ensure that a client sees its own writes + by synchronizing the WAL replay on the standby with the known point of + change on the primary. + </para> + + <para> + This is achieved by the following steps. After performing write + operations, the application retrieves the current WAL location using a + function call like this. + + <programlisting> +postgres=# SELECT pg_current_wal_insert_lsn(); +pg_current_wal_insert_lsn +-------------------- +0/306EE20 +(1 row) + </programlisting> + </para> + + <para> + The <acronym>LSN</acronym> obtained from the primary is then communicated + to the standby server. This can be managed at the application level or + via the connection pooler. On the standby, the application issues the + <xref linkend="sql-wait-for"/> command to block further processing until + the standby's WAL replay process reaches (or exceeds) the specified + <acronym>LSN</acronym>. + + <programlisting> +postgres=# WAIT FOR LSN '0/306EE20'; + RESULT STATUS +--------------- + success +(1 row) + </programlisting> + Once the command returns a status of success, it guarantees that all + changes up to the provided <acronym>LSN</acronym> have been applied, + ensuring that subsequent read queries will reflect the latest updates. + </para> + </sect2> + <sect2 id="continuous-archiving-in-standby"> <title>Continuous Archiving in Standby</title> diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml index f5be638867a..e167406c744 100644 --- a/doc/src/sgml/ref/allfiles.sgml +++ b/doc/src/sgml/ref/allfiles.sgml @@ -188,6 +188,7 @@ Complete list of usable sgml source files in this directory. <!ENTITY update SYSTEM "update.sgml"> <!ENTITY vacuum SYSTEM "vacuum.sgml"> <!ENTITY values SYSTEM "values.sgml"> +<!ENTITY waitFor SYSTEM "wait_for.sgml"> <!-- applications and utilities --> <!ENTITY clusterdb SYSTEM "clusterdb.sgml"> diff --git a/doc/src/sgml/ref/wait_for.sgml b/doc/src/sgml/ref/wait_for.sgml new file mode 100644 index 00000000000..8df1f2ab953 --- /dev/null +++ b/doc/src/sgml/ref/wait_for.sgml @@ -0,0 +1,234 @@ +<!-- +doc/src/sgml/ref/wait_for.sgml +PostgreSQL documentation +--> + +<refentry id="sql-wait-for"> + <indexterm zone="sql-wait-for"> + <primary>WAIT FOR</primary> + </indexterm> + + <refmeta> + <refentrytitle>WAIT FOR</refentrytitle> + <manvolnum>7</manvolnum> + <refmiscinfo>SQL - Language Statements</refmiscinfo> + </refmeta> + + <refnamediv> + <refname>WAIT FOR</refname> + <refpurpose>wait for target <acronym>LSN</acronym> to be replayed, optionally with a timeout</refpurpose> + </refnamediv> + + <refsynopsisdiv> +<synopsis> +WAIT FOR LSN '<replaceable class="parameter">lsn</replaceable>' [ [WITH] ( <replaceable class="parameter">option</replaceable> [, ...] ) ] + +<phrase>where <replaceable class="parameter">option</replaceable> can be:</phrase> + + TIMEOUT '<replaceable class="parameter">timeout</replaceable>' + NO_THROW +</synopsis> + </refsynopsisdiv> + + <refsect1> + <title>Description</title> + + <para> + Waits until recovery replays <parameter>lsn</parameter>. + If no <parameter>timeout</parameter> is specified or it is set to + zero, this command waits indefinitely for the + <parameter>lsn</parameter>. + On timeout, or if the server is promoted before + <parameter>lsn</parameter> is reached, an error is emitted, + unless <literal>NO_THROW</literal> is specified in the WITH clause. + If <parameter>NO_THROW</parameter> is specified, then the command + doesn't throw errors. + </para> + + <para> + The possible return values are <literal>success</literal>, + <literal>timeout</literal>, and <literal>not in recovery</literal>. + </para> + </refsect1> + + <refsect1> + <title>Parameters</title> + + <variablelist> + <varlistentry> + <term><replaceable class="parameter">lsn</replaceable></term> + <listitem> + <para> + Specifies the target <acronym>LSN</acronym> to wait for. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>WITH ( <replaceable class="parameter">option</replaceable> [, ...] )</literal></term> + <listitem> + <para> + This clause specifies optional parameters for the wait operation. + The following parameters are supported: + + <variablelist> + <varlistentry> + <term><literal>TIMEOUT</literal> '<replaceable class="parameter">timeout</replaceable>'</term> + <listitem> + <para> + When specified and <parameter>timeout</parameter> is greater than zero, + the command waits until <parameter>lsn</parameter> is reached or + the specified <parameter>timeout</parameter> has elapsed. + </para> + <para> + The <parameter>timeout</parameter> might be given as integer number of + milliseconds. Also it might be given as string literal with + integer number of milliseconds or a number with unit + (see <xref linkend="config-setting-names-values"/>). + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>NO_THROW</literal></term> + <listitem> + <para> + Specify to not throw an error in the case of timeout or + running on the primary. In this case the result status can be get from + the return value. + </para> + </listitem> + </varlistentry> + </variablelist> + </para> + </listitem> + </varlistentry> + </variablelist> + </refsect1> + + <refsect1> + <title>Outputs</title> + + <variablelist> + <varlistentry> + <term><literal>success</literal></term> + <listitem> + <para> + This return value denotes that we have successfully reached + the target <parameter>lsn</parameter>. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>timeout</literal></term> + <listitem> + <para> + This return value denotes that the timeout happened before reaching + the target <parameter>lsn</parameter>. + </para> + </listitem> + </varlistentry> + + <varlistentry> + <term><literal>not in recovery</literal></term> + <listitem> + <para> + This return value denotes that the database server is not in a recovery + state. This might mean either the database server was not in recovery + at the moment of receiving the command, or it was promoted before + reaching the target <parameter>lsn</parameter>. + </para> + </listitem> + </varlistentry> + </variablelist> + </refsect1> + + <refsect1> + <title>Notes</title> + + <para> + <command>WAIT FOR</command> command waits till + <parameter>lsn</parameter> to be replayed on standby. + That is, after this command execution, the value returned by + <function>pg_last_wal_replay_lsn</function> should be greater or equal + to the <parameter>lsn</parameter> value. This is useful to achieve + read-your-writes-consistency, while using async replica for reads and + primary for writes. In that case, the <acronym>lsn</acronym> of the last + modification should be stored on the client application side or the + connection pooler side. + </para> + + <para> + <command>WAIT FOR</command> command should be called on standby. + If a user runs <command>WAIT FOR</command> on primary, it + will error out unless <parameter>NO_THROW</parameter> is specified in the WITH clause. + However, if <command>WAIT FOR</command> is + called on primary promoted from standby and <literal>lsn</literal> + was already replayed, then the <command>WAIT FOR</command> command just + exits immediately. + </para> + +</refsect1> + + <refsect1> + <title>Examples</title> + + <para> + You can use <command>WAIT FOR</command> command to wait for + the <type>pg_lsn</type> value. For example, an application could update + the <literal>movie</literal> table and get the <acronym>lsn</acronym> after + changes just made. This example uses <function>pg_current_wal_insert_lsn</function> + on primary server to get the <acronym>lsn</acronym> given that + <varname>synchronous_commit</varname> could be set to + <literal>off</literal>. + + <programlisting> +postgres=# UPDATE movie SET genre = 'Dramatic' WHERE genre = 'Drama'; +UPDATE 100 +postgres=# SELECT pg_current_wal_insert_lsn(); +pg_current_wal_insert_lsn +-------------------- +0/306EE20 +(1 row) +</programlisting> + + Then an application could run <command>WAIT FOR</command> + with the <parameter>lsn</parameter> obtained from primary. After that the + changes made on primary should be guaranteed to be visible on replica. + +<programlisting> +postgres=# WAIT FOR LSN '0/306EE20'; + status +-------- + success +(1 row) +postgres=# SELECT * FROM movie WHERE genre = 'Drama'; + genre +------- +(0 rows) +</programlisting> + </para> + + <para> + If the target LSN is not reached before the timeout, the error is thrown. + +<programlisting> +postgres=# WAIT FOR LSN '0/306EE20' WITH (TIMEOUT '0.1s'); +ERROR: timed out while waiting for target LSN 0/306EE20 to be replayed; current replay LSN 0/306EA60 +</programlisting> + </para> + + <para> + The same example uses <command>WAIT FOR</command> with + <parameter>NO_THROW</parameter> option. +<programlisting> +postgres=# WAIT FOR LSN '0/306EE20' WITH (TIMEOUT '100ms', NO_THROW); + status +-------- + timeout +(1 row) +</programlisting> + </para> + </refsect1> +</refentry> diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml index ff85ace83fc..2cf02c37b17 100644 --- a/doc/src/sgml/reference.sgml +++ b/doc/src/sgml/reference.sgml @@ -216,6 +216,7 @@ &update; &vacuum; &values; + &waitFor; </reference> diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 2cf3d4e92b7..092e197eba3 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -31,6 +31,7 @@ #include "access/xloginsert.h" #include "access/xlogrecovery.h" #include "access/xlogutils.h" +#include "access/xlogwait.h" #include "catalog/index.h" #include "catalog/namespace.h" #include "catalog/pg_enum.h" @@ -2843,6 +2844,11 @@ AbortTransaction(void) */ LWLockReleaseAll(); + /* + * Cleanup waiting for LSN if any. + */ + WaitLSNCleanup(); + /* Clear wait information and command progress indicator */ pgstat_report_wait_end(); pgstat_progress_end_command(); diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index eceab341255..b5e07a724f5 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -62,6 +62,7 @@ #include "access/xlogreader.h" #include "access/xlogrecovery.h" #include "access/xlogutils.h" +#include "access/xlogwait.h" #include "backup/basebackup.h" #include "catalog/catversion.h" #include "catalog/pg_control.h" @@ -6225,6 +6226,12 @@ StartupXLOG(void) UpdateControlFile(); LWLockRelease(ControlFileLock); + /* + * Wake up all waiters for replay LSN. They need to report an error that + * recovery was ended before reaching the target LSN. + */ + WaitLSNWakeupReplay(InvalidXLogRecPtr); + /* * Shutdown the recovery environment. This must occur after * RecoverPreparedTransactions() (see notes in lock_twophase_recover()) diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 52ff4d119e6..1859d2084e8 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -40,6 +40,7 @@ #include "access/xlogreader.h" #include "access/xlogrecovery.h" #include "access/xlogutils.h" +#include "access/xlogwait.h" #include "backup/basebackup.h" #include "catalog/pg_control.h" #include "commands/tablespace.h" @@ -1838,6 +1839,16 @@ PerformWalRecovery(void) break; } + /* + * If we replayed an LSN that someone was waiting for then walk + * over the shared memory array and set latches to notify the + * waiters. + */ + if (waitLSNState && + (XLogRecoveryCtl->lastReplayedEndRecPtr >= + pg_atomic_read_u64(&waitLSNState->minWaitedReplayLSN))) + WaitLSNWakeupReplay(XLogRecoveryCtl->lastReplayedEndRecPtr); + /* Else, try to fetch the next WAL record */ record = ReadRecord(xlogprefetcher, LOG, false, replayTLI); } while (record != NULL); diff --git a/src/backend/access/transam/xlogwait.c b/src/backend/access/transam/xlogwait.c index a114738bddf..7c8134f1209 100644 --- a/src/backend/access/transam/xlogwait.c +++ b/src/backend/access/transam/xlogwait.c @@ -373,9 +373,10 @@ WaitLSNCleanup(void) * or replica got promoted before the target LSN replayed. */ WaitLSNResult -WaitForLSNReplay(XLogRecPtr targetLSN) +WaitForLSNReplay(XLogRecPtr targetLSN, int64 timeout) { XLogRecPtr currentLSN; + TimestampTz endtime = 0; int wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH; /* Shouldn't be called when shmem isn't initialized */ @@ -404,6 +405,12 @@ WaitForLSNReplay(XLogRecPtr targetLSN) return WAIT_LSN_RESULT_SUCCESS; } + if (timeout > 0) + { + endtime = TimestampTzPlusMilliseconds(GetCurrentTimestamp(), timeout); + wake_events |= WL_TIMEOUT; + } + /* * Add our process to the replay waiters heap. It might happen that * target LSN gets replayed before we do. Another check at the beginning @@ -438,6 +445,18 @@ WaitForLSNReplay(XLogRecPtr targetLSN) break; } + /* + * If the timeout value is specified, calculate the number of + * milliseconds before the timeout. Exit if the timeout is already + * reached. + */ + if (timeout > 0) + { + delay_ms = TimestampDifferenceMilliseconds(GetCurrentTimestamp(), endtime); + if (delay_ms <= 0) + break; + } + CHECK_FOR_INTERRUPTS(); rc = WaitLatch(MyLatch, wake_events, delay_ms, @@ -464,6 +483,12 @@ WaitForLSNReplay(XLogRecPtr targetLSN) */ deleteLSNWaiter(WAIT_LSN_REPLAY); + /* + * If we didn't reach the target LSN, we must be exited by timeout. + */ + if (targetLSN > currentLSN) + return WAIT_LSN_RESULT_TIMEOUT; + return WAIT_LSN_RESULT_SUCCESS; } diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile index cb2fbdc7c60..f99acfd2b4b 100644 --- a/src/backend/commands/Makefile +++ b/src/backend/commands/Makefile @@ -64,6 +64,7 @@ OBJS = \ vacuum.o \ vacuumparallel.o \ variable.o \ - view.o + view.o \ + wait.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/commands/meson.build b/src/backend/commands/meson.build index dd4cde41d32..9f640ad4810 100644 --- a/src/backend/commands/meson.build +++ b/src/backend/commands/meson.build @@ -53,4 +53,5 @@ backend_sources += files( 'vacuumparallel.c', 'variable.c', 'view.c', + 'wait.c', ) diff --git a/src/backend/commands/wait.c b/src/backend/commands/wait.c new file mode 100644 index 00000000000..44db2d71164 --- /dev/null +++ b/src/backend/commands/wait.c @@ -0,0 +1,212 @@ +/*------------------------------------------------------------------------- + * + * wait.c + * Implements WAIT FOR, which allows waiting for events such as + * time passing or LSN having been replayed on replica. + * + * Portions Copyright (c) 2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/commands/wait.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include <math.h> + +#include "access/xlogrecovery.h" +#include "access/xlogwait.h" +#include "commands/defrem.h" +#include "commands/wait.h" +#include "executor/executor.h" +#include "parser/parse_node.h" +#include "storage/proc.h" +#include "utils/builtins.h" +#include "utils/guc.h" +#include "utils/pg_lsn.h" +#include "utils/snapmgr.h" + + +void +ExecWaitStmt(ParseState *pstate, WaitStmt *stmt, DestReceiver *dest) +{ + XLogRecPtr lsn; + int64 timeout = 0; + WaitLSNResult waitLSNResult; + bool throw = true; + TupleDesc tupdesc; + TupOutputState *tstate; + const char *result = "<unset>"; + bool timeout_specified = false; + bool no_throw_specified = false; + + /* Parse and validate the mandatory LSN */ + lsn = DatumGetLSN(DirectFunctionCall1(pg_lsn_in, + CStringGetDatum(stmt->lsn_literal))); + + foreach_node(DefElem, defel, stmt->options) + { + if (strcmp(defel->defname, "timeout") == 0) + { + char *timeout_str; + const char *hintmsg; + double result; + + if (timeout_specified) + errorConflictingDefElem(defel, pstate); + timeout_specified = true; + + timeout_str = defGetString(defel); + + if (!parse_real(timeout_str, &result, GUC_UNIT_MS, &hintmsg)) + { + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("invalid timeout value: \"%s\"", timeout_str), + hintmsg ? errhint("%s", _(hintmsg)) : 0); + } + + /* + * Get rid of any fractional part in the input. This is so we + * don't fail on just-out-of-range values that would round + * into range. + */ + result = rint(result); + + /* Range check */ + if (unlikely(isnan(result) || !FLOAT8_FITS_IN_INT64(result))) + ereport(ERROR, + errcode(ERRCODE_NUMERIC_VALUE_OUT_OF_RANGE), + errmsg("timeout value is out of range")); + + if (result < 0) + ereport(ERROR, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("timeout cannot be negative")); + + timeout = (int64) result; + } + else if (strcmp(defel->defname, "no_throw") == 0) + { + if (no_throw_specified) + errorConflictingDefElem(defel, pstate); + + no_throw_specified = true; + + throw = !defGetBoolean(defel); + } + else + { + ereport(ERROR, + errcode(ERRCODE_SYNTAX_ERROR), + errmsg("option \"%s\" not recognized", + defel->defname), + parser_errposition(pstate, defel->location)); + } + } + + /* + * We are going to wait for the LSN replay. We should first care that we + * don't hold a snapshot and correspondingly our MyProc->xmin is invalid. + * Otherwise, our snapshot could prevent the replay of WAL records + * implying a kind of self-deadlock. This is the reason why + * WAIT FOR is a command, not a procedure or function. + * + * At first, we should check there is no active snapshot. According to + * PlannedStmtRequiresSnapshot(), even in an atomic context, CallStmt is + * processed with a snapshot. Thankfully, we can pop this snapshot, + * because PortalRunUtility() can tolerate this. + */ + if (ActiveSnapshotSet()) + PopActiveSnapshot(); + + /* + * At second, invalidate a catalog snapshot if any. And we should be done + * with the preparation. + */ + InvalidateCatalogSnapshot(); + + /* Give up if there is still an active or registered snapshot. */ + if (HaveRegisteredOrActiveSnapshot()) + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("WAIT FOR must be only called without an active or registered snapshot"), + errdetail("WAIT FOR cannot be executed from a function or a procedure or within a transaction with an isolation level higher than READ COMMITTED.")); + + /* + * As the result we should hold no snapshot, and correspondingly our xmin + * should be unset. + */ + Assert(MyProc->xmin == InvalidTransactionId); + + waitLSNResult = WaitForLSNReplay(lsn, timeout); + + /* + * Process the result of WaitForLSNReplay(). Throw appropriate error if + * needed. + */ + switch (waitLSNResult) + { + case WAIT_LSN_RESULT_SUCCESS: + /* Nothing to do on success */ + result = "success"; + break; + + case WAIT_LSN_RESULT_TIMEOUT: + if (throw) + ereport(ERROR, + errcode(ERRCODE_QUERY_CANCELED), + errmsg("timed out while waiting for target LSN %X/%08X to be replayed; current replay LSN %X/%08X", + LSN_FORMAT_ARGS(lsn), + LSN_FORMAT_ARGS(GetXLogReplayRecPtr(NULL)))); + else + result = "timeout"; + break; + + case WAIT_LSN_RESULT_NOT_IN_RECOVERY: + if (throw) + { + if (PromoteIsTriggered()) + { + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("recovery is not in progress"), + errdetail("Recovery ended before replaying target LSN %X/%08X; last replay LSN %X/%08X.", + LSN_FORMAT_ARGS(lsn), + LSN_FORMAT_ARGS(GetXLogReplayRecPtr(NULL)))); + } + else + ereport(ERROR, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("recovery is not in progress"), + errhint("Waiting for the replay LSN can only be executed during recovery.")); + } + else + result = "not in recovery"; + break; + } + + /* need a tuple descriptor representing a single TEXT column */ + tupdesc = WaitStmtResultDesc(stmt); + + /* prepare for projection of tuples */ + tstate = begin_tup_output_tupdesc(dest, tupdesc, &TTSOpsVirtual); + + /* Send it */ + do_text_output_oneline(tstate, result); + + end_tup_output(tstate); +} + +TupleDesc +WaitStmtResultDesc(WaitStmt *stmt) +{ + TupleDesc tupdesc; + + /* Need a tuple descriptor representing a single TEXT column */ + tupdesc = CreateTemplateTupleDesc(1); + TupleDescInitEntry(tupdesc, (AttrNumber) 1, "status", + TEXTOID, -1, 0); + return tupdesc; +} diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y index 21caf2d43bf..1d016df1f6b 100644 --- a/src/backend/parser/gram.y +++ b/src/backend/parser/gram.y @@ -308,7 +308,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); SecLabelStmt SelectStmt TransactionStmt TransactionStmtLegacy TruncateStmt UnlistenStmt UpdateStmt VacuumStmt VariableResetStmt VariableSetStmt VariableShowStmt - ViewStmt CheckPointStmt CreateConversionStmt + ViewStmt WaitStmt CheckPointStmt CreateConversionStmt DeallocateStmt PrepareStmt ExecuteStmt DropOwnedStmt ReassignOwnedStmt AlterTSConfigurationStmt AlterTSDictionaryStmt @@ -325,6 +325,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); %type <boolean> opt_concurrently %type <dbehavior> opt_drop_behavior %type <list> opt_utility_option_list +%type <list> opt_wait_with_clause %type <list> utility_option_list %type <defelt> utility_option_elem %type <str> utility_option_name @@ -678,7 +679,6 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); json_object_constructor_null_clause_opt json_array_constructor_null_clause_opt - /* * Non-keyword token types. These are hard-wired into the "flex" lexer. * They must be listed first so that their numeric codes do not depend on @@ -748,7 +748,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); LABEL LANGUAGE LARGE_P LAST_P LATERAL_P LEADING LEAKPROOF LEAST LEFT LEVEL LIKE LIMIT LISTEN LOAD LOCAL - LOCALTIME LOCALTIMESTAMP LOCATION LOCK_P LOCKED LOGGED + LOCALTIME LOCALTIMESTAMP LOCATION LOCK_P LOCKED LOGGED LSN_P MAPPING MATCH MATCHED MATERIALIZED MAXVALUE MERGE MERGE_ACTION METHOD MINUTE_P MINVALUE MODE MONTH_P MOVE @@ -792,7 +792,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query); VACUUM VALID VALIDATE VALIDATOR VALUE_P VALUES VARCHAR VARIADIC VARYING VERBOSE VERSION_P VIEW VIEWS VIRTUAL VOLATILE - WHEN WHERE WHITESPACE_P WINDOW WITH WITHIN WITHOUT WORK WRAPPER WRITE + WAIT WHEN WHERE WHITESPACE_P WINDOW WITH WITHIN WITHOUT WORK WRAPPER WRITE XML_P XMLATTRIBUTES XMLCONCAT XMLELEMENT XMLEXISTS XMLFOREST XMLNAMESPACES XMLPARSE XMLPI XMLROOT XMLSERIALIZE XMLTABLE @@ -1120,6 +1120,7 @@ stmt: | VariableSetStmt | VariableShowStmt | ViewStmt + | WaitStmt | /*EMPTY*/ { $$ = NULL; } ; @@ -16453,6 +16454,26 @@ xml_passing_mech: | BY VALUE_P ; +/***************************************************************************** + * + * WAIT FOR LSN + * + *****************************************************************************/ + +WaitStmt: + WAIT FOR LSN_P Sconst opt_wait_with_clause + { + WaitStmt *n = makeNode(WaitStmt); + n->lsn_literal = $4; + n->options = $5; + $$ = (Node *) n; + } + ; + +opt_wait_with_clause: + opt_with '(' utility_option_list ')' { $$ = $3; } + | /*EMPTY*/ { $$ = NIL; } + ; /* * Aggregate decoration clauses @@ -17940,6 +17961,7 @@ unreserved_keyword: | LOCK_P | LOCKED | LOGGED + | LSN_P | MAPPING | MATCH | MATCHED @@ -18110,6 +18132,7 @@ unreserved_keyword: | VIEWS | VIRTUAL | VOLATILE + | WAIT | WHITESPACE_P | WITHIN | WITHOUT @@ -18556,6 +18579,7 @@ bare_label_keyword: | LOCK_P | LOCKED | LOGGED + | LSN_P | MAPPING | MATCH | MATCHED @@ -18767,6 +18791,7 @@ bare_label_keyword: | VIEWS | VIRTUAL | VOLATILE + | WAIT | WHEN | WHITESPACE_P | WORK diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index 96f29aafc39..f8685fa9039 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -947,6 +947,11 @@ ProcKill(int code, Datum arg) */ LWLockReleaseAll(); + /* + * Cleanup waiting for LSN if any. + */ + WaitLSNCleanup(); + /* Cancel any pending condition variable sleep, too */ ConditionVariableCancelSleep(); diff --git a/src/backend/tcop/pquery.c b/src/backend/tcop/pquery.c index 08791b8f75e..07b2e2fa67b 100644 --- a/src/backend/tcop/pquery.c +++ b/src/backend/tcop/pquery.c @@ -1163,10 +1163,11 @@ PortalRunUtility(Portal portal, PlannedStmt *pstmt, MemoryContextSwitchTo(portal->portalContext); /* - * Some utility commands (e.g., VACUUM) pop the ActiveSnapshot stack from - * under us, so don't complain if it's now empty. Otherwise, our snapshot - * should be the top one; pop it. Note that this could be a different - * snapshot from the one we made above; see EnsurePortalSnapshotExists. + * Some utility commands (e.g., VACUUM, WAIT FOR) pop the ActiveSnapshot + * stack from under us, so don't complain if it's now empty. Otherwise, + * our snapshot should be the top one; pop it. Note that this could be a + * different snapshot from the one we made above; see + * EnsurePortalSnapshotExists. */ if (portal->portalSnapshot != NULL && ActiveSnapshotSet()) { @@ -1743,7 +1744,8 @@ PlannedStmtRequiresSnapshot(PlannedStmt *pstmt) IsA(utilityStmt, ListenStmt) || IsA(utilityStmt, NotifyStmt) || IsA(utilityStmt, UnlistenStmt) || - IsA(utilityStmt, CheckPointStmt)) + IsA(utilityStmt, CheckPointStmt) || + IsA(utilityStmt, WaitStmt)) return false; return true; diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c index 918db53dd5e..082967c0a86 100644 --- a/src/backend/tcop/utility.c +++ b/src/backend/tcop/utility.c @@ -56,6 +56,7 @@ #include "commands/user.h" #include "commands/vacuum.h" #include "commands/view.h" +#include "commands/wait.h" #include "miscadmin.h" #include "parser/parse_utilcmd.h" #include "postmaster/bgwriter.h" @@ -266,6 +267,7 @@ ClassifyUtilityCommandAsReadOnly(Node *parsetree) case T_PrepareStmt: case T_UnlistenStmt: case T_VariableSetStmt: + case T_WaitStmt: { /* * These modify only backend-local state, so they're OK to run @@ -1055,6 +1057,12 @@ standard_ProcessUtility(PlannedStmt *pstmt, break; } + case T_WaitStmt: + { + ExecWaitStmt(pstate, (WaitStmt *) parsetree, dest); + } + break; + default: /* All other statement types have event trigger support */ ProcessUtilitySlow(pstate, pstmt, queryString, @@ -2059,6 +2067,9 @@ UtilityReturnsTuples(Node *parsetree) case T_VariableShowStmt: return true; + case T_WaitStmt: + return true; + default: return false; } @@ -2114,6 +2125,9 @@ UtilityTupleDescriptor(Node *parsetree) return GetPGVariableResultDesc(n->name); } + case T_WaitStmt: + return WaitStmtResultDesc((WaitStmt *) parsetree); + default: return NULL; } @@ -3091,6 +3105,10 @@ CreateCommandTag(Node *parsetree) } break; + case T_WaitStmt: + tag = CMDTAG_WAIT; + break; + /* already-planned queries */ case T_PlannedStmt: { @@ -3689,6 +3707,10 @@ GetCommandLogLevel(Node *parsetree) lev = LOGSTMT_DDL; break; + case T_WaitStmt: + lev = LOGSTMT_ALL; + break; + /* already-planned queries */ case T_PlannedStmt: { diff --git a/src/include/access/xlogwait.h b/src/include/access/xlogwait.h index 441bf475b4d..2e33a1d22d0 100644 --- a/src/include/access/xlogwait.h +++ b/src/include/access/xlogwait.h @@ -27,6 +27,7 @@ typedef enum WAIT_LSN_RESULT_SUCCESS, /* Target LSN is reached */ WAIT_LSN_RESULT_NOT_IN_RECOVERY, /* Recovery ended before or during our * wait */ + WAIT_LSN_RESULT_TIMEOUT, /* Timeout occurred */ } WaitLSNResult; /* @@ -106,7 +107,7 @@ extern void WaitLSNShmemInit(void); extern void WaitLSNWakeupReplay(XLogRecPtr currentLSN); extern void WaitLSNWakeupFlush(XLogRecPtr currentLSN); extern void WaitLSNCleanup(void); -extern WaitLSNResult WaitForLSNReplay(XLogRecPtr targetLSN); +extern WaitLSNResult WaitForLSNReplay(XLogRecPtr targetLSN, int64 timeout); extern void WaitForLSNFlush(XLogRecPtr targetLSN); #endif /* XLOG_WAIT_H */ diff --git a/src/include/commands/wait.h b/src/include/commands/wait.h new file mode 100644 index 00000000000..ce332134fb3 --- /dev/null +++ b/src/include/commands/wait.h @@ -0,0 +1,22 @@ +/*------------------------------------------------------------------------- + * + * wait.h + * prototypes for commands/wait.c + * + * Portions Copyright (c) 2025, PostgreSQL Global Development Group + * + * src/include/commands/wait.h + * + *------------------------------------------------------------------------- + */ +#ifndef WAIT_H +#define WAIT_H + +#include "nodes/parsenodes.h" +#include "parser/parse_node.h" +#include "tcop/dest.h" + +extern void ExecWaitStmt(ParseState *pstate, WaitStmt *stmt, DestReceiver *dest); +extern TupleDesc WaitStmtResultDesc(WaitStmt *stmt); + +#endif /* WAIT_H */ diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h index dc09d1a3f03..c741099e186 100644 --- a/src/include/nodes/parsenodes.h +++ b/src/include/nodes/parsenodes.h @@ -4384,4 +4384,12 @@ typedef struct DropSubscriptionStmt DropBehavior behavior; /* RESTRICT or CASCADE behavior */ } DropSubscriptionStmt; +typedef struct WaitStmt +{ + NodeTag type; + char *lsn_literal; /* LSN string from grammar */ + List *options; /* List of DefElem nodes */ +} WaitStmt; + + #endif /* PARSENODES_H */ diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h index 84182eaaae2..5d4fe27ef96 100644 --- a/src/include/parser/kwlist.h +++ b/src/include/parser/kwlist.h @@ -270,6 +270,7 @@ PG_KEYWORD("location", LOCATION, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("lock", LOCK_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("locked", LOCKED, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("logged", LOGGED, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("lsn", LSN_P, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("mapping", MAPPING, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("match", MATCH, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("matched", MATCHED, UNRESERVED_KEYWORD, BARE_LABEL) @@ -496,6 +497,7 @@ PG_KEYWORD("view", VIEW, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("views", VIEWS, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("virtual", VIRTUAL, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("volatile", VOLATILE, UNRESERVED_KEYWORD, BARE_LABEL) +PG_KEYWORD("wait", WAIT, UNRESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("when", WHEN, RESERVED_KEYWORD, BARE_LABEL) PG_KEYWORD("where", WHERE, RESERVED_KEYWORD, AS_LABEL) PG_KEYWORD("whitespace", WHITESPACE_P, UNRESERVED_KEYWORD, BARE_LABEL) diff --git a/src/include/tcop/cmdtaglist.h b/src/include/tcop/cmdtaglist.h index d250a714d59..c4606d65043 100644 --- a/src/include/tcop/cmdtaglist.h +++ b/src/include/tcop/cmdtaglist.h @@ -217,3 +217,4 @@ PG_CMDTAG(CMDTAG_TRUNCATE_TABLE, "TRUNCATE TABLE", false, false, false) PG_CMDTAG(CMDTAG_UNLISTEN, "UNLISTEN", false, false, false) PG_CMDTAG(CMDTAG_UPDATE, "UPDATE", false, false, true) PG_CMDTAG(CMDTAG_VACUUM, "VACUUM", false, false, false) +PG_CMDTAG(CMDTAG_WAIT, "WAIT", false, false, false) diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index 52993c32dbb..523a5cd5b52 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -56,7 +56,8 @@ tests += { 't/045_archive_restartpoint.pl', 't/046_checkpoint_logical_slot.pl', 't/047_checkpoint_physical_slot.pl', - 't/048_vacuum_horizon_floor.pl' + 't/048_vacuum_horizon_floor.pl', + 't/049_wait_for_lsn.pl', ], }, } diff --git a/src/test/recovery/t/049_wait_for_lsn.pl b/src/test/recovery/t/049_wait_for_lsn.pl new file mode 100644 index 00000000000..62fdc7cd06c --- /dev/null +++ b/src/test/recovery/t/049_wait_for_lsn.pl @@ -0,0 +1,293 @@ +# Checks waiting for the lsn replay on standby using +# WAIT FOR procedure. +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More; + +# Initialize primary node +my $node_primary = PostgreSQL::Test::Cluster->new('primary'); +$node_primary->init(allows_streaming => 1); +$node_primary->start; + +# And some content and take a backup +$node_primary->safe_psql('postgres', + "CREATE TABLE wait_test AS SELECT generate_series(1,10) AS a"); +my $backup_name = 'my_backup'; +$node_primary->backup($backup_name); + +# Create a streaming standby with a 1 second delay from the backup +my $node_standby = PostgreSQL::Test::Cluster->new('standby'); +my $delay = 1; +$node_standby->init_from_backup($node_primary, $backup_name, + has_streaming => 1); +$node_standby->append_conf( + 'postgresql.conf', qq[ + recovery_min_apply_delay = '${delay}s' +]); +$node_standby->start; + +# 1. Make sure that WAIT FOR works: add new content to +# primary and memorize primary's insert LSN, then wait for that LSN to be +# replayed on standby. +$node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (generate_series(11, 20))"); +my $lsn1 = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); +my $output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn1}' WITH (timeout '1d'); + SELECT pg_lsn_cmp(pg_last_wal_replay_lsn(), '${lsn1}'::pg_lsn); +]); + +# Make sure the current LSN on standby is at least as big as the LSN we +# observed on primary's before. +ok((split("\n", $output))[-1] >= 0, + "standby reached the same LSN as primary after WAIT FOR"); + +# 2. Check that new data is visible after calling WAIT FOR +$node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (generate_series(21, 30))"); +my $lsn2 = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn2}'; + SELECT count(*) FROM wait_test; +]); + +# Make sure the count(*) on standby reflects the recent changes on primary +ok((split("\n", $output))[-1] eq 30, + "standby reached the same LSN as primary"); + +# 3. Check that waiting for unreachable LSN triggers the timeout. The +# unreachable LSN must be well in advance. So WAL records issued by +# the concurrent autovacuum could not affect that. +my $lsn3 = + $node_primary->safe_psql('postgres', + "SELECT pg_current_wal_insert_lsn() + 10000000000"); +my $stderr; +$node_standby->safe_psql('postgres', "WAIT FOR LSN '${lsn2}' WITH (timeout '10ms');"); +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${lsn3}' WITH (timeout '1000ms');", + stderr => \$stderr); +ok( $stderr =~ /timed out while waiting for target LSN/, + "get timeout on waiting for unreachable LSN"); + +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn2}' WITH (timeout '0.1s', no_throw);]); +ok($output eq "success", + "WAIT FOR returns correct status after successful waiting"); +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn3}' WITH (timeout '10ms', no_throw);]); +ok($output eq "timeout", "WAIT FOR returns correct status after timeout"); + +# 4. Check that WAIT FOR triggers an error if called on primary, +# within another function, or inside a transaction with an isolation level +# higher than READ COMMITTED. + +$node_primary->psql('postgres', "WAIT FOR LSN '${lsn3}';", + stderr => \$stderr); +ok( $stderr =~ /recovery is not in progress/, + "get an error when running on the primary"); + +$node_standby->psql( + 'postgres', + "BEGIN ISOLATION LEVEL REPEATABLE READ; SELECT 1; WAIT FOR LSN '${lsn3}';", + stderr => \$stderr); +ok( $stderr =~ + /WAIT FOR must be only called without an active or registered snapshot/, + "get an error when running in a transaction with an isolation level higher than REPEATABLE READ" +); + +$node_primary->safe_psql( + 'postgres', qq[ +CREATE FUNCTION pg_wal_replay_wait_wrap(target_lsn pg_lsn) RETURNS void AS \$\$ + BEGIN + EXECUTE format('WAIT FOR LSN %L;', target_lsn); + END +\$\$ +LANGUAGE plpgsql; +]); + +$node_primary->wait_for_catchup($node_standby); +$node_standby->psql( + 'postgres', + "SELECT pg_wal_replay_wait_wrap('${lsn3}');", + stderr => \$stderr); +ok( $stderr =~ + /WAIT FOR must be only called without an active or registered snapshot/, + "get an error when running within another function"); + +# Test parameter validation error cases on standby before promotion +my $test_lsn = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); + +# Test negative timeout +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (timeout '-1000ms');", + stderr => \$stderr); +ok($stderr =~ /timeout cannot be negative/, + "get error for negative timeout"); + +# Test unknown parameter with WITH clause +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (unknown_param 'value');", + stderr => \$stderr); +ok($stderr =~ /option "unknown_param" not recognized/, "get error for unknown parameter"); + +# Test duplicate TIMEOUT parameter with WITH clause +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (timeout '1000', timeout '2000');", + stderr => \$stderr); +ok( $stderr =~ /conflicting or redundant options/, + "get error for duplicate TIMEOUT parameter"); + +# Test duplicate NO_THROW parameter with WITH clause +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (no_throw, no_throw);", + stderr => \$stderr); +ok( $stderr =~ /conflicting or redundant options/, + "get error for duplicate NO_THROW parameter"); + +# Test syntax error - missing LSN +$node_standby->psql('postgres', "WAIT FOR TIMEOUT 1000;", stderr => \$stderr); +ok($stderr =~ /syntax error/, "get syntax error for missing LSN"); + +# Test invalid LSN format +$node_standby->psql( + 'postgres', + "WAIT FOR LSN 'invalid_lsn';", + stderr => \$stderr); +ok($stderr =~ /invalid input syntax for type pg_lsn/, + "get error for invalid LSN format"); + +# Test invalid timeout format +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (timeout 'invalid');", + stderr => \$stderr); +ok( $stderr =~ /invalid timeout value/, + "get error for invalid timeout format"); + +# Test new WITH clause syntax +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn2}' WITH (timeout '0.1s', no_throw);]); +ok($output eq "success", + "WAIT FOR WITH clause syntax works correctly"); + +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn3}' WITH (timeout 100, no_throw);]); +ok($output eq "timeout", "WAIT FOR WITH clause returns correct timeout status"); + +# Test WITH clause error case - invalid option +$node_standby->psql( + 'postgres', + "WAIT FOR LSN '${test_lsn}' WITH (invalid_option 'value');", + stderr => \$stderr); +ok($stderr =~ /option "invalid_option" not recognized/, + "get error for invalid WITH clause option"); + +# 5. Also, check the scenario of multiple LSN waiters. We make 5 background +# psql sessions each waiting for a corresponding insertion. When waiting is +# finished, stored procedures logs if there are visible as many rows as +# should be. +$node_primary->safe_psql( + 'postgres', qq[ +CREATE FUNCTION log_count(i int) RETURNS void AS \$\$ + DECLARE + count int; + BEGIN + SELECT count(*) FROM wait_test INTO count; + IF count >= 31 + i THEN + RAISE LOG 'count %', i; + END IF; + END +\$\$ +LANGUAGE plpgsql; +]); +$node_standby->safe_psql('postgres', "SELECT pg_wal_replay_pause();"); +my @psql_sessions; +for (my $i = 0; $i < 5; $i++) +{ + $node_primary->safe_psql('postgres', + "INSERT INTO wait_test VALUES (${i});"); + my $lsn = + $node_primary->safe_psql('postgres', + "SELECT pg_current_wal_insert_lsn()"); + $psql_sessions[$i] = $node_standby->background_psql('postgres'); + $psql_sessions[$i]->query_until( + qr/start/, qq[ + \\echo start + WAIT FOR LSN '${lsn}'; + SELECT log_count(${i}); + ]); +} +my $log_offset = -s $node_standby->logfile; +$node_standby->safe_psql('postgres', "SELECT pg_wal_replay_resume();"); +for (my $i = 0; $i < 5; $i++) +{ + $node_standby->wait_for_log("count ${i}", $log_offset); + $psql_sessions[$i]->quit; +} + +ok(1, 'multiple LSN waiters reported consistent data'); + +# 6. Check that the standby promotion terminates the wait on LSN. Start +# waiting for an unreachable LSN then promote. Check the log for the relevant +# error message. Also, check that waiting for already replayed LSN doesn't +# cause an error even after promotion. +my $lsn4 = + $node_primary->safe_psql('postgres', + "SELECT pg_current_wal_insert_lsn() + 10000000000"); +my $lsn5 = + $node_primary->safe_psql('postgres', "SELECT pg_current_wal_insert_lsn()"); +my $psql_session = $node_standby->background_psql('postgres'); +$psql_session->query_until( + qr/start/, qq[ + \\echo start + WAIT FOR LSN '${lsn4}'; +]); + +# Make sure standby will be promoted at least at the primary insert LSN we +# have just observed. Use pg_switch_wal() to force the insert LSN to be +# written then wait for standby to catchup. +$node_primary->safe_psql('postgres', 'SELECT pg_switch_wal();'); +$node_primary->wait_for_catchup($node_standby); + +$log_offset = -s $node_standby->logfile; +$node_standby->promote; +$node_standby->wait_for_log('recovery is not in progress', $log_offset); + +ok(1, 'got error after standby promote'); + +$node_standby->safe_psql('postgres', "WAIT FOR LSN '${lsn5}';"); + +ok(1, 'wait for already replayed LSN exits immediately even after promotion'); + +$output = $node_standby->safe_psql( + 'postgres', qq[ + WAIT FOR LSN '${lsn4}' WITH (timeout '10ms', no_throw);]); +ok($output eq "not in recovery", + "WAIT FOR returns correct status after standby promotion"); + + +$node_standby->stop; +$node_primary->stop; + +# If we send \q with $psql_session->quit the command can be sent to the session +# already closed. So \q is in initial script, here we only finish IPC::Run. +$psql_session->{run}->finish; + +done_testing(); diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 5290b91e83e..a2c93c0ef4e 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -3263,6 +3263,9 @@ WaitEventIO WaitEventIPC WaitEventSet WaitEventTimeout +WaitLSNProcInfo +WaitLSNResult +WaitLSNState WaitPMResult WalCloseMethod WalCompression -- 2.51.0
From 32dab7ed64eecb62adce6b1d124b1fa389515e74 Mon Sep 17 00:00:00 2001 From: alterego655 <[email protected]> Date: Fri, 10 Oct 2025 16:35:38 +0800 Subject: [PATCH v13 2/3] Add infrastructure for efficient LSN waiting Implement a new facility that allows processes to wait for WAL to reach specific LSNs, both on primary (waiting for flush) and standby (waiting for replay) servers. The implementation uses shared memory with per-backend information organized into pairing heaps, allowing O(1) access to the minimum waited LSN. This enables fast-path checks: after replaying or flushing WAL, the startup process or WAL writer can quickly determine if any waiters need to be awakened. Key components: - New xlogwait.c/h module with WaitForLSNReplay() and WaitForLSNFlush() - Separate pairing heaps for replay and flush waiters - WaitLSN lightweight lock for coordinating shared state - Wait events WAIT_FOR_WAL_REPLAY and WAIT_FOR_WAL_FLUSH for monitoring This infrastructure can be used by features that need to wait for WAL operations to complete. Discussion: https://www.postgresql.org/message-id/flat/CAPpHfdsjtZLVzxjGT8rJHCYbM0D5dwkO+BBjcirozJ6nYbOW8Q@mail.gmail.com https://www.postgresql.org/message-id/flat/CABPTF7UNft368x-RgOXkfj475OwEbp%2BVVO-wEXz7StgjD_%3D6sw%40mail.gmail.com Author: Kartyshov Ivan <[email protected]> Author: Alexander Korotkov <[email protected]> Author: Xuneng Zhou <[email protected]> Reviewed-by: Michael Paquier <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Dilip Kumar <[email protected]> Reviewed-by: Amit Kapila <[email protected]> Reviewed-by: Alexander Lakhin <[email protected]> Reviewed-by: Bharath Rupireddy <[email protected]> Reviewed-by: Euler Taveira <[email protected]> Reviewed-by: Heikki Linnakangas <[email protected]> Reviewed-by: Kyotaro Horiguchi <[email protected]> --- src/backend/access/transam/Makefile | 3 +- src/backend/access/transam/meson.build | 1 + src/backend/access/transam/xlogwait.c | 525 ++++++++++++++++++ src/backend/storage/ipc/ipci.c | 3 + .../utils/activity/wait_event_names.txt | 3 + src/include/access/xlogwait.h | 112 ++++ src/include/storage/lwlocklist.h | 1 + 7 files changed, 647 insertions(+), 1 deletion(-) create mode 100644 src/backend/access/transam/xlogwait.c create mode 100644 src/include/access/xlogwait.h diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile index 661c55a9db7..a32f473e0a2 100644 --- a/src/backend/access/transam/Makefile +++ b/src/backend/access/transam/Makefile @@ -36,7 +36,8 @@ OBJS = \ xlogreader.o \ xlogrecovery.o \ xlogstats.o \ - xlogutils.o + xlogutils.o \ + xlogwait.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/access/transam/meson.build b/src/backend/access/transam/meson.build index e8ae9b13c8e..74a62ab3eab 100644 --- a/src/backend/access/transam/meson.build +++ b/src/backend/access/transam/meson.build @@ -24,6 +24,7 @@ backend_sources += files( 'xlogrecovery.c', 'xlogstats.c', 'xlogutils.c', + 'xlogwait.c', ) # used by frontend programs to build a frontend xlogreader diff --git a/src/backend/access/transam/xlogwait.c b/src/backend/access/transam/xlogwait.c new file mode 100644 index 00000000000..4faed65765c --- /dev/null +++ b/src/backend/access/transam/xlogwait.c @@ -0,0 +1,525 @@ +/*------------------------------------------------------------------------- + * + * xlogwait.c + * Implements waiting for WAL operations to reach specific LSNs. + * Used by internal WAL reading operations. + * + * Copyright (c) 2025, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/access/transam/xlogwait.c + * + * NOTES + * This file implements waiting for WAL operations to reach specific LSNs + * on both physical standby and primary servers. The core idea is simple: + * every process that wants to wait publishes the LSN it needs to the + * shared memory, and the appropriate process (startup on standby, or + * WAL writer/backend on primary) wakes it once that LSN has been reached. + * + * The shared memory used by this module comprises a procInfos + * per-backend array with the information of the awaited LSN for each + * of the backend processes. The elements of that array are organized + * into a pairing heap waitersHeap, which allows for very fast finding + * of the least awaited LSN. + * + * In addition, the least-awaited LSN is cached as minWaitedLSN. The + * waiter process publishes information about itself to the shared + * memory and waits on the latch before it wakens up by the appropriate + * process, standby is promoted, or the postmaster dies. Then, it cleans + * information about itself in the shared memory. + * + * On standby servers: After replaying a WAL record, the startup process + * first performs a fast path check minWaitedLSN > replayLSN. If this + * check is negative, it checks waitersHeap and wakes up the backend + * whose awaited LSNs are reached. + * + * On primary servers: After flushing WAL, the WAL writer or backend + * process performs a similar check against the flush LSN and wakes up + * waiters whose target flush LSNs have been reached. + * + *------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include <float.h> +#include <math.h> + +#include "access/xlog.h" +#include "access/xlogrecovery.h" +#include "access/xlogwait.h" +#include "miscadmin.h" +#include "pgstat.h" +#include "storage/latch.h" +#include "storage/proc.h" +#include "storage/shmem.h" +#include "utils/fmgrprotos.h" +#include "utils/pg_lsn.h" +#include "utils/snapmgr.h" + + +static int waitlsn_replay_cmp(const pairingheap_node *a, const pairingheap_node *b, + void *arg); + +static int waitlsn_flush_cmp(const pairingheap_node *a, const pairingheap_node *b, + void *arg); + +struct WaitLSNState *waitLSNState = NULL; + +/* Report the amount of shared memory space needed for WaitLSNState. */ +Size +WaitLSNShmemSize(void) +{ + Size size; + + size = offsetof(WaitLSNState, procInfos); + size = add_size(size, mul_size(MaxBackends + NUM_AUXILIARY_PROCS, sizeof(WaitLSNProcInfo))); + return size; +} + +/* Initialize the WaitLSNState in the shared memory. */ +void +WaitLSNShmemInit(void) +{ + bool found; + + waitLSNState = (WaitLSNState *) ShmemInitStruct("WaitLSNState", + WaitLSNShmemSize(), + &found); + if (!found) + { + /* Initialize replay heap and tracking */ + pg_atomic_init_u64(&waitLSNState->minWaitedReplayLSN, PG_UINT64_MAX); + pairingheap_initialize(&waitLSNState->replayWaitersHeap, waitlsn_replay_cmp, (void *)(uintptr_t)WAIT_LSN_REPLAY); + + /* Initialize flush heap and tracking */ + pg_atomic_init_u64(&waitLSNState->minWaitedFlushLSN, PG_UINT64_MAX); + pairingheap_initialize(&waitLSNState->flushWaitersHeap, waitlsn_flush_cmp, (void *)(uintptr_t)WAIT_LSN_FLUSH); + + /* Initialize process info array */ + memset(&waitLSNState->procInfos, 0, + (MaxBackends + NUM_AUXILIARY_PROCS) * sizeof(WaitLSNProcInfo)); + } +} + +/* + * Comparison function for replay waiters heaps. Waiting processes are + * ordered by LSN, so that the waiter with smallest LSN is at the top. + */ +static int +waitlsn_replay_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg) +{ + const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, replayHeapNode, a); + const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, replayHeapNode, b); + + if (aproc->waitLSN < bproc->waitLSN) + return 1; + else if (aproc->waitLSN > bproc->waitLSN) + return -1; + else + return 0; +} + +/* + * Comparison function for flush waiters heaps. Waiting processes are + * ordered by LSN, so that the waiter with smallest LSN is at the top. + */ +static int +waitlsn_flush_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg) +{ + const WaitLSNProcInfo *aproc = pairingheap_const_container(WaitLSNProcInfo, flushHeapNode, a); + const WaitLSNProcInfo *bproc = pairingheap_const_container(WaitLSNProcInfo, flushHeapNode, b); + + if (aproc->waitLSN < bproc->waitLSN) + return 1; + else if (aproc->waitLSN > bproc->waitLSN) + return -1; + else + return 0; +} + +/* + * Update minimum waited LSN for the specified operation type + */ +static void +updateMinWaitedLSN(WaitLSNOperation operation) +{ + XLogRecPtr minWaitedLSN = PG_UINT64_MAX; + + if (operation == WAIT_LSN_REPLAY) + { + if (!pairingheap_is_empty(&waitLSNState->replayWaitersHeap)) + { + pairingheap_node *node = pairingheap_first(&waitLSNState->replayWaitersHeap); + WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, replayHeapNode, node); + minWaitedLSN = procInfo->waitLSN; + } + pg_atomic_write_u64(&waitLSNState->minWaitedReplayLSN, minWaitedLSN); + } + else /* WAIT_LSN_FLUSH */ + { + if (!pairingheap_is_empty(&waitLSNState->flushWaitersHeap)) + { + pairingheap_node *node = pairingheap_first(&waitLSNState->flushWaitersHeap); + WaitLSNProcInfo *procInfo = pairingheap_container(WaitLSNProcInfo, flushHeapNode, node); + minWaitedLSN = procInfo->waitLSN; + } + pg_atomic_write_u64(&waitLSNState->minWaitedFlushLSN, minWaitedLSN); + } +} + +/* + * Add current process to appropriate waiters heap based on operation type + */ +static void +addLSNWaiter(XLogRecPtr lsn, WaitLSNOperation operation) +{ + WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber]; + + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + procInfo->procno = MyProcNumber; + procInfo->waitLSN = lsn; + + if (operation == WAIT_LSN_REPLAY) + { + Assert(!procInfo->inReplayHeap); + pairingheap_add(&waitLSNState->replayWaitersHeap, &procInfo->replayHeapNode); + procInfo->inReplayHeap = true; + updateMinWaitedLSN(WAIT_LSN_REPLAY); + } + else /* WAIT_LSN_FLUSH */ + { + Assert(!procInfo->inFlushHeap); + pairingheap_add(&waitLSNState->flushWaitersHeap, &procInfo->flushHeapNode); + procInfo->inFlushHeap = true; + updateMinWaitedLSN(WAIT_LSN_FLUSH); + } + + LWLockRelease(WaitLSNLock); +} + +/* + * Remove current process from appropriate waiters heap based on operation type + */ +static void +deleteLSNWaiter(WaitLSNOperation operation) +{ + WaitLSNProcInfo *procInfo = &waitLSNState->procInfos[MyProcNumber]; + + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + if (operation == WAIT_LSN_REPLAY && procInfo->inReplayHeap) + { + pairingheap_remove(&waitLSNState->replayWaitersHeap, &procInfo->replayHeapNode); + procInfo->inReplayHeap = false; + updateMinWaitedLSN(WAIT_LSN_REPLAY); + } + else if (operation == WAIT_LSN_FLUSH && procInfo->inFlushHeap) + { + pairingheap_remove(&waitLSNState->flushWaitersHeap, &procInfo->flushHeapNode); + procInfo->inFlushHeap = false; + updateMinWaitedLSN(WAIT_LSN_FLUSH); + } + + LWLockRelease(WaitLSNLock); +} + +/* + * Size of a static array of procs to wakeup by WaitLSNWakeup() allocated + * on the stack. It should be enough to take single iteration for most cases. + */ +#define WAKEUP_PROC_STATIC_ARRAY_SIZE (16) + +/* + * Remove waiters whose LSN has been reached from the heap and set their + * latches. If InvalidXLogRecPtr is given, remove all waiters from the heap + * and set latches for all waiters. + * + * This function first accumulates waiters to wake up into an array, then + * wakes them up without holding a WaitLSNLock. The array size is static and + * equal to WAKEUP_PROC_STATIC_ARRAY_SIZE. That should be more than enough + * to wake up all the waiters at once in the vast majority of cases. However, + * if there are more waiters, this function will loop to process them in + * multiple chunks. + */ +static void +wakeupWaiters(WaitLSNOperation operation, XLogRecPtr currentLSN) +{ + ProcNumber wakeUpProcs[WAKEUP_PROC_STATIC_ARRAY_SIZE]; + int numWakeUpProcs; + int i; + pairingheap *heap; + + /* Select appropriate heap */ + heap = (operation == WAIT_LSN_REPLAY) ? + &waitLSNState->replayWaitersHeap : + &waitLSNState->flushWaitersHeap; + + do + { + numWakeUpProcs = 0; + LWLockAcquire(WaitLSNLock, LW_EXCLUSIVE); + + /* + * Iterate the waiters heap until we find LSN not yet reached. + * Record process numbers to wake up, but send wakeups after releasing lock. + */ + while (!pairingheap_is_empty(heap)) + { + pairingheap_node *node = pairingheap_first(heap); + WaitLSNProcInfo *procInfo; + + /* Get procInfo using appropriate heap node */ + if (operation == WAIT_LSN_REPLAY) + procInfo = pairingheap_container(WaitLSNProcInfo, replayHeapNode, node); + else + procInfo = pairingheap_container(WaitLSNProcInfo, flushHeapNode, node); + + if (!XLogRecPtrIsInvalid(currentLSN) && procInfo->waitLSN > currentLSN) + break; + + Assert(numWakeUpProcs < WAKEUP_PROC_STATIC_ARRAY_SIZE); + wakeUpProcs[numWakeUpProcs++] = procInfo->procno; + (void) pairingheap_remove_first(heap); + + /* Update appropriate flag */ + if (operation == WAIT_LSN_REPLAY) + procInfo->inReplayHeap = false; + else + procInfo->inFlushHeap = false; + + if (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE) + break; + } + + updateMinWaitedLSN(operation); + LWLockRelease(WaitLSNLock); + + /* + * Set latches for processes, whose waited LSNs are already reached. + * As the time consuming operations, we do this outside of + * WaitLSNLock. This is actually fine because procLatch isn't ever + * freed, so we just can potentially set the wrong process' (or no + * process') latch. + */ + for (i = 0; i < numWakeUpProcs; i++) + SetLatch(&GetPGProcByNumber(wakeUpProcs[i])->procLatch); + + } while (numWakeUpProcs == WAKEUP_PROC_STATIC_ARRAY_SIZE); +} + +/* + * Wake up processes waiting for replay LSN to reach currentLSN + */ +void +WaitLSNWakeupReplay(XLogRecPtr currentLSN) +{ + /* Fast path check */ + if (pg_atomic_read_u64(&waitLSNState->minWaitedReplayLSN) > currentLSN) + return; + + wakeupWaiters(WAIT_LSN_REPLAY, currentLSN); +} + +/* + * Wake up processes waiting for flush LSN to reach currentLSN + */ +void +WaitLSNWakeupFlush(XLogRecPtr currentLSN) +{ + /* Fast path check */ + if (pg_atomic_read_u64(&waitLSNState->minWaitedFlushLSN) > currentLSN) + return; + + wakeupWaiters(WAIT_LSN_FLUSH, currentLSN); +} + +/* + * Clean up LSN waiters for exiting process + */ +void +WaitLSNCleanup(void) +{ + if (waitLSNState) + { + /* + * We do a fast-path check of the heap flags without the lock. These + * flags are set to true only by the process itself. So, it's only possible + * to get a false positive. But that will be eliminated by a recheck + * inside deleteLSNWaiter(). + */ + if (waitLSNState->procInfos[MyProcNumber].inReplayHeap) + deleteLSNWaiter(WAIT_LSN_REPLAY); + if (waitLSNState->procInfos[MyProcNumber].inFlushHeap) + deleteLSNWaiter(WAIT_LSN_FLUSH); + } +} + +/* + * Wait using MyLatch till the given LSN is replayed, the replica gets + * promoted, or the postmaster dies. + * + * Returns WAIT_LSN_RESULT_SUCCESS if target LSN was replayed. + * Returns WAIT_LSN_RESULT_NOT_IN_RECOVERY if run not in recovery, + * or replica got promoted before the target LSN replayed. + */ +WaitLSNResult +WaitForLSNReplay(XLogRecPtr targetLSN) +{ + XLogRecPtr currentLSN; + int wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH; + + /* Shouldn't be called when shmem isn't initialized */ + Assert(waitLSNState); + + /* Should have a valid proc number */ + Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends); + + /* + * Add our process to the replay waiters heap. It might happen that + * target LSN gets replayed before we do. Another check at the beginning + * of the loop below prevents the race condition. + */ + addLSNWaiter(targetLSN, WAIT_LSN_REPLAY); + + for (;;) + { + int rc; + long delay_ms = 0; + + /* Recheck that recovery is still in-progress */ + if (!RecoveryInProgress()) + { + /* + * Recovery was ended, but recheck if target LSN was already + * replayed. See the comment regarding deleteLSNWaiter() below. + */ + deleteLSNWaiter(WAIT_LSN_REPLAY); + currentLSN = GetXLogReplayRecPtr(NULL); + if (PromoteIsTriggered() && targetLSN <= currentLSN) + return WAIT_LSN_RESULT_SUCCESS; + return WAIT_LSN_RESULT_NOT_IN_RECOVERY; + } + else + { + /* Check if the waited LSN has been replayed */ + currentLSN = GetXLogReplayRecPtr(NULL); + if (targetLSN <= currentLSN) + break; + } + + CHECK_FOR_INTERRUPTS(); + + rc = WaitLatch(MyLatch, wake_events, delay_ms, + WAIT_EVENT_WAIT_FOR_WAL_REPLAY); + + /* + * Emergency bailout if postmaster has died. This is to avoid the + * necessity for manual cleanup of all postmaster children. + */ + if (rc & WL_POSTMASTER_DEATH) + ereport(FATAL, + (errcode(ERRCODE_ADMIN_SHUTDOWN), + errmsg("terminating connection due to unexpected postmaster exit"), + errcontext("while waiting for LSN replay"))); + + if (rc & WL_LATCH_SET) + ResetLatch(MyLatch); + } + + /* + * Delete our process from the shared memory replay heap. We might + * already be deleted by the startup process. The 'inReplayHeap' flag prevents + * us from the double deletion. + */ + deleteLSNWaiter(WAIT_LSN_REPLAY); + + return WAIT_LSN_RESULT_SUCCESS; +} + +/* + * Wait until targetLSN has been flushed on a primary server. + * Returns only after the condition is satisfied or on FATAL exit. + */ +void +WaitForLSNFlush(XLogRecPtr targetLSN) +{ + XLogRecPtr currentLSN; + int wake_events = WL_LATCH_SET | WL_POSTMASTER_DEATH; + + /* Shouldn't be called when shmem isn't initialized */ + Assert(waitLSNState); + + /* Should have a valid proc number */ + Assert(MyProcNumber >= 0 && MyProcNumber < MaxBackends + NUM_AUXILIARY_PROCS); + + /* We can only wait for flush when we are not in recovery */ + Assert(!RecoveryInProgress()); + + /* Quick exit if already flushed */ + currentLSN = GetFlushRecPtr(NULL); + if (targetLSN <= currentLSN) + return; + + /* Add to flush waiters */ + addLSNWaiter(targetLSN, WAIT_LSN_FLUSH); + + /* Wait loop */ + for (;;) + { + int rc; + + /* Check if the waited LSN has been flushed */ + currentLSN = GetFlushRecPtr(NULL); + if (targetLSN <= currentLSN) + break; + + CHECK_FOR_INTERRUPTS(); + + rc = WaitLatch(MyLatch, wake_events, -1, + WAIT_EVENT_WAIT_FOR_WAL_FLUSH); + + /* + * Emergency bailout if postmaster has died. This is to avoid the + * necessity for manual cleanup of all postmaster children. + */ + if (rc & WL_POSTMASTER_DEATH) + ereport(FATAL, + (errcode(ERRCODE_ADMIN_SHUTDOWN), + errmsg("terminating connection due to unexpected postmaster exit"), + errcontext("while waiting for LSN flush"))); + + if (rc & WL_LATCH_SET) + ResetLatch(MyLatch); + } + + /* + * Delete our process from the shared memory flush heap. We might + * already be deleted by the waker process. The 'inFlushHeap' flag prevents + * us from the double deletion. + */ + deleteLSNWaiter(WAIT_LSN_FLUSH); + + return; +} diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index 2fa045e6b0f..10ffce8d174 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -24,6 +24,7 @@ #include "access/twophase.h" #include "access/xlogprefetcher.h" #include "access/xlogrecovery.h" +#include "access/xlogwait.h" #include "commands/async.h" #include "miscadmin.h" #include "pgstat.h" @@ -150,6 +151,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); size = add_size(size, AioShmemSize()); + size = add_size(size, WaitLSNShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -343,6 +345,7 @@ CreateOrAttachShmemStructs(void) WaitEventCustomShmemInit(); InjectionPointShmemInit(); AioShmemInit(); + WaitLSNShmemInit(); } /* diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt index 7553f6eacef..c1ac71ff7f2 100644 --- a/src/backend/utils/activity/wait_event_names.txt +++ b/src/backend/utils/activity/wait_event_names.txt @@ -89,6 +89,8 @@ LIBPQWALRECEIVER_CONNECT "Waiting in WAL receiver to establish connection to rem LIBPQWALRECEIVER_RECEIVE "Waiting in WAL receiver to receive data from remote server." SSL_OPEN_SERVER "Waiting for SSL while attempting connection." WAIT_FOR_STANDBY_CONFIRMATION "Waiting for WAL to be received and flushed by the physical standby." +WAIT_FOR_WAL_FLUSH "Waiting for WAL flush to reach a target LSN on a primary." +WAIT_FOR_WAL_REPLAY "Waiting for WAL replay to reach a target LSN on a standby." WAL_SENDER_WAIT_FOR_WAL "Waiting for WAL to be flushed in WAL sender process." WAL_SENDER_WRITE_DATA "Waiting for any activity when processing replies from WAL receiver in WAL sender process." @@ -355,6 +357,7 @@ DSMRegistry "Waiting to read or update the dynamic shared memory registry." InjectionPoint "Waiting to read or update information related to injection points." SerialControl "Waiting to read or update shared <filename>pg_serial</filename> state." AioWorkerSubmissionQueue "Waiting to access AIO worker submission queue." +WaitLSN "Waiting to read or update shared Wait-for-LSN state." # # END OF PREDEFINED LWLOCKS (DO NOT CHANGE THIS LINE) diff --git a/src/include/access/xlogwait.h b/src/include/access/xlogwait.h new file mode 100644 index 00000000000..441bf475b4d --- /dev/null +++ b/src/include/access/xlogwait.h @@ -0,0 +1,112 @@ +/*------------------------------------------------------------------------- + * + * xlogwait.h + * Declarations for LSN replay waiting routines. + * + * Copyright (c) 2025, PostgreSQL Global Development Group + * + * src/include/access/xlogwait.h + * + *------------------------------------------------------------------------- + */ +#ifndef XLOG_WAIT_H +#define XLOG_WAIT_H + +#include "lib/pairingheap.h" +#include "port/atomics.h" +#include "postgres.h" +#include "storage/procnumber.h" +#include "storage/spin.h" +#include "tcop/dest.h" + +/* + * Result statuses for WaitForLSNReplay(). + */ +typedef enum +{ + WAIT_LSN_RESULT_SUCCESS, /* Target LSN is reached */ + WAIT_LSN_RESULT_NOT_IN_RECOVERY, /* Recovery ended before or during our + * wait */ +} WaitLSNResult; + +/* + * Wait operation types for LSN waiting facility. + */ +typedef enum WaitLSNOperation +{ + WAIT_LSN_REPLAY, /* Waiting for replay on standby */ + WAIT_LSN_FLUSH /* Waiting for flush on primary */ +} WaitLSNOperation; + +/* + * WaitLSNProcInfo - the shared memory structure representing information + * about the single process, which may wait for LSN operations. An item of + * waitLSNState->procInfos array. + */ +typedef struct WaitLSNProcInfo +{ + /* LSN, which this process is waiting for */ + XLogRecPtr waitLSN; + + /* Process to wake up once the waitLSN is reached */ + ProcNumber procno; + + /* Type-safe heap membership flags */ + bool inReplayHeap; /* In replay waiters heap */ + bool inFlushHeap; /* In flush waiters heap */ + + /* Separate heap nodes for type safety */ + pairingheap_node replayHeapNode; + pairingheap_node flushHeapNode; +} WaitLSNProcInfo; + +/* + * WaitLSNState - the shared memory state for the LSN waiting facility. + */ +typedef struct WaitLSNState +{ + /* + * The minimum replay LSN value some process is waiting for. Used for the + * fast-path checking if we need to wake up any waiters after replaying a + * WAL record. Could be read lock-less. Update protected by WaitLSNLock. + */ + pg_atomic_uint64 minWaitedReplayLSN; + + /* + * A pairing heap of replay waiting processes ordered by LSN values (least LSN is + * on top). Protected by WaitLSNLock. + */ + pairingheap replayWaitersHeap; + + /* + * The minimum flush LSN value some process is waiting for. Used for the + * fast-path checking if we need to wake up any waiters after flushing + * WAL. Could be read lock-less. Update protected by WaitLSNLock. + */ + pg_atomic_uint64 minWaitedFlushLSN; + + /* + * A pairing heap of flush waiting processes ordered by LSN values (least LSN is + * on top). Protected by WaitLSNLock. + */ + pairingheap flushWaitersHeap; + + /* + * An array with per-process information, indexed by the process number. + * Protected by WaitLSNLock. + */ + WaitLSNProcInfo procInfos[FLEXIBLE_ARRAY_MEMBER]; +} WaitLSNState; + + +extern PGDLLIMPORT WaitLSNState *waitLSNState; + +extern Size WaitLSNShmemSize(void); +extern void WaitLSNShmemInit(void); +extern void WaitLSNWakeupReplay(XLogRecPtr currentLSN); +extern void WaitLSNWakeupFlush(XLogRecPtr currentLSN); +extern void WaitLSNCleanup(void); +extern WaitLSNResult WaitForLSNReplay(XLogRecPtr targetLSN); +extern void WaitForLSNFlush(XLogRecPtr targetLSN); + +#endif /* XLOG_WAIT_H */ diff --git a/src/include/storage/lwlocklist.h b/src/include/storage/lwlocklist.h index 06a1ffd4b08..5b0ce383408 100644 --- a/src/include/storage/lwlocklist.h +++ b/src/include/storage/lwlocklist.h @@ -85,6 +85,7 @@ PG_LWLOCK(50, DSMRegistry) PG_LWLOCK(51, InjectionPoint) PG_LWLOCK(52, SerialControl) PG_LWLOCK(53, AioWorkerSubmissionQueue) +PG_LWLOCK(54, WaitLSN) /* * There also exist several built-in LWLock tranches. As with the predefined -- 2.51.0
From 48abb92fb33628f6eba5bbe865b3b19c24fb716d Mon Sep 17 00:00:00 2001 From: alterego655 <[email protected]> Date: Thu, 9 Oct 2025 10:29:05 +0800 Subject: [PATCH v13 1/3] Add pairingheap_initialize() for shared memory usage The existing pairingheap_allocate() uses palloc(), which allocates from process-local memory. For shared memory use cases, the pairingheap structure must be allocated via ShmemAlloc() or embedded in a shared memory struct. Add pairingheap_initialize() to initialize an already- allocated pairingheap structure in-place, enabling shared memory usage. Discussion: https://www.postgresql.org/message-id/flat/CAPpHfdsjtZLVzxjGT8rJHCYbM0D5dwkO+BBjcirozJ6nYbOW8Q@mail.gmail.com https://www.postgresql.org/message-id/flat/CABPTF7UNft368x-RgOXkfj475OwEbp%2BVVO-wEXz7StgjD_%3D6sw%40mail.gmail.com Author: Kartyshov Ivan <[email protected]> Author: Alexander Korotkov <[email protected]> Reviewed-by: Michael Paquier <[email protected]> Reviewed-by: Peter Eisentraut <[email protected]> Reviewed-by: Dilip Kumar <[email protected]> Reviewed-by: Amit Kapila <[email protected]> Reviewed-by: Alexander Lakhin <[email protected]> Reviewed-by: Bharath Rupireddy <[email protected]> Reviewed-by: Euler Taveira <[email protected]> Reviewed-by: Heikki Linnakangas <[email protected]> Reviewed-by: Kyotaro Horiguchi <[email protected]> Reviewed-by: Xuneng Zhou <[email protected]> --- src/backend/lib/pairingheap.c | 18 ++++++++++++++++-- src/include/lib/pairingheap.h | 3 +++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/backend/lib/pairingheap.c b/src/backend/lib/pairingheap.c index 0aef8a88f1b..fa8431f7946 100644 --- a/src/backend/lib/pairingheap.c +++ b/src/backend/lib/pairingheap.c @@ -44,12 +44,26 @@ pairingheap_allocate(pairingheap_comparator compare, void *arg) pairingheap *heap; heap = (pairingheap *) palloc(sizeof(pairingheap)); + pairingheap_initialize(heap, compare, arg); + + return heap; +} + +/* + * pairingheap_initialize + * + * Same as pairingheap_allocate(), but initializes the pairing heap in-place + * rather than allocating a new chunk of memory. Useful to store the pairing + * heap in a shared memory. + */ +void +pairingheap_initialize(pairingheap *heap, pairingheap_comparator compare, + void *arg) +{ heap->ph_compare = compare; heap->ph_arg = arg; heap->ph_root = NULL; - - return heap; } /* diff --git a/src/include/lib/pairingheap.h b/src/include/lib/pairingheap.h index 3c57d3fda1b..567586f2ecf 100644 --- a/src/include/lib/pairingheap.h +++ b/src/include/lib/pairingheap.h @@ -77,6 +77,9 @@ typedef struct pairingheap extern pairingheap *pairingheap_allocate(pairingheap_comparator compare, void *arg); +extern void pairingheap_initialize(pairingheap *heap, + pairingheap_comparator compare, + void *arg); extern void pairingheap_free(pairingheap *heap); extern void pairingheap_add(pairingheap *heap, pairingheap_node *node); extern pairingheap_node *pairingheap_first(pairingheap *heap); -- 2.51.0
