Do Shuffle Hash Joins.

For simplicity, contains many limits. For example, it is utilizes hash partitioning only and sequental scan of relations.
In this version we do not calculate cost of shuffle paths, use fixed values. And so on.

Author: danolivo

contrib/pg_exchange/dmq.c

@@ -62,16 +62,6 @@
 #define DMQ_CONNSTR_MAX_LEN 1024
 
 #define DMQ_MAX_SUBS_PER_BACKEND 100
-#define DMQ_MAX_DESTINATIONS 127
-#define DMQ_MAX_RECEIVERS 100
-
-typedef enum
-{
-	Idle,			/* upon init or falure */
-	Connecting,		/* upon PQconnectStart */
-	Negotiating,	/* upon PQconnectPoll == OK */
-	Active,			/* upon dmq_receiver_loop() response */
-} DmqConnState;
 
 typedef struct {
 	bool			active;
@@ -1676,12 +1666,13 @@ DmqConnState
 dmq_get_destination_status(DmqDestinationId dest_id)
 {
 	DmqConnState state;
+	DmqDestination *dest;
 
 	if ((dest_id < 0) || (dest_id >= DMQ_MAX_DESTINATIONS))
 		return -2;
 
 	LWLockAcquire(dmq_state->lock, LW_EXCLUSIVE);
-	DmqDestination *dest = &(dmq_state->destinations[dest_id]);
+	dest = &(dmq_state->destinations[dest_id]);
 	if (!dest->active)
 	{
 		LWLockRelease(dmq_state->lock);

contrib/pg_exchange/dmq.h

@@ -4,15 +4,26 @@
 #include "libpq-fe.h"
 #include "lib/stringinfo.h"
 
+typedef enum
+{
+	Idle,			/* upon init or falure */
+	Connecting,		/* upon PQconnectStart */
+	Negotiating,	/* upon PQconnectPoll == OK */
+	Active,			/* upon dmq_receiver_loop() response */
+} DmqConnState;
+
 typedef int8 DmqDestinationId;
 typedef int8 DmqSenderId;
 
 #define DMQ_NAME_MAXLEN 32
+#define DMQ_MAX_DESTINATIONS 127
+#define DMQ_MAX_RECEIVERS 100
 
 extern void dmq_init(const char *library_name);
 
 extern DmqDestinationId dmq_destination_add(char *connstr, char *sender_name,
 											char *receiver_name, int ping_period);
+extern DmqConnState dmq_get_destination_status(DmqDestinationId dest_id);
 extern void dmq_destination_drop(char *receiver_name);
 
 extern DmqSenderId dmq_attach_receiver(const char *sender_name, int mask_pos);