notify-standby.diff

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index e22bdac..1b2eca3 100644
*** a/src/backend/access/transam/xact.c
--- b/src/backend/access/transam/xact.c
*************** RecordTransactionCommit(void)
*** 988,994 ****
  		/*
  		 * Do we need the long commit record? If not, use the compact format.
  		 */
! 		if (nrels > 0 || nmsgs > 0 || RelcacheInitFileInval || forceSyncCommit)
  		{
  			XLogRecData rdata[4];
  			int			lastrdata = 0;
--- 988,995 ----
  		/*
  		 * Do we need the long commit record? If not, use the compact format.
  		 */
! 		if (nrels > 0 || nmsgs > 0 || RelcacheInitFileInval || forceSyncCommit
! 			|| backendHasSentNotifications)
  		{
  			XLogRecData rdata[4];
  			int			lastrdata = 0;
*************** RecordTransactionCommit(void)
*** 1009,1014 ****
--- 1010,1016 ----
  			xlrec.nrels = nrels;
  			xlrec.nsubxacts = nchildren;
  			xlrec.nmsgs = nmsgs;
+ 			xlrec.sent_notifications = backendHasSentNotifications;
  			rdata[0].data = (char *) (&xlrec);
  			rdata[0].len = MinSizeOfXactCommit;
  			rdata[0].buffer = InvalidBuffer;
*************** RecordTransactionAbort(bool isSubXact)
*** 1409,1414 ****
--- 1411,1417 ----
  	}
  	xlrec.nrels = nrels;
  	xlrec.nsubxacts = nchildren;
+ 	xlrec.sent_notifications = backendHasSentNotifications;
  	rdata[0].data = (char *) (&xlrec);
  	rdata[0].len = MinSizeOfXactAbort;
  	rdata[0].buffer = InvalidBuffer;
*************** xact_redo_commit_internal(TransactionId
*** 4510,4516 ****
  					TransactionId *sub_xids, int nsubxacts,
  					SharedInvalidationMessage *inval_msgs, int nmsgs,
  					RelFileNode *xnodes, int nrels,
! 					Oid dbId, Oid tsId,
  					uint32 xinfo)
  {
  	TransactionId max_xid;
--- 4513,4519 ----
  					TransactionId *sub_xids, int nsubxacts,
  					SharedInvalidationMessage *inval_msgs, int nmsgs,
  					RelFileNode *xnodes, int nrels,
! 					Oid dbId, Oid tsId, bool sent_notifications,
  					uint32 xinfo)
  {
  	TransactionId max_xid;
*************** xact_redo_commit_internal(TransactionId
*** 4616,4621 ****
--- 4619,4626 ----
  	if (XactCompletionForceSyncCommit(xinfo))
  		XLogFlush(lsn);
  
+ 	if (sent_notifications)
+ 		AtEOXact_HS_Notify();
  }
  /*
   * Utility function to call xact_redo_commit_internal after breaking down xlrec
*************** xact_redo_commit(xl_xact_commit *xlrec,
*** 4637,4642 ****
--- 4642,4648 ----
  								xlrec->xnodes, xlrec->nrels,
  								xlrec->dbId,
  								xlrec->tsId,
+ 								xlrec->sent_notifications,
  								xlrec->xinfo);
  }
  
*************** xact_redo_commit_compact(xl_xact_commit_
*** 4652,4657 ****
--- 4658,4664 ----
  								NULL, 0,		/* relfilenodes */
  								InvalidOid,		/* dbId */
  								InvalidOid,		/* tsId */
+ 								false,			/* sent_notifications */
  								0);				/* xinfo */
  }
  
*************** xact_redo_abort(xl_xact_abort *xlrec, Tr
*** 4740,4745 ****
--- 4747,4755 ----
  		}
  		smgrclose(srel);
  	}
+ 
+ 	if (xlrec->sent_notifications)
+ 		AtEOXact_HS_Notify();
  }
  
  void
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 8e65962..78d8905 100644
*** a/src/backend/access/transam/xlog.c
--- b/src/backend/access/transam/xlog.c
***************
*** 35,40 ****
--- 35,41 ----
  #include "catalog/catversion.h"
  #include "catalog/pg_control.h"
  #include "catalog/pg_database.h"
+ #include "commands/async.h"
  #include "libpq/pqsignal.h"
  #include "miscadmin.h"
  #include "pgstat.h"
*************** RequestXLogSwitch(void)
*** 8306,8311 ****
--- 8307,8326 ----
  	return RecPtr;
  }
  
+ void
+ XLogInsertNotifyData(NotifyData *NotifyRecord)
+ {
+ 	XLogRecData	rdata;
+ 
+ 	rdata.buffer = InvalidBuffer;
+ 	rdata.data = (char *) NotifyRecord;
+ 	/* -1 because one character is already contained in sizeof(NotifyData) */
+ 	rdata.len = sizeof(NotifyData) + NotifyRecord->notify_len - 1;
+ 	rdata.next = NULL;
+ 
+ 	(void) XLogInsert(RM_XLOG_ID, XLOG_NOTIFY, &rdata);
+ }
+ 
  /*
   * Write a RESTORE POINT record
   */
