Some corrections on optimizations

Author: danolivo

contrib/pg_exchange/Makefile

@@ -6,7 +6,7 @@ EXTVERSION = 0.1
 PGFILEDESC = "pg_exchange - an exchange custom node and rules for the planner"
 MODULES = pg_exchange
 OBJS =	common.o dmq.o exchange.o expath.o hooks.o nodeDistPlanExec.o \
-		nodeDummyscan.o partutils.o pg_exchange.o stream.o $(WIN32RES)
+		nodeDummyscan.o partutils.o pg_exchange.o sbuf.o stream.o $(WIN32RES)
 
 fdw_srcdir = $(top_srcdir)/contrib/postgres_fdw/
 

contrib/pg_exchange/dmq.c

@@ -47,7 +47,10 @@
 #include "utils/dynahash.h"
 #include "utils/ps_status.h"
 
-#define DMQ_MQ_SIZE  ((Size) 65536)
+//#define DMQ_MQ_SIZE  ((Size) 65536)
+#define DMQ_MQ_SIZE  ((Size) 1048576) /* 1 MB */
+//#define DMQ_MQ_SIZE  ((Size) 8388608) /* 8 MB */
+
 #define DMQ_MQ_MAGIC 0x646d71
 
 // XXX: move to common
@@ -222,6 +225,7 @@ dmq_shmem_size(void)
 {
 	Size	size = 0;
 
+//	size = add_size(size, DMQ_MQ_SIZE * DMQ_MAX_DESTINATIONS * 2);
 	size = add_size(size, sizeof(struct DmqSharedState));
 	size = add_size(size, hash_estimate_size(DMQ_MAX_SUBS_PER_BACKEND*MaxBackends,
 											 sizeof(DmqStreamSubscription)));

contrib/pg_exchange/stream.c

@@ -3,12 +3,13 @@
  *
  */
 
+#include "sbuf.h"
 #include "stream.h"
 #include "miscadmin.h"
 #include "unistd.h"
 #include "utils/memutils.h" /* MemoryContexts */
 
-#define IsDeliveryMessage(msg)	(msg->tot_len == MinSizeOfSendBuf)
+#define IsDeliveryMessage(msg)	(msg->datalen == 0)
 
 static List *istreams = NIL;
 static List *ostreams = NIL;
@@ -41,6 +42,8 @@ get_stream(List *streams, const char *name)
 	for (lc = list_head(streams); lc != NULL; lc = lnext(lc))
 	{
 		char *streamName = (char *) lfirst(lc);
+
+		/* streamName is a first field of IStream and OStream structures. */
 		if (strcmp(name, streamName) == 0)
 			return streamName;
 	}
@@ -153,12 +156,13 @@ RecvIfAny(void)
 			DmqDestinationId dest_id;
 
 			/* If message is not delivery message, send delivery. */
-			dbuf.tot_len = MinSizeOfSendBuf;
+			dbuf.datalen = 0;
 			dbuf.index = msg->index;
 			dest_id = dmq_dest_id(sender_id);
 			Assert(dest_id >= 0);
 
-			dmq_push_buffer(dest_id, istream->streamName, &dbuf, dbuf.tot_len, false);
+			dmq_push_buffer(dest_id, istream->streamName, &dbuf,
+													MinSizeOfSendBuf, false);
 		}
 	}
 }
@@ -193,17 +197,18 @@ checkDelivery(OStream *ostream)
 	{
 		RecvBuf *buf = lfirst(lc);
 
-		istream->msgs = list_delete_ptr(istream->msgs, buf);
+		istream->deliveries = list_delete_ptr(istream->deliveries, buf);
 		pfree(buf);
 	}
+	list_free(temp);
 	return found;
 }
 
 static void
 StreamRepeatSend(OStream *ostream)
 {
 	while (!dmq_push_buffer(ostream->dest_id, ostream->streamName, ostream->buf,
-			ostream->buf->tot_len, true))
+												buf_len(ostream->buf), true))
 			RecvIfAny();
 }
 
