From 1b756a897eee84e068e039333666278112c235b0 Mon Sep 17 00:00:00 2001 From: Takamichi Osumi 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 pg_stat_subscription_workerspg_stat_subscription_workers - One row per subscription worker, showing statistics about errors - that occurred on that subscription worker. + One row per subscription worker, showing statistics about transactions + and errors that occurred on that subscription worker. See pg_stat_subscription_workers for details. @@ -3123,6 +3123,36 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i + commit_count bigint + + + Number of transactions successfully applied in this subscription. + COMMIT and COMMIT PREPARED increments this counter. + + + + + + error_count bigint + + + Number of transactions that failed to be applied by the table + sync worker or main apply worker in this subscription. + + + + + + abort_count bigint + + + Number of transactions aborted in this subscription. + ROLLBACK PREPARED increments this counter. + + + + + last_error_relid oid 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