public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Masahiko Sawada' <[email protected]>
To: Amit Kapila <[email protected]>
Cc: vignesh C <[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: Wed, 1 Dec 2021 09:34:05 +0000
Message-ID: <TYCPR01MB8373380ED159E8BD52F2D9A5ED689@TYCPR01MB8373.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAD21AoC_p=gQKhhGwhDLzEU9rGwCWSzTk+a89XGG1JQKn-PiKw@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>
On Friday, November 19, 2021 11:11 PM Masahiko Sawada <[email protected]> wrote:
> Besides that, I’m not sure how useful commit_bytes, abort_bytes, and
> error_bytes are. I originally thought these statistics track the size of received
> data, i.g., how much data is transferred from the publisher and processed on
> the subscriber. But what the view currently has is how much memory is used in
> the subscription worker. The subscription worker emulates
> ReorderBufferChangeSize() on the subscriber side but, as the comment of
> update_apply_change_size() mentions, the size in the view is not accurate:
...
> I guess that the purpose of these values is to compare them to total_bytes,
> stream_byte, and spill_bytes but if the calculation is not accurate, does it mean
> that the more stats are updated, the more the stats will be getting inaccurate?
Thanks for your comment !
I tried to solve your concerns about byte columns but there are really difficult issues to solve.
For example, to begin with the messages of apply worker are different from those of
reorder buffer.
Therefore, I decided to split the previous patch and make counter columns go first.
v14 was checked by pgperltidy and pgindent.
This patch can be applied to the PG whose commit id is after 8d74fc9 (introduction of
pg_stat_subscription_workers).
Best Regards,
Takamichi Osumi
Attachments:
[application/octet-stream] v14-0001-Extend-pg_stat_subscription_workers-to-include-g.patch (22.5K, ../TYCPR01MB8373380ED159E8BD52F2D9A5ED689@TYCPR01MB8373.jpnprd01.prod.outlook.com/2-v14-0001-Extend-pg_stat_subscription_workers-to-include-g.patch)
download | inline diff:
From 5869058de564558af1c84d185f74f24211d24035 Mon Sep 17 00:00:00 2001
From: Takamichi Osumi <[email protected]>
Date: Wed, 1 Dec 2021 09:01:09 +0000
Subject: [PATCH v14] 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.
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
Discussed & 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 | 74 +++++++++++++
src/backend/replication/logical/tablesync.c | 5 +
src/backend/replication/logical/worker.c | 16 +++
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 | 131 +++++++++++++++++++++++
src/tools/pgindent/typedefs.list | 1 +
11 files changed, 316 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..a6cb0e0 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,31 @@ 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, Oid subrel, LogicalRepMsgType command)
+{
+ PgStat_MsgSubWorkerXactEnd msg;
+
+ Assert(command == 0 /* table sync worker */ ||
+ 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_subrelid = subrel;
+ msg.m_command = command;
+ pgstat_send(&msg, sizeof(PgStat_MsgSubWorkerXactEnd));
+}
+
+/* ----------
* pgstat_report_subworker_error() -
*
* Tell the collector about the subscription worker error.
@@ -3746,6 +3772,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 +3995,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 +6186,46 @@ 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,
+ msg->m_subrelid, true);
+ Assert(wentry);
+
+ /* table sync worker */
+ if (msg->m_command == 0)
+ wentry->commit_count++;
+ else
+ {
+ /* apply worker */
+ 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.
@@ -6186,6 +6259,7 @@ pgstat_recv_subworker_error(PgStat_MsgSubWorkerError *msg, int len)
}
/* Otherwise, update the error information */
+ subwentry->error_count++;
subwentry->last_error_relid = msg->m_relid;
subwentry->last_error_command = msg->m_command;
subwentry->last_error_xid = msg->m_xid;
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 */ );
+
/*
* 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..0a68de3 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -818,6 +818,10 @@ apply_handle_commit(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ InvalidOid,
+ LOGICAL_REP_MSG_COMMIT);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
@@ -963,6 +967,10 @@ apply_handle_commit_prepared(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(prepare_data.end_lsn);
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ InvalidOid,
+ LOGICAL_REP_MSG_COMMIT_PREPARED);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
@@ -1014,6 +1022,10 @@ apply_handle_rollback_prepared(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(rollback_data.rollback_end_lsn);
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ InvalidOid,
+ LOGICAL_REP_MSG_ROLLBACK_PREPARED);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
}
@@ -1438,6 +1450,10 @@ apply_handle_stream_commit(StringInfo s)
/* Process any tables that are being synchronized in parallel. */
process_syncing_tables(commit_data.end_lsn);
+ pgstat_report_subworker_xact_end(MyLogicalRepWorker->subid,
+ InvalidOid,
+ LOGICAL_REP_MSG_STREAM_COMMIT);
+
pgstat_report_activity(STATE_IDLE, NULL);
reset_apply_error_context_info();
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..f976651 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 or the table sync worker
+ * to report successful 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, Oid subrel, 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..04e0abf
--- /dev/null
+++ b/src/test/subscription/t/027_worker_xact_stats.pl
@@ -0,0 +1,131 @@
+
+# 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
+$node_publisher->safe_psql('postgres', "CREATE TABLE test_tab (a int)");
+$node_subscriber->safe_psql('postgres',
+ "CREATE TABLE test_tab (a int primary key)");
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION tap_pub FOR TABLE test_tab");
+
+my $appname = 'tap_sub';
+my $subopts = 'streaming = on, two_phase = on, copy_data = false';
+$node_subscriber->safe_psql(
+ 'postgres', qq[
+CREATE SUBSCRIPTION tap_sub
+CONNECTION '$publisher_connstr application_name=$appname'
+PUBLICATION tap_pub WITH ($subopts);
+]);
+
+$node_publisher->wait_for_catchup($appname);
+
+# 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');
+
+# COMMIT
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab VALUES (1); COMMIT;");
+
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers where commit_count = 1"
+) or die "didn't get updates of xact stats by commit";
+
+# PREPARE & COMMIT PREPARED
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab 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 = 2"
+) or die "didn't get updates of xact stats by commit prepared";
+
+# STREAM COMMIT
+$node_publisher->safe_psql('postgres',
+ "BEGIN; INSERT INTO test_tab VALUES(generate_series(1001, 2000)); 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 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_tab 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 = 4"
+ )
+ 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_tab 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";
+
+# error stats (by duplication error)
+$node_publisher->safe_psql(
+ 'postgres', q[
+BEGIN;
+INSERT INTO test_tab VALUES (1);
+COMMIT;
+]);
+
+$node_subscriber->poll_query_until('postgres',
+ "SELECT count(1) = 1 FROM pg_stat_subscription_workers WHERE error_count > 0"
+) 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: <TYCPR01MB8373380ED159E8BD52F2D9A5ED689@TYCPR01MB8373.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