public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: 'Amit Kapila' <[email protected]>
To: Masahiko Sawada <[email protected]>
Cc: [email protected] <[email protected]>
Subject: RE: Failed transaction statistics to measure the logical replication progress
Date: Thu, 2 Sep 2021 02:22:42 +0000
Message-ID: <TY2PR01MB48905729E8CB81805CCF64C0EDCE9@TY2PR01MB4890.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1KMT8biciVqTBoZ9gYV-Gf297JFeNhJaxZNmFrZL8m2jA@mail.gmail.com>
References: <OSBPR01MB48887CA8F40C8D984A6DC00CED199@OSBPR01MB4888.jpnprd01.prod.outlook.com>
<CAD21AoAixHnpjG-TtnSejJ2Dv1VsrzGr3oVPSRFhYjz3Z8_XZA@mail.gmail.com>
<OSBPR01MB4888D9048D3D7E7C3D035D54EDEF9@OSBPR01MB4888.jpnprd01.prod.outlook.com>
<CAD21AoCbj6YBNmdG9gAXf=dPE+bdY95ENwkh4q5wPL0q=qMzVw@mail.gmail.com>
<CAA4eK1+tOV-+ssGjj1zq+nAL8a9LfPsxbtyupZGvZ0U7nV0A7g@mail.gmail.com>
<CAD21AoAO0004L1kv9+uArrhT7d=vFj=2ZUKFbUfpZnHJku-37g@mail.gmail.com>
<CAA4eK1+CfMrJeA8TR__uFMQVZk50L+OGUEW-8h5G0hD_erfaSw@mail.gmail.com>
<CAD21AoDtn1bY4CYkBDXMgxpyBQZTAh45HMTXfAjQ3nY_G_=7Mg@mail.gmail.com>
<CAA4eK1KNqNnWq=wep5OALjqRRKbqp3pRjnqBJi2CPfMYJMUjrQ@mail.gmail.com>
<CAD21AoA6X7UtbewgRVetcwh2tuAp=umR5n_bE69ZBj3mN6R31Q@mail.gmail.com>
<CAA4eK1KMT8biciVqTBoZ9gYV-Gf297JFeNhJaxZNmFrZL8m2jA@mail.gmail.com>
Hi
I've made a new patch to extend pg_stat_subscription
as suggested in [1] to have columns
xact_commit, xact_error and independent xact_abort mentioned in [2].
Also, during discussion, we touched a topic if we should
include data sizes for each column above and concluded that it's
better to have ones. Accordingly, I've implemented corresponding
columns to show the data sizes as well.
Note that this patch depends on v12 patchset of apply error callback
provided in [3]. Therfore, applying the patchset first is required,
if you would like to test my patch.
[1] - https://www.postgresql.org/message-id/CAA4eK1%2BtOV-%2BssGjj1zq%2BnAL8a9LfPsxbtyupZGvZ0U7nV0A7g%40ma...
[2] - https://www.postgresql.org/message-id/CAA4eK1KMT8biciVqTBoZ9gYV-Gf297JFeNhJaxZNmFrZL8m2jA%40mail.gma...
[3] - https://www.postgresql.org/message-id/CAD21AoA5HrhXqwbYLpSobGzV6rWoJmH3-NB9J3YarKDwARBj4w%40mail.gma...
Best Regards,
Takamichi Osumi
Attachments:
[application/octet-stream] extend_subscription_stats_of_transaction_v02.patch (24.0K, ../TY2PR01MB48905729E8CB81805CCF64C0EDCE9@TY2PR01MB4890.jpnprd01.prod.outlook.com/2-extend_subscription_stats_of_transaction_v02.patch)
download | inline diff:
diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml
index 0c02e46..ee36c3b 100644
--- a/doc/src/sgml/monitoring.sgml
+++ b/doc/src/sgml/monitoring.sgml
@@ -3046,6 +3046,60 @@ 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>xact_commit</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions successfully applied in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>xact_commit_bytes</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Amount of transactions data successfully applied in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>xact_error</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions unsuccessfully applied in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>xact_error_bytes</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Amount of transactions data unsuccessfully applied in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>xact_abort</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Number of transactions aborted in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>xact_abort_bytes</structfield> <type>bigint</type>
+ </para>
+ <para>
+ Amount of transactions data aborted in this subscription
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
<structfield>latest_end_time</structfield> <type>timestamp with time zone</type>
</para>
<para>
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 449692a..109ba92 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -909,10 +909,37 @@ CREATE VIEW pg_stat_subscription AS
st.last_msg_send_time,
st.last_msg_receipt_time,
st.latest_end_lsn,
+ pg_stat_get_subscription_xact_commit(su.oid, st.relid) AS xact_commit,
+ pg_stat_get_subscription_xact_commit_bytes(su.oid, st.relid) AS xact_commit_bytes,
+ pg_stat_get_subscription_xact_error(su.oid, st.relid) AS xact_error,
+ pg_stat_get_subscription_xact_error_bytes(su.oid, st.relid) AS xact_error_bytes,
+ pg_stat_get_subscription_xact_abort(su.oid, st.relid) AS xact_abort,
+ pg_stat_get_subscription_xact_abort_bytes(su.oid, st.relid) AS xact_abort_bytes,
st.latest_end_time
FROM pg_subscription su
- LEFT JOIN pg_stat_get_subscription(NULL) st
- ON (st.subid = su.oid);
+ LEFT JOIN
+ (SELECT
+ s.subid,
+ s.relid,
+ s.pid,
+ s.received_lsn,
+ s.last_msg_send_time,
+ s.last_msg_receipt_time,
+ s.latest_end_lsn,
+ s.latest_end_time
+ FROM pg_stat_get_subscription(NULL) s
+ UNION -- acquire relid of table sync worker
+ SELECT
+ r.srsubid,
+ r.srrelid,
+ NULL as pid,
+ NULL as received_lsn,
+ NULL as last_msg_send_time,
+ NULL as last_msg_receipt_time,
+ NULL as latest_end_lsn,
+ NULL as latest_end_time
+ FROM pg_subscription_rel r WHERE r.srsubstate <> 'r') st
+ ON (st.subid = su.oid);
CREATE VIEW pg_stat_ssl AS
SELECT
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 778e409..bca3c41 100644
--- a/src/backend/postmaster/pgstat.c
+++ b/src/backend/postmaster/pgstat.c
@@ -55,6 +55,7 @@
#include "postmaster/postmaster.h"
#include "replication/slot.h"
#include "replication/walsender.h"
+#include "replication/logicalworker.h"
#include "storage/backendid.h"
#include "storage/dsm.h"
#include "storage/fd.h"
@@ -378,6 +379,7 @@ static void pgstat_recv_checksum_failure(PgStat_MsgChecksumFailure *msg, int len
static void pgstat_recv_connstat(PgStat_MsgConn *msg, int len);
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_success(PgStat_MsgSubscriptionErr *msg, int len);
static void pgstat_recv_subscription_error(PgStat_MsgSubscriptionErr *msg, int len);
static void pgstat_recv_subscription_error_purge(PgStat_MsgSubscriptionErrPurge *msg,
int len);
@@ -2038,6 +2040,40 @@ pgstat_report_replslot_drop(const char *slotname)
}
/* ----------
+ * pgstat_report_subscription_success() -
+ *
+ * Tell the collector about the subscription success.
+ * ----------
+ */
+void
+pgstat_report_subscription_success(Oid subid, Oid subrel,
+ LogicalRepMsgType command, PgStat_Counter bytes)
+{
+ PgStat_MsgSubscriptionErr msg;
+
+ Assert(command == 0 /* table sync worker */ ||
+ command == LOGICAL_REP_MSG_COMMIT ||
+ command == LOGICAL_REP_MSG_COMMIT_PREPARED ||
+ command == LOGICAL_REP_MSG_STREAM_ABORT);
+
+ pgstat_setheader(&msg.m_hdr, PGSTAT_MTYPE_SUBSCRIPTIONSUCCESS);
+ msg.m_subid = subid;
+ msg.m_subrelid = subrel;
+ msg.m_clear = false;
+ msg.m_reset = false;
+ msg.m_databaseid = InvalidOid;
+ msg.m_relid = InvalidOid;
+ msg.m_command = command;
+ msg.m_xid = InvalidTransactionId;
+ msg.m_bytes = bytes;
+ msg.m_failure_time = GetCurrentTimestamp();
+ msg.m_errmsg[0] = '\0';
+ pgstat_send(&msg, sizeof(PgStat_MsgSubscriptionErr));
+
+ reset_apply_error_context_xact_size();
+}
+
+/* ----------
* pgstat_report_subscription_error() -
*
* Tell the collector about the subscription error.
@@ -2063,10 +2099,13 @@ pgstat_report_subscription_error(Oid subid, Oid subrelid, Oid relid,
msg.m_relid = relid;
msg.m_command = command;
msg.m_xid = xid;
+ msg.m_bytes = get_apply_error_context_xact_size();
msg.m_failure_time = GetCurrentTimestamp();
strlcpy(msg.m_errmsg, errmsg, PGSTAT_SUBSCRIPTIONERR_MSGLEN);
pgstat_send(&msg, len);
+
+ reset_apply_error_context_xact_size();
}
/* ----------
@@ -3730,6 +3769,10 @@ PgstatCollectorMain(int argc, char *argv[])
pgstat_recv_connstat(&msg.msg_conn, len);
break;
+ case PGSTAT_MTYPE_SUBSCRIPTIONSUCCESS:
+ pgstat_recv_subscription_success(&msg.msg_subscriptionerr, len);
+ break;
+
case PGSTAT_MTYPE_SUBSCRIPTIONERR:
pgstat_recv_subscription_error(&msg.msg_subscriptionerr, len);
break;
@@ -6153,6 +6196,45 @@ pgstat_recv_funcpurge(PgStat_MsgFuncpurge *msg, int len)
}
/* ----------
+ * pgstat_recv_subscription_success() -
+ *
+ * Process a SUBSCRIPTIONERR message.
+ * ----------
+ */
+static void
+pgstat_recv_subscription_success(PgStat_MsgSubscriptionErr *msg, int len)
+{
+ PgStat_StatSubErrEntry *errent;
+ bool create = !(msg->m_reset || msg->m_clear);
+
+ errent = pgstat_get_subscription_error_entry(msg->m_subid,
+ msg->m_subrelid,
+ create);
+
+ /* msg from table sync worker */
+ if (msg->m_command == 0)
+ errent->xact_commit++;
+ else
+ {
+ switch(msg->m_command)
+ {
+ case LOGICAL_REP_MSG_COMMIT:
+ case LOGICAL_REP_MSG_COMMIT_PREPARED:
+ errent->xact_commit++;
+ errent->xact_commit_bytes += msg->m_bytes;
+ break;
+ case LOGICAL_REP_MSG_STREAM_ABORT:
+ errent->xact_abort++;
+ errent->xact_abort_bytes = msg->m_bytes;
+ break;
+ default:
+ elog(ERROR, "unexpected command type");
+ }
+ }
+ errent->last_failure = msg->m_failure_time;
+}
+
+/* ----------
* pgstat_recv_subscription_error() -
*
* Process a SUBSCRIPTIONERR message.
@@ -6202,6 +6284,8 @@ pgstat_recv_subscription_error(PgStat_MsgSubscriptionErr *msg, int len)
errent->relid = msg->m_relid;
errent->command = msg->m_command;
errent->xid = msg->m_xid;
+ errent->xact_error++;
+ errent->xact_error_bytes += msg->m_bytes;
errent->failure_count++;
errent->last_failure = msg->m_failure_time;
strlcpy(errent->last_errmsg, msg->m_errmsg, PGSTAT_SUBSCRIPTIONERR_MSGLEN);
@@ -6465,6 +6549,12 @@ pgstat_get_subscription_error_entry(Oid subid, Oid subrelid, bool create)
errent->relid = InvalidOid;
errent->command = 0;
errent->xid = InvalidTransactionId;
+ errent->xact_commit = 0;
+ errent->xact_commit_bytes = 0;
+ errent->xact_error = 0;
+ errent->xact_error_bytes = 0;
+ errent->xact_abort = 0;
+ errent->xact_abort_bytes = 0;
errent->failure_count = 0;
errent->last_failure = 0;
errent->last_errmsg[0] = '\0';
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 9f5bf4b..64d18c6 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -17,6 +17,7 @@
#include "catalog/pg_type.h"
#include "libpq/pqformat.h"
#include "replication/logicalproto.h"
+#include "replication/logicalworker.h"
#include "utils/lsyscache.h"
#include "utils/syscache.h"
@@ -872,6 +873,9 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
case LOGICALREP_COLUMN_TEXT:
len = pq_getmsgint(in, 4); /* read length */
+ /* accumulate data length for transaction stats */
+ add_apply_error_context_xact_size(len);
+
/* and data */
value->data = palloc(len + 1);
pq_copymsgbytes(in, value->data, len);
@@ -884,6 +888,9 @@ logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
case LOGICALREP_COLUMN_BINARY:
len = pq_getmsgint(in, 4); /* read length */
+ /* accumulate data length for transaction stats */
+ add_apply_error_context_xact_size(len);
+
/* and data */
value->data = palloc(len + 1);
pq_copymsgbytes(in, value->data, len);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index f07983a..7ba7486 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1150,6 +1150,15 @@ copy_table_done:
SpinLockRelease(&MyLogicalRepWorker->relmutex);
/*
+ * Update the stats for table sync. We don't record the bytes of
+ * table synchronization.
+ */
+ pgstat_report_subscription_success(MyLogicalRepWorker->subid,
+ MyLogicalRepWorker->relid,
+ 0, /* no corresponding logical message type */
+ 0);
+
+ /*
* 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 34ed8e4..a3f9ce3 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -222,6 +222,9 @@ typedef struct ApplyErrorCallbackArg
LogicalRepMsgType command; /* 0 if invalid */
LogicalRepRelMapEntry *rel;
+ /* data size of this transaction */
+ PgStat_Counter bytes;
+
/* Remote node information */
int remote_attnum; /* -1 if invalid */
TransactionId remote_xid;
@@ -232,6 +235,7 @@ static ApplyErrorCallbackArg apply_error_callback_arg =
{
.command = 0,
.rel = NULL,
+ .bytes = 0,
.remote_attnum = -1,
.remote_xid = InvalidTransactionId,
.ts = 0,
@@ -863,6 +867,10 @@ apply_handle_commit(StringInfo s)
process_syncing_tables(commit_data.end_lsn);
pgstat_report_activity(STATE_IDLE, NULL);
+ pgstat_report_subscription_success(MySubscription->oid,
+ InvalidOid,
+ LOGICAL_REP_MSG_COMMIT,
+ apply_error_callback_arg.bytes);
reset_apply_error_context_info();
}
@@ -1016,6 +1024,10 @@ apply_handle_commit_prepared(StringInfo s)
process_syncing_tables(prepare_data.end_lsn);
pgstat_report_activity(STATE_IDLE, NULL);
+ pgstat_report_subscription_success(MySubscription->oid,
+ InvalidOid,
+ LOGICAL_REP_MSG_COMMIT_PREPARED,
+ apply_error_callback_arg.bytes);
reset_apply_error_context_info();
}
@@ -1350,6 +1362,10 @@ apply_handle_stream_abort(StringInfo s)
if (is_skipping_changes())
stop_skipping_changes(InvalidXLogRecPtr, 0);
+ pgstat_report_subscription_success(MySubscription->oid,
+ InvalidOid,
+ LOGICAL_REP_MSG_STREAM_ABORT,
+ apply_error_callback_arg.bytes);
reset_apply_error_context_info();
}
@@ -3380,6 +3396,9 @@ stream_write_change(char action, StringInfo s)
len = (s->len - s->cursor);
BufFileWrite(stream_fd, &s->data[s->cursor], len);
+
+ /* accumulate the bytes for stats */
+ add_apply_error_context_xact_size(len);
}
/*
@@ -3744,6 +3763,27 @@ set_apply_error_context_xact(TransactionId xid, TimestampTz ts)
apply_error_callback_arg.ts = ts;
}
+/* Exported so that stats collector can utilize this value */
+int64
+get_apply_error_context_xact_size(void)
+{
+ return apply_error_callback_arg.bytes;
+}
+
+/* Add size to apply error callback bytes */
+void
+add_apply_error_context_xact_size(int64 size)
+{
+ apply_error_callback_arg.bytes += size;
+}
+
+/* Reset bytes information of apply error callback */
+void
+reset_apply_error_context_xact_size(void)
+{
+ apply_error_callback_arg.bytes = 0;
+}
+
/* Reset all information of apply error callback */
static inline void
reset_apply_error_context_info(void)
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index c454e2f..0b4ad21 100644
--- a/src/backend/utils/adt/pgstatfuncs.c
+++ b/src/backend/utils/adt/pgstatfuncs.c
@@ -2492,3 +2492,136 @@ pg_stat_get_subscription_error(PG_FUNCTION_ARGS)
/* Returns the record as Datum */
PG_RETURN_DATUM(HeapTupleGetDatum(heap_form_tuple(tupdesc, values, nulls)));
}
+
+Datum
+pg_stat_get_subscription_xact_commit(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_commit);
+
+ PG_RETURN_INT64(result);
+}
+
+Datum
+pg_stat_get_subscription_xact_commit_bytes(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_commit_bytes);
+
+ PG_RETURN_INT64(result);
+}
+
+
+Datum
+pg_stat_get_subscription_xact_error(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_error);
+
+ PG_RETURN_INT64(result);
+}
+
+Datum
+pg_stat_get_subscription_xact_error_bytes(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_error_bytes);
+
+ PG_RETURN_INT64(result);
+}
+
+Datum
+pg_stat_get_subscription_xact_abort(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_abort);
+
+ PG_RETURN_INT64(result);
+}
+
+Datum
+pg_stat_get_subscription_xact_abort_bytes(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+ Oid relid;
+ int64 result;
+ PgStat_StatSubErrEntry *errent;
+
+ if (PG_ARGISNULL(1))
+ relid = InvalidOid;
+ else
+ relid = PG_GETARG_OID(1);
+
+ /* Get subscription error entry */
+ if ((errent = pgstat_fetch_subscription_error(subid, relid)) == NULL)
+ result = 0;
+ else
+ result = (int64) (errent->xact_abort_bytes);
+
+ PG_RETURN_INT64(result);
+}
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index ea4e6d2..aeb935c 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -5393,6 +5393,30 @@
proargmodes => '{i,i,o,o,o,o,o,o,o,o,o,o}',
proargnames => '{subid,relid,datid,subid,relid,command,xid,failure_source,failure_count,last_failure,last_failure_message,stats_reset}',
prosrc => 'pg_stat_get_subscription_error' },
+{ oid => '8525', descr => 'statistics: number of transactions commit for a subscription',
+ proname => 'pg_stat_get_subscription_xact_commit', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_commit' },
+{ oid => '8526', descr => 'statistics: bytes of transactions commit for a subscription',
+ proname => 'pg_stat_get_subscription_xact_commit_bytes', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_commit_bytes' },
+{ oid => '8527', descr => 'statistics: number of transactions error for a subscription',
+ proname => 'pg_stat_get_subscription_xact_error', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_error' },
+{ oid => '8528', descr => 'statistics: bytes of transactions error for a subscription',
+ proname => 'pg_stat_get_subscription_xact_error_bytes', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_error_bytes' },
+{ oid => '8529', descr => 'statistics: number of transactions abort for a subscription',
+ proname => 'pg_stat_get_subscription_xact_abort', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_abort' },
+{ oid => '8530', descr => 'statistics: bytes of transactions abort for a subscription',
+ proname => 'pg_stat_get_subscription_xact_abort_bytes', proisstrict => 'f', provolatile => 's',
+ proparallel => 'r', prorettype => 'int8', proargtypes => 'oid oid',
+ prosrc => 'pg_stat_get_subscription_xact_abort_bytes' },
{ oid => '6118', descr => 'statistics: information about subscription',
proname => 'pg_stat_get_subscription', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', proparallel => 'r',
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index 6775736..2a43135 100644
--- a/src/include/pgstat.h
+++ b/src/include/pgstat.h
@@ -68,6 +68,7 @@ typedef enum StatMsgType
PGSTAT_MTYPE_RESETSLRUCOUNTER,
PGSTAT_MTYPE_RESETREPLSLOTCOUNTER,
PGSTAT_MTYPE_SUBSCRIPTIONERR,
+ PGSTAT_MTYPE_SUBSCRIPTIONSUCCESS,
PGSTAT_MTYPE_SUBSCRIPTIONERRPURGE,
PGSTAT_MTYPE_SUBSCRIPTIONPURGE,
PGSTAT_MTYPE_AUTOVAC_START,
@@ -565,6 +566,7 @@ typedef struct PgStat_MsgSubscriptionErr
Oid m_relid;
LogicalRepMsgType m_command;
TransactionId m_xid;
+ PgStat_Counter m_bytes;
TimestampTz m_failure_time;
char m_errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
} PgStat_MsgSubscriptionErr;
@@ -1002,8 +1004,12 @@ typedef struct PgStat_StatSubEntry
* (subrelid is InvalidOid) or by the table sync worker (subrelid is a valid OID).
* The error reported by the table sync worker is removed also when the table
* synchronization process completed.
+ *
+ * Other general stats for transaction on this subscription are stored in this entry
+ * as well, aligned with transaction error and its total bytes. Holding the size of
+ * computed xact_error_bytes during replay depends on the apply error callback and
+ * having general stats in the same place reduces code complexity.
*/
-
typedef struct PgStat_StatSubErrEntry
{
Oid subrelid; /* InvalidOid if the apply worker, otherwise
@@ -1014,7 +1020,16 @@ typedef struct PgStat_StatSubErrEntry
* case. */
LogicalRepMsgType command;
TransactionId xid;
- PgStat_Counter failure_count;
+
+ PgStat_Counter xact_commit;
+ PgStat_Counter xact_commit_bytes;
+ PgStat_Counter xact_error;
+ PgStat_Counter xact_error_bytes; /* total error counts of this
+ subscription */
+ PgStat_Counter xact_abort;
+ PgStat_Counter xact_abort_bytes;
+ PgStat_Counter failure_count; /* total error counts of one
+ specific error */
TimestampTz last_failure;
char last_errmsg[PGSTAT_SUBSCRIPTIONERR_MSGLEN];
TimestampTz stat_reset_timestamp;
@@ -1129,10 +1144,11 @@ 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_subscription_success(Oid subid, Oid subrel,
+ LogicalRepMsgType command, PgStat_Counter bytes);
extern void pgstat_report_subscription_error(Oid subid, Oid subrelid, Oid relid,
LogicalRepMsgType command,
TransactionId xid, const char *errmsg);
-
extern void pgstat_initialize(void);
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 2ad61a0..923c8ce 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -15,5 +15,9 @@
extern void ApplyWorkerMain(Datum main_arg);
extern bool IsLogicalWorker(void);
+extern int64 get_apply_error_context_xact_size(void);
+extern void add_apply_error_context_xact_size(int64 size);
+extern void reset_apply_error_context_xact_size(void);
+
#endif /* LOGICALWORKER_H */
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 66b185f..89c68f1 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -2091,9 +2091,34 @@ pg_stat_subscription| SELECT su.oid AS subid,
st.last_msg_send_time,
st.last_msg_receipt_time,
st.latest_end_lsn,
+ pg_stat_get_subscription_xact_commit(su.oid, st.relid) AS xact_commit,
+ pg_stat_get_subscription_xact_commit_bytes(su.oid, st.relid) AS xact_commit_bytes,
+ pg_stat_get_subscription_xact_error(su.oid, st.relid) AS xact_error,
+ pg_stat_get_subscription_xact_error_bytes(su.oid, st.relid) AS xact_error_bytes,
+ pg_stat_get_subscription_xact_abort(su.oid, st.relid) AS xact_abort,
+ pg_stat_get_subscription_xact_abort_bytes(su.oid, st.relid) AS xact_abort_bytes,
st.latest_end_time
FROM (pg_subscription su
- LEFT JOIN pg_stat_get_subscription(NULL::oid) st(subid, relid, pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time) ON ((st.subid = su.oid)));
+ LEFT JOIN ( SELECT s.subid,
+ s.relid,
+ s.pid,
+ s.received_lsn,
+ s.last_msg_send_time,
+ s.last_msg_receipt_time,
+ s.latest_end_lsn,
+ s.latest_end_time
+ FROM pg_stat_get_subscription(NULL::oid) s(subid, relid, pid, received_lsn, last_msg_send_time, last_msg_receipt_time, latest_end_lsn, latest_end_time)
+ UNION
+ SELECT r.srsubid,
+ r.srrelid,
+ NULL::integer AS pid,
+ NULL::pg_lsn AS received_lsn,
+ NULL::timestamp with time zone AS last_msg_send_time,
+ NULL::timestamp with time zone AS last_msg_receipt_time,
+ NULL::pg_lsn AS latest_end_lsn,
+ NULL::timestamp with time zone AS latest_end_time
+ FROM pg_subscription_rel r
+ WHERE (r.srsubstate <> 'r'::"char")) st ON ((st.subid = su.oid)));
pg_stat_subscription_errors| SELECT d.datname,
sr.subid,
s.subname,
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]
Subject: RE: Failed transaction statistics to measure the logical replication progress
In-Reply-To: <TY2PR01MB48905729E8CB81805CCF64C0EDCE9@TY2PR01MB4890.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