*************** xlog_redo(XLogRecPtr lsn, XLogRecord *re
*** 8529,8534 ****
--- 8544,8554 ----
  	{
  		/* nothing to do here */
  	}
+ 	else if (info == XLOG_NOTIFY)
+ 	{
+ 		NotifyData *rdata = (NotifyData *) XLogRecGetData(record);
+ 		asyncGetNotifyFromXLog(rdata, record->xl_xid);
+ 	}
  	else if (info == XLOG_BACKUP_END)
  	{
  		XLogRecPtr	startpoint;
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 50ba20c..051595c 100644
*** a/src/backend/catalog/system_views.sql
--- b/src/backend/catalog/system_views.sql
*************** CREATE VIEW pg_stat_database_conflicts A
*** 585,591 ****
              pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
              pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
              pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
!             pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock
      FROM pg_database D;
  
  CREATE VIEW pg_stat_user_functions AS
--- 585,592 ----
              pg_stat_get_db_conflict_lock(D.oid) AS confl_lock,
              pg_stat_get_db_conflict_snapshot(D.oid) AS confl_snapshot,
              pg_stat_get_db_conflict_bufferpin(D.oid) AS confl_bufferpin,
!             pg_stat_get_db_conflict_startup_deadlock(D.oid) AS confl_deadlock,
!             pg_stat_get_db_conflict_notify_queue(D.oid) AS confl_notify
      FROM pg_database D;
  
  CREATE VIEW pg_stat_user_functions AS
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index a81f7c7..b9528b4 100644
*** a/src/backend/commands/async.c
--- b/src/backend/commands/async.c
***************
*** 118,123 ****
--- 118,124 ----
  #include "access/slru.h"
  #include "access/transam.h"
  #include "access/xact.h"
+ #include "access/xlog.h"
  #include "catalog/pg_database.h"
  #include "commands/async.h"
  #include "funcapi.h"
***************
*** 128,133 ****
--- 129,135 ----
  #include "storage/lmgr.h"
  #include "storage/procsignal.h"
  #include "storage/sinval.h"
+ #include "storage/standby.h"
  #include "tcop/tcopprot.h"
  #include "utils/builtins.h"
  #include "utils/memutils.h"
***************
*** 160,165 ****
--- 162,168 ----
  typedef struct AsyncQueueEntry
  {
  	int			length;			/* total allocated length of entry */
+ 	int			str_length;		/* only the length of the two string fields */
  	Oid			dboid;			/* sender's database OID */
  	TransactionId xid;			/* sender's XID */
  	int32		srcPid;			/* sender's PID */
*************** static volatile sig_atomic_t notifyInter
*** 351,357 ****
  static bool unlistenExitRegistered = false;
  
  /* has this backend sent notifications in the current transaction? */
! static bool backendHasSentNotifications = false;
  
  /* has this backend executed its first LISTEN in the current transaction? */
  static bool backendHasExecutedInitialListen = false;
--- 354,360 ----
  static bool unlistenExitRegistered = false;
  
  /* has this backend sent notifications in the current transaction? */
! bool backendHasSentNotifications = false;
  
  /* has this backend executed its first LISTEN in the current transaction? */
  static bool backendHasExecutedInitialListen = false;
*************** static bool IsListeningOn(const char *ch
*** 371,378 ****
  static void asyncQueueUnregister(void);
  static bool asyncQueueIsFull(void);
  static bool asyncQueueAdvance(QueuePosition *position, int entryLength);
! static void asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe);
! static ListCell *asyncQueueAddEntries(ListCell *nextNotify);
  static void asyncQueueFillWarning(void);
  static bool SignalBackends(void);
  static void asyncQueueReadAllNotifications(void);
--- 374,388 ----
  static void asyncQueueUnregister(void);
  static bool asyncQueueIsFull(void);
  static bool asyncQueueAdvance(QueuePosition *position, int entryLength);
! static void asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe,
! 										  Oid dboid, TransactionId xid);
! static ListCell *asyncQueueAddEntries(ListCell *nextNotify, Oid dboid,
! 									  TransactionId xid);
! static void asyncAddNotifyXLog(ListCell *HSStart, ListCell *HSEnd,
! 							   uint32 xlogLength, Oid dboid, TransactionId xid,
! 							   bool isLast);
! static void asyncAddNotificationsIntoQueue(Oid dboid, TransactionId xid);
! static int32 asyncQueueGetSlowestPid(int *backendId);
  static void asyncQueueFillWarning(void);
  static bool SignalBackends(void);
  static void asyncQueueReadAllNotifications(void);
*************** PreCommit_Notify(void)
*** 768,773 ****
--- 778,784 ----
  	if (Trace_notify)
  		elog(DEBUG1, "PreCommit_Notify");
  
+ 	Assert(backendHasSentNotifications == false);
  	Assert(backendHasExecutedInitialListen == false);
  
  	/* Preflight for any pending listen/unlisten actions */
*************** PreCommit_Notify(void)
*** 788,853 ****
  				break;
  		}
  	}
- 
  	/* Queue any pending notifies */
! 	if (pendingNotifies)
! 	{
! 		ListCell   *nextNotify;
  
! 		/*
! 		 * Make sure that we have an XID assigned to the current transaction.
! 		 * GetCurrentTransactionId is cheap if we already have an XID, but not
! 		 * so cheap if we don't, and we'd prefer not to do that work while
! 		 * holding AsyncQueueLock.
! 		 */
! 		(void) GetCurrentTransactionId();
  
  		/*
! 		 * Serialize writers by acquiring a special lock that we hold till
! 		 * after commit.  This ensures that queue entries appear in commit
! 		 * order, and in particular that there are never uncommitted queue
! 		 * entries ahead of committed ones, so an uncommitted transaction
! 		 * can't block delivery of deliverable notifications.
  		 *
! 		 * We use a heavyweight lock so that it'll automatically be released
! 		 * after either commit or abort.  This also allows deadlocks to be
! 		 * detected, though really a deadlock shouldn't be possible here.
  		 *
! 		 * The lock is on "database 0", which is pretty ugly but it doesn't
! 		 * seem worth inventing a special locktag category just for this.
! 		 * (Historical note: before PG 9.0, a similar lock on "database 0" was
! 		 * used by the flatfiles mechanism.)
  		 */
! 		LockSharedObject(DatabaseRelationId, InvalidOid, 0,
! 						 AccessExclusiveLock);
  
! 		/* Now push the notifications into the queue */
! 		backendHasSentNotifications = true;
  
! 		nextNotify = list_head(pendingNotifies);
! 		while (nextNotify != NULL)
  		{
- 			/*
- 			 * Add the pending notifications to the queue.	We acquire and
- 			 * release AsyncQueueLock once per page, which might be overkill
- 			 * but it does allow readers to get in while we're doing this.
- 			 *
- 			 * A full queue is very uncommon and should really not happen,
- 			 * given that we have so much space available in the SLRU pages.
- 			 * Nevertheless we need to deal with this possibility. Note that
- 			 * when we get here we are in the process of committing our
- 			 * transaction, but we have not yet committed to clog, so at this
- 			 * point in time we can still roll the transaction back.
- 			 */
- 			LWLockAcquire(AsyncQueueLock, LW_EXCLUSIVE);
- 			asyncQueueFillWarning();
  			if (asyncQueueIsFull())
  				ereport(ERROR,
  						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
  					  errmsg("too many notifications in the NOTIFY queue")));
- 			nextNotify = asyncQueueAddEntries(nextNotify);
- 			LWLockRelease(AsyncQueueLock);
  		}
  	}
  }
  
--- 799,919 ----
  				break;
  		}
  	}
  	/* Queue any pending notifies */
! 	asyncAddNotificationsIntoQueue(MyDatabaseId, InvalidTransactionId);
! }
  
