On Wed, 5 Nov 2025 at 15:11, shveta malik <[email protected]> wrote:
>
> Please find a few comments on 003 patch (seqsync_error_count)
>
> 1)
> + /*
> + * Report the worker failed during either sequence synchronization or
> + * table synchronization or apply.
> + */
>
> Shall we tweak it slightly to:
> Report the worker failed during sequence synchronization, table
> synchronization, or apply.
Modified
> 2)
>
> + SELECT count(1) = 1 FROM pg_stat_subscription_stats
> + WHERE subname = '$sub_name' and seq_sync_error_count > 0 and
> sync_error_count > 0
> ])
> or die
> qq(Timed out while waiting for tablesync errors for subscription
> '$sub_name');
>
> Since we are checking both table-sync and seq-sync errors here, we
> shall update the failure-message.
Modified
> 3)
> + # Change the sequence start value on the subscriber so that it
> doesn't error out.
> + $node_subscriber->safe_psql($db,
> + qq(ALTER SEQUENCE $sequence_name INCREMENT 1));
>
> Please mention in comment 'Change the sequence start value to default....'.
> Otherwise it is not clear why changing to 1 is helping here as the
> previous creation of seq on pub did not mention any 'INCREMENT' value
> at all.
Modified
> 4)
>
> - # Wait for initial tablesync to finish.
> + # Wait for initial sync to finish.
> $node_subscriber->poll_query_until(
> $db,
> qq[
> - SELECT count(1) = 1 FROM pg_subscription_rel
> - WHERE srrelid = '$table_name'::regclass AND srsubstate in ('r', 's')
> + SELECT count(1) = 2 FROM pg_subscription_rel
> + WHERE srrelid IN ('$table_name'::regclass,
> '$sequence_name'::regclass) AND srsubstate in ('r', 's')
> ])
> or die
> qq(Timed out while waiting for subscriber to synchronize data for
> table '$table_name'.);
>
>
> a) Will it be better to separate the 2 queries as the table-sync can
> be 'r' and 's', while seq-sync has to be 'r'.
>
> b) If we plan to keep the same as above, the failure-message needs to
> be changed as it mentions only table.
I have separated the query to check individually for table sync and
sequence sync.
The attached v20251105 version patch has the changes for the same.
Regards,
Vignesh
From e8a1f0e18155126738fbdd1963c2680d991e08a7 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 27 Oct 2025 09:18:07 +0530
Subject: [PATCH v20251105 2/2] Documentation for sequence synchronization
feature.
Documentation for sequence synchronization feature.
Author: Vignesh C <[email protected]>
Reviewed-by: Amit Kapila <[email protected]>
Reviewed-by: shveta malik <[email protected]>
Reviewed-by: Hou Zhijie <[email protected]>
Reviewed-by: Masahiko Sawada <[email protected]>
Reviewed-by: Hayato Kuroda <[email protected]>
Reviewed-by: Dilip Kumar <[email protected]>
Reviewed-by: Peter Smith <[email protected]>
Reviewed-by: Nisha Moond <[email protected]>
Reviewed-by: Shlok Kyal <[email protected]>
Discussion: https://www.postgresql.org/message-id/caa4ek1lc+kjiaksrpe_nwvndidw9f2os7geruesxskv71gx...@mail.gmail.co
---
doc/src/sgml/catalogs.sgml | 2 +-
doc/src/sgml/config.sgml | 14 +-
doc/src/sgml/func/func-sequence.sgml | 24 +++
doc/src/sgml/logical-replication.sgml | 224 ++++++++++++++++++++--
doc/src/sgml/monitoring.sgml | 5 +-
doc/src/sgml/ref/alter_subscription.sgml | 15 ++
doc/src/sgml/ref/create_subscription.sgml | 27 ++-
7 files changed, 279 insertions(+), 32 deletions(-)
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 6c8a0f173c9..2fc63442980 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6568,7 +6568,7 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
(references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>oid</structfield>)
</para>
<para>
- Reference to relation
+ Reference to table or sequence
</para></entry>
</row>
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 06d1e4403b5..f5cbc68b938 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -5191,9 +5191,9 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
is taken into account.
</para>
<para>
- In logical replication, this parameter also limits how often a failing
- replication apply worker or table synchronization worker will be
- respawned.
+ In logical replication, this parameter also limits how quickly a
+ failing replication apply worker, or table/sequence synchronization
+ worker will be respawned.
</para>
</listitem>
</varlistentry>
@@ -5334,8 +5334,8 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
<listitem>
<para>
Specifies maximum number of logical replication workers. This includes
- leader apply workers, parallel apply workers, and table synchronization
- workers.
+ leader apply workers, parallel apply workers, and table/sequence
+ synchronization workers.
</para>
<para>
Logical replication workers are taken from the pool defined by
@@ -5359,9 +5359,11 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
Maximum number of synchronization workers per subscription. This
parameter controls the amount of parallelism of the initial data copy
during the subscription initialization or when new tables are added.
+ One additional worker is also needed for sequence synchronization.
</para>
<para>
- Currently, there can be only one synchronization worker per table.
+ Currently, there can be only one table synchronization worker per table
+ and one sequence synchronization worker to synchronize all sequences.
</para>
<para>
The synchronization workers are taken from the pool defined by
diff --git a/doc/src/sgml/func/func-sequence.sgml b/doc/src/sgml/func/func-sequence.sgml
index e9f5b4e8e6b..80e51e9e365 100644
--- a/doc/src/sgml/func/func-sequence.sgml
+++ b/doc/src/sgml/func/func-sequence.sgml
@@ -143,6 +143,30 @@ SELECT setval('myseq', 42, false); <lineannotation>Next <function>nextval</fu
or <literal>SELECT</literal> privilege on the last used sequence.
</para></entry>
</row>
+
+ <row>
+ <entry role="func_table_entry"><para role="func_signature">
+ <indexterm>
+ <primary>pg_get_sequence_data</primary>
+ </indexterm>
+ <function>pg_get_sequence_data</function> ( <type>regclass</type> )
+ <returnvalue>record</returnvalue>
+ ( <parameter>last_value</parameter> <type>bigint</type>,
+ <parameter>is_called</parameter> <type>bool</type>,
+ <parameter>page_lsn</parameter> <type>pg_lsn</type> )
+ </para>
+ <para>
+ Returns information about the sequence. <literal>last_value</literal>
+ indicates last sequence value set in sequence by nextval or setval,
+ <literal>is_called</literal> indicates whether the sequence has been
+ used, and <literal>page_lsn</literal> is the LSN corresponding to the
+ most recent WAL record that modified this sequence relation.
+ </para>
+ <para>
+ This function requires <literal>USAGE</literal>
+ or <literal>SELECT</literal> privilege on the sequence.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index b01f5e998b2..9e78c2f0465 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -113,7 +113,9 @@
Publications may currently only contain tables or sequences. Objects must be
added explicitly, except when a publication is created using
<literal>FOR TABLES IN SCHEMA</literal>, <literal>FOR ALL TABLES</literal>,
- or <literal>FOR ALL SEQUENCES</literal>.
+ or <literal>FOR ALL SEQUENCES</literal>. Unlike tables, sequences can be
+ synchronized at any time. For more information, see
+ <xref linkend="logical-replication-sequences"/>.
</para>
<para>
@@ -1745,6 +1747,194 @@ Publications:
</note>
</sect1>
+ <sect1 id="logical-replication-sequences">
+ <title>Replicating Sequences</title>
+
+ <para>
+ To synchronize sequences from a publisher to a subscriber, first publish
+ them using <link linkend="sql-createpublication-params-for-all-sequences">
+ <command>CREATE PUBLICATION ... FOR ALL SEQUENCES</command></link> and then
+ on the subscriber:
+ </para>
+
+ <para>
+ <itemizedlist>
+ <listitem>
+ <para>
+ use <link linkend="sql-createsubscription"><command>CREATE SUBSCRIPTION</command></link>
+ to initially synchronize the published sequences.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ use <link linkend="sql-altersubscription-params-refresh-publication">
+ <command>ALTER SUBSCRIPTION ... REFRESH PUBLICATION</command></link>
+ to synchronize only newly added sequences.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ use <link linkend="sql-altersubscription-params-refresh-sequences">
+ <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>
+ to re-synchronize all sequences currently known to the subscription.
+ </para>
+ </listitem>
+ </itemizedlist>
+ </para>
+
+ <para>
+ A new <firstterm>sequence synchronization worker</firstterm> will be started
+ after executing any of the above subscriber commands, and will exit once the
+ sequences are synchronized.
+ </para>
+ <para>
+ The ability to launch a sequence synchronization worker is limited by the
+ <link linkend="guc-max-sync-workers-per-subscription">
+ <varname>max_sync_workers_per_subscription</varname></link>
+ configuration.
+ </para>
+
+ <sect2 id="sequence-definition-mismatches">
+ <title>Sequence Definition Mismatches</title>
+ <para>
+ The sequence synchronization worker validates that sequence definitions
+ match between publisher and subscriber. If mismatches exist, the worker
+ logs an error identifying them and exits. The apply worker continues
+ respawning the sequence synchronization worker until synchronization
+ succeeds. See also
+ <link linkend="guc-wal-retrieve-retry-interval"><varname>wal_retrieve_retry_interval</varname></link>.
+ </para>
+ <para>
+ To resolve this, use
+ <link linkend="sql-altersequence"><command>ALTER SEQUENCE</command></link>
+ to align the subscriber's sequence parameters with those of the publisher.
+ </para>
+ </sect2>
+
+ <sect2 id="sequences-out-of-sync">
+ <title>Refreshing Stale Sequences</title>
+ <para>
+ Subscriber side sequence values may frequently become out of sync due to
+ updates on the publisher.
+ </para>
+ <para>
+ Subscriber sequence values drift out of sync as the publisher advances
+ them. Compare values between publisher and subscriber, then run
+ <link linkend="sql-altersubscription-params-refresh-sequences">
+ <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link> to
+ resynchronize if necessary.
+ </para>
+ </sect2>
+
+ <sect2 id="logical-replication-sequences-examples">
+ <title>Examples</title>
+
+ <para>
+ Create some sequences on the publisher.
+<programlisting>
+/* pub # */ CREATE SEQUENCE s1 START WITH 10 INCREMENT BY 1;
+/* pub # */ CREATE SEQUENCE s2 START WITH 100 INCREMENT BY 10;
+</programlisting></para>
+
+ <para>
+ Create the same sequences on the subscriber.
+<programlisting>
+/* sub # */ CREATE SEQUENCE s1 START WITH 10 INCREMENT BY 1
+/* sub # */ CREATE SEQUENCE s2 START WITH 100 INCREMENT BY 10;
+</programlisting></para>
+
+ <para>
+ Update the sequences at the publisher side a few times.
+<programlisting>
+/* pub # */ SELECT nextval('s1');
+ nextval
+---------
+ 10
+(1 row)
+/* pub # */ SELECT nextval('s1');
+ nextval
+---------
+ 11
+(1 row)
+/* pub # */ SELECT nextval('s2');
+ nextval
+---------
+ 100
+(1 row)
+/* pub # */ SELECT nextval('s2');
+ nextval
+---------
+ 110
+(1 row)
+</programlisting></para>
+
+ <para>
+ Create a publication for the sequences.
+<programlisting>
+/* pub # */ CREATE PUBLICATION pub1 FOR ALL SEQUENCES;
+</programlisting></para>
+
+ <para>
+ Subscribe to the publication.
+<programlisting>
+/* sub # */ CREATE SUBSCRIPTION sub1
+/* sub - */ CONNECTION 'host=localhost dbname=test_pub application_name=sub1'
+/* sub - */ PUBLICATION pub1;
+</programlisting></para>
+
+ <para>
+ Observe that initial sequence values are synchronized.
+<programlisting>
+/* sub # */ SELECT * FROM s1;
+ last_value | log_cnt | is_called
+------------+---------+-----------
+ 11 | 31 | t
+(1 row)
+
+/* sub # */ SELECT * FROM s2;
+ last_value | log_cnt | is_called
+------------+---------+-----------
+ 110 | 31 | t
+(1 row)
+</programlisting></para>
+
+ <para>
+ Update the sequences at the publisher side.
+<programlisting>
+/* pub # */ SELECT nextval('s1');
+ nextval
+---------
+ 12
+(1 row)
+/* pub # */ SELECT nextval('s2');
+ nextval
+---------
+ 120
+(1 row)
+</programlisting></para>
+
+ <para>
+ Re-synchronize all sequences known to the subscriber using
+ <link linkend="sql-altersubscription-params-refresh-sequences">
+ <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>.
+<programlisting>
+/* sub # */ ALTER SUBSCRIPTION sub1 REFRESH SEQUENCES;
+
+/* sub # */ SELECT * FROM s1;
+ last_value | log_cnt | is_called
+------------+---------+-----------
+ 12 | 30 | t
+(1 row)
+
+/* sub # */ SELECT * FROM s2
+ last_value | log_cnt | is_called
+------------+---------+-----------
+ 120 | 30 | t
+(1 row)
+</programlisting></para>
+ </sect2>
+ </sect1>
+
<sect1 id="logical-replication-conflicts">
<title>Conflicts</title>
@@ -2090,16 +2280,19 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<listitem>
<para>
- Sequence data is not replicated. The data in serial or identity columns
- backed by sequences will of course be replicated as part of the table,
- but the sequence itself would still show the start value on the
- subscriber. If the subscriber is used as a read-only database, then this
- should typically not be a problem. If, however, some kind of switchover
- or failover to the subscriber database is intended, then the sequences
- would need to be updated to the latest values, either by copying the
- current data from the publisher (perhaps
- using <command>pg_dump</command>) or by determining a sufficiently high
- value from the tables themselves.
+ Incremental sequence changes are not replicated. Although the data in
+ serial or identity columns backed by sequences will be replicated as part
+ of the table, the sequences themselves do not replicate ongoing changes.
+ On the subscriber, a sequence will retain the last value it synchronized
+ from the publisher. If the subscriber is used as a read-only database,
+ then this should typically not be a problem. If, however, some kind of
+ switchover or failover to the subscriber database is intended, then the
+ sequences would need to be updated to the latest values, either by
+ executing <link linkend="sql-altersubscription-params-refresh-sequences">
+ <command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>
+ or by copying the current data from the publisher (perhaps using
+ <command>pg_dump</command>) or by determining a sufficiently high value
+ from the tables themselves.
</para>
</listitem>
@@ -2423,8 +2616,8 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<para>
<link linkend="guc-max-logical-replication-workers"><varname>max_logical_replication_workers</varname></link>
must be set to at least the number of subscriptions (for leader apply
- workers), plus some reserve for the table synchronization workers and
- parallel apply workers.
+ workers), plus some reserve for the parallel apply workers, and
+ table/sequence synchronization workers.
</para>
<para>
@@ -2437,8 +2630,9 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
<para>
<link linkend="guc-max-sync-workers-per-subscription"><varname>max_sync_workers_per_subscription</varname></link>
- controls the amount of parallelism of the initial data copy during the
- subscription initialization or when new tables are added.
+ controls how many tables can be synchronized in parallel during
+ subscription initialization or when new tables are added. One additional
+ worker is also needed for sequence synchronization.
</para>
<para>
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 02e4400b4a8..1dc0024ab92 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2045,8 +2045,9 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para>
<para>
Type of the subscription worker process. Possible types are
- <literal>apply</literal>, <literal>parallel apply</literal>, and
- <literal>table synchronization</literal>.
+ <literal>apply</literal>, <literal>parallel apply</literal>,
+ <literal>table synchronization</literal>, and
+ <literal>sequence synchronization</literal>.
</para></entry>
</row>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 8ab3b7fbd37..27c06439f4f 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -195,6 +195,12 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
use <link linkend="sql-altersubscription-params-refresh-sequences">
<command>ALTER SUBSCRIPTION ... REFRESH SEQUENCES</command></link>.
</para>
+ <para>
+ See <xref linkend="sequence-definition-mismatches"/> for recommendations on how
+ to handle any warnings about sequence definition differences between
+ the publisher and the subscriber, which might occur when
+ <literal>copy_data = true</literal>.
+ </para>
<para>
See <xref linkend="sql-createsubscription-notes"/> for details of
how <literal>copy_data = true</literal> can interact with the
@@ -225,6 +231,15 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
data for all currently subscribed sequences. It does not add or remove
sequences from the subscription to match the publication.
</para>
+ <para>
+ See <xref linkend="sequence-definition-mismatches"/> for
+ recommendations on how to handle any warnings about sequence definition
+ differences between the publisher and the subscriber.
+ </para>
+ <para>
+ See <xref linkend="sequences-out-of-sync"/> for recommendations on how to
+ identify and handle out-of-sync sequences.
+ </para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index ed82cf1809e..442d370b426 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -127,10 +127,10 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<para>
Since no connection is made when this option is
- <literal>false</literal>, no tables are subscribed. To initiate
- replication, you must manually create the replication slot, enable
- the failover if required, enable the subscription, and refresh the
- subscription. See
+ <literal>false</literal>, no tables and sequences are subscribed. To
+ initiate replication, you must manually create the replication slot,
+ enable the failover if required, enable the subscription, and refresh
+ the subscription. See
<xref linkend="logical-replication-subscription-examples-deferred-slot"/>
for examples.
</para>
@@ -228,7 +228,7 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
the initial synchronization requires all data types to have binary
send and receive functions, otherwise the synchronization will fail
(see <xref linkend="sql-createtype"/> for more about send/receive
- functions).
+ functions). This parameter has no effect for sequences.
</para>
<para>
@@ -265,6 +265,12 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<literal>copy_data = true</literal> can interact with the
<literal>origin</literal> parameter.
</para>
+ <para>
+ See <xref linkend="sequence-definition-mismatches"/>
+ for recommendations on how to handle any warnings about sequence
+ definition differences between the publisher and the subscriber,
+ which might occur when <literal>copy_data = true</literal>.
+ </para>
</listitem>
</varlistentry>
@@ -280,6 +286,7 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
temporary files and applied after the transaction is committed. Note
that if an error happens in a parallel apply worker, the finish LSN
of the remote transaction might not be reported in the server log.
+ This parameter has no effect for sequences.
</para>
<caution>
@@ -310,7 +317,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
The value of this parameter overrides the
<xref linkend="guc-synchronous-commit"/> setting within this
subscription's apply worker processes. The default value
- is <literal>off</literal>.
+ is <literal>off</literal>. This parameter has no effect for
+ sequences.
</para>
<para>
@@ -340,7 +348,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<listitem>
<para>
Specifies whether two-phase commit is enabled for this subscription.
- The default is <literal>false</literal>.
+ The default is <literal>false</literal>. This parameter has no effect
+ for sequences.
</para>
<para>
@@ -417,6 +426,7 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
changes that don't have an origin. Setting <literal>origin</literal>
to <literal>any</literal> means that the publisher sends changes
regardless of their origin. The default is <literal>any</literal>.
+ This parameter has no effect for sequences.
</para>
<para>
See <xref linkend="sql-createsubscription-notes"/> for details of how
@@ -449,7 +459,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<xref linkend="conflict-update-deleted"/> is enabled, and a physical
replication slot named <quote><literal>pg_conflict_detection</literal></quote>
is created on the subscriber to prevent the information for detecting
- conflicts from being removed.
+ conflicts from being removed. This parameter has no effect for
+ sequences.
</para>
<para>
--
2.43.0
From 74f67efcd95df7421c80a3fc160deb396152c783 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 3 Nov 2025 19:50:16 +0530
Subject: [PATCH v20251105 1/2] Add seq_sync_error_count to subscription
statistics.
This commit introduces a new column seq_sync_error_count to subscription
statistics. The new field tracks the number of errors encountered during
sequence synchronization for each subscription.
---
doc/src/sgml/monitoring.sgml | 9 ++
src/backend/catalog/system_views.sql | 1 +
.../replication/logical/sequencesync.c | 3 +
src/backend/replication/logical/tablesync.c | 3 +-
src/backend/replication/logical/worker.c | 18 ++--
.../utils/activity/pgstat_subscription.c | 27 ++++--
src/backend/utils/adt/pgstatfuncs.c | 27 +++---
src/include/catalog/pg_proc.dat | 6 +-
src/include/pgstat.h | 6 +-
src/test/regress/expected/rules.out | 3 +-
src/test/subscription/t/026_stats.pl | 89 +++++++++++++------
11 files changed, 133 insertions(+), 59 deletions(-)
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index f3bf527d5b4..02e4400b4a8 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -2192,6 +2192,15 @@ description | Waiting for a newly initialized WAL file to reach durable storage
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>sequence_sync_error_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of times an error occurred during the sequence synchronization
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>sync_error_count</structfield> <type>bigint</type>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index dec8df4f8ee..059e8778ca7 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1415,6 +1415,7 @@ CREATE VIEW pg_stat_subscription_stats AS
ss.subid,
s.subname,
ss.apply_error_count,
+ ss.seq_sync_error_count,
ss.sync_error_count,
ss.confl_insert_exists,
ss.confl_update_origin_differs,
diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c
index 717c82328f2..161ff1177d9 100644
--- a/src/backend/replication/logical/sequencesync.c
+++ b/src/backend/replication/logical/sequencesync.c
@@ -725,6 +725,9 @@ start_sequence_sync()
* idle state.
*/
AbortOutOfAnyTransaction();
+ pgstat_report_subscription_error(MySubscription->oid,
+ WORKERTYPE_SEQUENCESYNC);
+
PG_RE_THROW();
}
}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index e5a2856fd17..dcc6124cc73 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1530,7 +1530,8 @@ start_table_sync(XLogRecPtr *origin_startpos, char **slotname)
* idle state.
*/
AbortOutOfAnyTransaction();
- pgstat_report_subscription_error(MySubscription->oid, false);
+ pgstat_report_subscription_error(MySubscription->oid,
+ WORKERTYPE_TABLESYNC);
PG_RE_THROW();
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index e1c757c911e..56658273ffc 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -5606,7 +5606,8 @@ start_apply(XLogRecPtr origin_startpos)
* idle state.
*/
AbortOutOfAnyTransaction();
- pgstat_report_subscription_error(MySubscription->oid, !am_tablesync_worker());
+ pgstat_report_subscription_error(MySubscription->oid,
+ MyLogicalRepWorker->type);
PG_RE_THROW();
}
@@ -5953,15 +5954,12 @@ DisableSubscriptionAndExit(void)
RESUME_INTERRUPTS();
- if (am_leader_apply_worker() || am_tablesync_worker())
- {
- /*
- * Report the worker failed during either table synchronization or
- * apply.
- */
- pgstat_report_subscription_error(MyLogicalRepWorker->subid,
- !am_tablesync_worker());
- }
+ /*
+ * Report the worker failed during sequence synchronization, table
+ * synchronization, or apply.
+ */
+ pgstat_report_subscription_error(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->type);
/* Disable the subscription */
StartTransactionCommand();
diff --git a/src/backend/utils/activity/pgstat_subscription.c b/src/backend/utils/activity/pgstat_subscription.c
index f9a1c831a07..35916772b9d 100644
--- a/src/backend/utils/activity/pgstat_subscription.c
+++ b/src/backend/utils/activity/pgstat_subscription.c
@@ -17,6 +17,7 @@
#include "postgres.h"
+#include "replication/worker_internal.h"
#include "utils/pgstat_internal.h"
@@ -24,7 +25,7 @@
* Report a subscription error.
*/
void
-pgstat_report_subscription_error(Oid subid, bool is_apply_error)
+pgstat_report_subscription_error(Oid subid, LogicalRepWorkerType wtype)
{
PgStat_EntryRef *entry_ref;
PgStat_BackendSubEntry *pending;
@@ -33,10 +34,25 @@ pgstat_report_subscription_error(Oid subid, bool is_apply_error)
InvalidOid, subid, NULL);
pending = entry_ref->pending;
- if (is_apply_error)
- pending->apply_error_count++;
- else
- pending->sync_error_count++;
+ switch (wtype)
+ {
+ case WORKERTYPE_APPLY:
+ pending->apply_error_count++;
+ break;
+
+ case WORKERTYPE_SEQUENCESYNC:
+ pending->seq_sync_error_count++;
+ break;
+
+ case WORKERTYPE_TABLESYNC:
+ pending->sync_error_count++;
+ break;
+
+ default:
+ /* Should never happen. */
+ Assert(0);
+ break;
+ }
}
/*
@@ -115,6 +131,7 @@ pgstat_subscription_flush_cb(PgStat_EntryRef *entry_ref, bool nowait)
#define SUB_ACC(fld) shsubent->stats.fld += localent->fld
SUB_ACC(apply_error_count);
+ SUB_ACC(seq_sync_error_count);
SUB_ACC(sync_error_count);
for (int i = 0; i < CONFLICT_NUM_TYPES; i++)
SUB_ACC(conflict_count[i]);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index a710508979e..1521d6e2ab4 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2203,7 +2203,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 12
+#define PG_STAT_GET_SUBSCRIPTION_STATS_COLS 13
Oid subid = PG_GETARG_OID(0);
TupleDesc tupdesc;
Datum values[PG_STAT_GET_SUBSCRIPTION_STATS_COLS] = {0};
@@ -2221,25 +2221,27 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 2, "apply_error_count",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 3, "sync_error_count",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 3, "seq_sync_error_count",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 4, "confl_insert_exists",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 4, "sync_error_count",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_update_origin_differs",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 5, "confl_insert_exists",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_exists",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 6, "confl_update_origin_differs",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_deleted",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 7, "confl_update_exists",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_update_missing",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 8, "confl_update_deleted",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_delete_origin_differs",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 9, "confl_update_missing",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_delete_missing",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 10, "confl_delete_origin_differs",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 11, "confl_multiple_unique_conflicts",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 11, "confl_delete_missing",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 12, "stats_reset",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 12, "confl_multiple_unique_conflicts",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 13, "stats_reset",
TIMESTAMPTZOID, -1, 0);
BlessTupleDesc(tupdesc);
@@ -2256,6 +2258,9 @@ pg_stat_get_subscription_stats(PG_FUNCTION_ARGS)
/* apply_error_count */
values[i++] = Int64GetDatum(subentry->apply_error_count);
+ /* seq_sync_error_count */
+ values[i++] = Int64GetDatum(subentry->seq_sync_error_count);
+
/* sync_error_count */
values[i++] = Int64GetDatum(subentry->sync_error_count);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 34b7fddb0e7..5cf9e12fcb9 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5704,9 +5704,9 @@
{ oid => '6231', descr => 'statistics: information about subscription stats',
proname => 'pg_stat_get_subscription_stats', provolatile => 's',
proparallel => 'r', prorettype => 'record', proargtypes => 'oid',
- proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
- proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{subid,subid,apply_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_deleted,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
+ proallargtypes => '{oid,oid,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,int8,timestamptz}',
+ proargmodes => '{i,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{subid,subid,apply_error_count,seq_sync_error_count,sync_error_count,confl_insert_exists,confl_update_origin_differs,confl_update_exists,confl_update_deleted,confl_update_missing,confl_delete_origin_differs,confl_delete_missing,confl_multiple_unique_conflicts,stats_reset}',
prosrc => 'pg_stat_get_subscription_stats' },
{ oid => '6118', descr => 'statistics: information about subscription',
proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 7ae503e71a2..a0610bb3e31 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -16,6 +16,7 @@
#include "portability/instr_time.h"
#include "postmaster/pgarch.h" /* for MAX_XFN_CHARS */
#include "replication/conflict.h"
+#include "replication/worker_internal.h"
#include "utils/backend_progress.h" /* for backward compatibility */ /* IWYU pragma: export */
#include "utils/backend_status.h" /* for backward compatibility */ /* IWYU pragma: export */
#include "utils/pgstat_kind.h"
@@ -108,6 +109,7 @@ typedef struct PgStat_FunctionCallUsage
typedef struct PgStat_BackendSubEntry
{
PgStat_Counter apply_error_count;
+ PgStat_Counter seq_sync_error_count;
PgStat_Counter sync_error_count;
PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
} PgStat_BackendSubEntry;
@@ -416,6 +418,7 @@ typedef struct PgStat_SLRUStats
typedef struct PgStat_StatSubEntry
{
PgStat_Counter apply_error_count;
+ PgStat_Counter seq_sync_error_count;
PgStat_Counter sync_error_count;
PgStat_Counter conflict_count[CONFLICT_NUM_TYPES];
TimestampTz stat_reset_timestamp;
@@ -769,7 +772,8 @@ extern PgStat_SLRUStats *pgstat_fetch_slru(void);
* Functions in pgstat_subscription.c
*/
-extern void pgstat_report_subscription_error(Oid subid, bool is_apply_error);
+extern void pgstat_report_subscription_error(Oid subid,
+ LogicalRepWorkerType wtype);
extern void pgstat_report_subscription_conflict(Oid subid, ConflictType type);
extern void pgstat_create_subscription(Oid subid);
extern void pgstat_drop_subscription(Oid subid);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 2bf968ae3d3..7c52181cbcb 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2191,6 +2191,7 @@ pg_stat_subscription| SELECT su.oid AS subid,
pg_stat_subscription_stats| SELECT ss.subid,
s.subname,
ss.apply_error_count,
+ ss.seq_sync_error_count,
ss.sync_error_count,
ss.confl_insert_exists,
ss.confl_update_origin_differs,
@@ -2202,7 +2203,7 @@ pg_stat_subscription_stats| SELECT ss.subid,
ss.confl_multiple_unique_conflicts,
ss.stats_reset
FROM pg_subscription s,
- LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_deleted, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, confl_multiple_unique_conflicts, stats_reset);
+ LATERAL pg_stat_get_subscription_stats(s.oid) ss(subid, apply_error_count, seq_sync_error_count, sync_error_count, confl_insert_exists, confl_update_origin_differs, confl_update_exists, confl_update_deleted, confl_update_missing, confl_delete_origin_differs, confl_delete_missing, confl_multiple_unique_conflicts, stats_reset);
pg_stat_sys_indexes| SELECT relid,
indexrelid,
schemaname,
diff --git a/src/test/subscription/t/026_stats.pl b/src/test/subscription/t/026_stats.pl
index 00a1c2fcd48..38a6ec542cf 100644
--- a/src/test/subscription/t/026_stats.pl
+++ b/src/test/subscription/t/026_stats.pl
@@ -21,7 +21,8 @@ $node_subscriber->start;
sub create_sub_pub_w_errors
{
- my ($node_publisher, $node_subscriber, $db, $table_name) = @_;
+ my ($node_publisher, $node_subscriber, $db, $table_name, $sequence_name)
+ = @_;
# Initial table setup on both publisher and subscriber. On subscriber we
# create the same tables but with primary keys. Also, insert some data that
# will conflict with the data replicated from publisher later.
@@ -32,6 +33,7 @@ sub create_sub_pub_w_errors
CREATE TABLE $table_name(a int);
ALTER TABLE $table_name REPLICA IDENTITY FULL;
INSERT INTO $table_name VALUES (1);
+ CREATE SEQUENCE $sequence_name;
COMMIT;
]);
$node_subscriber->safe_psql(
@@ -40,40 +42,62 @@ sub create_sub_pub_w_errors
BEGIN;
CREATE TABLE $table_name(a int primary key);
INSERT INTO $table_name VALUES (1);
+ CREATE SEQUENCE $sequence_name INCREMENT BY 10;
COMMIT;
]);
# Set up publication.
my $pub_name = $table_name . '_pub';
+ my $pub_seq_name = $sequence_name . '_pub';
my $publisher_connstr = $node_publisher->connstr . qq( dbname=$db);
- $node_publisher->safe_psql($db,
- qq(CREATE PUBLICATION $pub_name FOR TABLE $table_name));
+ $node_publisher->safe_psql(
+ $db,
+ qq[
+ CREATE PUBLICATION $pub_name FOR TABLE $table_name;
+ CREATE PUBLICATION $pub_seq_name FOR ALL SEQUENCES;
+ ]);
# Create subscription. The tablesync for table on subscription will enter into
- # infinite error loop due to violating the unique constraint.
+ # infinite error loop due to violating the unique constraint. The sequencesync
+ # will also fail due to different sequence increment values on publisher and
+ # subscriber.
my $sub_name = $table_name . '_sub';
$node_subscriber->safe_psql($db,
- qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name)
+ qq(CREATE SUBSCRIPTION $sub_name CONNECTION '$publisher_connstr' PUBLICATION $pub_name, $pub_seq_name)
);
$node_publisher->wait_for_catchup($sub_name);
- # Wait for the tablesync error to be reported.
+ # Wait for the tablesync and sequencesync error to be reported.
$node_subscriber->poll_query_until(
$db,
qq[
- SELECT sync_error_count > 0
- FROM pg_stat_subscription_stats
- WHERE subname = '$sub_name'
+ SELECT count(1) = 1 FROM pg_stat_subscription_stats
+ WHERE subname = '$sub_name' AND seq_sync_error_count > 0 AND sync_error_count > 0
+ ])
+ or die
+ qq(Timed out while waiting for sequencesync errors and tablesync errors for subscription '$sub_name');
+
+ # Change the sequence start value back to the default on the subscriber so
+ # it doesn't error out.
+ $node_subscriber->safe_psql($db,
+ qq(ALTER SEQUENCE $sequence_name INCREMENT 1));
+
+ # Wait for sequencesync to finish.
+ $node_subscriber->poll_query_until(
+ $db,
+ qq[
+ SELECT count(1) = 1 FROM pg_subscription_rel
+ WHERE srrelid = '$sequence_name'::regclass AND srsubstate = 'r'
])
or die
- qq(Timed out while waiting for tablesync errors for subscription '$sub_name');
+ qq(Timed out while waiting for subscriber to synchronize data for sequence '$sequence_name'.);
# Truncate test_tab1 so that tablesync worker can continue.
$node_subscriber->safe_psql($db, qq(TRUNCATE $table_name));
- # Wait for initial tablesync to finish.
+ # Wait for initial tablesync sync to finish.
$node_subscriber->poll_query_until(
$db,
qq[
@@ -136,14 +160,17 @@ is($result, qq(0),
# Create the publication and subscription with sync and apply errors
my $table1_name = 'test_tab1';
+my $sequence1_name = 'test_seq1';
my ($pub1_name, $sub1_name) =
create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
- $table1_name);
+ $table1_name, $sequence1_name);
-# Apply errors, sync errors, and conflicts are > 0 and stats_reset timestamp is NULL
+# Apply errors, sequencesync errors, tablesync errors, and conflicts are > 0 and stats_reset
+# timestamp is NULL.
is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count > 0,
+ seq_sync_error_count > 0,
sync_error_count > 0,
confl_insert_exists > 0,
confl_delete_missing > 0,
@@ -151,8 +178,8 @@ is( $node_subscriber->safe_psql(
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t|t|t),
- qq(Check that apply errors, sync errors, and conflicts are > 0 and stats_reset is NULL for subscription '$sub1_name'.)
+ qq(t|t|t|t|t|t),
+ qq(Check that apply errors, sequencesync errors, tablesync errors, and conflicts are > 0 and stats_reset is NULL for subscription '$sub1_name'.)
);
# Reset a single subscription
@@ -160,10 +187,12 @@ $node_subscriber->safe_psql($db,
qq(SELECT pg_stat_reset_subscription_stats((SELECT subid FROM pg_stat_subscription_stats WHERE subname = '$sub1_name')))
);
-# Apply errors, sync errors, and conflicts are 0 and stats_reset timestamp is not NULL
+# Apply errors, sequencesync errors, tablesync errors, and conflicts are 0 and
+# stats_reset timestamp is not NULL.
is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
+ seq_sync_error_count = 0,
sync_error_count = 0,
confl_insert_exists = 0,
confl_delete_missing = 0,
@@ -171,8 +200,8 @@ is( $node_subscriber->safe_psql(
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t|t|t),
- qq(Confirm that apply errors, sync errors, and conflicts are 0 and stats_reset is not NULL after reset for subscription '$sub1_name'.)
+ qq(t|t|t|t|t|t),
+ qq(Confirm that apply errors, sequencesync errors, tablesync errors, and conflicts are 0 and stats_reset is not NULL after reset for subscription '$sub1_name'.)
);
# Get reset timestamp
@@ -198,14 +227,17 @@ is( $node_subscriber->safe_psql(
# Make second subscription and publication
my $table2_name = 'test_tab2';
+my $sequence2_name = 'test_seq2';
my ($pub2_name, $sub2_name) =
create_sub_pub_w_errors($node_publisher, $node_subscriber, $db,
- $table2_name);
+ $table2_name, $sequence2_name);
-# Apply errors, sync errors, and conflicts are > 0 and stats_reset timestamp is NULL
+# Apply errors, sequencesync errors, tablesync errors, and conflicts are > 0
+# and stats_reset timestamp is NULL
is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count > 0,
+ seq_sync_error_count > 0,
sync_error_count > 0,
confl_insert_exists > 0,
confl_delete_missing > 0,
@@ -213,18 +245,20 @@ is( $node_subscriber->safe_psql(
FROM pg_stat_subscription_stats
WHERE subname = '$sub2_name')
),
- qq(t|t|t|t|t),
- qq(Confirm that apply errors, sync errors, and conflicts are > 0 and stats_reset is NULL for sub '$sub2_name'.)
+ qq(t|t|t|t|t|t),
+ qq(Confirm that apply errors, sequencesync errors, tablesync errors, and conflicts are > 0 and stats_reset is NULL for sub '$sub2_name'.)
);
# Reset all subscriptions
$node_subscriber->safe_psql($db,
qq(SELECT pg_stat_reset_subscription_stats(NULL)));
-# Apply errors, sync errors, and conflicts are 0 and stats_reset timestamp is not NULL
+# Apply errors, sequencesync errors, tablesync errors, and conflicts are 0 and
+# stats_reset timestamp is not NULL.
is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
+ seq_sync_error_count = 0,
sync_error_count = 0,
confl_insert_exists = 0,
confl_delete_missing = 0,
@@ -232,13 +266,14 @@ is( $node_subscriber->safe_psql(
FROM pg_stat_subscription_stats
WHERE subname = '$sub1_name')
),
- qq(t|t|t|t|t),
- qq(Confirm that apply errors, sync errors, and conflicts are 0 and stats_reset is not NULL for sub '$sub1_name' after reset.)
+ qq(t|t|t|t|t|t),
+ qq(Confirm that apply errors, sequencesync errors, tablesync errors, and conflicts are 0 and stats_reset is not NULL for sub '$sub1_name' after reset.)
);
is( $node_subscriber->safe_psql(
$db,
qq(SELECT apply_error_count = 0,
+ seq_sync_error_count = 0,
sync_error_count = 0,
confl_insert_exists = 0,
confl_delete_missing = 0,
@@ -246,8 +281,8 @@ is( $node_subscriber->safe_psql(
FROM pg_stat_subscription_stats
WHERE subname = '$sub2_name')
),
- qq(t|t|t|t|t),
- qq(Confirm that apply errors, sync errors, and conflicts are 0 and stats_reset is not NULL for sub '$sub2_name' after reset.)
+ qq(t|t|t|t|t|t),
+ qq(Confirm that apply errors, sequencesync errors, tablesync errors, errors, and conflicts are 0 and stats_reset is not NULL for sub '$sub2_name' after reset.)
);
$reset_time1 = $node_subscriber->safe_psql($db,
--
2.43.0
