From 3028e49dd656fc820c32ffc725b55a9201898291 Mon Sep 17 00:00:00 2001 From: Takamichi Osumi Date: Tue, 7 Dec 2021 08:53:47 +0000 Subject: [PATCH v16 2/2] 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 | 43 +++++++++- src/backend/catalog/system_views.sql | 3 + src/backend/postmaster/pgstat.c | 75 ++++++++++++++++++ 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/026_worker_stats.pl | 118 +++++++++++++++++++++++++++- src/tools/pgindent/typedefs.list | 1 + 11 files changed, 313 insertions(+), 19 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 62f2a33..1feb075 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. @@ -3074,8 +3074,10 @@ SELECT pid, wait_event_type, wait_event FROM pg_stat_activity WHERE wait_event i The pg_stat_subscription_workers view will contain one row per subscription worker on which errors have occurred, for workers applying logical replication changes and workers handling the initial data - copy of the subscribed tables. The statistics entry is removed when the - corresponding subscription is dropped. + copy of the subscribed tables. Also, the row corresponding to the apply + worker shows all transaction statistics of both types of workers on the + subscription. The statistics entry is removed when the corresponding + subscription is dropped. @@ -3123,6 +3125,39 @@ 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, COMMIT PREPARED and successful table copy of table sync + worker increment 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. This + counter is updated after confirming the error is not same as + the previous one. + + + + + + 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..fef9181 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,16 @@ 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, by swapping the stats + * entries. + */ + if (OidIsValid(msg->m_subrelid)) + subwentry = pgstat_get_subworker_entry(dbentry, msg->m_subid, + InvalidOid, true); + subwentry->error_count++; } /* ---------- 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/026_worker_stats.pl b/src/test/subscription/t/026_worker_stats.pl index 8005c54..1a952bd 100644 --- a/src/test/subscription/t/026_worker_stats.pl +++ b/src/test/subscription/t/026_worker_stats.pl @@ -1,7 +1,7 @@ # Copyright (c) 2021, PostgreSQL Global Development Group -# Tests for subscription error stats. +# Tests for subscription stats. use strict; use warnings; use PostgreSQL::Test::Cluster; @@ -44,9 +44,32 @@ WHERE last_error_relid = '$relname'::regclass or die "Timed out while waiting for " . $msg; } +# Test the update of general transaction stats which satisfy +# the expected condition has been detected. +sub confirm_transaction_stats_update +{ + my ($node, $condition, $msg) = @_; + + # Check only the stats of the apply worker + my $sql = qq[ +SELECT count(1) = 1 +FROM pg_stat_subscription_workers +WHERE subrelid IS NULL AND $condition]; + + $node->poll_query_until('postgres', $sql) + or die "Timed out while waiting for " . $msg; +} + # 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. @@ -58,6 +81,7 @@ $node_subscriber->init(allows_streaming => 'logical'); $node_subscriber->append_conf( 'postgresql.conf', qq[ +max_prepared_transactions = 10 wal_retrieve_retry_interval = 2s ]); $node_subscriber->start; @@ -98,7 +122,7 @@ is($result, qq(0), 'check no subscription error'); # Create subscription. The table sync for test_tab2 on tap_sub will enter into # infinite error loop due to violating the unique constraint. $node_subscriber->safe_psql('postgres', - "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub;" + "CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr' PUBLICATION tap_pub WITH (streaming = on, two_phase = on);" ); $node_publisher->wait_for_catchup('tap_sub'); @@ -116,6 +140,17 @@ $result = $node_subscriber->safe_psql('postgres', "SELECT count(a) FROM test_tab1"); is($result, q(1), 'check initial data are copied to subscriber'); +# Check the update of stats counters. +confirm_transaction_stats_update( + $node_subscriber, + 'commit_count = 1', + 'the commit_count increment by table sync'); + +confirm_transaction_stats_update( + $node_subscriber, + 'error_count = 1', + 'the error_count increment by table sync'); + # Insert more data to test_tab1, raising an error on the subscriber due to # violation of the unique constraint on test_tab1. my $xid = $node_publisher->safe_psql( @@ -132,6 +167,11 @@ test_subscription_error( qq(duplicate key value violates unique constraint), 'error reported by the apply worker'); +confirm_transaction_stats_update( + $node_subscriber, + 'error_count = 2', + 'the error_count increment by the apply worker'); + # Check the table sync worker's error in the view. test_subscription_error( $node_subscriber, 'test_tab2', '', '', @@ -150,11 +190,81 @@ $node_subscriber->poll_query_until('postgres', $node_subscriber->poll_query_until('postgres', "SELECT count(1) > 0 FROM test_tab2"); -# There shouldn't be any errors in the view after dropping the subscription. +confirm_transaction_stats_update( + $node_subscriber, + 'commit_count = 3', + 'the commit_count increments by both workers'); + +# Some more tests for transaction stats +# PREPARE & COMMIT PREPARED +$node_publisher->safe_psql( + 'postgres', qq[ +BEGIN; +INSERT INTO test_tab1 VALUES (2); +PREPARE TRANSACTION 'gid1' +]); +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gid1'"); +confirm_transaction_stats_update( + $node_subscriber, + 'commit_count = 4', + 'the commit_count increment by commit prepared'); + +# STREAM COMMIT +$node_publisher->safe_psql( + 'postgres', qq[ +BEGIN; +INSERT INTO test_tab1 VALUES(generate_series(1001, 2000)); +COMMIT; +]); +confirm_transaction_stats_update( + $node_subscriber, + 'commit_count = 5', + 'the commit_count increment by stream commit'); + +# STREAM PREPARE & COMMIT PREPARED +$node_publisher->safe_psql( + 'postgres', qq[ +BEGIN; +INSERT INTO test_tab1 VALUES(generate_series(2001, 3000)); +PREPARE TRANSACTION 'gid2' +]); +$node_publisher->safe_psql('postgres', "COMMIT PREPARED 'gid2'"); +confirm_transaction_stats_update( + $node_subscriber, + 'commit_count = 6', + 'the commit_count increment by streamed commit prepared'); + +# ROLLBACK PREPARED +$node_publisher->safe_psql( + 'postgres', qq[ +BEGIN; +INSERT INTO test_tab1 VALUES (3); +PREPARE TRANSACTION 'gid3'; +]); +$node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'gid3'"); +confirm_transaction_stats_update( + $node_subscriber, + 'abort_count = 1', + 'the abort_count increment by rollback prepared'); + +# STREAM PREPARE & ROLLBACK PREPARED +$node_publisher->safe_psql( + 'postgres', qq[ +BEGIN; +INSERT INTO test_tab1 VALUES (generate_series(3001, 4000)); +PREPARE TRANSACTION 'gid4'; +]); +$node_publisher->safe_psql('postgres', "ROLLBACK PREPARED 'gid4'"); +confirm_transaction_stats_update( + $node_subscriber, + 'abort_count = 2', + 'the abort_count increment by streamed rollback prepared'); + +# There shouldn't be any records in the view after dropping the subscription. $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub;"); $result = $node_subscriber->safe_psql('postgres', "SELECT count(1) FROM pg_stat_subscription_workers"); -is($result, q(0), 'no error after dropping subscription'); +is($result, q(0), 'no record after dropping subscription'); $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