! static void
! asyncAddNotificationsIntoQueue(Oid dboid, TransactionId xid)
! {
! 	ListCell   *nextNotify;
! 
! 	if (pendingNotifies == NIL)
! 		return;
! 
! 	/*
! 	 * Make sure that we have an XID assigned to the current transaction.
! 	 * GetCurrentTransactionId is cheap if we already have an XID, but not
! 	 * so cheap if we don't, and we'd prefer not to do that work while
! 	 * holding AsyncQueueLock.
! 	 */
! 	if (!TransactionIdIsValid(xid))
! 		xid = GetCurrentTransactionId();
  
+ 	/*
+ 	 * Serialize writers by acquiring a special lock that we hold till
+ 	 * after commit.  This ensures that queue entries appear in commit
+ 	 * order, and in particular that there are never uncommitted queue
+ 	 * entries ahead of committed ones, so an uncommitted transaction
+ 	 * can't block delivery of deliverable notifications.
+ 	 *
+ 	 * We use a heavyweight lock so that it'll automatically be released
+ 	 * after either commit or abort.  This also allows deadlocks to be
+ 	 * detected, though really a deadlock shouldn't be possible here.
+ 	 *
+ 	 * The lock is on "database 0", which is pretty ugly but it doesn't
+ 	 * seem worth inventing a special locktag category just for this.
+ 	 * (Historical note: before PG 9.0, a similar lock on "database 0" was
+ 	 * used by the flatfiles mechanism.)
+ 	 */
+ 	if (!RecoveryInProgress())
+ 		LockSharedObject(DatabaseRelationId, InvalidOid, 0,
+ 						 AccessExclusiveLock);
+ 
+ 	/* Now push the notifications into the queue */
+ 	backendHasSentNotifications = true;
+ 
+ 	nextNotify = list_head(pendingNotifies);
+ 	while (nextNotify != NULL)
+ 	{
  		/*
! 		 * Add the pending notifications to the queue.	We acquire and
! 		 * release AsyncQueueLock once per page, which might be overkill
! 		 * but it does allow readers to get in while we're doing this.
  		 *
! 		 * A full queue is very uncommon and should really not happen,
! 		 * given that we have so much space available in the SLRU pages.
! 		 * Nevertheless we need to deal with this possibility.
  		 *
! 		 * On the primary, note that when we get here we are in the process of
! 		 * committing our transaction, but we have not yet committed to clog,
! 		 * so at this point in time we can still roll the transaction back.
! 		 *
! 		 * On the standby if we wouldn't have listening processes, we'd never
! 		 * run into the issue because if the master overflows, then we don't
! 		 * even get that XLOG record that causes the overflow. So the only
! 		 * reason why it can happen on the standby is that there are backend
! 		 * processes that are busy and haven't had the time to catch up. We
! 		 * need to kill these now so that we can go on.
  		 */
! 		LWLockAcquire(AsyncQueueLock, LW_EXCLUSIVE);
! 		asyncQueueFillWarning();
! 		if (RecoveryInProgress())
! 		{
! 			while (asyncQueueIsFull())
! 			{
! 				int32 pid;
! 				int backendId = 1;
! 				bool found = false;
! 				while ((pid = asyncQueueGetSlowestPid(&backendId)) != InvalidPid)
! 				{
! 					SendRecoveryConflictWithNotifyQueue(pid);
! 					/* continue checking the next backend and not the same one again */
! 					backendId++;
! 					found = true;
! 				}
  
! 				/*
! 				 * This shouldn't really happen. It could happen if the master
! 				 * had a larger queue than the standby and if we filled up the queue
! 				 * in one step (for only one transaction) or if we do only have
! 				 * uncommitted transactions in the queue. But since the master
! 				 * always has the same queue, such a transaction would have
! 				 * errored out on the master already.
! 				 */
! 				if (!found)
! 					elog(PANIC, "Notify queue is full but nobody is blocking it");
  
! 				/* Need to release the lock so that others can catch up */
! 				LWLockRelease(AsyncQueueLock);
! 				/*
! 				 * wait a bit (250ms) for other processes to terminate their
! 				 * transactions and catch up.
! 				 */
! 				pg_usleep(250000);
! 				/* Grab the lock again before looping */
! 				LWLockAcquire(AsyncQueueLock, LW_EXCLUSIVE);
! 			}
! 		}
! 		else
  		{
  			if (asyncQueueIsFull())
  				ereport(ERROR,
  						(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
  					  errmsg("too many notifications in the NOTIFY queue")));
  		}
+ 		nextNotify = asyncQueueAddEntries(nextNotify, dboid, xid);
+ 		LWLockRelease(AsyncQueueLock);
  	}
  }
  
*************** AtCommit_Notify(void)
*** 903,908 ****
--- 969,993 ----
  }
  
  /*
+  * AtEOXact_HS_Notify
+  *
+  *		This is called at transaction commit/abort, after committing/aborting
+  *		to clog.
+  *
+  *		Update listenChannels and clear transaction-local state.
+  */
+ void
+ AtEOXact_HS_Notify(void)
+ {
+ 	/*
+ 	 * Send signals to other backends if we have sent some out, if there's
+ 	 * no other backend listening, advance the queue pointer ourselves.
+ 	 */
+ 	if (!SignalBackends())
+ 		asyncQueueAdvanceTail();
+ }
+ 
+ /*
   * Exec_ListenPreCommit --- subroutine for PreCommit_Notify
   *
   * This function must make sure we are ready to catch any incoming messages.
*************** asyncQueueAdvance(QueuePosition *positio
*** 1245,1251 ****
   * Fill the AsyncQueueEntry at *qe with an outbound notification message.
   */
  static void
! asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe)
  {
  	size_t		channellen = strlen(n->channel);
  	size_t		payloadlen = strlen(n->payload);
--- 1330,1337 ----
   * Fill the AsyncQueueEntry at *qe with an outbound notification message.
   */
  static void
! asyncQueueNotificationToEntry(Notification *n, AsyncQueueEntry *qe,
! 							  Oid dboid, TransactionId xid)
  {
  	size_t		channellen = strlen(n->channel);
  	size_t		payloadlen = strlen(n->payload);
*************** asyncQueueNotificationToEntry(Notificati
*** 1255,1265 ****
  	Assert(payloadlen < NOTIFY_PAYLOAD_MAX_LENGTH);
  
  	/* The terminators are already included in AsyncQueueEntryEmptySize */
! 	entryLength = AsyncQueueEntryEmptySize + payloadlen + channellen;
  	entryLength = QUEUEALIGN(entryLength);
  	qe->length = entryLength;
! 	qe->dboid = MyDatabaseId;
! 	qe->xid = GetCurrentTransactionId();
  	qe->srcPid = MyProcPid;
  	memcpy(qe->data, n->channel, channellen + 1);
  	memcpy(qe->data + channellen + 1, n->payload, payloadlen + 1);
--- 1341,1352 ----
  	Assert(payloadlen < NOTIFY_PAYLOAD_MAX_LENGTH);
  
  	/* The terminators are already included in AsyncQueueEntryEmptySize */
! 	qe->str_length = channellen + payloadlen;
! 	entryLength = AsyncQueueEntryEmptySize + qe->str_length;
  	entryLength = QUEUEALIGN(entryLength);
  	qe->length = entryLength;
! 	qe->dboid = dboid;
! 	qe->xid = xid;
  	qe->srcPid = MyProcPid;
  	memcpy(qe->data, n->channel, channellen + 1);
  	memcpy(qe->data + channellen + 1, n->payload, payloadlen + 1);
*************** asyncQueueNotificationToEntry(Notificati
*** 1282,1293 ****
   * locally in this function.
   */
  static ListCell *
! asyncQueueAddEntries(ListCell *nextNotify)
  {
  	AsyncQueueEntry qe;
  	int			pageno;
  	int			offset;
  	int			slotno;
  
  	/* We hold both AsyncQueueLock and AsyncCtlLock during this operation */
  	LWLockAcquire(AsyncCtlLock, LW_EXCLUSIVE);
--- 1369,1383 ----
   * locally in this function.
   */
  static ListCell *
! asyncQueueAddEntries(ListCell *nextNotify, Oid dboid, TransactionId xid)
  {
  	AsyncQueueEntry qe;
  	int			pageno;
  	int			offset;
  	int			slotno;
+ 	ListCell   *HSStart = NULL;
+ 	ListCell   *HSEnd = NULL;
+ 	uint32		xlogLength = 0;
  
  	/* We hold both AsyncQueueLock and AsyncCtlLock during this operation */
  	LWLockAcquire(AsyncCtlLock, LW_EXCLUSIVE);
*************** asyncQueueAddEntries(ListCell *nextNotif
*** 1301,1315 ****
  	while (nextNotify != NULL)
  	{
  		Notification *n = (Notification *) lfirst(nextNotify);
  
  		/* Construct a valid queue entry in local variable qe */
! 		asyncQueueNotificationToEntry(n, &qe);
  
  		offset = QUEUE_POS_OFFSET(QUEUE_HEAD);
  
  		/* Check whether the entry really fits on the current page */
  		if (offset + qe.length <= QUEUE_PAGESIZE)
  		{
  			/* OK, so advance nextNotify past this item */
  			nextNotify = lnext(nextNotify);
  		}
--- 1391,1410 ----
  	while (nextNotify != NULL)
  	{
  		Notification *n = (Notification *) lfirst(nextNotify);
+ 		if (!HSStart)
+ 			HSStart = nextNotify;
  
  		/* Construct a valid queue entry in local variable qe */
! 		asyncQueueNotificationToEntry(n, &qe, dboid, xid);
  
  		offset = QUEUE_POS_OFFSET(QUEUE_HEAD);
  
  		/* Check whether the entry really fits on the current page */
  		if (offset + qe.length <= QUEUE_PAGESIZE)
  		{
+ 			xlogLength += qe.str_length + 2;
+ 			HSEnd = nextNotify;
+ 
  			/* OK, so advance nextNotify past this item */
  			nextNotify = lnext(nextNotify);
  		}
*************** asyncQueueAddEntries(ListCell *nextNotif
*** 1350,1360 ****
--- 1445,1489 ----
  
  	LWLockRelease(AsyncCtlLock);
  
+ 	/*
+ 	 * Send out the notifications for the hot standby servers, we do one xlog
+ 	 * update per notfication slru page, we could send more or less but this is
+ 	 * most convenient.
+ 	 * If HSStart == NULL: No notifications sent
+ 	 * If HSEnd == NULL: No more space on this page, so no notifications for this page
+ 	 * Only one entry if HSStart == HSEnd.
+ 	 */
+ 	if (!RecoveryInProgress() && HSStart && HSEnd)
+ 		asyncAddNotifyXLog(HSStart, HSEnd, xlogLength, dboid, xid, nextNotify == NULL);
+ 
  	return nextNotify;
  }
  
  /*
+  * Report pids that have their pointer into the queue set to the same position
+  * as the global tail pointer.
+  * The function receives a pointer to the backend id from where it should start
+  * scanning and updates it as it iterates over the backends. When it returns a
+  * pid, then the backend's id that has this pid is in *backendId.
+  */
+ static int32
+ asyncQueueGetSlowestPid(int *backendId)
+ {
+ 	Assert(backendId != NULL);
+ 	/* backend ids start with 1, 0 is unused, we expect to always get at least 1. */
+ 	Assert(*backendId >= 1);
+ 
+ 	for (; *backendId <= MaxBackends; (*backendId)++)
+ 		if (QUEUE_BACKEND_PID(*backendId) != InvalidPid)
+ 			if (QUEUE_POS_EQUAL(QUEUE_TAIL, QUEUE_BACKEND_POS(*backendId)))
+ 				return QUEUE_BACKEND_PID(*backendId);
+ 
+ 	return InvalidPid;
+ }
+ 
+ /*
   * Check whether the queue is at least half full, and emit a warning if so.
+  * Return the pid of one of the lazy backends that aren't catching up.
   *
   * This is unlikely given the size of the queue, but possible.
   * The warnings show up at most once every QUEUE_FULL_WARN_INTERVAL.
*************** asyncQueueFillWarning(void)
*** 1391,1410 ****
  	if (TimestampDifferenceExceeds(asyncQueueControl->lastQueueFillWarn,
  								   t, QUEUE_FULL_WARN_INTERVAL))
  	{
! 		QueuePosition min = QUEUE_HEAD;
! 		int32		minPid = InvalidPid;
! 		int			i;
! 
! 		for (i = 1; i <= MaxBackends; i++)
! 		{
! 			if (QUEUE_BACKEND_PID(i) != InvalidPid)
! 			{
! 				min = QUEUE_POS_MIN(min, QUEUE_BACKEND_POS(i));
! 				if (QUEUE_POS_EQUAL(min, QUEUE_BACKEND_POS(i)))
! 					minPid = QUEUE_BACKEND_PID(i);
! 			}
! 		}
! 
  		ereport(WARNING,
  				(errmsg("NOTIFY queue is %.0f%% full", fillDegree * 100),
  				 (minPid != InvalidPid ?
--- 1520,1528 ----
  	if (TimestampDifferenceExceeds(asyncQueueControl->lastQueueFillWarn,
  								   t, QUEUE_FULL_WARN_INTERVAL))
  	{
! 		/* backend ids start at 1, slot 0 is unused */
! 		int backendId = 1;
! 		int32 minPid = asyncQueueGetSlowestPid(&backendId);
  		ereport(WARNING,
  				(errmsg("NOTIFY queue is %.0f%% full", fillDegree * 100),
  				 (minPid != InvalidPid ?
*************** asyncQueueFillWarning(void)
*** 1418,1423 ****
--- 1536,1650 ----
  	}
  }
  
+ /*
+  * Create an Xlog entry from the notifications of HSStart up to and including HSEnd.
+  * The XLOG entry has a header and then
+  * channel\0payload\0channel\0payload\0...
+  */
+ static void
+ asyncAddNotifyXLog(ListCell *HSStart, ListCell *HSEnd, uint32 xlogLength,
+ 				   Oid dboid, TransactionId xid, bool isLast)
+ {
+ 	/* -1 because NotifyData has already a buffer[1] */
+ 	NotifyData *rdata = (NotifyData *) palloc(sizeof(NotifyData) + xlogLength - 1);
+ 	char	   *dataPtr = rdata->notify_buffer;
+ 	ListCell   *nextNotify = HSStart;
+ 	char	   *p;
+ 
+ 	Assert(HSEnd != NULL);
+ 
+ 	for (;;)
+ 	{
+ 		Notification *n = (Notification *) lfirst(nextNotify);
+ 
+ 		p = n->channel;
+ 		while (*p)
+ 			*dataPtr++ = *p++;
+ 		*dataPtr++ = *p++;
+ 
+ 		p = n->payload;
+ 		while (*p)
+ 			*dataPtr++ = *p++;
+ 		*dataPtr++ = *p++;
+ 
+ 		if (nextNotify == HSEnd)
+ 			break;
+ 
+ 		nextNotify = lnext(nextNotify);
+ 	}
+ 
+ 	Assert(dataPtr == rdata->notify_buffer + xlogLength);
+ 
+ 	rdata->notify_dboid = dboid;
+ 	rdata->notify_len = xlogLength;
+ 	rdata->notify_isLast = isLast;
+ 
+ 	XLogInsertNotifyData(rdata);
+ }
+ 
+ /*
+  * This function is only run from the recovery process (which is why we can use
+  * a static variable here). The function always runs in NotifyCtx but only for
+  * the time the function is run.
+  * It continues to use its snapshot until notify_isLast is set.
+  */
+ void
+ asyncGetNotifyFromXLog(NotifyData *rdata, TransactionId xid)
+ {
+ 	static MemoryContext NotifyCtx = NULL;
+ 	MemoryContext oldctx;
+ 	char	   *data = rdata->notify_buffer;
+ 
+ 	Assert(CurrentMemoryContext != NotifyCtx);
+ 
+ 	if (!NotifyCtx)
+ 	{
+ 		NotifyCtx = AllocSetContextCreate(TopMemoryContext,
+ 										  "NotifyXLogMemoryContext",
+ 										  ALLOCSET_DEFAULT_MINSIZE,
+ 										  ALLOCSET_DEFAULT_INITSIZE,
+ 										  ALLOCSET_DEFAULT_MAXSIZE);
+ 	}
+ 	oldctx = MemoryContextSwitchTo(NotifyCtx);
+ 
+ 	while ((data - rdata->notify_buffer) < rdata->notify_len)
+ 	{
+ 		Notification *n = (Notification *) palloc(sizeof(Notification));
+ 
+ 		n->channel = data;
+ 		while (*data)
+ 			data++;
+ 		data++;
+ 
+ 		n->payload = data;
+ 		while (*data)
+ 			data++;
+ 		data++;
+ 
+ 		pendingNotifies = lappend(pendingNotifies, n);
+ 	}
+ 
+ 	/* We should hit it exactly */
+ 	Assert((data - rdata->notify_buffer) == rdata->notify_len);
+ 
+ 	asyncAddNotificationsIntoQueue(rdata->notify_dboid, xid);
+ 	pendingNotifies = NIL;
+ 
+ 	MemoryContextSwitchTo(oldctx);
+ 
+ 	if (rdata->notify_isLast)
+ 	{
+ 		/*
+ 		 * XXX Maybe we should only do a MemoryContextReset for performance
+ 		 * reasons. However then we keep the space allocated that the
+ 		 * transaction with most/biggest notifications needed. And that might be
+ 		 * carried on forever.
+ 		 */
+ 		MemoryContextDelete(NotifyCtx);
+ 		NotifyCtx = NULL;
+ 	}
+ }
+ 
  /*
   * Send signals to all listening backends (except our own).
   *
diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c
index 323d42b..21c10fe 100644
*** a/src/backend/postmaster/pgstat.c
--- b/src/backend/postmaster/pgstat.c
*************** pgstat_get_db_entry(Oid databaseid, bool
*** 3266,3271 ****
--- 3266,3272 ----
  		result->n_conflict_snapshot = 0;
  		result->n_conflict_bufferpin = 0;
  		result->n_conflict_startup_deadlock = 0;
+ 		result->n_conflict_notify_queue = 0;
  
  		result->stat_reset_timestamp = GetCurrentTimestamp();
  
*************** pgstat_recv_recoveryconflict(PgStat_MsgR
*** 4399,4404 ****
--- 4400,4408 ----
  		case PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK:
  			dbentry->n_conflict_startup_deadlock++;
  			break;
+ 		case PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE:
+ 			dbentry->n_conflict_notify_queue++;
+ 			break;
  	}
  }
  
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 6ea0a28..a802992 100644
*** a/src/backend/storage/ipc/procarray.c
--- b/src/backend/storage/ipc/procarray.c
*************** GetCurrentVirtualXIDs(TransactionId limi
*** 1999,2004 ****
--- 1999,2021 ----
  }
  
  /*
+  * GetVirtualXIDfromPid -- returns the current VXID of a certain backend pid
+  */
+ VirtualTransactionId
+ GetVirtualXIDfromPid(int pid)
+ {
+ 	VirtualTransactionId vxid;
+ 	PGPROC *proc = BackendPidGetProc(pid);
+ 
+ 	if (proc)
+ 		GET_VXID_FROM_PGPROC(vxid, *proc);
+ 	else
+ 		SetInvalidVirtualTransactionId(vxid);
+ 
+ 	return vxid;
+ }
+ 
+ /*
   * GetConflictingVirtualXIDs -- returns an array of currently active VXIDs.
   *
   * Usage is limited to conflict resolution during recovery on standby servers.
diff --git a/src/backend/storage/ipc/procsignal.c b/src/backend/storage/ipc/procsignal.c
index 3d7e85f..7f0bf52 100644
*** a/src/backend/storage/ipc/procsignal.c
--- b/src/backend/storage/ipc/procsignal.c
*************** procsignal_sigusr1_handler(SIGNAL_ARGS)
*** 276,281 ****
--- 276,284 ----
  	if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN))
  		RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_BUFFERPIN);
  
+ 	if (CheckProcSignal(PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE))
+ 		RecoveryConflictInterrupt(PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE);
+ 
  	latch_sigusr1_handler();
  
  	errno = save_errno;
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index c88557c..535f456 100644
*** a/src/backend/storage/ipc/standby.c
--- b/src/backend/storage/ipc/standby.c
*************** SendRecoveryConflictWithBufferPin(ProcSi
*** 458,463 ****
--- 458,472 ----
  	CancelDBBackends(InvalidOid, reason, false);
  }
  
+ void
+ SendRecoveryConflictWithNotifyQueue(int pid)
+ {
+ 	VirtualTransactionId vxid = GetVirtualXIDfromPid(pid);
+ 	if (!VirtualTransactionIdIsValid(vxid))
+ 		return;
+ 	CancelVirtualTransaction(vxid, PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE);
+ }
+ 
  /*
   * In Hot Standby perform early deadlock detection.  We abort the lock
   * wait if we are about to sleep while holding the buffer pin that Startup
diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c
index 59a287f..29664e9 100644
*** a/src/backend/tcop/postgres.c
--- b/src/backend/tcop/postgres.c
*************** errdetail_recovery_conflict(void)
*** 2196,2201 ****
--- 2196,2204 ----
  		case PROCSIG_RECOVERY_CONFLICT_DATABASE:
  			errdetail("User was connected to a database that must be dropped.");
  			break;
+ 		case PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE:
+ 			errdetail("User was not processing notify messages promptly enough.");
+ 			break;
  		default:
  			break;
  			/* no errdetail */
*************** RecoveryConflictInterrupt(ProcSignalReas
*** 2698,2703 ****
--- 2701,2707 ----
  			case PROCSIG_RECOVERY_CONFLICT_LOCK:
  			case PROCSIG_RECOVERY_CONFLICT_TABLESPACE:
  			case PROCSIG_RECOVERY_CONFLICT_SNAPSHOT:
+ 			case PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE:
  
  				/*
  				 * If we aren't in a transaction any longer then ignore.
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 923da6a..f538290 100644
*** a/src/backend/tcop/utility.c
--- b/src/backend/tcop/utility.c
*************** standard_ProcessUtility(Node *parsetree,
*** 971,977 ****
  			{
  				ListenStmt *stmt = (ListenStmt *) parsetree;
  
- 				PreventCommandDuringRecovery("LISTEN");
  				CheckRestrictedOperation("LISTEN");
  				Async_Listen(stmt->conditionname);
  			}
--- 971,976 ----
*************** standard_ProcessUtility(Node *parsetree,
*** 981,987 ****
  			{
  				UnlistenStmt *stmt = (UnlistenStmt *) parsetree;
  
- 				PreventCommandDuringRecovery("UNLISTEN");
  				CheckRestrictedOperation("UNLISTEN");
  				if (stmt->conditionname)
  					Async_Unlisten(stmt->conditionname);
--- 980,985 ----
diff --git a/src/backend/utils/adt/pgstatfuncs.c b/src/backend/utils/adt/pgstatfuncs.c
index b4986d8..bd25f67 100644
*** a/src/backend/utils/adt/pgstatfuncs.c
--- b/src/backend/utils/adt/pgstatfuncs.c
*************** extern Datum pg_stat_get_db_conflict_loc
*** 77,82 ****
--- 77,83 ----
  extern Datum pg_stat_get_db_conflict_snapshot(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_db_conflict_bufferpin(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_db_conflict_startup_deadlock(PG_FUNCTION_ARGS);
+ extern Datum pg_stat_get_db_conflict_notify_queue(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS);
  extern Datum pg_stat_get_db_stat_reset_time(PG_FUNCTION_ARGS);
  
*************** pg_stat_get_db_conflict_startup_deadlock
*** 1253,1258 ****
--- 1254,1274 ----
  
  	PG_RETURN_INT64(result);
  }
+ 
+ Datum
+ pg_stat_get_db_conflict_notify_queue(PG_FUNCTION_ARGS)
+ {
+ 	Oid			dbid = PG_GETARG_OID(0);
+ 	int64		result;
+ 	PgStat_StatDBEntry *dbentry;
+ 
+ 	if ((dbentry = pgstat_fetch_stat_dbentry(dbid)) == NULL)
+ 		result = 0;
+ 	else
+ 		result = (int64) (dbentry->n_conflict_notify_queue);
+ 
+ 	PG_RETURN_INT64(result);
+ }
  
  Datum
  pg_stat_get_db_conflict_all(PG_FUNCTION_ARGS)
diff --git a/src/include/access/xact.h b/src/include/access/xact.h
index 5f063a9..333350f 100644
*** a/src/include/access/xact.h
--- b/src/include/access/xact.h
*************** typedef struct xl_xact_commit
*** 135,140 ****
--- 135,141 ----
  	int			nmsgs;			/* number of shared inval msgs */
  	Oid			dbId;			/* MyDatabaseId */
  	Oid			tsId;			/* MyDatabaseTableSpace */
+ 	bool		sent_notifications;  /* true if notifications have been sent */
  	/* Array of RelFileNode(s) to drop at commit */
  	RelFileNode xnodes[1];		/* VARIABLE LENGTH ARRAY */
  	/* ARRAY OF COMMITTED SUBTRANSACTION XIDs FOLLOWS */
*************** typedef struct xl_xact_abort
*** 163,168 ****
--- 164,170 ----
  	TimestampTz xact_time;		/* time of abort */
  	int			nrels;			/* number of RelFileNodes */
  	int			nsubxacts;		/* number of subtransaction XIDs */
+ 	bool		sent_notifications;  /* true if notifications have been sent */
  	/* Array of RelFileNode(s) to drop at abort */
  	RelFileNode xnodes[1];		/* VARIABLE LENGTH ARRAY */
  	/* ARRAY OF ABORTED SUBTRANSACTION XIDs FOLLOWS */
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 93622c4..c95902d 100644
*** a/src/include/access/xlog.h
--- b/src/include/access/xlog.h
*************** typedef struct CheckpointStatsData
*** 263,268 ****
--- 263,277 ----
  
  extern CheckpointStatsData CheckpointStats;
  
+ typedef struct NotifyData
+ {
+ 	Oid			notify_dboid;
+ 	bool		notify_isLast;
+ 	uint32		notify_len;
+ 	char	    notify_buffer[1];
+ 	/* notify_buffer with variable-length down here */
+ } NotifyData;
+ 
  extern XLogRecPtr XLogInsert(RmgrId rmid, uint8 info, XLogRecData *rdata);
  extern void XLogFlush(XLogRecPtr RecPtr);
  extern void XLogBackgroundFlush(void);
*************** extern void InitXLOGAccess(void);
*** 306,311 ****
--- 315,321 ----
  extern void CreateCheckPoint(int flags);
  extern bool CreateRestartPoint(int flags);
  extern void XLogPutNextOid(Oid nextOid);
+ extern void XLogInsertNotifyData(NotifyData *notifyRecord);
  extern XLogRecPtr XLogRestorePoint(const char *rpName);
  extern XLogRecPtr GetRedoRecPtr(void);
  extern XLogRecPtr GetInsertRecPtr(void);
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index f7af5fd..a2764d4 100644
*** a/src/include/catalog/catversion.h
--- b/src/include/catalog/catversion.h
***************
*** 52,57 ****
--- 52,58 ----
   * catalog changes on the same day...)
   */
  
+ /* XXX Catversion needs new version with the Notify-to-Standby feature */
  /*							yyyymmddN */
  #define CATALOG_VERSION_NO	201112241
  
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h
index d0d2e9e..a2fe517 100644
*** a/src/include/catalog/pg_control.h
--- b/src/include/catalog/pg_control.h
*************** typedef struct CheckPoint
*** 60,65 ****
--- 60,66 ----
  #define XLOG_BACKUP_END					0x50
  #define XLOG_PARAMETER_CHANGE			0x60
  #define XLOG_RESTORE_POINT				0x70
+ #define XLOG_NOTIFY						0x80
  
  
  /*
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 355c61a..c7be249 100644
*** a/src/include/catalog/pg_proc.h
--- b/src/include/catalog/pg_proc.h
*************** DATA(insert OID = 3068 (  pg_stat_get_db
*** 2629,2634 ****
--- 2629,2636 ----
  DESCR("statistics: recovery conflicts in database caused by shared buffer pin");
  DATA(insert OID = 3069 (  pg_stat_get_db_conflict_startup_deadlock PGNSP PGUID 12 1 0 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_db_conflict_startup_deadlock _null_ _null_ _null_ ));
  DESCR("statistics: recovery conflicts in database caused by buffer deadlock");
+ DATA(insert OID = 3144 (  pg_stat_get_db_conflict_notify_queue PGNSP PGUID 12 1 0 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_db_conflict_notify_queue _null_ _null_ _null_ ));
+ DESCR("statistics: recovery conflicts in database caused by the notify queue");
  DATA(insert OID = 3070 (  pg_stat_get_db_conflict_all PGNSP PGUID 12 1 0 0 0 f f f t f s 1 0 20 "26" _null_ _null_ _null_ _null_ pg_stat_get_db_conflict_all _null_ _null_ _null_ ));
  DESCR("statistics: recovery conflicts in database");
  DATA(insert OID = 3074 (  pg_stat_get_db_stat_reset_time PGNSP PGUID 12 1 0 0 0 f f f t f s 1 0 1184 "26" _null_ _null_ _null_ _null_ pg_stat_get_db_stat_reset_time _null_ _null_ _null_ ));
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index 3d1cf68..3da9bbe 100644
*** a/src/include/commands/async.h
--- b/src/include/commands/async.h
***************
*** 21,26 ****
--- 21,27 ----
  #define NUM_ASYNC_BUFFERS	8
  
  extern bool Trace_notify;
+ extern bool backendHasSentNotifications;
  
  extern Size AsyncShmemSize(void);
  extern void AsyncShmemInit(void);
*************** extern Datum pg_notify(PG_FUNCTION_ARGS)
*** 38,43 ****
--- 39,45 ----
  /* perform (or cancel) outbound notify processing at transaction commit */
  extern void PreCommit_Notify(void);
  extern void AtCommit_Notify(void);
+ extern void AtEOXact_HS_Notify(void);
  extern void AtAbort_Notify(void);
  extern void AtSubStart_Notify(void);
  extern void AtSubCommit_Notify(void);
*************** extern void HandleNotifyInterrupt(void);
*** 56,59 ****
--- 58,67 ----
  extern void EnableNotifyInterrupt(void);
  extern bool DisableNotifyInterrupt(void);
  
+ /*
+  * Called from redo_xlog in xlog.c to inject notifications into a replicated
+  * database.
+  */
+ void asyncGetNotifyFromXLog(NotifyData *rdata, TransactionId xid);
+ 
  #endif   /* ASYNC_H */
diff --git a/src/include/pgstat.h b/src/include/pgstat.h
index b8c6d82..a14f733 100644
*** a/src/include/pgstat.h
--- b/src/include/pgstat.h
*************** typedef struct PgStat_StatDBEntry
*** 507,512 ****
--- 507,513 ----
  	PgStat_Counter n_conflict_snapshot;
  	PgStat_Counter n_conflict_bufferpin;
  	PgStat_Counter n_conflict_startup_deadlock;
+ 	PgStat_Counter n_conflict_notify_queue;
  	TimestampTz stat_reset_timestamp;
  
  
diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h
index 0b0aa35..64f9433 100644
*** a/src/include/storage/procarray.h
--- b/src/include/storage/procarray.h
*************** extern bool IsBackendPid(int pid);
*** 62,67 ****
--- 62,68 ----
  extern VirtualTransactionId *GetCurrentVirtualXIDs(TransactionId limitXmin,
  					  bool excludeXmin0, bool allDbs, int excludeVacuum,
  					  int *nvxids);
+ extern VirtualTransactionId GetVirtualXIDfromPid(int pid);
  extern VirtualTransactionId *GetConflictingVirtualXIDs(TransactionId limitXmin, Oid dbOid);
  extern pid_t CancelVirtualTransaction(VirtualTransactionId vxid, ProcSignalReason sigmode);
  
diff --git a/src/include/storage/procsignal.h b/src/include/storage/procsignal.h
index 715e5be..181dbde 100644
*** a/src/include/storage/procsignal.h
--- b/src/include/storage/procsignal.h
*************** typedef enum
*** 39,44 ****
--- 39,45 ----
  	PROCSIG_RECOVERY_CONFLICT_SNAPSHOT,
  	PROCSIG_RECOVERY_CONFLICT_BUFFERPIN,
  	PROCSIG_RECOVERY_CONFLICT_STARTUP_DEADLOCK,
+ 	PROCSIG_RECOVERY_CONFLICT_NOTIFY_QUEUE,
  
  	NUM_PROCSIGNALS				/* Must be last! */
  } ProcSignalReason;
diff --git a/src/include/storage/standby.h b/src/include/storage/standby.h
index a539ec2..671fe0d 100644
*** a/src/include/storage/standby.h
--- b/src/include/storage/standby.h
*************** extern void ResolveRecoveryConflictWithD
*** 35,40 ****
--- 35,41 ----
  
  extern void ResolveRecoveryConflictWithBufferPin(void);
  extern void SendRecoveryConflictWithBufferPin(ProcSignalReason reason);
+ extern void SendRecoveryConflictWithNotifyQueue(int pid);
  extern void CheckRecoveryConflictDeadlock(void);
  
  /*
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 454e1f9..a273d64 100644
*** a/src/test/regress/expected/rules.out
--- b/src/test/regress/expected/rules.out
*************** SELECT viewname, definition FROM pg_view
*** 1297,1303 ****
   pg_stat_all_tables              | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_numscans(c.oid) AS seq_scan, pg_stat_get_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_tuples_hot_updated(c.oid) AS n_tup_hot_upd, pg_stat_get_live_tuples(c.oid) AS n_live_tup, pg_stat_get_dead_tuples(c.oid) AS n_dead_tup, pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum, pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum, pg_stat_get_last_analyze_time(c.oid) AS last_analyze, pg_stat_get_last_autoanalyze_time(c.oid) AS last_autoanalyze, pg_stat_get_vacuum_count(c.oid) AS vacuum_count, pg_stat_get_autovacuum_count(c.oid) AS autovacuum_count, pg_stat_get_analyze_count(c.oid) AS analyze_count, pg_stat_get_autoanalyze_count(c.oid) AS autoanalyze_count FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
   pg_stat_bgwriter                | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_fsync_backend() AS buffers_backend_fsync, pg_stat_get_buf_alloc() AS buffers_alloc, pg_stat_get_bgwriter_stat_reset_time() AS stats_reset;
   pg_stat_database                | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted, pg_stat_get_db_conflict_all(d.oid) AS conflicts, pg_stat_get_db_stat_reset_time(d.oid) AS stats_reset FROM pg_database d;
!  pg_stat_database_conflicts      | SELECT d.oid AS datid, d.datname, pg_stat_get_db_conflict_tablespace(d.oid) AS confl_tablespace, pg_stat_get_db_conflict_lock(d.oid) AS confl_lock, pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot, pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin, pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock FROM pg_database d;
   pg_stat_replication             | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_hostname, s.client_port, s.backend_start, w.state, w.sent_location, w.write_location, w.flush_location, w.replay_location, w.sync_priority, w.sync_state FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_hostname, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location, write_location, flush_location, replay_location, sync_priority, sync_state) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
   pg_stat_sys_indexes             | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
   pg_stat_sys_tables              | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze, pg_stat_all_tables.vacuum_count, pg_stat_all_tables.autovacuum_count, pg_stat_all_tables.analyze_count, pg_stat_all_tables.autoanalyze_count FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));
--- 1297,1303 ----
   pg_stat_all_tables              | SELECT c.oid AS relid, n.nspname AS schemaname, c.relname, pg_stat_get_numscans(c.oid) AS seq_scan, pg_stat_get_tuples_returned(c.oid) AS seq_tup_read, (sum(pg_stat_get_numscans(i.indexrelid)))::bigint AS idx_scan, ((sum(pg_stat_get_tuples_fetched(i.indexrelid)))::bigint + pg_stat_get_tuples_fetched(c.oid)) AS idx_tup_fetch, pg_stat_get_tuples_inserted(c.oid) AS n_tup_ins, pg_stat_get_tuples_updated(c.oid) AS n_tup_upd, pg_stat_get_tuples_deleted(c.oid) AS n_tup_del, pg_stat_get_tuples_hot_updated(c.oid) AS n_tup_hot_upd, pg_stat_get_live_tuples(c.oid) AS n_live_tup, pg_stat_get_dead_tuples(c.oid) AS n_dead_tup, pg_stat_get_last_vacuum_time(c.oid) AS last_vacuum, pg_stat_get_last_autovacuum_time(c.oid) AS last_autovacuum, pg_stat_get_last_analyze_time(c.oid) AS last_analyze, pg_stat_get_last_autoanalyze_time(c.oid) AS last_autoanalyze, pg_stat_get_vacuum_count(c.oid) AS vacuum_count, pg_stat_get_autovacuum_count(c.oid) AS autovacuum_count, pg_stat_get_analyze_count(c.oid) AS analyze_count, pg_stat_get_autoanalyze_count(c.oid) AS autoanalyze_count FROM ((pg_class c LEFT JOIN pg_index i ON ((c.oid = i.indrelid))) LEFT JOIN pg_namespace n ON ((n.oid = c.relnamespace))) WHERE (c.relkind = ANY (ARRAY['r'::"char", 't'::"char"])) GROUP BY c.oid, n.nspname, c.relname;
   pg_stat_bgwriter                | SELECT pg_stat_get_bgwriter_timed_checkpoints() AS checkpoints_timed, pg_stat_get_bgwriter_requested_checkpoints() AS checkpoints_req, pg_stat_get_bgwriter_buf_written_checkpoints() AS buffers_checkpoint, pg_stat_get_bgwriter_buf_written_clean() AS buffers_clean, pg_stat_get_bgwriter_maxwritten_clean() AS maxwritten_clean, pg_stat_get_buf_written_backend() AS buffers_backend, pg_stat_get_buf_fsync_backend() AS buffers_backend_fsync, pg_stat_get_buf_alloc() AS buffers_alloc, pg_stat_get_bgwriter_stat_reset_time() AS stats_reset;
   pg_stat_database                | SELECT d.oid AS datid, d.datname, pg_stat_get_db_numbackends(d.oid) AS numbackends, pg_stat_get_db_xact_commit(d.oid) AS xact_commit, pg_stat_get_db_xact_rollback(d.oid) AS xact_rollback, (pg_stat_get_db_blocks_fetched(d.oid) - pg_stat_get_db_blocks_hit(d.oid)) AS blks_read, pg_stat_get_db_blocks_hit(d.oid) AS blks_hit, pg_stat_get_db_tuples_returned(d.oid) AS tup_returned, pg_stat_get_db_tuples_fetched(d.oid) AS tup_fetched, pg_stat_get_db_tuples_inserted(d.oid) AS tup_inserted, pg_stat_get_db_tuples_updated(d.oid) AS tup_updated, pg_stat_get_db_tuples_deleted(d.oid) AS tup_deleted, pg_stat_get_db_conflict_all(d.oid) AS conflicts, pg_stat_get_db_stat_reset_time(d.oid) AS stats_reset FROM pg_database d;
!  pg_stat_database_conflicts      | SELECT d.oid AS datid, d.datname, pg_stat_get_db_conflict_tablespace(d.oid) AS confl_tablespace, pg_stat_get_db_conflict_lock(d.oid) AS confl_lock, pg_stat_get_db_conflict_snapshot(d.oid) AS confl_snapshot, pg_stat_get_db_conflict_bufferpin(d.oid) AS confl_bufferpin, pg_stat_get_db_conflict_startup_deadlock(d.oid) AS confl_deadlock, pg_stat_get_db_conflict_notify_queue(d.oid) AS confl_notify FROM pg_database d;
   pg_stat_replication             | SELECT s.procpid, s.usesysid, u.rolname AS usename, s.application_name, s.client_addr, s.client_hostname, s.client_port, s.backend_start, w.state, w.sent_location, w.write_location, w.flush_location, w.replay_location, w.sync_priority, w.sync_state FROM pg_stat_get_activity(NULL::integer) s(datid, procpid, usesysid, application_name, current_query, waiting, xact_start, query_start, backend_start, client_addr, client_hostname, client_port), pg_authid u, pg_stat_get_wal_senders() w(procpid, state, sent_location, write_location, flush_location, replay_location, sync_priority, sync_state) WHERE ((s.usesysid = u.oid) AND (s.procpid = w.procpid));
   pg_stat_sys_indexes             | SELECT pg_stat_all_indexes.relid, pg_stat_all_indexes.indexrelid, pg_stat_all_indexes.schemaname, pg_stat_all_indexes.relname, pg_stat_all_indexes.indexrelname, pg_stat_all_indexes.idx_scan, pg_stat_all_indexes.idx_tup_read, pg_stat_all_indexes.idx_tup_fetch FROM pg_stat_all_indexes WHERE ((pg_stat_all_indexes.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_indexes.schemaname ~ '^pg_toast'::text));
   pg_stat_sys_tables              | SELECT pg_stat_all_tables.relid, pg_stat_all_tables.schemaname, pg_stat_all_tables.relname, pg_stat_all_tables.seq_scan, pg_stat_all_tables.seq_tup_read, pg_stat_all_tables.idx_scan, pg_stat_all_tables.idx_tup_fetch, pg_stat_all_tables.n_tup_ins, pg_stat_all_tables.n_tup_upd, pg_stat_all_tables.n_tup_del, pg_stat_all_tables.n_tup_hot_upd, pg_stat_all_tables.n_live_tup, pg_stat_all_tables.n_dead_tup, pg_stat_all_tables.last_vacuum, pg_stat_all_tables.last_autovacuum, pg_stat_all_tables.last_analyze, pg_stat_all_tables.last_autoanalyze, pg_stat_all_tables.vacuum_count, pg_stat_all_tables.autovacuum_count, pg_stat_all_tables.analyze_count, pg_stat_all_tables.autoanalyze_count FROM pg_stat_all_tables WHERE ((pg_stat_all_tables.schemaname = ANY (ARRAY['pg_catalog'::name, 'information_schema'::name])) OR (pg_stat_all_tables.schemaname ~ '^pg_toast'::text));