public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'vignesh C' <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Greg Nancarrow <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: Failed transaction statistics to measure the logical replication progress
Date: Sat, 4 Dec 2021 13:01:58 +0000
Message-ID: <TYWPR01MB8362480F9148248F3D1B1213ED6B9@TYWPR01MB8362.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CALDaNm32HHjdwmuoF+Nw5CU70r819Kq+Tmat3bzkkvSv_2u=gA@mail.gmail.com>
References: <TYCPR01MB8373DC922E549FF3F031E9DAED8E9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAJcOf-cMU_1VJ1c+9fwEBgimkj+=hHcCjj_SUfbHvKTuN=1tuw@mail.gmail.com>
<TYCPR01MB8373B1614522B90BA96883E0ED929@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CALDaNm3-QTZnd4rxbrUfKMM4u_vgb23LO+KW3O9LX7X6b8mr8A@mail.gmail.com>
<CALDaNm130T4jr2RvpgY19SCY87UQ75wVbjrhjGj43bHV=9-LpQ@mail.gmail.com>
<TYCPR01MB8373FEB287F733C81C1E4D42ED989@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<TYCPR01MB8373D7AE0B6C62E5CDB03C6CED989@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<TYCPR01MB8373CFDCD60CCDB1C35CFA45ED999@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAD21AoBAREVDtAM+JGR3DcnwnOi1Nw7E_5m68ZwFUGkOkfDn1Q@mail.gmail.com>
<TYCPR01MB8373271BE5D4619302C05B00ED9A9@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CAA4eK1JCbhxuO8fEaAaw8e30=5kEDaFRXpV2+opAdMoEOH5LEg@mail.gmail.com>
<TYWPR01MB8362FCFAAD8D90424BEA2300ED9A9@TYWPR01MB8362.jpnprd01.prod.outlook.com>
<CAA4eK1KR41bRUuPeNBSGv2+q7ROKukS3myeAUqrZMD8MEwR0DQ@mail.gmail.com>
<CAD21AoC_p=gQKhhGwhDLzEU9rGwCWSzTk+a89XGG1JQKn-PiKw@mail.gmail.com>
<TYCPR01MB8373380ED159E8BD52F2D9A5ED689@TYCPR01MB8373.jpnprd01.prod.outlook.com>
<CALDaNm32HHjdwmuoF+Nw5CU70r819Kq+Tmat3bzkkvSv_2u=gA@mail.gmail.com>
On Friday, December 3, 2021 3:12 PM vignesh C <[email protected]> wrote:
> Thanks for the updated patch.
> Currently we are storing the commit count, error_count and abort_count for
> each table of the table sync operation. If we have thousands of tables, we will
> be storing the information for each of the tables.
> Shouldn't we be storing the consolidated information in this case.
> diff --git a/src/backend/replication/logical/tablesync.c
> b/src/backend/replication/logical/tablesync.c
> index f07983a..02e9486 100644
> --- a/src/backend/replication/logical/tablesync.c
> +++ b/src/backend/replication/logical/tablesync.c
> @@ -1149,6 +1149,11 @@ copy_table_done:
> MyLogicalRepWorker->relstate_lsn = *origin_startpos;
> SpinLockRelease(&MyLogicalRepWorker->relmutex);
>
> + /* Report the success of table sync. */
> + pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
> +
> MyLogicalRepWorker->relid,
> +
> 0 /* no logical message type */ );
Okay.
I united all stats into that of apply worker.
In line with this change, I fixed the TAP tests as well
to cover the updates of stats done by table sync workers.
Also, during my self-review, I noticed that
I should call pgstat_report_subworker_xact_end() before
process_syncing_tables() because it can lead to process
exit, which results in missing one increment of the stats columns.
I noted this point in a comment as well.
Best Regards,
Takamichi Osumi
Attachments:
[application/octet-stream] v15-0001-Extend-pg_stat_subscription_workers-to-include-g.patch (24.3K, ../TYWPR01MB8362480F9148248F3D1B1213ED6B9@TYWPR01MB8362.jpnprd01.prod.outlook.com/2-v15-0001-Extend-pg_stat_subscription_workers-to-include-g.patch)
download | inline diff:
From 1b756a897eee84e068e039333666278112c235b0 Mon Sep 17 00:00:00 2001
From: Takamichi Osumi <[email protected]>
Date: Sat, 4 Dec 2021 12:27:10 +0000
Subject: [PATCH v15] Extend pg_stat_subscription_workers to include general
transaction statistics
Categorize transactions of logical replication subscriber
into three types (commit, error, abort) and introduce
cumulative columns of those numbers.
The stats for table sync worker is merged into those of
the apply worker.
At present, there's no special consideration to statistics
of spool file logic such as amount of data spooled to disk,
its count or corresponding statistics associated with
STREAM ABORT. However, those can be added later.
Author: Takamichi Osumi
Reviewed-by: Amit Kapila, Masahiko Sawada, Hou Zhijie, Greg Nancarrow, Vignesh C, Ajin Cherian
Discussion: https://www.postgresql.org/message-id/OSBPR01MB48887CA8F40C8D984A6DC00CED199%40OSBPR01MB4888.jpnprd01.prod.outlook.com
---
doc/src/sgml/monitoring.sgml | 34 ++++-
src/backend/catalog/system_views.sql | 3 +
src/backend/postmaster/pgstat.c | 79 +++++++++++
src/backend/replication/logical/tablesync.c | 7 +
src/backend/replication/logical/worker.c | 20 +++
src/backend/utils/adt/pgstatfuncs.c | 25 +++-
src/include/catalog/pg_proc.dat | 6 +-
src/include/pgstat.h | 29 ++++
src/test/regress/expected/rules.out | 5 +-
src/test/subscription/t/027_worker_xact_stats.pl | 162 +++++++++++++++++++++++
src/tools/pgindent/typedefs.list | 1 +
11 files changed, 358 insertions(+), 13 deletions(-)
create mode 100644 src/test/subscription/t/027_worker_xact_stats.pl
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 62f2a33..e10ca98 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -629,8 +629,8 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser
<row>
<entry><structname>pg_stat_subscription_workers</structname><indexterm><primary>pg_stat_subscription_workers</primary></indexterm></entry>
- <entry>One row per subscription worker, showing statistics about errors
- that occurred on that subscription worker.
+ <entry>One row per subscription worker, showing statistics about transactions
+ and errors that occurred on that subscription worker.
See <link linkend="monitoring-pg-stat-subscription-workers">
<structname>pg_stat_subscription_workers</structname></link> for details.
</entry>
@@ -3123,6 +3123,36 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i
<row>
<entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>commit_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions successfully applied in this subscription.
+ COMMIT and COMMIT PREPARED increments this counter.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>error_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions that failed to be applied by the table
+ sync worker or main apply worker in this subscription.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>abort_count</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions aborted in this subscription.
+ ROLLBACK PREPARED increments this counter.
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
<structfield>last_error_relid</structfield> <type>oid</type>
</para>
<para>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 61b515c..d2ca10a 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1267,6 +1267,9 @@ CREATE VIEW pg_stat_subscription_workers AS
w.subid,
s.subname,
w.subrelid,
+ w.commit_count,
+ w.error_count,
+ w.abort_count,
w.last_error_relid,
w.last_error_command,
w.last_error_xid,
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 7264d2c..5fe49ae 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -382,6 +382,7 @@ static void pgstat_recv_replslot(PgStat_MsgReplSlot *msg, int len);
static void pgstat_recv_tempfile(PgStat_MsgTempFile *msg, int len);
static void pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len);
static void pgstat_recv_subworker_error(PgStat_MsgSubWorkerError *msg, int len);
+static void pgstat_recv_subworker_xact_end(PgStat_MsgSubWorkerXactEnd *msg, int len);
/* ------------------------------------------------------------
* Public functions called from postmaster follow
@@ -1949,6 +1950,30 @@ pgstat_report_replslot_drop(const char *slotname)
}
/* ----------
+ * pgstat_report_subworker_xact_end() -
+ *
+ * Tell the collector that worker transaction has successfully completed.
+ * ----------
+ */
+void
+pgstat_report_subworker_xact_end(Oid subid, LogicalRepMsgType command)
+{
+ PgStat_MsgSubWorkerXactEnd msg;
+
+ /* LOGICAL_REP_MSG_COMMIT cares about table sync's commit too */
+ Assert(command == LOGICAL_REP_MSG_COMMIT ||
+ command == LOGICAL_REP_MSG_STREAM_COMMIT ||
+ command == LOGICAL_REP_MSG_COMMIT_PREPARED ||
+ command == LOGICAL_REP_MSG_ROLLBACK_PREPARED);
+
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBWORKERXACTEND);
+ msg.m_databaseid = MyDatabaseId;
+ msg.m_subid = subid;
+ msg.m_command = command;
+ pgstat_send(&msg, sizeof(PgStat_MsgSubWorkerXactEnd));
+}
+
+/* ----------
* pgstat_report_subworker_error() -
*
* Tell the collector about the subscription worker error.
@@ -3746,6 +3771,10 @@ PgstatCollectorMain(int argc, char *argv[])
pgstat_recv_subworker_error(&msg.msg_subworkererror, len);
break;
+ case PGSTAT_MTYPE_SUBWORKERXACTEND:
+ pgstat_recv_subworker_xact_end(&msg.msg_subworkerxactend, len);
+ break;
+
default:
break;
}
@@ -3965,6 +3994,9 @@ pgstat_get_subworker_entry(PgStat_StatDBEntry *dbentry, Oid subid, Oid subrelid,
/* If not found, initialize the new one */
if (!found)
{
+ subwentry->commit_count = 0;
+ subwentry->error_count = 0;
+ subwentry->abort_count = 0;
subwentry->last_error_relid = InvalidOid;
subwentry->last_error_command = 0;
subwentry->last_error_xid = InvalidTransactionId;
@@ -6153,6 +6185,39 @@ pgstat_recv_subscription_purge(PgStat_MsgSubscriptionPurge *msg, int len)
}
/* ----------
+ * pgstat_recv_subworker_xact_end() -
+ *
+ * Process a SUBWORKERXACTEND message.
+ * ----------
+ */
+static void
+pgstat_recv_subworker_xact_end(PgStat_MsgSubWorkerXactEnd *msg, int len)
+{
+ PgStat_StatDBEntry *dbentry;
+ PgStat_StatSubWorkerEntry *wentry;
+
+ dbentry = pgstat_get_db_entry(msg->m_databaseid, true);
+ wentry = pgstat_get_subworker_entry(dbentry, msg->m_subid,
+ InvalidOid, true);
+ Assert(wentry);
+
+ switch (msg->m_command)
+ {
+ case LOGICAL_REP_MSG_COMMIT:
+ case LOGICAL_REP_MSG_STREAM_COMMIT:
+ case LOGICAL_REP_MSG_COMMIT_PREPARED:
+ wentry->commit_count++;
+ break;
+ case LOGICAL_REP_MSG_ROLLBACK_PREPARED:
+ wentry->abort_count++;
+ break;
+ default:
+ elog(ERROR, "unexpected logical message type as transaction end");
+ break;
+ }
+}
+
+/* ----------
* pgstat_recv_subworker_error() -
*
* Process a SUBWORKERERROR message.
@@ -6193,6 +6258,20 @@ pgstat_recv_subworker_error(PgStat_MsgSubWorkerError *msg, int len)
subwentry->last_error_time = msg->m_timestamp;
strlcpy(subwentry->last_error_message, msg->m_message,
PGSTAT_SUBWORKERERROR_MSGLEN);
+
+ /*
+ * If this is a new error reported by table sync worker, consolidate this
+ * error count into the entry of apply worker.
+ */
+ if (OidIsValid(msg->m_subrelid))
+ {
+ /* Gain the apply worker stats */
+ subwentry = pgstat_get_subworker_entry(dbentry, msg->m_subid,
+ InvalidOid, true);
+ subwentry->error_count++;
+ }
+ else
+ subwentry->error_count++; /* increment the apply worker's counter. */
}
/* ----------
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a..e2bfefd 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1150,6 +1150,13 @@ copy_table_done:
SpinLockRelease(&MyLogicalRepWorker->relmutex);
/*
+ * Report the success of table sync as one commit to consolidate all
+ * transaction stats into one record.
+ */
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ LOGICAL_REP_MSG_COMMIT);
+
+ /*
* Finally, wait until the main apply worker tells us to catch up and then
* return to let LogicalRepApplyLoop do it.
*/
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 2e79302..d897e63 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -815,6 +815,14 @@ apply_handle_commit(StringInfo s)
apply_handle_commit_internal(&commit_data);
+ /*
+ * Report ealier than the call of process_syncing_tables() not to miss an
+ * increment of commit_count in case it leads to the process exit. See
+ * process_syncing_tables_for_apply().
+ */
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ LOGICAL_REP_MSG_COMMIT);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
@@ -960,6 +968,10 @@ apply_handle_commit_prepared(StringInfo s)
store_flush_position(prepare_data.end_lsn);
in_remote_transaction = false;
+ /* Update stats */
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ LOGICAL_REP_MSG_COMMIT_PREPARED);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
@@ -1011,6 +1023,10 @@ apply_handle_rollback_prepared(StringInfo s)
store_flush_position(rollback_data.rollback_end_lsn);
in_remote_transaction = false;
+ /* Update stats */
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ LOGICAL_REP_MSG_ROLLBACK_PREPARED);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(rollback_data.rollback_end_lsn);
@@ -1435,6 +1451,10 @@ apply_handle_stream_commit(StringInfo s)
/* unlink the files with serialized changes and subxact info */
stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+ /* Update stats */
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ LOGICAL_REP_MSG_STREAM_COMMIT);
+
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index f529c15..25a9fd2 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2415,7 +2415,7 @@ pg_stat_get_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_stat_get_subscription_worker(PG_FUNCTION_ARGS)
{
-#define PG_STAT_GET_SUBSCRIPTION_WORKER_COLS 8
+#define PG_STAT_GET_SUBSCRIPTION_WORKER_COLS 11
Oid subid = PG_GETARG_OID(0);
Oid subrelid;
TupleDesc tupdesc;
@@ -2442,17 +2442,23 @@ pg_stat_get_subscription_worker(PG_FUNCTION_ARGS)
OIDOID, -1, 0);
TupleDescInitEntry(tupdesc, (AttrNumber) 2, "subrelid",
OIDOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 3, "last_error_relid",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 3, "commit_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 4, "error_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 5, "abort_count",
+ INT8OID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_error_relid",
OIDOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 4, "last_error_command",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 7, "last_error_command",
TEXTOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 5, "last_error_xid",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_error_xid",
XIDOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 6, "last_error_count",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 9, "last_error_count",
INT8OID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 7, "last_error_message",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 10, "last_error_message",
TEXTOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 8, "last_error_time",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 11, "last_error_time",
TIMESTAMPTZOID, -1, 0);
BlessTupleDesc(tupdesc);
@@ -2470,6 +2476,11 @@ pg_stat_get_subscription_worker(PG_FUNCTION_ARGS)
else
nulls[i++] = true;
+ /* transaction stats */
+ values[i++] = Int64GetDatum(wentry->commit_count);
+ values[i++] = Int64GetDatum(wentry->error_count);
+ values[i++] = Int64GetDatum(wentry->abort_count);
+
/* last_error_relid */
if (OidIsValid(wentry->last_error_relid))
values[i++] = ObjectIdGetDatum(wentry->last_error_relid);
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 79d787c..e78a4a2 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5393,9 +5393,9 @@
proname => 'pg_stat_get_subscription_worker', prorows => '1', proisstrict => 'f',
proretset => 't', provolatile => 's', proparallel => 'r',
prorettype => 'record', proargtypes => 'oid oid',
- proallargtypes => '{oid,oid,oid,oid,oid,text,xid,int8,text,timestamptz}',
- proargmodes => '{i,i,o,o,o,o,o,o,o,o}',
- proargnames => '{subid,subrelid,subid,subrelid,last_error_relid,last_error_command,last_error_xid,last_error_count,last_error_message,last_error_time}',
+ proallargtypes => '{oid,oid,oid,oid,int8,int8,int8,oid,text,xid,int8,text,timestamptz}',
+ proargmodes => '{i,i,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{subid,subrelid,subid,subrelid,commit_count,error_count,abort_count,last_error_relid,last_error_command,last_error_xid,last_error_count,last_error_message,last_error_time}',
prosrc => 'pg_stat_get_subscription_worker' },
{ 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 5b51b58..94ad151 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -86,6 +86,7 @@ typedef enum StatMsgType
PGSTAT_MTYPE_DISCONNECT,
PGSTAT_MTYPE_SUBSCRIPTIONPURGE,
PGSTAT_MTYPE_SUBWORKERERROR,
+ PGSTAT_MTYPE_SUBWORKERXACTEND,
} StatMsgType;
/* ----------
@@ -558,6 +559,25 @@ typedef struct PgStat_MsgSubscriptionPurge
} PgStat_MsgSubscriptionPurge;
/* ----------
+ * PgStat_MsgSubscriptionXactEnd Sent by the apply worker to report transaction
+ * ends.
+ * ----------
+ */
+typedef struct PgStat_MsgSubWorkerXactEnd
+{
+ PgStat_MsgHdr m_hdr;
+
+ /* determine the worker entry */
+ Oid m_databaseid;
+ Oid m_subid;
+ Oid m_subrelid;
+
+ /* necessary to determine column to increment */
+ LogicalRepMsgType m_command;
+
+} PgStat_MsgSubWorkerXactEnd;
+
+/* ----------
* PgStat_MsgSubWorkerError Sent by the apply worker or the table sync
* worker to report the error occurred while
* processing changes.
@@ -769,6 +789,7 @@ typedef union PgStat_Msg
PgStat_MsgDisconnect msg_disconnect;
PgStat_MsgSubscriptionPurge msg_subscriptionpurge;
PgStat_MsgSubWorkerError msg_subworkererror;
+ PgStat_MsgSubWorkerXactEnd msg_subworkerxactend;
} PgStat_Msg;
@@ -1010,6 +1031,13 @@ typedef struct PgStat_StatSubWorkerEntry
PgStat_StatSubWorkerKey key; /* hash key (must be first) */
/*
+ * Cumulative transaction statistics of subscription worker
+ */
+ PgStat_Counter commit_count;
+ PgStat_Counter error_count;
+ PgStat_Counter abort_count;
+
+ /*
* Subscription worker error statistics representing an error that
* occurred during application of changes or the initial table
* synchronization.
@@ -1131,6 +1159,7 @@ extern void pgstat_report_checksum_failure(void);
extern void pgstat_report_replslot(const PgStat_StatReplSlotEntry *repSlotStat);
extern void pgstat_report_replslot_create(const char *slotname);
extern void pgstat_report_replslot_drop(const char *slotname);
+extern void pgstat_report_subworker_xact_end(Oid subid, LogicalRepMsgType command);
extern void pgstat_report_subworker_error(Oid subid, Oid subrelid, Oid relid,
LogicalRepMsgType command,
TransactionId xid, const char *errmsg);
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index b58b062..6f62f0d 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2097,6 +2097,9 @@ pg_stat_subscription| SELECT su.oid AS subid,
pg_stat_subscription_workers| SELECT w.subid,
s.subname,
w.subrelid,
+ w.commit_count,
+ w.error_count,
+ w.abort_count,
w.last_error_relid,
w.last_error_command,
w.last_error_xid,
@@ -2110,7 +2113,7 @@ pg_stat_subscription_workers| SELECT w.subid,
SELECT pg_subscription_rel.srsubid AS subid,
pg_subscription_rel.srrelid AS relid
FROM pg_subscription_rel) sr,
- (LATERAL pg_stat_get_subscription_worker(sr.subid, sr.relid) w(subid, subrelid, last_error_relid, last_error_command, last_error_xid, last_error_count, last_error_message, last_error_time)
+ (LATERAL pg_stat_get_subscription_worker(sr.subid, sr.relid) w(subid, subrelid, commit_count, error_count, abort_count, last_error_relid, last_error_command, last_error_xid, last_error_count, last_error_message, last_error_time)
JOIN pg_subscription s ON ((w.subid = s.oid)));
pg_stat_sys_indexes| SELECT pg_stat_all_indexes.relid,
pg_stat_all_indexes.indexrelid,
diff --git a/src/test/subscription/t/027_worker_xact_stats.pl b/src/test/subscription/t/027_worker_xact_stats.pl
new file mode 100644
index 0000000..31dbea1
--- /dev/null
+++ b/src/test/subscription/t/027_worker_xact_stats.pl
@@ -0,0 +1,162 @@
+
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+# Tests for subscription worker statistics during apply.
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More tests => 1;
+
+# Create publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->append_conf(
+ 'postgresql.conf', qq[
+logical_decoding_work_mem = 64kB
+max_prepared_transactions = 10
+max_wal_senders = 10
+wal_sender_timeout = 0
+]);
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->append_conf(
+ 'postgresql.conf', qq[
+max_prepared_transactions = 10
+]);
+$node_subscriber->start;
+
+# Setup structures on the publisher and the subscriber.
+# Insert one record that it will cause duplication error
+# during the initial table sync.
+$node_publisher->safe_psql(
+ 'postgres', qq[
+CREATE TABLE test_tab1 (a int);
+CREATE TABLE test_tab2 (a int);
+INSERT INTO test_tab1 VALUES (0);
+INSERT INTO test_tab2 VALUES (0);
+]);
+$node_subscriber->safe_psql(
+ 'postgres', qq[
+CREATE TABLE test_tab1 (a int primary key);
+CREATE TABLE test_tab2 (a int primary key);
+INSERT INTO test_tab2 VALUES (0);
+]);
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE test_tab1, test_tab2");
+
+# There's no entry at the beginning
+my $result = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) FROM pg_stat_subscription_workers");
+is($result, q(0), 'no entry for transaction stats yet');
+
+my $appname = 'tap_sub';
+my $subopts = 'streaming = on, two_phase = on';
+$node_subscriber->safe_psql(
+ 'postgres', qq[
+CREATE SUBSCRIPTION tap_sub
+CONNECTION '$publisher_connstr application_name=$appname'
+PUBLICATION tap_pub WITH ($subopts);
+]);
+
+# Table synchronization tests. The commit_count gets
+# incremented by the number of successful sync completions.
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 1"
+) or die "didn't get table sync commit.";
+
+# Resolve the error after confirming error_count has been incremented.
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where error_count > 0;"
+) or die "didn't get any error of table sync.";
+$node_subscriber->safe_psql('postgres', "DELETE FROM test_tab2");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 2;"
+) or die "didn't get the 2nd commit of table sync.";
+
+# COMMIT
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab1 VALUES (1); COMMIT;");
+
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers WHERE commit_count = 3"
+) or die "didn't get updates of xact stats by commit";
+
+# PREPARE & COMMIT PREPARED
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab1 VALUES (2); PREPARE TRANSACTION 'gid1'");
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gid1';");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 4"
+) or die "didn't get updates of xact stats by commit prepared";
+
+# STREAM COMMIT
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab1 VALUES(generate_series(1001, 2000)); COMMIT;"
+);
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 5"
+) or die "didn't get updates of xact stats by stream commit";
+
+# STREAM PREPARE & COMMIT PREPARED
+# This should restore the xact size before the shutdown.
+$node_publisher->safe_psql(
+ 'postgres', q[
+BEGIN;
+INSERT INTO test_tab1 VALUES(generate_series(2001, 3000));
+PREPARE TRANSACTION 'gid2';
+]);
+
+# Commit prepared increments the commit_count.
+$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gid2'");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 6"
+ )
+ or die
+ "didn't get updates of xact stats by stream prepare and commit prepared";
+
+# ROLLBACK PREPARED
+$node_publisher->safe_psql(
+ 'postgres', q[
+BEGIN;
+INSERT INTO test_tab1 VALUES (4);
+PREPARE TRANSACTION 'gid3';
+]);
+$node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'gid3'");
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where abort_count = 1"
+) or die "didn't get updates of xact stats by rollback prepared";
+
+# Test of error stats by duplication error.
+# Obtain the current error_count that was increased by table sync error
+# and compare it with the result after making some errors of apply.
+my $prev_err_count = $node_subscriber->safe_psql('postgres',
+ "SELECT error_count FROM pg_stat_subscription_workers WHERE subrelid IS NULL"
+);
+
+$node_publisher->safe_psql(
+ 'postgres', q[
+BEGIN;
+INSERT INTO test_tab1 VALUES (1);
+COMMIT;
+]);
+
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers WHERE subrelid IS NULL AND error_count > $prev_err_count"
+) or die "didn't get updates of xact stats by error";
+
+# Check the cleanup of DROP SUBSCRIPTION
+$node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
+
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 0 FROM pg_stat_subscription_workers")
+ or die "drop subscription failed to clean up the worker stats";
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index f41ef0d..2d195ae 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1945,6 +1945,7 @@ PgStat_MsgResetslrucounter
PgStat_MsgSLRU
PgStat_MsgSubscriptionPurge
PgStat_MsgSubWorkerError
+PgStat_MsgSubWorkerXactEnd
PgStat_MsgTabpurge
PgStat_MsgTabstat
PgStat_MsgTempFile
--
1.8.3.1
view thread (113+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Failed transaction statistics to measure the logical replication progress
In-Reply-To: <TYWPR01MB8362480F9148248F3D1B1213ED6B9@TYWPR01MB8362.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox