diff --git a/contrib/tcn/tcn.c b/contrib/tcn/tcn.c
index 0c274322bd..e29b581824 100644
--- a/contrib/tcn/tcn.c
+++ b/contrib/tcn/tcn.c
@@ -161,7 +161,7 @@ triggered_change_notification(PG_FUNCTION_ARGS)
strcpy_quoted(payload, SPI_getvalue(trigtuple, tupdesc, colno), '\'');
}
- Async_Notify(channel, payload->data);
+ Async_Notify(channel, payload->data, "maybe");
}
ReleaseSysCache(indexTuple);
break;
diff --git a/doc/src/sgml/ref/notify.sgml b/doc/src/sgml/ref/notify.sgml
index e0e125a2a2..96e0d7a990 100644
--- a/doc/src/sgml/ref/notify.sgml
+++ b/doc/src/sgml/ref/notify.sgml
@@ -21,7 +21,8 @@ PostgreSQL documentation
-NOTIFY channel [ , payload ]
+NOTIFY channel [ , payload [ , collapse_mode ] ]
+
@@ -93,20 +94,6 @@ NOTIFY channel [ ,
-
- If the same channel name is signaled multiple times from the same
- transaction with identical payload strings, the
- database server can decide to deliver a single notification only.
- On the other hand, notifications with distinct payload strings will
- always be delivered as distinct notifications. Similarly, notifications from
- different transactions will never get folded into one notification.
- Except for dropping later instances of duplicate notifications,
- NOTIFY guarantees that notifications from the same
- transaction get delivered in the order they were sent. It is also
- guaranteed that messages from different transactions are delivered in
- the order in which the transactions committed.
-
-
It is common for a client that executes NOTIFY
to be listening on the same notification channel itself. In that case
@@ -121,6 +108,41 @@ NOTIFY channel [ ,
+
+
+
+
+
+ Ordering and collapsing of notifications
+
+
+ If the same channel name is signaled multiple times from the same
+ transaction with identical payload strings, the
+ database server can decide to deliver a single notification only,
+ when the value of the collapse_mode parameter is
+ 'maybe' or '' (the empty string).
+
+ If the 'never' collapse mode is specified, the server will
+ deliver all notifications, including duplicates. Turning off deduplication
+ in this way can considerably speed up transactions that emit large numbers
+ of notifications.
+
+ Removal of duplicate notifications takes place within transaction block,
+ finished with COMMIT, END or SAVEPOINT.
+
+
+
+ Notifications with distinct payload strings will
+ always be delivered as distinct notifications. Similarly, notifications from
+ different transactions will never get folded into one notification.
+ Except for dropping later instances of duplicate notifications,
+ NOTIFY guarantees that notifications from the same
+ transaction get delivered in the order they were sent. It is also
+ guaranteed that messages from different transactions are delivered in
+ the order in which the transactions committed.
+
+
+
@@ -147,6 +169,16 @@ NOTIFY channel [ ,
+
+ collapse_mode
+
+
+ The collapse mode to apply when identical notifications are issued within
+ a transaction. The acceptable values are 'maybe' (the
+ default) and 'never'.
+
+
+
@@ -190,6 +222,11 @@ NOTIFY channel [ , NOTIFY command if you need to work with
non-constant channel names and payloads.
+
+ There is a three-argument version, pg_notify(text,
+ text, text) where the third argument takes
+ the value of the collapse_mode parameter.
+
@@ -210,6 +247,21 @@ Asynchronous notification "virtual" with payload "This is the payload" received
LISTEN foo;
SELECT pg_notify('fo' || 'o', 'pay' || 'load');
Asynchronous notification "foo" with payload "payload" received from server process with PID 14728.
+
+/* Identical messages from same (sub-) transaction can be eliminated - unless you use the 'never' collapse mode */
+LISTEN bar;
+BEGIN;
+NOTIFY bar, 'Coffee please';
+NOTIFY bar, 'Coffee please';
+NOTIFY bar, 'Milk please';
+NOTIFY bar, 'Milk please', 'never';
+SAVEPOINT s;
+NOTIFY bar, 'Coffee please';
+COMMIT;
+Asynchronous notification "bar" with payload "Coffee please" received from server process with PID 31517.
+Asynchronous notification "bar" with payload "Milk please" received from server process with PID 31517.
+Asynchronous notification "bar" with payload "Milk please" received from server process with PID 31517.
+Asynchronous notification "bar" with payload "Coffee please" received from server process with PID 31517.
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index ee7c6d41b4..3f206b0143 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -524,7 +524,42 @@ pg_notify(PG_FUNCTION_ARGS)
/* For NOTIFY as a statement, this is checked in ProcessUtility */
PreventCommandDuringRecovery("NOTIFY");
- Async_Notify(channel, payload);
+ Async_Notify(channel, payload, "maybe");
+
+ PG_RETURN_VOID();
+}
+
+
+/*
+ * pg_notify_3args
+ * SQL function to send a notification event, 3-argument version
+ */
+Datum
+pg_notify_3args(PG_FUNCTION_ARGS)
+{
+ const char *channel;
+ const char *payload;
+ const char *collapse_mode;
+
+ if (PG_ARGISNULL(0))
+ channel = "";
+ else
+ channel = text_to_cstring(PG_GETARG_TEXT_PP(0));
+
+ if (PG_ARGISNULL(1))
+ payload = "";
+ else
+ payload = text_to_cstring(PG_GETARG_TEXT_PP(1));
+
+ if (PG_ARGISNULL(2))
+ collapse_mode = "maybe";
+ else
+ collapse_mode = text_to_cstring(PG_GETARG_TEXT_PP(2));
+
+ /* For NOTIFY as a statement, this is checked in ProcessUtility */
+ PreventCommandDuringRecovery("NOTIFY");
+
+ Async_Notify(channel, payload, collapse_mode);
PG_RETURN_VOID();
}
@@ -540,10 +575,11 @@ pg_notify(PG_FUNCTION_ARGS)
* ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
*/
void
-Async_Notify(const char *channel, const char *payload)
+Async_Notify(const char *channel, const char *payload, const char *collapse_mode)
{
Notification *n;
MemoryContext oldcontext;
+ bool removeDuplicates = true;
if (IsParallelWorker())
elog(ERROR, "cannot send notifications from a parallel worker");
@@ -570,9 +606,23 @@ Async_Notify(const char *channel, const char *payload)
errmsg("payload string too long")));
}
- /* no point in making duplicate entries in the list ... */
- if (AsyncExistsPendingNotify(channel, payload))
- return;
+ if (strlen(collapse_mode) != 0) {
+ if (strcmp(collapse_mode, "never") == 0)
+ {
+ removeDuplicates = false;
+ }
+ else if (strcmp(collapse_mode, "maybe") != 0)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("invalid collapse_mode value '%s'", collapse_mode)));
+ }
+ }
+
+ if (removeDuplicates)
+ /* remove duplicate entries in the list */
+ if (AsyncExistsPendingNotify(channel, payload))
+ return;
/*
* The notification list needs to live until end of transaction, so store
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index ce60e99cff..f0a6056e0d 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -521,7 +521,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
%type opt_varying opt_timezone opt_no_inherit
%type Iconst SignedIconst
-%type Sconst comment_text notify_payload
+%type Sconst comment_text notify_payload notify_collapse_mode
%type RoleId opt_boolean_or_string
%type var_list
%type ColId ColLabel var_name type_function_name param_name
@@ -9784,18 +9784,32 @@ opt_instead:
*
*****************************************************************************/
-NotifyStmt: NOTIFY ColId notify_payload
+NotifyStmt:
+ NOTIFY ColId
+ {
+ NotifyStmt *n = makeNode(NotifyStmt);
+ n->conditionname = $2;
+ n->payload = NULL;
+ n->collapse_mode = "";
+ $$ = (Node *)n;
+ }
+ | NOTIFY ColId notify_payload notify_collapse_mode
{
NotifyStmt *n = makeNode(NotifyStmt);
n->conditionname = $2;
n->payload = $3;
+ n->collapse_mode = $4;
$$ = (Node *)n;
}
;
notify_payload:
',' Sconst { $$ = $2; }
- | /*EMPTY*/ { $$ = NULL; }
+ ;
+
+notify_collapse_mode:
+ ',' Sconst { $$ = $2; }
+ | /*EMPTY*/ { $$ = ""; }
;
ListenStmt: LISTEN ColId
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index b5804f64ad..64eeb6cb8d 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -608,7 +608,7 @@ standard_ProcessUtility(PlannedStmt *pstmt,
NotifyStmt *stmt = (NotifyStmt *) parsetree;
PreventCommandDuringRecovery("NOTIFY");
- Async_Notify(stmt->conditionname, stmt->payload);
+ Async_Notify(stmt->conditionname, stmt->payload, stmt->collapse_mode);
}
break;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 8e4145f42b..27bba69b62 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -7670,6 +7670,11 @@
proname => 'pg_notify', proisstrict => 'f', provolatile => 'v',
proparallel => 'r', prorettype => 'void', proargtypes => 'text text',
prosrc => 'pg_notify' },
+{ oid => '3423', descr => 'send a notification event',
+ proname => 'pg_notify', proisstrict => 'f', provolatile => 'v',
+ proparallel => 'r', prorettype => 'void', proargtypes => 'text text text',
+ prosrc => 'pg_notify_3args' },
+
{ oid => '3296',
descr => 'get the fraction of the asynchronous notification queue currently in use',
proname => 'pg_notification_queue_usage', provolatile => 'v',
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index d5868c42a0..3cd78096bd 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -33,7 +33,7 @@ extern void NotifyMyFrontEnd(const char *channel,
int32 srcPid);
/* notify-related SQL statements */
-extern void Async_Notify(const char *channel, const char *payload);
+extern void Async_Notify(const char *channel, const char *payload, const char *collapse_mode);
extern void Async_Listen(const char *channel);
extern void Async_Unlisten(const char *channel);
extern void Async_UnlistenAll(void);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 200df8e659..464138e8bc 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2942,6 +2942,7 @@ typedef struct NotifyStmt
NodeTag type;
char *conditionname; /* condition name to notify */
char *payload; /* the payload string, or NULL if none */
+ char *collapse_mode; /* the collapse mode (empty string by default, which is equivalent to 'maybe') */
} NotifyStmt;
/* ----------------------
diff --git a/src/test/regress/expected/async.out b/src/test/regress/expected/async.out
index 19cbe38e63..beff62b895 100644
--- a/src/test/regress/expected/async.out
+++ b/src/test/regress/expected/async.out
@@ -8,6 +8,18 @@ SELECT pg_notify('notify_async1','sample message1');
(1 row)
+SELECT pg_notify('notify_async1','sample message1','maybe');
+ pg_notify
+-----------
+
+(1 row)
+
+SELECT pg_notify('notify_async1','sample_message1','never');
+ pg_notify
+-----------
+
+(1 row)
+
SELECT pg_notify('notify_async1','');
pg_notify
-----------
@@ -29,9 +41,14 @@ SELECT pg_notify('notify_async_channel_name_too_long____________________________
ERROR: channel name too long
--Should work. Valid NOTIFY/LISTEN/UNLISTEN commands
NOTIFY notify_async2;
+NOTIFY notify_async2, '', 'maybe';
+NOTIFY notify_async2, '', 'never';
LISTEN notify_async2;
UNLISTEN notify_async2;
UNLISTEN *;
+--Should fail. Invalid collapse mode
+NOTIFY notify_async2, '', 'foobar';
+ERROR: invalid collapse_mode value 'foobar'
-- Should return zero while there are no pending notifications.
-- src/test/isolation/specs/async-notify.spec tests for actual usage.
SELECT pg_notification_queue_usage();
diff --git a/src/test/regress/sql/async.sql b/src/test/regress/sql/async.sql
index 40f6e01538..f95292e3e4 100644
--- a/src/test/regress/sql/async.sql
+++ b/src/test/regress/sql/async.sql
@@ -4,6 +4,8 @@
--Should work. Send a valid message via a valid channel name
SELECT pg_notify('notify_async1','sample message1');
+SELECT pg_notify('notify_async1','sample message1','maybe');
+SELECT pg_notify('notify_async1','sample_message1','never');
SELECT pg_notify('notify_async1','');
SELECT pg_notify('notify_async1',NULL);
@@ -14,10 +16,15 @@ SELECT pg_notify('notify_async_channel_name_too_long____________________________
--Should work. Valid NOTIFY/LISTEN/UNLISTEN commands
NOTIFY notify_async2;
+NOTIFY notify_async2, '', 'maybe';
+NOTIFY notify_async2, '', 'never';
LISTEN notify_async2;
UNLISTEN notify_async2;
UNLISTEN *;
+--Should fail. Invalid collapse mode
+NOTIFY notify_async2, '', 'foobar';
+
-- Should return zero while there are no pending notifications.
-- src/test/isolation/specs/async-notify.spec tests for actual usage.
SELECT pg_notification_queue_usage();