@@ -215,36 +220,31 @@ ISendTuple(DmqDestinationId dest_id, char *stream, TupleTableSlot *slot,
 	int tupsize;
 	SendBuf *buf;
 	OStream	*ostream;
+	int tot_len;
 
 	RecvIfAny();
 
 	ostream = (OStream *) get_stream(ostreams, stream);
 	Assert(ostream && !ostream->buf);
 
-	if (!TupIsNull(slot))
-	{
-		int tot_len;
+	Assert(!TupIsNull(slot));
 
-		if (slot->tts_tuple == NULL)
-			ExecMaterializeSlot(slot);
+	if (slot->tts_tuple == NULL)
+		ExecMaterializeSlot(slot);
 
-		tuple = slot->tts_tuple;
-		tupsize = offsetof(HeapTupleData, t_data);
-
-		tot_len = MinSizeOfSendBuf + tupsize + tuple->t_len;
-		buf = palloc(tot_len);
-		buf->tot_len = tot_len;
-		memcpy(buf->data, tuple, tupsize);
-		memcpy(buf->data + tupsize, tuple->t_data, tuple->t_len);
-	}
-	else
-		Assert(0);
+	tuple = slot->tts_tuple;
+	tupsize = offsetof(HeapTupleData, t_data);
+	tot_len = MinSizeOfSendBuf + tupsize + tuple->t_len;
+	buf = palloc(tot_len);
+	buf->datalen = tot_len - MinSizeOfSendBuf;
+	memcpy(buf->data, tuple, tupsize);
+	memcpy(buf->data + tupsize, tuple->t_data, tuple->t_len);
 
 	buf->index = ++(ostream->index);
 	buf->needConfirm = needConfirm;
 	ostream->dest_id = dest_id;
 
-	while (!dmq_push_buffer(dest_id, stream, buf, buf->tot_len, true))
+	while (!dmq_push_buffer(dest_id, stream, buf, buf_len(buf), true))
 		RecvIfAny();
 
 	if (buf->needConfirm)
@@ -284,15 +284,15 @@ SendByteMessage(DmqDestinationId dest_id, char *stream, char tag)
 	Assert(ostream && !ostream->buf);
 
 	buf = palloc(MinSizeOfSendBuf + 1);
-	buf->tot_len = MinSizeOfSendBuf + 1;
+	buf->datalen = 1;
 	buf->data[0] = tag;
 	buf->index = ++(ostream->index);
 	buf->needConfirm = true;
 
 	ostream->buf = buf;
 	ostream->dest_id = dest_id;
 
-	while (!dmq_push_buffer(dest_id, stream, buf, buf->tot_len, true))
+	while (!dmq_push_buffer(dest_id, stream, buf, buf_len(buf), true))
 		RecvIfAny();
 
 	wait_for_delivery(ostream);
@@ -335,7 +335,7 @@ RecvByteMessage(const char *streamName, const char *sender)
  */
 void
 SendTuple(DmqDestinationId dest_id, char *stream, TupleTableSlot *slot,
-		bool needConfirm)
+		  bool needConfirm)
 {
 	OStream *ostream;
 

contrib/pg_exchange/stream.h

@@ -21,7 +21,7 @@
 typedef struct SendBuf
 {
 	uint32 index;
-	uint32 tot_len;
+	uint32 datalen;
 	bool needConfirm;
 	char data[FLEXIBLE_ARRAY_MEMBER];
 } SendBuf;
@@ -35,12 +35,13 @@ typedef struct RecvBuf
 } RecvBuf;
 
 #define MinSizeOfSendBuf offsetof(SendBuf, data)
+#define buf_len(buf) (MinSizeOfSendBuf + buf->datalen)
 #define MinSizeOfRecvBuf offsetof(RecvBuf, data)
 
 typedef struct
 {
 	char streamName[STREAM_NAME_MAX_LEN];
-	uint64 index;
+	uint32 index;
 	SendBuf *buf;
 	DmqDestinationId dest_id;
 } OStream;

src/backend/optimizer/path/joinpath.c

@@ -68,6 +68,9 @@ static void consider_parallel_mergejoin(PlannerInfo *root,
 							JoinType jointype,
 							JoinPathExtraData *extra,
 							Path *inner_cheapest_total);
+static void hash_inner_and_outer(PlannerInfo *root, RelOptInfo *joinrel,
+					 RelOptInfo *outerrel, RelOptInfo *innerrel,
+					 JoinType jointype, JoinPathExtraData *extra);
 static List *select_mergejoin_clauses(PlannerInfo *root,
 						 RelOptInfo *joinrel,
 						 RelOptInfo *outerrel,
@@ -1670,7 +1673,7 @@ consider_parallel_nestloop(PlannerInfo *root,
  * 'jointype' is the type of join to do
  * 'extra' contains additional input values
  */
-void
+static void
 hash_inner_and_outer(PlannerInfo *root,
 					 RelOptInfo *joinrel,
 					 RelOptInfo *outerrel,

src/backend/optimizer/plan/planner.c

@@ -1182,7 +1182,7 @@ inheritance_planner(PlannerInfo *root)
 	PlannerInfo **parent_roots = NULL;
 
 	Assert(parse->commandType != CMD_INSERT);
-elog(LOG, "inheritance_plnner()");
+
 	/*
 	 * We generate a modified instance of the original Query for each target
 	 * relation, plan that, and put all the plans into a list that will be

src/backend/storage/ipc/dsm_impl.c

@@ -341,7 +341,7 @@ dsm_impl_posix(dsm_op op, dsm_handle handle, Size request_size,
 		 */
 		if (errno == EINTR && elevel >= ERROR)
 			CHECK_FOR_INTERRUPTS();
-
+Assert(0);
 		ereport(elevel,
 				(errcode_for_dynamic_shared_memory(),
 				 errmsg("could not resize shared memory segment \"%s\" to %zu bytes: %m",

src/backend/storage/ipc/latch.c

@@ -43,7 +43,6 @@
 #include <poll.h>
 #endif
 
-//#include "common/pg_socket.h"
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "port/atomics.h"

src/backend/tcop/postgres.c

@@ -2738,7 +2738,6 @@ StatementCancelHandler(SIGNAL_ARGS)
 void
 FloatExceptionHandler(SIGNAL_ARGS)
 {
-	Assert(0);
 	/* We're not returning, so no need to save errno */
 	ereport(ERROR,
 			(errcode(ERRCODE_FLOATING_POINT_EXCEPTION),
@@ -3190,7 +3189,6 @@ check_stack_depth(void)
 {
 	if (stack_is_too_deep())
 	{
-		Assert(0);
 		ereport(ERROR,
 				(errcode(ERRCODE_STATEMENT_TOO_COMPLEX),
 				 errmsg("stack depth limit exceeded"),

src/backend/utils/cache/typcache.c

@@ -1621,7 +1621,6 @@ lookup_rowtype_tupdesc_internal(Oid type_id, int32 typmod, bool noError)
 			ereport(ERROR,
 					(errcode(ERRCODE_WRONG_OBJECT_TYPE),
 					 errmsg("record type has not been registered")));
-
 		return NULL;
 	}
 }