On Mon, Feb 8, 2016 at 1:52 PM, Craig Ringer <cr...@2ndquadrant.com> wrote:
> Would it be correct to say that if ALL is specified then a message is queued
> no matter what. If DISTINCT is specified then it is only queued if no
> message with the same channel and argument is already queued for delivery.
Yes, exactly.
> Using DISTINCT can never decrease the total number of messages to be sent.
This sentence does not sound true. DISTINCT is the default, old
behaviour. It *can* decrease total number of messages (by
deduplication)
> I've found the deduplication functionality of NOTIFY very frustrating in the
> past
> and I see this as a significant improvement. Sometimes the *number of times*
> something happened is significant too...
yep, same idea here.
Here is my next try, after suggestions from -perf and -hackers list:
* no GUC
* small addition to NOTIFY grammar: NOTIFY ALL/DISTINCT
* corresponding, 3-argument version of pg_notify(text,text,bool)
* updated the docs to include new syntax and clarify behavior
* no hashtable in AsyncExistsPendingNotify
(I don't see much sense in that part; and it can be well done
separately from this)
diff --git a/doc/src/sgml/ref/notify.sgml b/doc/src/sgml/ref/notify.sgml
index 4dd5608..933c76c 100644
--- a/doc/src/sgml/ref/notify.sgml
+++ b/doc/src/sgml/ref/notify.sgml
@@ -21,7 +21,7 @@ PostgreSQL documentation
<refsynopsisdiv>
<synopsis>
-NOTIFY <replaceable class="PARAMETER">channel</replaceable> [ , <replaceable class="PARAMETER">payload</replaceable> ]
+NOTIFY [ ALL | DISTINCT ] <replaceable class="PARAMETER">channel</replaceable> [ , <replaceable class="PARAMETER">payload</replaceable> ]
</synopsis>
</refsynopsisdiv>
@@ -105,6 +105,10 @@ NOTIFY <replaceable class="PARAMETER">channel</replaceable> [ , <replaceable cla
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.
+ If <literal>ALL</> is specified (contrary to <literal>DISTINCT</>, the
+ default), the server will deliver all notifications, including duplicates.
+ Removal of duplicate notifications takes place within transaction block,
+ finished with <literal>COMMIT</>, <literal>END</> or <literal>SAVEPOINT</>.
</para>
<para>
@@ -190,6 +194,12 @@ NOTIFY <replaceable class="PARAMETER">channel</replaceable> [ , <replaceable cla
to use than the <command>NOTIFY</command> command if you need to work with
non-constant channel names and payloads.
</para>
+ <para>
+ There is a three-argument version, <literal>pg_notify(<type>text</>,
+ <type>text</>, <type>boolean</>)</literal>. The third argument acts like
+ the <literal>ALL</> keyword when set to <literal>true</>, and
+ <literal>DISTINCT</> when set to <literal>false</>.
+ </para>
</refsect2>
</refsect1>
@@ -210,6 +220,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 ALL keyword */
+LISTEN bar;
+BEGIN;
+NOTIFY bar, 'Coffee please';
+NOTIFY bar, 'Coffee please';
+NOTIFY bar, 'Milk please';
+NOTIFY ALL bar, 'Milk please';
+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.
</programlisting></para>
</refsect1>
diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index c39ac3a..38a8246 100644
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -510,6 +510,7 @@ pg_notify(PG_FUNCTION_ARGS)
{
const char *channel;
const char *payload;
+ bool use_all;
if (PG_ARGISNULL(0))
channel = "";
@@ -521,10 +522,15 @@ pg_notify(PG_FUNCTION_ARGS)
else
payload = text_to_cstring(PG_GETARG_TEXT_PP(1));
+ if (PG_NARGS() > 2 && ! PG_ARGISNULL(2))
+ use_all = PG_GETARG_BOOL(2);
+ else
+ use_all = false;
+
/* For NOTIFY as a statement, this is checked in ProcessUtility */
PreventCommandDuringRecovery("NOTIFY");
- Async_Notify(channel, payload);
+ Async_Notify(channel, payload, use_all);
PG_RETURN_VOID();
}
@@ -540,7 +546,7 @@ pg_notify(PG_FUNCTION_ARGS)
* ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
*/
void
-Async_Notify(const char *channel, const char *payload)
+Async_Notify(const char *channel, const char *payload, bool use_all)
{
Notification *n;
MemoryContext oldcontext;
@@ -570,9 +576,10 @@ 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 (!use_all)
+ /* 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 b307b48..7203f4a 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -8528,11 +8528,12 @@ DropRuleStmt:
*
*****************************************************************************/
-NotifyStmt: NOTIFY ColId notify_payload
+NotifyStmt: NOTIFY all_or_distinct ColId notify_payload
{
NotifyStmt *n = makeNode(NotifyStmt);
- n->conditionname = $2;
- n->payload = $3;
+ n->use_all = $2;
+ n->conditionname = $3;
+ n->payload = $4;
$$ = (Node *)n;
}
;
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 045f7f0..0e50561 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -599,7 +599,7 @@ standard_ProcessUtility(Node *parsetree,
NotifyStmt *stmt = (NotifyStmt *) parsetree;
PreventCommandDuringRecovery("NOTIFY");
- Async_Notify(stmt->conditionname, stmt->payload);
+ Async_Notify(stmt->conditionname, stmt->payload, stmt->use_all);
}
break;
diff --git a/src/include/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index 1c0ef9a..53ed138 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -4005,6 +4005,8 @@ DESCR("trigger description with pretty-print option");
/* asynchronous notifications */
DATA(insert OID = 3035 ( pg_listening_channels PGNSP PGUID 12 1 10 0 0 f f f f t t s r 0 0 25 "" _null_ _null_ _null_ _null_ _null_ pg_listening_channels _null_ _null_ _null_ ));
DESCR("get the channels that the current backend listens to");
+DATA(insert OID = 2561 ( pg_notify PGNSP PGUID 12 1 0 0 0 f f f f f f v r 3 0 2278 "25 25 16" _null_ _null_ _null_ _null_ _null_ pg_notify _null_ _null_ _null_ ));
+DESCR("send a notification event");
DATA(insert OID = 3036 ( pg_notify PGNSP PGUID 12 1 0 0 0 f f f f f f v r 2 0 2278 "25 25" _null_ _null_ _null_ _null_ _null_ pg_notify _null_ _null_ _null_ ));
DESCR("send a notification event");
DATA(insert OID = 3296 ( pg_notification_queue_usage PGNSP PGUID 12 1 0 0 0 f f f f t f v s 0 0 701 "" _null_ _null_ _null_ _null_ _null_ pg_notification_queue_usage _null_ _null_ _null_ ));
diff --git a/src/include/commands/async.h b/src/include/commands/async.h
index b4c13fa..3a2c1c2 100644
--- a/src/include/commands/async.h
+++ b/src/include/commands/async.h
@@ -29,7 +29,7 @@ extern Size AsyncShmemSize(void);
extern void AsyncShmemInit(void);
/* notify-related SQL statements */
-extern void Async_Notify(const char *channel, const char *payload);
+extern void Async_Notify(const char *channel, const char *payload, bool use_all);
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 2fd0629..7f1f01d 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -2590,6 +2590,7 @@ typedef struct NotifyStmt
NodeTag type;
char *conditionname; /* condition name to notify */
char *payload; /* the payload string, or NULL if none */
+ bool use_all; /* ALL option */
} NotifyStmt;
/* ----------------------
diff --git a/src/test/regress/expected/async.out b/src/test/regress/expected/async.out
index 19cbe38..c650b90 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',false);
+ pg_notify
+-----------
+
+(1 row)
+
+SELECT pg_notify('notify_async1','sample_message1',true);
+ pg_notify
+-----------
+
+(1 row)
+
SELECT pg_notify('notify_async1','');
pg_notify
-----------
@@ -29,6 +41,8 @@ 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 DISTINCT notify_async2;
+NOTIFY ALL notify_async2;
LISTEN notify_async2;
UNLISTEN notify_async2;
UNLISTEN *;
diff --git a/src/test/regress/sql/async.sql b/src/test/regress/sql/async.sql
index 40f6e01..6e53b86 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',false);
+SELECT pg_notify('notify_async1','sample_message1',true);
SELECT pg_notify('notify_async1','');
SELECT pg_notify('notify_async1',NULL);
@@ -14,6 +16,8 @@ SELECT pg_notify('notify_async_channel_name_too_long____________________________
--Should work. Valid NOTIFY/LISTEN/UNLISTEN commands
NOTIFY notify_async2;
+NOTIFY DISTINCT notify_async2;
+NOTIFY ALL notify_async2;
LISTEN notify_async2;
UNLISTEN notify_async2;
UNLISTEN *;
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
