0011-Introduce-wal-decoding-via-catalog-timetravel.patch

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index d025ff7..7765cae 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -53,6 +53,7 @@
 #include "access/xact.h"
 #include "access/xlogutils.h"
 #include "catalog/catalog.h"
+#include "catalog/index.h"
 #include "catalog/namespace.h"
 #include "miscadmin.h"
 #include "pgstat.h"
@@ -86,6 +87,7 @@ static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup,
 static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf,
 				ItemPointerData from, Buffer newbuf, HeapTuple newtup,
 				bool all_visible_cleared, bool new_all_visible_cleared);
+static XLogRecPtr log_heap_new_cid(Relation relation, HeapTuple tup);
 static bool HeapSatisfiesHOTUpdate(Relation relation, Bitmapset *hot_attrs,
 					   HeapTuple oldtup, HeapTuple newtup);
 
@@ -1618,10 +1620,16 @@ heap_hot_search_buffer(ItemPointer tid, Relation relation, Buffer buffer,
 		 */
 		if (!skip)
 		{
+			/* setup the redirected t_self for the benefit of timetravel access */
+			ItemPointerSet(&(heapTuple->t_self), BufferGetBlockNumber(buffer), offnum);
+
 			/* If it's visible per the snapshot, we must return it */
 			valid = HeapTupleSatisfiesVisibility(heapTuple, snapshot, buffer);
 			CheckForSerializableConflictOut(valid, relation, heapTuple,
 											buffer, snapshot);
+			/* reset original, non-redirected, tid */
+			heapTuple->t_self = *tid;
+
 			if (valid)
 			{
 				ItemPointerSetOffsetNumber(tid, offnum);
@@ -1960,10 +1968,24 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 		xl_heap_insert xlrec;
 		xl_heap_header xlhdr;
 		XLogRecPtr	recptr;
-		XLogRecData rdata[3];
+		XLogRecData rdata[4];
 		Page		page = BufferGetPage(buffer);
 		uint8		info = XLOG_HEAP_INSERT;
 
+		/*
+		 * For the logical replication case we need the tuple even if were
+		 * doing a full page write. We could alternatively store a pointer into
+		 * the fpw though.
+		 * For that to work we add another rdata entry for the buffer in that
+		 * case.
+		 */
+		bool        need_tuple_data = wal_level >= WAL_LEVEL_LOGICAL
+			&& RelationGetRelid(relation)  >= FirstNormalObjectId;
+
+		/* For logical decode we need combocids to properly decode the catalog */
+		if (wal_level >= WAL_LEVEL_LOGICAL && RelationGetRelid(relation)  < FirstNormalObjectId)
+			log_heap_new_cid(relation, heaptup);
+
 		xlrec.all_visible_cleared = all_visible_cleared;
 		xlrec.target.node = relation->rd_node;
 		xlrec.target.tid = heaptup->t_self;
@@ -1983,18 +2005,33 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 		 */
 		rdata[1].data = (char *) &xlhdr;
 		rdata[1].len = SizeOfHeapHeader;
-		rdata[1].buffer = buffer;
+		rdata[1].buffer = need_tuple_data ? InvalidBuffer : buffer;
 		rdata[1].buffer_std = true;
 		rdata[1].next = &(rdata[2]);
 
 		/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
 		rdata[2].data = (char *) heaptup->t_data + offsetof(HeapTupleHeaderData, t_bits);
 		rdata[2].len = heaptup->t_len - offsetof(HeapTupleHeaderData, t_bits);
-		rdata[2].buffer = buffer;
+		rdata[2].buffer = need_tuple_data ? InvalidBuffer : buffer;
 		rdata[2].buffer_std = true;
 		rdata[2].next = NULL;
 
 		/*
+		 * add record for the buffer without actual content thats removed if
+		 * fpw is done for that buffer
+		 */
+		if (need_tuple_data)
+		{
+			rdata[2].next = &(rdata[3]);
+
+			rdata[3].data = NULL;
+			rdata[3].len = 0;
+			rdata[3].buffer = buffer;
+			rdata[3].buffer_std = true;
+			rdata[3].next = NULL;
+		}
+
+		/*
 		 * If this is the single and first tuple on page, we can reinit the
 		 * page instead of restoring the whole thing.  Set flag, and hide
 		 * buffer references from XLogInsert.
@@ -2003,7 +2040,7 @@ heap_insert(Relation relation, HeapTuple tup, CommandId cid,
 			PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
 		{
 			info |= XLOG_HEAP_INIT_PAGE;
-			rdata[1].buffer = rdata[2].buffer = InvalidBuffer;
+			rdata[1].buffer = rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
 		}
 
 		recptr = XLogInsert(RM_HEAP_ID, info, rdata);
@@ -2123,6 +2160,10 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 	Page		page;
 	bool		needwal;
 	Size		saveFreeSpace;
+	bool        need_tuple_data = wal_level >= WAL_LEVEL_LOGICAL
+		&& RelationGetRelid(relation)  >= FirstNormalObjectId;
+	bool        need_cids = wal_level >= WAL_LEVEL_LOGICAL &&
+		RelationGetRelid(relation)  < FirstNormalObjectId;
 
 	needwal = !(options & HEAP_INSERT_SKIP_WAL) && RelationNeedsWAL(relation);
 	saveFreeSpace = RelationGetTargetPageFreeSpace(relation,
@@ -2205,7 +2246,7 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 		{
 			XLogRecPtr	recptr;
 			xl_heap_multi_insert *xlrec;
-			XLogRecData rdata[2];
+			XLogRecData rdata[3];
 			uint8		info = XLOG_HEAP2_MULTI_INSERT;
 			char	   *tupledata;
 			int			totaldatalen;
@@ -2267,6 +2308,15 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 					   datalen);
 				tuphdr->datalen = datalen;
 				scratchptr += datalen;
+
+				/*
+				 * We don't use heap_multi_insert for catalog tuples yet, but
+				 * better be prepared...
+				 */
+				if (need_cids)
+				{
+					log_heap_new_cid(relation, heaptup);
+				}
 			}
 			totaldatalen = scratchptr - tupledata;
 			Assert((scratchptr - scratch) < BLCKSZ);
@@ -2278,17 +2328,32 @@ heap_multi_insert(Relation relation, HeapTuple *tuples, int ntuples,
 
 			rdata[1].data = tupledata;
 			rdata[1].len = totaldatalen;
-			rdata[1].buffer = buffer;
+			rdata[1].buffer = need_tuple_data ? InvalidBuffer : buffer;
 			rdata[1].buffer_std = true;
 			rdata[1].next = NULL;
 
 			/*
+			 * add record for the buffer without actual content thats removed if
+			 * fpw is done for that buffer
+			 */
+			if (need_tuple_data)
+			{
+				rdata[1].next = &(rdata[2]);
+
+				rdata[2].data = NULL;
+				rdata[2].len = 0;
+				rdata[2].buffer = buffer;
+				rdata[2].buffer_std = true;
+				rdata[2].next = NULL;
+			}
+
+			/*
 			 * If we're going to reinitialize the whole page using the WAL
 			 * record, hide buffer reference from XLogInsert.
 			 */
 			if (init)
 			{
-				rdata[1].buffer = InvalidBuffer;
+				rdata[1].buffer = rdata[2].buffer = InvalidBuffer;
 				info |= XLOG_HEAP_INIT_PAGE;
 			}
 
@@ -2595,7 +2660,14 @@ l1:
 	{
 		xl_heap_delete xlrec;
 		XLogRecPtr	recptr;
-		XLogRecData rdata[2];
+		XLogRecData rdata[4];
+
+		bool need_tuple_data = wal_level >= WAL_LEVEL_LOGICAL &&
+			RelationGetRelid(relation) >= FirstNormalObjectId;
+
+		/* For logical decode we need combocids to properly decode the catalog */
+		if (wal_level >= WAL_LEVEL_LOGICAL && RelationGetRelid(relation)  < FirstNormalObjectId)
+			log_heap_new_cid(relation, &tp);
 
 		xlrec.all_visible_cleared = all_visible_cleared;
 		xlrec.target.node = relation->rd_node;
@@ -2611,6 +2683,76 @@ l1:
 		rdata[1].buffer_std = true;
 		rdata[1].next = NULL;
 
+		/*
+		 * XXX: We could decide not to log changes when the origin is not the
+		 * local node, that should reduce redundant logging.
+		 */
+		if (need_tuple_data)
+		{
+			xl_heap_header xlhdr;
+
+			Oid indexoid = InvalidOid;
+			int16 pknratts;
+			int16 pkattnum[INDEX_MAX_KEYS];
+			Oid pktypoid[INDEX_MAX_KEYS];
+			Oid pkopclass[INDEX_MAX_KEYS];
+			TupleDesc desc = RelationGetDescr(relation);
+			Relation index_rel;
+			TupleDesc indexdesc;
+			int natt;
+
+			Datum idxvals[INDEX_MAX_KEYS];
+			bool idxisnull[INDEX_MAX_KEYS];
+			HeapTuple idxtuple;
+
+			MemSet(pkattnum, 0, sizeof(pkattnum));
+			MemSet(pktypoid, 0, sizeof(pktypoid));
+			MemSet(pkopclass, 0, sizeof(pkopclass));
+			MemSet(idxvals, 0, sizeof(idxvals));
+			MemSet(idxisnull, 0, sizeof(idxisnull));
+			relationFindPrimaryKey(relation, &indexoid, &pknratts, pkattnum, pktypoid, pkopclass);
+
+			if (!indexoid)
+			{
+				elog(WARNING, "Could not find primary key for table with oid %u",
+					 RelationGetRelid(relation));
+				goto no_index_found;
+			}
+
+			index_rel = index_open(indexoid, AccessShareLock);
+
+			indexdesc = RelationGetDescr(index_rel);
+
+			for (natt = 0; natt < indexdesc->natts; natt++)
+			{
+				idxvals[natt] =
+					fastgetattr(&tp, pkattnum[natt], desc, &idxisnull[natt]);
+				Assert(!idxisnull[natt]);
+			}
+
+			idxtuple = heap_form_tuple(indexdesc, idxvals, idxisnull);
+
+			xlhdr.t_infomask2 = idxtuple->t_data->t_infomask2;
+			xlhdr.t_infomask = idxtuple->t_data->t_infomask;
+			xlhdr.t_hoff = idxtuple->t_data->t_hoff;
+
+			rdata[1].next = &(rdata[2]);
+			rdata[2].data = (char*)&xlhdr;
+			rdata[2].len = SizeOfHeapHeader;
+			rdata[2].buffer = InvalidBuffer;
+			rdata[2].next = NULL;
+
+			rdata[2].next = &(rdata[3]);
+			rdata[3].data = (char *) idxtuple->t_data + offsetof(HeapTupleHeaderData, t_bits);
+			rdata[3].len = idxtuple->t_len - offsetof(HeapTupleHeaderData, t_bits);
+			rdata[3].buffer = InvalidBuffer;
+			rdata[3].next = NULL;
+
+			heap_close(index_rel, NoLock);
+		no_index_found:
+			;
+		}
+
 		recptr = XLogInsert(RM_HEAP_ID, XLOG_HEAP_DELETE, rdata);
 
 		PageSetLSN(page, recptr);
@@ -3203,10 +3345,20 @@ l2:
 	/* XLOG stuff */
 	if (RelationNeedsWAL(relation))
 	{
-		XLogRecPtr	recptr = log_heap_update(relation, buffer, oldtup.t_self,
-											 newbuf, heaptup,
-											 all_visible_cleared,
-											 all_visible_cleared_new);
+		XLogRecPtr	recptr;
+
+		/* For logical decode we need combocids to properly decode the catalog */
+		if (wal_level >= WAL_LEVEL_LOGICAL &&
+			RelationGetRelid(relation)  < FirstNormalObjectId)
+		{
+			log_heap_new_cid(relation, &oldtup);
+			log_heap_new_cid(relation, heaptup);
+		}
+
+		recptr = log_heap_update(relation, buffer, oldtup.t_self,
+								 newbuf, heaptup,
+								 all_visible_cleared,
+								 all_visible_cleared_new);
 
 		if (newbuf != buffer)
 		{
@@ -4445,9 +4597,15 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
 	xl_heap_header xlhdr;
 	uint8		info;
 	XLogRecPtr	recptr;
-	XLogRecData rdata[4];
+	XLogRecData rdata[5];
 	Page		page = BufferGetPage(newbuf);
 
+	/*
+	 * Just as for XLOG_HEAP_INSERT we need to make sure the tuple
+	 */
+	bool        need_tuple_data = wal_level >= WAL_LEVEL_LOGICAL
+		&& RelationGetRelid(reln) >= FirstNormalObjectId;
+
 	/* Caller should not call me on a non-WAL-logged relation */
 	Assert(RelationNeedsWAL(reln));
 
@@ -4478,28 +4636,44 @@ log_heap_update(Relation reln, Buffer oldbuf, ItemPointerData from,
 	xlhdr.t_hoff = newtup->t_data->t_hoff;
 
 	/*
-	 * As with insert records, we need not store the rdata[2] segment if we
-	 * decide to store the whole buffer instead.
+	 * As with insert's logging , we need not store the the Datum containing
+	 * tuples separately from the buffer if we do logical replication that
+	 * is...
 	 */
 	rdata[2].data = (char *) &xlhdr;
 	rdata[2].len = SizeOfHeapHeader;
-	rdata[2].buffer = newbuf;
+	rdata[2].buffer = need_tuple_data ? InvalidBuffer : newbuf;
 	rdata[2].buffer_std = true;
 	rdata[2].next = &(rdata[3]);
 
 	/* PG73FORMAT: write bitmap [+ padding] [+ oid] + data */
 	rdata[3].data = (char *) newtup->t_data + offsetof(HeapTupleHeaderData, t_bits);
 	rdata[3].len = newtup->t_len - offsetof(HeapTupleHeaderData, t_bits);
-	rdata[3].buffer = newbuf;
+	rdata[3].buffer = need_tuple_data ? InvalidBuffer : newbuf;
 	rdata[3].buffer_std = true;
 	rdata[3].next = NULL;
 
+	/*
+	 * separate storage for the buffer reference of the new page in the
+	 * wal_level >= logical case
+	*/
+	if(need_tuple_data)
+	{
+		rdata[3].next = &(rdata[4]);
+
+		rdata[4].data = NULL,
+		rdata[4].len = 0;
+		rdata[4].buffer = newbuf;
+		rdata[4].buffer_std = true;
+		rdata[4].next = NULL;
+	}
+
 	/* If new tuple is the single and first tuple on page... */
 	if (ItemPointerGetOffsetNumber(&(newtup->t_self)) == FirstOffsetNumber &&
 		PageGetMaxOffsetNumber(page) == FirstOffsetNumber)
 	{
 		info |= XLOG_HEAP_INIT_PAGE;
-		rdata[2].buffer = rdata[3].buffer = InvalidBuffer;
+		rdata[2].buffer = rdata[3].buffer = rdata[4].buffer = InvalidBuffer;
 	}
 
 	recptr = XLogInsert(RM_HEAP_ID, info, rdata);
@@ -4608,6 +4782,64 @@ log_newpage_buffer(Buffer buffer)
 }
 
 /*
+ * Perform XLogInsert of a XLOG_HEAP2_NEW_CID record
+ *
+ * The HeapTuple really needs to already have a ComboCid set otherwise we
+ * cannot detect combocid/cmin/cmax.
+ *
+ * This is only used in wal_level >= WAL_LEVEL_LOGICAL
+ */
+static XLogRecPtr
+log_heap_new_cid(Relation relation, HeapTuple tup)
+{
+	xl_heap_new_cid xlrec;
+
+	XLogRecPtr	recptr;
+	XLogRecData rdata[1];
+	HeapTupleHeader hdr = tup->t_data;
+
+	Assert(ItemPointerIsValid(&tup->t_self));
+	Assert(tup->t_tableOid != InvalidOid);
+
+	xlrec.top_xid = GetTopTransactionId();
+	xlrec.target.node = relation->rd_node;
+	xlrec.target.tid = tup->t_self;
+
+	if (hdr->t_infomask & HEAP_COMBOCID)
+	{
+		xlrec.cmin = HeapTupleHeaderGetCmin(hdr);
+		xlrec.cmax = HeapTupleHeaderGetCmax(hdr);
+		xlrec.combocid = HeapTupleHeaderGetRawCommandId(hdr);
+	}
+	else
+	{
+		/* tuple inserted */
+		if (hdr->t_infomask & HEAP_XMAX_INVALID)
+		{
+			xlrec.cmin = HeapTupleHeaderGetRawCommandId(hdr);
+			xlrec.cmax = InvalidCommandId;
+		}
+		/* tuple from a different tx updated or deleted */
+		else
+		{
+			xlrec.cmin = InvalidCommandId;
+			xlrec.cmax = HeapTupleHeaderGetRawCommandId(hdr);
+
+		}
+		xlrec.combocid = InvalidCommandId;
+	}
+
+	rdata[0].data = (char *) &xlrec;
+	rdata[0].len = SizeOfHeapNewCid;
+	rdata[0].buffer = InvalidBuffer;
+	rdata[0].next = NULL;
+
+	recptr = XLogInsert(RM_HEAP2_ID, XLOG_HEAP2_NEW_CID, rdata);
+
+	return recptr;
+}
+
+/*
  * Handles CLEANUP_INFO
  */
 static void
@@ -5676,6 +5908,9 @@ heap2_redo(XLogRecPtr lsn, XLogRecord *record)
 		case XLOG_HEAP2_MULTI_INSERT:
 			heap_xlog_multi_insert(lsn, record);
 			break;
+		case XLOG_HEAP2_NEW_CID:
+			/* nothing to do on a real replay, only during logical decoding */
+			break;
 		default:
 			elog(PANIC, "heap2_redo: unknown op code %u", info);
 	}
@@ -5825,6 +6060,15 @@ heap2_desc(StringInfo buf, uint8 xl_info, char *rec)
 				xlrec->node.spcNode, xlrec->node.dbNode, xlrec->node.relNode,
 						 xlrec->blkno, xlrec->ntuples);
 	}
+	else if (info == XLOG_HEAP2_NEW_CID)
+	{
+		xl_heap_new_cid *xlrec = (xl_heap_new_cid *) rec;
+
+		appendStringInfo(buf, "new_cid: ");
+		out_target(buf, &(xlrec->target));
+		appendStringInfo(buf, "; cmin: %u, cmax: %u, combo: %u",
+						 xlrec->cmin, xlrec->cmax, xlrec->combocid);
+	}
 	else
 		appendStringInfo(buf, "UNKNOWN");
 }
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 1749f46..e6fb04e 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -106,6 +106,7 @@ const struct config_enum_entry wal_level_options[] = {
 	{"minimal", WAL_LEVEL_MINIMAL, false},
 	{"archive", WAL_LEVEL_ARCHIVE, false},
 	{"hot_standby", WAL_LEVEL_HOT_STANDBY, false},
+	{"logical", WAL_LEVEL_LOGICAL, false},
 	{NULL, 0, false}
 };
 
diff --git a/src/backend/catalog/index.c b/src/backend/catalog/index.c
index 18d0c5a..1d86b87 100644
--- a/src/backend/catalog/index.c
+++ b/src/backend/catalog/index.c
@@ -50,6 +50,7 @@
 #include "nodes/nodeFuncs.h"
 #include "optimizer/clauses.h"
 #include "parser/parser.h"
+#include "parser/parse_relation.h"
 #include "storage/bufmgr.h"
 #include "storage/lmgr.h"
 #include "storage/predicate.h"
@@ -3426,3 +3427,76 @@ ResetReindexPending(void)
 {
 	pendingReindexedIndexes = NIL;
 }
+
+/*
+ * relationFindPrimaryKey
+ *		Find primary key for a relation if it exists.
+ *
+ * If no primary key is found *indexOid is set to InvalidOid
+ *
+ * This is quite similar to tablecmd.c's transformFkeyGetPrimaryKey.
+ *
+ * XXX: It might be a good idea to change pg_class.relhaspkey into an bool to
+ * make this more efficient.
+ */
+void
+relationFindPrimaryKey(Relation pkrel, Oid *indexOid,
+                       int16 *nratts, int16 *attnums, Oid *atttypids,
+                       Oid *opclasses){
+	List *indexoidlist;
+	ListCell *indexoidscan;
+	HeapTuple indexTuple = NULL;
+	Datum indclassDatum;
+	bool isnull;
+	oidvector  *indclass;
+	int i;
+	Form_pg_index indexStruct = NULL;
+
+	*indexOid = InvalidOid;
+
+	indexoidlist = RelationGetIndexList(pkrel);
+
+	foreach(indexoidscan, indexoidlist)
+	{
+		Oid indexoid = lfirst_oid(indexoidscan);
+
+		indexTuple = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexoid));
+		if(!HeapTupleIsValid(indexTuple))
+			elog(ERROR, "cache lookup failed for index %u", indexoid);
+
+		indexStruct = (Form_pg_index) GETSTRUCT(indexTuple);
+		if(indexStruct->indisprimary && indexStruct->indimmediate)
+		{
+			*indexOid = indexoid;
+			break;
+		}
+		ReleaseSysCache(indexTuple);
+
+	}
+	list_free(indexoidlist);
+
+	if (!OidIsValid(*indexOid))
+		return;
+
+	/* Must get indclass the hard way */
+	indclassDatum = SysCacheGetAttr(INDEXRELID, indexTuple,
+									Anum_pg_index_indclass, &isnull);
+	Assert(!isnull);
+	indclass = (oidvector *) DatumGetPointer(indclassDatum);
+
+	*nratts = indexStruct->indnatts;
+	/*
+	 * Now build the list of PK attributes from the indkey definition (we
+	 * assume a primary key cannot have expressional elements)
+	 */
+	for (i = 0; i < indexStruct->indnatts; i++)
+	{
+		int			pkattno = indexStruct->indkey.values[i];
+
+		attnums[i] = pkattno;
+		atttypids[i] = attnumTypeId(pkrel, pkattno);
+		opclasses[i] = indclass->values[i];
+	}
+
+	ReleaseSysCache(indexTuple);
+}
diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile
index 2dde011..2e13e27 100644
--- a/src/backend/replication/Makefile
+++ b/src/backend/replication/Makefile
@@ -17,6 +17,8 @@ override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 OBJS = walsender.o walreceiverfuncs.o walreceiver.o basebackup.o \
 	repl_gram.o syncrep.o
 
+SUBDIRS = logical
+
 include $(top_srcdir)/src/backend/common.mk
 
 # repl_scanner is compiled as part of repl_gram
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
new file mode 100644
index 0000000..cf040ef
--- /dev/null
+++ b/src/backend/replication/logical/Makefile
@@ -0,0 +1,19 @@
+#-------------------------------------------------------------------------
+#
+# Makefile--
+#    Makefile for src/backend/replication/logical
+#
+# IDENTIFICATION
+#    src/backend/replication/logical/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/backend/replication/logical
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
+
+OBJS = decode.o logicalfuncs.o reorderbuffer.o snapbuild.o
+
+include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
new file mode 100644
index 0000000..15d261b
--- /dev/null
+++ b/src/backend/replication/logical/decode.c
@@ -0,0 +1,496 @@
+/*-------------------------------------------------------------------------
+ *
+ * decode.c
+ *		Decodes wal records from an xlogreader.h callback into an reorderbuffer
+ *		while building an appropriate snapshots to decode those
+ *
+ * NOTE:
+ * Its possible that the separation between decode.c and snapbuild.c is a
+ * bit too strict, in the end they just about have the same switch.
+ *
+ * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/decode.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/heapam_xlog.h"
+#include "access/transam.h"
+#include "access/xlog_internal.h"
+#include "access/xact.h"
+#include "access/xlogreader.h"
+
+#include "catalog/pg_control.h"
+
+#include "replication/reorderbuffer.h"
+#include "replication/decode.h"
+#include "replication/snapbuild.h"
+#include "replication/logicalfuncs.h"
+
+#include "utils/memutils.h"
+#include "utils/syscache.h"
+#include "utils/lsyscache.h"
+
+static void DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple);
+
+static void DecodeInsert(ReorderBuffer *cache, XLogRecordBuffer *buf);
+
+static void DecodeUpdate(ReorderBuffer *cache, XLogRecordBuffer *buf);
+
+static void DecodeDelete(ReorderBuffer *cache, XLogRecordBuffer *buf);
+
+static void DecodeMultiInsert(ReorderBuffer *cache, XLogRecordBuffer *buf);
+
+static void DecodeCommit(ReaderApplyState *state, XLogRecordBuffer *buf, TransactionId xid,
+						 TransactionId *sub_xids, int nsubxacts);
+
+static void DecodeAbort(ReorderBuffer * cache, XLogRecPtr lsn, TransactionId xid,
+						TransactionId *sub_xids, int nsubxacts);
+
+
+void DecodeRecordIntoReorderBuffer(XLogReaderState *reader,
+								   ReaderApplyState *state,
+								   XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+	uint8 info = r->xl_info & ~XLR_INFO_MASK;
+	ReorderBuffer *reorder = state->reorderbuffer;
+	SnapBuildAction action;
+
+	/*
+	 * FIXME: The existance of the snapshot builder is pretty obvious to the
+	 * outside right now, that doesn't seem to be very good...
+	 */
+	if (!state->snapstate)
+	{
+		state->snapstate = AllocateSnapshotBuilder(reorder);
+	}
+
+	/*---------
+	 * Call the snapshot builder. It needs to be called before we analyze
+	 * tuples for two reasons:
+	 *
+	 * * Only in the snapshot building logic we know whether we have enough
+	 *   information to decode a particular tuple
+	 *
+	 * * The Snapshot/CommandIds computed by the SnapshotBuilder need to be
+	 *   added to the ReorderBuffer before we add tuples using them
+	 *---------
+	 */
+	action = SnapBuildDecodeCallback(reorder, state->snapstate, buf);
+
+	if (state->stop_after_consistent && state->snapstate->state == SNAPBUILD_CONSISTENT)
+	{
+		//Assert(action == SNAPBUILD_SKIP);
+		reader->stop_at_record_boundary = true;
+		elog(WARNING, "reached consistent point, stopping!");
+		return;
+	}
+
+	if (action == SNAPBUILD_SKIP)
+		return;
+
+	switch (r->xl_rmid)
+	{
+		case RM_HEAP_ID:
+			{
+				info &= XLOG_HEAP_OPMASK;
+				switch (info)
+				{
+					case XLOG_HEAP_INSERT:
+						DecodeInsert(reorder, buf);
+						break;
+
+						/*
+						 * no guarantee that we get an HOT update again, so
+						 * handle it as a normal update
+						 */
+					case XLOG_HEAP_HOT_UPDATE:
+					case XLOG_HEAP_UPDATE:
+						DecodeUpdate(reorder, buf);
+						break;
+
+					case XLOG_HEAP_NEWPAGE:
+						/*
+						 * XXX: There doesn't seem to be a usecase for decoding
+						 * HEAP_NEWPAGE's. Its only used in various indexam's
+						 * and CLUSTER, neither of which should be relevant for
+						 * the logical changestream.
+						 */
+						break;
+
+					case XLOG_HEAP_DELETE:
+						DecodeDelete(reorder, buf);
+						break;
+					default:
+						break;
+				}
+				break;
+			}
+		case RM_HEAP2_ID:
+			{
+				info &= XLOG_HEAP_OPMASK;
+				switch (info)
+				{
+					case XLOG_HEAP2_MULTI_INSERT:
+						DecodeMultiInsert(reorder, buf);
+						break;
+					default:
+						/*
+						 * everything else here is just physical stuff were not
+						 * interested in
+						 */
+						break;
+				}
+				break;
+			}
+
+		case RM_XACT_ID:
+			{
+				switch (info)
+				{
+					case XLOG_XACT_COMMIT:
+						{
+							TransactionId *sub_xids;
+							xl_xact_commit *xlrec =
+								(xl_xact_commit *) buf->record_data;
+
+							/*
+							 * FIXME: theoretically computing this address is
+							 * not really allowed if there are no
+							 * subtransactions
+							 */
+							sub_xids = (TransactionId *) &(
+								xlrec->xnodes[xlrec->nrels]);
+
+							DecodeCommit(state, buf, r->xl_xid,
+										 sub_xids, xlrec->nsubxacts);
+
+
+							break;
+						}
+					case XLOG_XACT_COMMIT_PREPARED:
+						{
+							TransactionId *sub_xids;
+							xl_xact_commit_prepared *xlrec =
+								(xl_xact_commit_prepared*) buf->record_data;
+
+							sub_xids = (TransactionId *) &(
+								xlrec->crec.xnodes[xlrec->crec.nrels]);
+
+							DecodeCommit(state, buf, r->xl_xid, sub_xids,
+										 xlrec->crec.nsubxacts);
+
+							break;
+						}
+					case XLOG_XACT_COMMIT_COMPACT:
+						{
+							xl_xact_commit_compact *xlrec =
+								(xl_xact_commit_compact *) buf->record_data;
+							DecodeCommit(state, buf, r->xl_xid,
+										 xlrec->subxacts, xlrec->nsubxacts);
+							break;
+						}
+					case XLOG_XACT_ABORT:
+						{
+							TransactionId *sub_xids;
+							xl_xact_abort *xlrec =
+								(xl_xact_abort *) buf->record_data;
+
+							sub_xids = (TransactionId *) &(
+								xlrec->xnodes[xlrec->nrels]);
+
+							DecodeAbort(reorder, r->xl_xid, buf->origptr,
+										sub_xids, xlrec->nsubxacts);
+							break;
+						}
+					case XLOG_XACT_ABORT_PREPARED:
+						{
+							TransactionId *sub_xids;
+							xl_xact_abort_prepared *xlrec =
+								(xl_xact_abort_prepared *)buf->record_data;
+							xl_xact_abort *arec = &xlrec->arec;
+
+							sub_xids = (TransactionId *) &(
+								arec->xnodes[arec->nrels]);
+
+							DecodeAbort(reorder, xlrec->xid, buf->origptr,
+										sub_xids, arec->nsubxacts);
+							/* XXX: any reason for also aborting r->xl_xid? */
+							break;
+						}
+
+					case XLOG_XACT_ASSIGNMENT:
+						/*
+						 * XXX: We could reassign transactions to the parent
+						 * here to save space and effort when merging
+						 * transactions at commit.
+						 */
+						break;
+					case XLOG_XACT_PREPARE:
+						/*
+						 * FXIME: we should replay the transaction and prepare
+						 * it as well.
+						 */
+						break;
+					default:
+						break;
+						;
+				}
+				break;
+			}
+		case RM_XLOG_ID:
+			{
+				switch (info)
+				{
+					/* this is also used in END_OF_RECOVERY checkpoints */
+					case XLOG_CHECKPOINT_SHUTDOWN:
+						/*
+						 * abort all transactions that still are in progress,
+						 * they aren't in progress anymore.  do not abort
+						 * prepared transactions that have been prepared for
+						 * commit.
+						 *
+						 * FIXME: implement.
+						 */
+						break;
+				}
+			}
+		default:
+			break;
+	}
+}
+
+static void
+DecodeCommit(ReaderApplyState *state, XLogRecordBuffer *buf, TransactionId xid,
+			 TransactionId *sub_xids, int nsubxacts)
+{
+	int i;
+
+	/* not interested in that part of the stream */
+	if (XLByteLE(buf->origptr, state->snapstate->transactions_after))
+	{
+		DecodeAbort(state->reorderbuffer, buf->origptr, xid,
+					sub_xids, nsubxacts);
+		return;
+	}
+
+	for (i = 0; i < nsubxacts; i++)
+	{
+		ReorderBufferCommitChild(state->reorderbuffer, xid, *sub_xids,
+								 buf->origptr);
+		sub_xids++;
+	}
+
+	/* replay actions of all transaction + subtransactions in order */
+	ReorderBufferCommit(state->reorderbuffer, xid, buf->origptr);
+}
+
+static void
+DecodeAbort(ReorderBuffer *reorder, XLogRecPtr lsn, TransactionId xid,
+			TransactionId *sub_xids, int nsubxacts)
+{
+	int i;
+
+	elog(WARNING, "ABORT %u", xid);
+
+	for (i = 0; i < nsubxacts; i++)
+	{
+		ReorderBufferAbort(reorder, *sub_xids, lsn);
+		sub_xids++;
+	}
+
+	ReorderBufferAbort(reorder, xid, lsn);
+}
+
+static void
+DecodeInsert(ReorderBuffer *reorder, XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+	xl_heap_insert *xlrec = (xl_heap_insert *) buf->record_data;
+
+	ReorderBufferChange *change;
+
+	if (r->xl_info & XLR_BKP_BLOCK(0)
+		&& r->xl_len < (SizeOfHeapUpdate + SizeOfHeapHeader))
+	{
+		elog(DEBUG2, "huh, no tuple data on wal_level = logical?");
+		return;
+	}
+
+	change = ReorderBufferGetChange(reorder);
+	change->action = REORDER_BUFFER_CHANGE_INSERT;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	change->newtuple = ReorderBufferGetTupleBuf(reorder);
+
+	DecodeXLogTuple((char *) xlrec + SizeOfHeapInsert,
+					r->xl_len - SizeOfHeapInsert,
+					change->newtuple);
+
+	ReorderBufferAddChange(reorder, r->xl_xid, buf->origptr, change);
+}
+
+static void
+DecodeUpdate(ReorderBuffer *reorder, XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+	xl_heap_update *xlrec = (xl_heap_update *) buf->record_data;
+
+
+	ReorderBufferChange *change;
+
+	if ((r->xl_info & XLR_BKP_BLOCK(0) || r->xl_info & XLR_BKP_BLOCK(1)) &&
+		(r->xl_len < (SizeOfHeapUpdate + SizeOfHeapHeader)))
+	{
+		elog(DEBUG2, "huh, no tuple data on wal_level = logical?");
+		return;
+	}
+
+	change = ReorderBufferGetChange(reorder);
+	change->action = REORDER_BUFFER_CHANGE_UPDATE;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	/*
+	 * FIXME: need to get/save the old tuple as well if we want primary key
+	 * changes to work.
+	 */
+	change->newtuple = ReorderBufferGetTupleBuf(reorder);
+
+	DecodeXLogTuple((char *) xlrec + SizeOfHeapUpdate,
+					r->xl_len - SizeOfHeapUpdate,
+					change->newtuple);
+
+	ReorderBufferAddChange(reorder, r->xl_xid, buf->origptr, change);
+}
+
+static void
+DecodeDelete(ReorderBuffer *reorder, XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+
+	xl_heap_delete *xlrec = (xl_heap_delete *) buf->record_data;
+
+	ReorderBufferChange *change;
+
+	change = ReorderBufferGetChange(reorder);
+	change->action = REORDER_BUFFER_CHANGE_DELETE;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	if (r->xl_len <= (SizeOfHeapDelete + SizeOfHeapHeader))
+	{
+		elog(DEBUG2, "huh, no primary key for a delete on wal_level = logical?");
+		return;
+	}
+
+	change->oldtuple = ReorderBufferGetTupleBuf(reorder);
+
+	DecodeXLogTuple((char *) xlrec + SizeOfHeapDelete,
+					r->xl_len - SizeOfHeapDelete,
+					change->oldtuple);
+
+	ReorderBufferAddChange(reorder, r->xl_xid, buf->origptr, change);
+}
+
+/*
+ * Decode xl_heap_multi_insert record into multiple changes.
+ *
+ * Due to slightly different layout we can't reuse DecodeXLogTuple without
+ * making that even harder to understand than already is.
+ */
+static void
+DecodeMultiInsert(ReorderBuffer *reorder, XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+	xl_heap_multi_insert *xlrec = (xl_heap_multi_insert *)buf->record_data;
+	int i;
+	char *data = buf->record_data;
+	bool		isinit = (r->xl_info & XLOG_HEAP_INIT_PAGE) != 0;
+
+	data += SizeOfHeapMultiInsert;
+
+	/* OffsetNumber's are only stored if its not a HEAP_INIT_PAGE record */
+	if (!isinit)
+		data += sizeof(OffsetNumber) * xlrec->ntuples;
+
+	for (i = 0; i < xlrec->ntuples; i++)
+	{
+		ReorderBufferChange *change;
+		xl_multi_insert_tuple *xlhdr;
+		int datalen;
+		ReorderBufferTupleBuf *tuple;
+
+		change = ReorderBufferGetChange(reorder);
+		change->action = REORDER_BUFFER_CHANGE_INSERT;
+		change->newtuple = ReorderBufferGetTupleBuf(reorder);
+		memcpy(&change->relnode, &xlrec->node, sizeof(RelFileNode));
+
+		tuple = change->newtuple;
+		/* not a disk based tuple */
+		ItemPointerSetInvalid(&tuple->tuple.t_self);
+
+		xlhdr = (xl_multi_insert_tuple *) SHORTALIGN(data);
+		data = ((char *) xlhdr) + SizeOfMultiInsertTuple;
+		datalen = xlhdr->datalen;
+
+		/* we can only figure this out after reassembling the transactions */
+		tuple->tuple.t_tableOid = InvalidOid;
+		tuple->tuple.t_data = &tuple->header;
+		tuple->tuple.t_len = datalen + offsetof(HeapTupleHeaderData, t_bits);
+
+		memset(&tuple->header, 0, sizeof(HeapTupleHeaderData));
+
+		memcpy((char *) &tuple->header + offsetof(HeapTupleHeaderData, t_bits),
+			   (char *) data,
+			   datalen);
+		data += datalen;
+
+		tuple->header.t_infomask = xlhdr->t_infomask;
+		tuple->header.t_infomask2 = xlhdr->t_infomask2;
+		tuple->header.t_hoff = xlhdr->t_hoff;
+
+		ReorderBufferAddChange(reorder, r->xl_xid, buf->origptr, change);
+	}
+}
+
+
+static void
+DecodeXLogTuple(char *data, Size len, ReorderBufferTupleBuf *tuple)
+{
+	xl_heap_header xlhdr;
+	int datalen = len - SizeOfHeapHeader;
+
+	Assert(datalen >= 0);
+	Assert(datalen <= MaxHeapTupleSize);
+
+	tuple->tuple.t_len = datalen + offsetof(HeapTupleHeaderData, t_bits);
+
+	/* not a disk based tuple */
+	ItemPointerSetInvalid(&tuple->tuple.t_self);
+
+	/* we can only figure this out after reassembling the transactions */
+	tuple->tuple.t_tableOid = InvalidOid;
+	tuple->tuple.t_data = &tuple->header;
+
+	/* data is not stored aligned */
+	memcpy((char *) &xlhdr,
+		   data,
+		   SizeOfHeapHeader);
+
+	memset(&tuple->header, 0, sizeof(HeapTupleHeaderData));
+
+	memcpy((char *) &tuple->header + offsetof(HeapTupleHeaderData, t_bits),
+		   data + SizeOfHeapHeader,
+		   datalen);
+
+	tuple->header.t_infomask = xlhdr.t_infomask;
+	tuple->header.t_infomask2 = xlhdr.t_infomask2;
+	tuple->header.t_hoff = xlhdr.t_hoff;
+}
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
new file mode 100644
index 0000000..41f2ec8
--- /dev/null
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -0,0 +1,247 @@
+/*-------------------------------------------------------------------------
+ *
+ * logicalfuncs.c
+ *
+ *     Support functions for using xlog decoding
+ *
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logicalfuncs.c
+ *
+ */
+
+#include "postgres.h"
+
+#include "fmgr.h"
+
+#include "replication/logicalfuncs.h"
+
+#include "catalog/pg_class.h"
+#include "catalog/pg_type.h"
+
+#include "replication/reorderbuffer.h"
+#include "replication/decode.h"
+/*FIXME: XLogRead*/
+#include "replication/walsender_private.h"
+#include "replication/snapbuild.h"
+
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+#include "utils/typcache.h"
+
+
+/*
+ * XLogReader callbacks
+ */
+static bool
+replay_record_is_interesting(XLogReaderState* state, XLogRecord* r)
+{
+	return true;
+}
+
+static void
+replay_writeout_data(XLogReaderState* state, char* data, Size len)
+{
+	return;
+}
+
+static void
+replay_finished_record(XLogReaderState* state, XLogRecordBuffer* buf)
+{
+	ReaderApplyState* apply_state = state->private_data;
+	DecodeRecordIntoReorderBuffer(state, apply_state, buf);
+}
+
+static void
+replay_read_page(XLogReaderState* state, char* cur_page, XLogRecPtr startptr)
+{
+	XLogPageHeader page_header;
+
+	Assert((startptr % XLOG_BLCKSZ) == 0);
+
+	/* FIXME: more sensible/efficient implementation */
+	XLogRead(cur_page, startptr, XLOG_BLCKSZ);
+
+	page_header = (XLogPageHeader)cur_page;
+
+	if (page_header->xlp_magic != XLOG_PAGE_MAGIC)
+	{
+		elog(FATAL, "page header magic %x, should be %x at %X/%X", page_header->xlp_magic,
+		     XLOG_PAGE_MAGIC, (uint32)(startptr >> 32), (uint32)startptr);
+	}
+}
+
+/*
+ * Callbacks for ReorderBuffer which add in some more information and then call
+ * output_plugin.h plugins.
+ */
+static void
+begin_txn_wrapper(ReorderBuffer* cache, ReorderBufferTXN* txn)
+{
+	ReaderApplyState *state = cache->private_data;
+	bool send;
+
+	resetStringInfo(state->out);
+	WalSndPrepareWrite(state->out, txn->lsn);
+
+	send = state->begin_cb(state->user_private, state->out, txn);
+
+	if (send)
+	{
+		WalSndWriteData(state->out);
+	}
+}
+
+static void
+commit_txn_wrapper(ReorderBuffer* cache, ReorderBufferTXN* txn, XLogRecPtr commit_lsn)
+{
+	ReaderApplyState *state = cache->private_data;
+	bool send;
+
+	resetStringInfo(state->out);
+	WalSndPrepareWrite(state->out, commit_lsn);
+
+	send = state->commit_cb(state->user_private, state->out, txn, commit_lsn);
+
+	if (send)
+	{
+		WalSndWriteData(state->out);
+	}
+}
+
+static void
+change_wrapper(ReorderBuffer* cache, ReorderBufferTXN* txn, ReorderBufferChange* change)
+{
+	ReaderApplyState *state = cache->private_data;
+	bool send;
+	HeapTuple table;
+	Oid reloid;
+
+	resetStringInfo(state->out);
+	WalSndPrepareWrite(state->out, change->lsn);
+
+	table = LookupTableByRelFileNode(&change->relnode);
+	Assert(table);
+	reloid = HeapTupleHeaderGetOid(table->t_data);
+	ReleaseSysCache(table);
+
+
+	send = state->change_cb(state->user_private, state->out, txn,
+							reloid, change);
+
+	if (send)
+	{
+		WalSndWriteData(state->out);
+	}
+}
+
+/*
+ * Build a snapshot reader that doesn't ever outputs/decodes anything, but just
+ * waits for the first point in the LSN stream where it reaches a consistent
+ * state.
+ */
+XLogReaderState *
+initial_snapshot_reader()
+{
+	ReorderBuffer *reorder;
+	XLogReaderState *xlogreader_state = XLogReaderAllocate();
+	ReaderApplyState *apply_state;
+
+	xlogreader_state->is_record_interesting = replay_record_is_interesting;
+	xlogreader_state->finished_record = replay_finished_record;
+	xlogreader_state->writeout_data = replay_writeout_data;
+	xlogreader_state->read_page = replay_read_page;
+	xlogreader_state->private_data = calloc(1, sizeof(ReaderApplyState));
+
+	if (!xlogreader_state->private_data)
+		elog(ERROR, "Could not allocate the ReaderApplyState struct");
+
+	apply_state = (ReaderApplyState*)xlogreader_state->private_data;
+
+	reorder = ReorderBufferAllocate();
+	reorder->begin = NULL; /* not decoding yet */
+	reorder->apply_change = NULL;
+	reorder->commit = NULL;
+	reorder->private_data = xlogreader_state->private_data;
+
+	apply_state->reorderbuffer = reorder;
+	apply_state->stop_after_consistent = true;
+
+	return xlogreader_state;
+}
+
+/*
+ * Build a snapshot reader with callbacks found in the shared library "plugin"
+ * under the symbol names found in output_plugin.h.
+ * It wraps those callbacks so they send out their changes via an logical
+ * walsender.
+ */
+XLogReaderState *
+normal_snapshot_reader(char *plugin, XLogRecPtr valid_after)
+{
+	ReorderBuffer *reorder;
+	XLogReaderState *xlogreader_state = XLogReaderAllocate();
+	ReaderApplyState *apply_state;
+
+	xlogreader_state->is_record_interesting = replay_record_is_interesting;
+	xlogreader_state->finished_record = replay_finished_record;
+	xlogreader_state->writeout_data = replay_writeout_data;
+	xlogreader_state->read_page = replay_read_page;
+	xlogreader_state->private_data = calloc(1, sizeof(ReaderApplyState));
+
+	if (!xlogreader_state->private_data)
+		elog(ERROR, "Could not allocate the ReaderApplyState struct");
+
+	apply_state = (ReaderApplyState*)xlogreader_state->private_data;
+
+	reorder = ReorderBufferAllocate();
+
+	apply_state->reorderbuffer = reorder;
+	apply_state->stop_after_consistent = false;
+
+	apply_state->snapstate = AllocateSnapshotBuilder(reorder);
+	apply_state->snapstate->transactions_after = valid_after;
+
+	reorder->begin = begin_txn_wrapper;
+	reorder->apply_change = change_wrapper;
+	reorder->commit = commit_txn_wrapper;
+
+	/* lookup symbols in the shared libarary */
+
+	/* optional */
+	apply_state->init_cb = (LogicalDecodeInitCB)
+		load_external_function(plugin, "pg_decode_init", false, NULL);
+
+	apply_state->begin_cb = (LogicalDecodeBeginCB)
+		load_external_function(plugin, "pg_decode_begin_txn", true, NULL);
+
+	apply_state->change_cb = (LogicalDecodeChangeCB)
+		load_external_function(plugin, "pg_decode_change", true, NULL);
+
+	apply_state->commit_cb = (LogicalDecodeCommitCB)
+		load_external_function(plugin, "pg_decode_commit_txn", true, NULL);
+
+	/* optional */
+	apply_state->cleanup_cb = (LogicalDecodeCleanupCB)
+		load_external_function(plugin, "pg_decode_clean", false, NULL);
+
+	reorder->private_data = xlogreader_state->private_data;
+
+	apply_state->out = makeStringInfo();
+
+	/* initialize output plugin */
+	if (apply_state->init_cb)
+		apply_state->init_cb(&apply_state->user_private);
+
+	return xlogreader_state;
+}
+
+/* has the initial snapshot found a consistent state? */
+bool
+initial_snapshot_ready(XLogReaderState *reader)
+{
+	ReaderApplyState* state = reader->private_data;
+	return state->snapstate->state == SNAPBUILD_CONSISTENT;
+}
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
new file mode 100644
index 0000000..b80b054
--- /dev/null
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -0,0 +1,1156 @@
+/*-------------------------------------------------------------------------
+ *
+ * reorderbuffer.c
+ *
+ * PostgreSQL logical replay "cache" management
+ *
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/reorderbuffer.c
+ *
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/xact.h"
+
+#include "catalog/pg_class.h"
+#include "catalog/pg_control.h"
+
+#include "lib/simpleheap.h"
+
+#include "replication/reorderbuffer.h"
+#include "replication/snapbuild.h"
+
+#include "storage/sinval.h"
+#include "storage/bufmgr.h"
+
+#include "utils/builtins.h"
+#include "utils/combocid.h"
+#include "utils/memutils.h"
+#include "utils/relcache.h"
+#include "utils/tqual.h"
+#include "utils/syscache.h"
+
+
+
+const Size max_memtries = 1 << 16;
+
+const size_t max_cached_changes = 1024;
+const size_t max_cached_tuplebufs = 1024; /* ~8MB */
+const size_t max_cached_transactions = 512;
+
+/* entry for a hash table we use to map from xid to our transaction state */
+typedef struct ReorderBufferTXNByIdEnt
+{
+	TransactionId xid;
+	ReorderBufferTXN *txn;
+}  ReorderBufferTXNByIdEnt;
+
+typedef struct ReorderBufferTupleCidKey
+{
+	RelFileNode relnode;
+	ItemPointerData tid;
+} ReorderBufferTupleCidKey;
+
+typedef struct ReorderBufferTupleCidEnt
+{
+	ReorderBufferTupleCidKey key;
+	CommandId cmin;
+	CommandId cmax;
+	CommandId combocid;
+} ReorderBufferTupleCidEnt;
+
+/*
+ * For efficiency and simplicity reasons we want to keep Snapshots, CommandIds
+ * and ComboCids in the same list with the user visible INSERT/UPDATE/DELETE
+ * changes. We don't want to leak those internal values to external users
+ * though (they would just use switch()...default:) because that would make it
+ * harder to add to new user visible values.
+ *
+ * This needs to be synchronized with ReorderBufferChangeType! Adjust the
+ * StatisAssertExpr's in ReorderBufferAllocate if you add anything!
+ */
+enum ReorderBufferChangeTypeInternal
+{
+	REORDER_BUFFER_CHANGE_INTERNAL_INSERT,
+	REORDER_BUFFER_CHANGE_INTERNAL_UPDATE,
+	REORDER_BUFFER_CHANGE_INTERNAL_DELETE,
+	REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,
+	REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID,
+	REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID
+};
+
+/* Get an unused, but potentially cached, ReorderBufferTXN entry */
+static ReorderBufferTXN *ReorderBufferGetTXN(ReorderBuffer *cache);
+
+/* Return an unused ReorderBufferTXN entry */
+static void ReorderBufferReturnTXN(ReorderBuffer *cache, ReorderBufferTXN *txn);
+
+static ReorderBufferTXN *ReorderBufferTXNByXid(ReorderBuffer *cache, TransactionId xid,
+                                         bool create, bool *is_new);
+
+
+/*
+ * support functions for lsn-order iterating over the ->changes of a
+ * transaction and its subtransactions
+ */
+
+/*
+ * used for iteration over the k-way heap merge of a transaction and its
+ * subtransactions
+ */
+typedef struct ReorderBufferIterTXNState
+{
+	simpleheap *heap;
+} ReorderBufferIterTXNState;
+
+/* allocate & initialize an iterator */
+static ReorderBufferIterTXNState *
+ReorderBufferIterTXNInit(ReorderBuffer *cache, ReorderBufferTXN *txn);
+
+/* get the next change */
+static ReorderBufferChange *
+ReorderBufferIterTXNNext(ReorderBuffer *cache, ReorderBufferIterTXNState *state);
+
+/* deallocate iterator */
+static void
+ReorderBufferIterTXNFinish(ReorderBuffer *cache, ReorderBufferIterTXNState *state);
+
+/* where to put this? */
+static void
+ReorderBufferProcessInvalidations(ReorderBuffer *cache, ReorderBufferTXN *txn);
+
+ReorderBuffer *
+ReorderBufferAllocate(void)
+{
+	ReorderBuffer *cache = (ReorderBuffer *) malloc(sizeof(ReorderBuffer));
+	HASHCTL hash_ctl;
+
+	StaticAssertExpr((int)REORDER_BUFFER_CHANGE_INTERNAL_INSERT == (int)REORDER_BUFFER_CHANGE_INSERT, "out of sync enums");
+	StaticAssertExpr((int)REORDER_BUFFER_CHANGE_INTERNAL_UPDATE == (int)REORDER_BUFFER_CHANGE_UPDATE, "out of sync enums");
+	StaticAssertExpr((int)REORDER_BUFFER_CHANGE_INTERNAL_DELETE == (int)REORDER_BUFFER_CHANGE_DELETE, "out of sync enums");
+
+	if (!cache)
+		elog(ERROR, "Could not allocate the ReorderBuffer");
+
+	cache->build_snapshots = true;
+
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+
+	cache->context = AllocSetContextCreate(TopMemoryContext,
+	                                       "ReorderBuffer",
+	                                       ALLOCSET_DEFAULT_MINSIZE,
+	                                       ALLOCSET_DEFAULT_INITSIZE,
+	                                       ALLOCSET_DEFAULT_MAXSIZE);
+
+	hash_ctl.keysize = sizeof(TransactionId);
+	hash_ctl.entrysize = sizeof(ReorderBufferTXNByIdEnt);
+	hash_ctl.hash = tag_hash;
+	hash_ctl.hcxt = cache->context;
+
+	cache->by_txn = hash_create("ReorderBufferByXid", 1000, &hash_ctl,
+	                            HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
+
+	cache->nr_cached_transactions = 0;
+	cache->nr_cached_changes = 0;
+	cache->nr_cached_tuplebufs = 0;
+
+	dlist_init(&cache->cached_transactions);
+	dlist_init(&cache->cached_changes);
+	slist_init(&cache->cached_tuplebufs);
+
+	return cache;
+}
+
+/*
+ * Free a ReorderBuffer
+ */
+void
+ReorderBufferFree(ReorderBuffer *cache)
+{
+	/* FIXME: check for in-progress transactions */
+	/* FIXME: clean up cached transaction */
+	/* FIXME: clean up cached changes */
+	/* FIXME: clean up cached tuplebufs */
+	hash_destroy(cache->by_txn);
+	free(cache);
+}
+
+/*
+ * Get a unused, possibly preallocated, ReorderBufferTXN.
+ */
+static ReorderBufferTXN *
+ReorderBufferGetTXN(ReorderBuffer *cache)
+{
+	ReorderBufferTXN *txn;
+
+	if (cache->nr_cached_transactions)
+	{
+		cache->nr_cached_transactions--;
+		txn = dlist_container(ReorderBufferTXN, node,
+		                      dlist_pop_head_node(&cache->cached_transactions));
+	}
+	else
+	{
+		txn = (ReorderBufferTXN *)
+			malloc(sizeof(ReorderBufferTXN));
+
+		if (!txn)
+			elog(ERROR, "Could not allocate a ReorderBufferTXN struct");
+	}
+
+	memset(txn, 0, sizeof(ReorderBufferTXN));
+
+	dlist_init(&txn->changes);
+	dlist_init(&txn->tuplecids);
+	dlist_init(&txn->subtxns);
+
+	return txn;
+}
+
+/*
+ * Free an ReorderBufferTXN. Deallocation might be delayed for efficiency
+ * purposes.
+ */
+void
+ReorderBufferReturnTXN(ReorderBuffer *cache, ReorderBufferTXN *txn)
+{
+	if (txn->tuplecid_hash != NULL)
+	{
+		hash_destroy(txn->tuplecid_hash);
+		txn->tuplecid_hash = NULL;
+	}
+
+	if (txn->invalidations)
+	{
+		free(txn->invalidations);
+		txn->invalidations = NULL;
+	}
+
+
+	if (cache->nr_cached_transactions < max_cached_transactions)
+	{
+		cache->nr_cached_transactions++;
+		dlist_push_head(&cache->cached_transactions, &txn->node);
+	}
+	else
+	{
+		free(txn);
+	}
+}
+
+/*
+ * Get a unused, possibly preallocated, ReorderBufferChange.
+ */
+ReorderBufferChange *
+ReorderBufferGetChange(ReorderBuffer *cache)
+{
+	ReorderBufferChange *change;
+
+	if (cache->nr_cached_changes)
+	{
+		cache->nr_cached_changes--;
+		change = dlist_container(ReorderBufferChange, node,
+								 dlist_pop_head_node(&cache->cached_changes));
+	}
+	else
+	{
+		change = (ReorderBufferChange *) malloc(sizeof(ReorderBufferChange));
+
+		if (!change)
+			elog(ERROR, "Could not allocate a ReorderBufferChange struct");
+	}
+
+
+	memset(change, 0, sizeof(ReorderBufferChange));
+	return change;
+}
+
+/*
+ * Free an ReorderBufferChange. Deallocation might be delayed for efficiency
+ * purposes.
+ */
+void
+ReorderBufferReturnChange(ReorderBuffer *cache, ReorderBufferChange *change)
+{
+	switch ((enum ReorderBufferChangeTypeInternal)change->action_internal)
+	{
+		case REORDER_BUFFER_CHANGE_INTERNAL_INSERT:
+		case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE:
+		case REORDER_BUFFER_CHANGE_INTERNAL_DELETE:
+			if (change->newtuple)
+			{
+				ReorderBufferReturnTupleBuf(cache, change->newtuple);
+				change->newtuple = NULL;
+			}
+
+			if (change->oldtuple)
+			{
+				ReorderBufferReturnTupleBuf(cache, change->oldtuple);
+				change->oldtuple = NULL;
+			}
+			break;
+		case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
+			if (change->snapshot)
+			{
+				SnapBuildSnapDecRefcount(change->snapshot);
+				change->snapshot = NULL;
+			}
+		case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
+			break;
+		case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+			break;
+	}
+
+	if (cache->nr_cached_changes < max_cached_changes)
+	{
+		cache->nr_cached_changes++;
+		dlist_push_head(&cache->cached_changes, &change->node);
+	}
+	else
+	{
+		free(change);
+	}
+}
+
+
+/*
+ * Get a unused, possibly preallocated, ReorderBufferTupleBuf
+ */
+ReorderBufferTupleBuf *
+ReorderBufferGetTupleBuf(ReorderBuffer *cache)
+{
+	ReorderBufferTupleBuf *tuple;
+
+	if (cache->nr_cached_tuplebufs)
+	{
+		cache->nr_cached_tuplebufs--;
+		tuple = slist_container(ReorderBufferTupleBuf, node,
+		                        slist_pop_head_node(&cache->cached_tuplebufs));
+	}
+	else
+	{
+		tuple =
+			(ReorderBufferTupleBuf *) malloc(sizeof(ReorderBufferTupleBuf));
+
+		if (!tuple)
+			elog(ERROR, "Could not allocate a ReorderBufferTupleBuf struct");
+	}
+
+	return tuple;
+}
+
+/*
+ * Free an ReorderBufferTupleBuf. Deallocation might be delayed for efficiency
+ * purposes.
+ */
+void
+ReorderBufferReturnTupleBuf(ReorderBuffer *cache, ReorderBufferTupleBuf *tuple)
+{
+	if (cache->nr_cached_tuplebufs < max_cached_tuplebufs)
+	{
+		cache->nr_cached_tuplebufs++;
+		slist_push_head(&cache->cached_tuplebufs, &tuple->node);
+	}
+	else
+	{
+		free(tuple);
+	}
+}
+
+/*
+ * Access the transactions being processed via xid. Optionally create a new
+ * entry.
+ */
+static
+ReorderBufferTXN *
+ReorderBufferTXNByXid(ReorderBuffer *cache, TransactionId xid, bool create, bool *is_new)
+{
+	ReorderBufferTXNByIdEnt *ent;
+	bool found;
+
+	/* FIXME: add one entry fast-path cache */
+
+	ent = (ReorderBufferTXNByIdEnt *)
+		hash_search(cache->by_txn,
+		            (void *)&xid,
+		            (create ? HASH_ENTER : HASH_FIND),
+		            &found);
+
+	if (found)
+	{
+#ifdef VERBOSE_DEBUG
+		elog(LOG, "found cache entry for %u at %p", xid, ent);
+#endif
+	}
+	else
+	{
+#ifdef VERBOSE_DEBUG
+		elog(LOG, "didn't find cache entry for %u in %p at %p, creating %u",
+		     xid, cache, ent, create);
+#endif
+	}
+
+	if (!found && !create)
+		return NULL;
+
+	if (!found)
+	{
+		ent->txn = ReorderBufferGetTXN(cache);
+		ent->txn->xid = xid;
+	}
+
+	if (is_new)
+		*is_new = !found;
+
+	return ent->txn;
+}
+
+/*
+ * Queue a change into a transaction so it can be replayed uppon commit.
+ */
+void
+ReorderBufferAddChange(ReorderBuffer *cache, TransactionId xid, XLogRecPtr lsn,
+                    ReorderBufferChange *change)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+	txn->lsn = lsn;
+	dlist_push_tail(&txn->changes, &change->node);
+}
+
+
+/*
+ * Associate a subtransaction with its toplevel transaction.
+ */
+void
+ReorderBufferCommitChild(ReorderBuffer *cache, TransactionId xid,
+						 TransactionId subxid, XLogRecPtr lsn)
+{
+	ReorderBufferTXN *txn;
+	ReorderBufferTXN *subtxn;
+
+	subtxn = ReorderBufferTXNByXid(cache, subxid, false, NULL);
+
+	/*
+	 * No need to do anything if that subtxn didn't contain any changes
+	 */
+	if (!subtxn)
+		return;
+
+	subtxn->lsn = lsn;
+
+	txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+
+	dlist_push_tail(&txn->subtxns, &subtxn->node);
+	txn->nsubtxns++;
+}
+
+
+/*
+ * Support for efficiently iterating over a transaction's and its
+ * subtransactions' changes.
+ *
+ * We do by doing aa k-way merge between transactions/subtransactions. For that
+ * we model the current heads of the different transactions as a binary heap so
+ * we easily know which (sub-)transaction has the change with the smalles lsn
+ * next.
+ * Luckily the changes in individual transactions are already sorted by LSN.
+ */
+
+/*
+ * Binary heap comparison function.
+ */
+static int
+ReorderBufferIterCompare(simpleheap_kv *a, simpleheap_kv *b)
+{
+	ReorderBufferChange *change_a = dlist_container(ReorderBufferChange, node,
+													(dlist_node*)a->key);
+	ReorderBufferChange *change_b = dlist_container(ReorderBufferChange, node,
+													(dlist_node*)b->key);
+
+	if (change_a->lsn < change_b->lsn)
+		return -1;
+
+	else if (change_a->lsn == change_b->lsn)
+		return 0;
+
+	return 1;
+}
+
+/*
+ * Initialize an iterator which iterates in lsn order over a transaction and
+ * all its subtransactions.
+ */
+static ReorderBufferIterTXNState *
+ReorderBufferIterTXNInit(ReorderBuffer *cache, ReorderBufferTXN *txn)
+{
+	size_t nr_txns = 0; /* main txn */
+	ReorderBufferIterTXNState *state;
+	dlist_iter cur_txn_i;
+	ReorderBufferTXN *cur_txn;
+	ReorderBufferChange *cur_change;
+
+	if (!dlist_is_empty(&txn->changes))
+		nr_txns++;
+
+	/* count how large our heap must be */
+	dlist_foreach(cur_txn_i, &txn->subtxns)
+	{
+		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
+
+		if (!dlist_is_empty(&cur_txn->changes))
+			nr_txns++;
+	}
+
+	/*
+	 * FIXME: Add fastpath for the rather common nr_txns=1 case, no need to
+	 * allocate/build a heap in that case.
+	 */
+
+	/* allocate iteration state */
+	state = calloc(1, sizeof(ReorderBufferIterTXNState));
+
+	/* allocate heap */
+	state->heap = simpleheap_allocate(nr_txns);
+	state->heap->compare = ReorderBufferIterCompare;
+
+	/*
+	 * fill array with elements, heap condition not yet fullfilled. Properly
+	 * building the heap afterwards is more efficient.
+	 */
+
+	/* add toplevel transaction if it contains changes*/
+	if (!dlist_is_empty(&txn->changes))
+	{
+		cur_change = dlist_head_element(ReorderBufferChange, node, &txn->changes);
+
+		simpleheap_add_unordered(state->heap, &cur_change->node, txn);
+	}
+
+	/* add subtransactions if they contain changes */
+	dlist_foreach(cur_txn_i, &txn->subtxns)
+	{
+		cur_txn = dlist_container(ReorderBufferTXN, node, cur_txn_i.cur);
+
+		if (!dlist_is_empty(&cur_txn->changes))
+		{
+			cur_change = dlist_head_element(ReorderBufferChange, node,
+											&cur_txn->changes);
+
+			simpleheap_add_unordered(state->heap, &cur_change->node, txn);
+		}
+	}
+
+	/* make the array fullfill the heap property */
+	simpleheap_build(state->heap);
+	return state;
+}
+
+/*
+ * Return the next change when iterating over a transaction and its
+ * subtransaction.
+ */
+static ReorderBufferChange *
+ReorderBufferIterTXNNext(ReorderBuffer *cache, ReorderBufferIterTXNState *state)
+{
+	ReorderBufferTXN *txn = NULL;
+	ReorderBufferChange *change;
+	simpleheap_kv *kv;
+
+	/* nothing there anymore */
+	if (state->heap->size == 0)
+		return NULL;
+
+	kv = simpleheap_first(state->heap);
+
+	change = dlist_container(ReorderBufferChange, node, (dlist_node*)kv->key);
+
+	txn = (ReorderBufferTXN *) kv->value;
+
+	if (!dlist_has_next(&txn->changes, &change->node))
+	{
+		simpleheap_remove_first(state->heap);
+	}
+	else
+	{
+		simpleheap_change_key(state->heap, change->node.next);
+	}
+	return change;
+}
+
+/*
+ * Deallocate the iterator
+ */
+static void
+ReorderBufferIterTXNFinish(ReorderBuffer *cache, ReorderBufferIterTXNState *state)
+{
+	simpleheap_free(state->heap);
+	free(state);
+}
+
+
+/*
+ * Cleanup the contents of a transaction, usually after the transaction
+ * committed or aborted.
+ */
+static void
+ReorderBufferCleanupTXN(ReorderBuffer *cache, ReorderBufferTXN *txn)
+{
+	bool found;
+	dlist_mutable_iter cur_change;
+	dlist_mutable_iter cur_txn;
+
+	/* cleanup subtransactions & their changes */
+	dlist_foreach_modify(cur_txn, &txn->subtxns)
+	{
+		ReorderBufferTXN *subtxn = dlist_container(ReorderBufferTXN, node, cur_txn.cur);
+
+		dlist_foreach_modify(cur_change, &subtxn->changes)
+		{
+			ReorderBufferChange *change =
+				dlist_container(ReorderBufferChange, node, cur_change.cur);
+
+			ReorderBufferReturnChange(cache, change);
+		}
+		ReorderBufferReturnTXN(cache, subtxn);
+	}
+
+	/* cleanup changes in the toplevel txn */
+	dlist_foreach_modify(cur_change, &txn->changes)
+	{
+		ReorderBufferChange *change =
+			dlist_container(ReorderBufferChange, node, cur_change.cur);
+
+		ReorderBufferReturnChange(cache, change);
+	}
+
+	/*
+	 * cleanup the tuplecids we stored timetravel access. They are always
+	 * stored in the toplevel transaction.
+	 */
+	dlist_foreach_modify(cur_change, &txn->tuplecids)
+	{
+		ReorderBufferChange *change =
+			dlist_container(ReorderBufferChange, node, cur_change.cur);
+		Assert(change->action_internal == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
+		ReorderBufferReturnChange(cache, change);
+	}
+
+	/* now remove reference from cache */
+	hash_search(cache->by_txn,
+	            (void *)&txn->xid,
+	            HASH_REMOVE,
+	            &found);
+	Assert(found);
+
+	ReorderBufferReturnTXN(cache, txn);
+}
+
+/*
+ * Build a hash with a (relfilenode, itempoint) -> (cmin, cmax) mapping for use
+ * by tqual.c's HeapTupleSatisfiesMVCCDuringDecoding.
+ */
+static void
+ReorderBufferBuildTupleCidHash(ReorderBuffer *cache, ReorderBufferTXN *txn)
+{
+	dlist_iter cur_change;
+	HASHCTL hash_ctl;
+
+	if (!txn->does_timetravel || dlist_is_empty(&txn->tuplecids))
+		return;
+
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+
+	hash_ctl.keysize = sizeof(ReorderBufferTupleCidKey);
+	hash_ctl.entrysize = sizeof(ReorderBufferTupleCidEnt);
+	hash_ctl.hash = tag_hash;
+	hash_ctl.hcxt = cache->context;
+
+	/*
+	 * create the hash with the exact number of to-be-stored tuplecids from the
+	 * start
+	 */
+	txn->tuplecid_hash =
+		hash_create("ReorderBufferTupleCid", txn->ntuplecids, &hash_ctl,
+					HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
+
+	dlist_foreach(cur_change, &txn->tuplecids)
+	{
+		ReorderBufferTupleCidKey key;
+		ReorderBufferTupleCidEnt *ent;
+		bool found;
+		ReorderBufferChange *change =
+			dlist_container(ReorderBufferChange, node, cur_change.cur);
+
+		Assert(change->action_internal == REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID);
+
+		/* be careful about padding */
+		memset(&key, 0, sizeof(ReorderBufferTupleCidKey));
+
+		key.relnode = change->tuplecid.node;
+
+		ItemPointerCopy(&change->tuplecid.tid,
+						&key.tid);
+
+		ent = (ReorderBufferTupleCidEnt *)
+			hash_search(txn->tuplecid_hash,
+						(void *)&key,
+						HASH_ENTER|HASH_FIND,
+						&found);
+		if (!found)
+		{
+			ent->cmin = change->tuplecid.cmin;
+			ent->cmax = change->tuplecid.cmax;
+			ent->combocid = change->tuplecid.combocid;
+		}
+		else
+		{
+			Assert(ent->cmin == change->tuplecid.cmin);
+			Assert(ent->cmax == InvalidCommandId ||
+				   ent->cmax == change->tuplecid.cmax);
+			/*
+			 * if the tuple got valid in this transaction and now got deleted
+			 * we already have a valid cmin stored. The cmax will be
+			 * InvalidCommandId though.
+			 */
+			ent->cmax = change->tuplecid.cmax;
+		}
+	}
+}
+
+/*
+ * Copy a provided snapshot so we can modify it privately. This is needed so
+ * that catalog modifying transactions can look into intermediate catalog
+ * states.
+ */
+static Snapshot
+ReorderBufferCopySnap(ReorderBuffer *cache, Snapshot orig_snap,
+					  ReorderBufferTXN *txn, CommandId cid)
+{
+	Snapshot snap;
+	dlist_iter sub_txn_i;
+	ReorderBufferTXN *sub_txn;
+	int i = 0;
+	Size size = sizeof(SnapshotData) +
+		sizeof(TransactionId) * orig_snap->xcnt +
+		sizeof(TransactionId) * (txn->nsubtxns + 1)
+		;
+
+	elog(DEBUG1, "copying a non-transaction-specific snapshot into timetravel tx %u", txn->xid);
+
+	/* we only want to start with snapshots as provided by snapbuild.c */
+	Assert(!orig_snap->subxip);
+	Assert(!orig_snap->copied);
+
+	snap = calloc(1, size);
+	memcpy(snap, orig_snap, sizeof(SnapshotData));
+
+	snap->copied = true;
+	snap->active_count = 0;
+	snap->regd_count = 0;
+	snap->xip = (TransactionId *) (snap + 1);
+
+	memcpy(snap->xip, orig_snap->xip, sizeof(TransactionId) * snap->xcnt);
+
+	/*
+	 * ->subxip contains all txids that belong to our transaction which we need
+	 * to check via cmin/cmax. Thats why we store the toplevel transaction in
+	 * there as well.
+	 */
+	snap->subxip = snap->xip + snap->xcnt;
+	snap->subxip[i++] = txn->xid;
+	snap->subxcnt = txn->nsubtxns + 1;
+
+	dlist_foreach(sub_txn_i, &txn->subtxns)
+	{
+		sub_txn = dlist_container(ReorderBufferTXN, node, sub_txn_i.cur);
+		snap->subxip[i++] = sub_txn->xid;
+	}
+
+	/* bsearch()ability */
+	qsort(snap->subxip, snap->subxcnt,
+		  sizeof(TransactionId), xidComparator);
+
+	/*
+	 * store the specified current CommandId
+	 */
+	snap->curcid = cid;
+
+	return snap;
+}
+
+/*
+ * Free a previously ReorderBufferCopySnap'ed snapshot
+ */
+static void
+ReorderBufferFreeSnap(ReorderBuffer *cache, Snapshot snap)
+{
+	Assert(snap->copied);
+	free(snap);
+}
+
+/*
+ * Commit a transaction and replay all actions that previously have been
+ * ReorderBufferAddChange'd in the toplevel TX or any of the subtransactions
+ * assigned via ReorderBufferCommitChild.
+ */
+void
+ReorderBufferCommit(ReorderBuffer *cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, false, NULL);
+	ReorderBufferIterTXNState *iterstate = NULL;
+	ReorderBufferChange *change;
+	CommandId command_id = FirstCommandId;
+	Snapshot snapshot_mvcc = NULL;
+	Snapshot snapshot_now = NULL;
+	bool snapshot_copied = false;
+
+	if (!txn)
+		return;
+
+	txn->lsn = lsn;
+
+	cache->begin(cache, txn);
+
+	PG_TRY();
+	{
+		ReorderBufferBuildTupleCidHash(cache, txn);
+
+		iterstate = ReorderBufferIterTXNInit(cache, txn);
+		while ((change = ReorderBufferIterTXNNext(cache, iterstate)))
+		{
+			switch ((enum ReorderBufferChangeTypeInternal)change->action_internal)
+			{
+				case REORDER_BUFFER_CHANGE_INTERNAL_INSERT:
+				case REORDER_BUFFER_CHANGE_INTERNAL_UPDATE:
+				case REORDER_BUFFER_CHANGE_INTERNAL_DELETE:
+					Assert(snapshot_mvcc != NULL);
+					if (!SnapBuildHasCatalogChanges(NULL, xid, &change->relnode))
+						cache->apply_change(cache, txn, change);
+					break;
+				case REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT:
+					/* XXX: we could skip snapshots in non toplevel txns */
+					if (snapshot_copied)
+					{
+						ReorderBufferFreeSnap(cache, snapshot_now);
+						snapshot_now = ReorderBufferCopySnap(cache, change->snapshot,
+															 txn, command_id);
+					}
+					else
+					{
+						snapshot_now = change->snapshot;
+					}
+
+					/*
+					 * the first snapshot seen in a transaction is its mvcc
+					 * snapshot
+					 */
+					if (!snapshot_mvcc)
+						snapshot_mvcc = snapshot_now;
+					else
+						RevertFromDecodingSnapshots();
+
+					SetupDecodingSnapshots(snapshot_now, txn->tuplecid_hash);
+					break;
+
+				case REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID:
+					if (!snapshot_copied && snapshot_now)
+					{
+						/* we don't use the global one anymore */
+						snapshot_copied = true;
+						snapshot_now = ReorderBufferCopySnap(cache, snapshot_now,
+														  txn, command_id);
+					}
+
+					command_id = Max(command_id, change->command_id);
+
+					if (snapshot_now && command_id != InvalidCommandId)
+					{
+						snapshot_now->curcid = command_id;
+
+						RevertFromDecodingSnapshots();
+						SetupDecodingSnapshots(snapshot_now, txn->tuplecid_hash);
+					}
+
+					/*
+					 * everytime the CommandId is incremented, we could see new
+					 * catalog contents
+					 */
+					ReorderBufferProcessInvalidations(cache, txn);
+
+					break;
+
+				case REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID:
+					elog(ERROR, "tuplecid value in normal queue");
+					break;
+			}
+		}
+
+		ReorderBufferIterTXNFinish(cache, iterstate);
+
+		cache->commit(cache, txn, lsn);
+
+		RevertFromDecodingSnapshots();
+		ReorderBufferProcessInvalidations(cache, txn);
+
+		ReorderBufferCleanupTXN(cache, txn);
+
+
+		if (snapshot_copied)
+		{
+			ReorderBufferFreeSnap(cache, snapshot_now);
+		}
+	}
+	PG_CATCH();
+	{
+		if (iterstate)
+			ReorderBufferIterTXNFinish(cache, iterstate);
+
+		/*
+		 * XXX: do we want to do this here?
+		 * ReorderBufferCleanupTXN(cache, txn);
+		 */
+
+		RevertFromDecodingSnapshots();
+		ReorderBufferProcessInvalidations(cache, txn);
+
+		if (snapshot_copied)
+		{
+			ReorderBufferFreeSnap(cache, snapshot_now);
+		}
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+}
+
+/*
+ * Abort a transaction that possibly has previous changes. Needs to be done
+ * independently for toplevel and subtransactions.
+ */
+void
+ReorderBufferAbort(ReorderBuffer *cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, false, NULL);
+
+	/* no changes in this commit */
+	if (!txn)
+		return;
+
+	ReorderBufferCleanupTXN(cache, txn);
+}
+
+/*
+ * Check whether a transaction is already known in this module
+ */
+bool
+ReorderBufferIsXidKnown(ReorderBuffer *cache, TransactionId xid)
+{
+	bool is_new;
+	/*
+	 * FIXME: for efficiency reasons we create the xid here, that doesn't seem
+	 * like a good idea though
+	 */
+	ReorderBufferTXNByXid(cache, xid, true, &is_new);
+
+	/* no changes in this commit */
+	return !is_new;
+}
+
+/*
+ * Add a new snapshot to this transaction which is the "base" of snapshots we
+ * modify if this is a catalog modifying transaction.
+ */
+void
+ReorderBufferAddBaseSnapshot(ReorderBuffer *cache, TransactionId xid,
+							 XLogRecPtr lsn, Snapshot snap)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+	ReorderBufferChange *change = ReorderBufferGetChange(cache);
+
+	change->snapshot = snap;
+	change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT;
+
+	if (lsn == InvalidXLogRecPtr)
+		txn->has_base_snapshot = true;
+
+	ReorderBufferAddChange(cache, xid, lsn, change);
+}
+
+/*
+ * Access the catalog with this CommandId at this point in the changestream.
+ */
+void
+ReorderBufferAddNewCommandId(ReorderBuffer *cache, TransactionId xid,
+							 XLogRecPtr lsn, CommandId cid)
+{
+	ReorderBufferChange *change = ReorderBufferGetChange(cache);
+
+	change->command_id = cid;
+	change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID;
+
+	ReorderBufferAddChange(cache, xid, lsn, change);
+}
+
+
+/*
+ * Add new (relfilenode, tid) -> (cmin, cmax) mappings.
+ */
+void
+ReorderBufferAddNewTupleCids(ReorderBuffer *cache, TransactionId xid, XLogRecPtr lsn,
+							 RelFileNode node, ItemPointerData tid,
+							 CommandId cmin, CommandId cmax, CommandId combocid)
+{
+	ReorderBufferChange *change = ReorderBufferGetChange(cache);
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+
+	change->tuplecid.node = node;
+	change->tuplecid.tid = tid;
+	change->tuplecid.cmin = cmin;
+	change->tuplecid.cmax = cmax;
+	change->tuplecid.combocid = combocid;
+	change->lsn = lsn;
+	change->action_internal = REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID;
+	dlist_push_tail(&txn->tuplecids, &change->node);
+	txn->ntuplecids++;
+}
+
+/*
+ * Setup the invalidation of the toplevel transaction.
+ *
+ * This needs to be done before ReorderBufferCommit is called!
+ */
+void
+ReorderBufferAddInvalidations(ReorderBuffer *cache, TransactionId xid, XLogRecPtr lsn,
+                                Size nmsgs, SharedInvalidationMessage* msgs)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+
+	if (txn->ninvalidations)
+		elog(ERROR, "only ever add one set of invalidations");
+	/* FIXME: free */
+	txn->invalidations = malloc(sizeof(SharedInvalidationMessage) * nmsgs);
+
+	if (!txn->invalidations)
+		elog(ERROR, "could not allocate memory for invalidations");
+
+	memcpy(txn->invalidations, msgs, sizeof(SharedInvalidationMessage) * nmsgs);
+	txn->ninvalidations = nmsgs;
+}
+
+/*
+ * Apply all invalidations we know. Possibly we only need parts at this point
+ * in the changestream but we don't know which those are.
+ */
+static void
+ReorderBufferProcessInvalidations(ReorderBuffer *cache, ReorderBufferTXN *txn)
+{
+	int i;
+	for (i = 0; i < txn->ninvalidations; i++)
+	{
+		LocalExecuteInvalidationMessage(&txn->invalidations[i]);
+	}
+}
+
+/*
+ * Mark a transaction as doing timetravel.
+ */
+void
+ReorderBufferXidSetTimetravel(ReorderBuffer *cache, TransactionId xid)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+	txn->does_timetravel = true;
+}
+
+/*
+ * Query whether a transaction is *known* to be doing timetravel. This can be
+ * wrong until directly before the commit!
+ */
+bool
+ReorderBufferXidDoesTimetravel(ReorderBuffer *cache, TransactionId xid)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+	return txn->does_timetravel;
+}
+
+/*
+ * Have we already added the first snapshot?
+ */
+bool
+ReorderBufferXidHasBaseSnapshot(ReorderBuffer *cache, TransactionId xid)
+{
+	ReorderBufferTXN *txn = ReorderBufferTXNByXid(cache, xid, true, NULL);
+	return txn->has_base_snapshot;
+}
+
+/*
+ * Visibility support routines
+ */
+
+/*-------------------------------------------------------------------------
+ * Lookup actual cmin/cmax values during timetravel access. We can't always
+ * rely on stored cmin/cmax values because of two scenarios:
+ *
+ * * A tuple got changed multiple times during a single transaction and thus
+ *   has got a combocid. Combocid's are only valid for the duration of a single
+ *   transaction.
+ * * A tuple with a cmin but no cmax (and thus no combocid) got deleted/updated
+ *   in another transaction than the one which created it which we are looking
+ *   at right now. As only one of cmin, cmax or combocid is actually stored in
+ *   the heap we don't have access to the the value we need anymore.
+ *
+ * To resolve those problems we have a per-transaction hash of (cmin, cmax)
+ * tuples keyed by (relfilenode, ctid) which contains the actual (cmin, cmax)
+ * values. That also takes care of combocids by simply not caring about them at
+ * all. As we have the real cmin/cmax values thats enough.
+ *
+ * As we only care about catalog tuples here the overhead of this hashtable
+ * should be acceptable.
+ * -------------------------------------------------------------------------
+ */
+extern bool
+ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data,
+							  HeapTuple htup, Buffer buffer,
+							  CommandId *cmin, CommandId *cmax)
+{
+	ReorderBufferTupleCidKey key;
+	ReorderBufferTupleCidEnt* ent;
+	ForkNumber forkno;
+	BlockNumber blockno;
+
+	/* be careful about padding */
+	memset(&key, 0, sizeof(key));
+
+	Assert(!BufferIsLocal(buffer));
+
+	/*
+	 * get relfilenode from the buffer, no convenient way to access it other
+	 * than that.
+	 */
+	BufferGetTag(buffer, &key.relnode, &forkno, &blockno);
+
+	/* tuples can only be in the main fork */
+	Assert(forkno == MAIN_FORKNUM);
+	Assert(blockno == ItemPointerGetBlockNumber(&htup->t_self));
+
+	ItemPointerCopy(&htup->t_self,
+					&key.tid);
+
+	ent = (ReorderBufferTupleCidEnt *)
+		hash_search(tuplecid_data,
+					(void *)&key,
+					HASH_FIND,
+					NULL);
+
+	if (ent == NULL)
+		return false;
+
+	if (cmin)
+		*cmin = ent->cmin;
+	if (cmax)
+		*cmax = ent->cmax;
+	return true;
+}
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
new file mode 100644
index 0000000..df24b33
--- /dev/null
+++ b/src/backend/replication/logical/snapbuild.c
@@ -0,0 +1,1144 @@
+/*-------------------------------------------------------------------------
+ *
+ * snapbuild.c
+ *
+ *	  Support for building timetravel snapshots based on the contents of the
+ *	  wal
+ *
+ * NOTES:
+ *
+ * We build snapshots which can *only* be used to read catalog contents by
+ * reading the wal stream. The aim is to provide mvcc and SnapshotNow snapshots
+ * that behave the same as their respective counterparts would have at the time
+ * the XLogRecord was generated. This is done to provide a reliable environment
+ * for decoding those records into every format that pleases the author of an
+ * output plugin.
+ *
+ * To build the snapshots we reuse the infrastructure built for hot
+ * standby. The snapshots we build look different than HS' because we have
+ * different needs. To successfully decode data from the WAL we only need to
+ * access catalogs/(sys|rel|cat)cache, not the actual user tables. And we need
+ * to build multiple, vastly different, ones, without being able to fully rely
+ * on the clog for information about committed transactions because they might
+ * commit in the future from the POV of the wal entry were currently decoding.
+
+ * As the percentage of transactions modifying the catalog normally is fairly
+ * small, instead of keeping track of all running transactions and treating
+ * everything inside (xmin, xmax) thats not known to be running as commited we
+ * do the contrary. That is we keep a list of transactions between
+ * snapshot->(xmin, xmax) that we consider committed, everything else is
+ * considered aborted/in progress.
+ * That also allows us not to care about subtransactions before they have
+ * committed.
+ *
+ * Classic SnapshotNow behaviour - which is mainly used for efficiency, not for
+ * correctness - is not actually required by any of the routines that we need
+ * during decoding and is hard to emulate fully. Instead we build snapshots
+ * with MVCC behaviour that are updated whenever another transaction commits.
+ *
+ * One additional complexity of doing this is that to handle mixed DDL/DML
+ * transactions we need Snapshots that see intermediate states in a
+ * transaction. In normal operation this is achieved by using
+ * CommandIds/cmin/cmax. The problem with this however is that for space
+ * efficiency reasons only one value of that is stored (cf. combocid.c). To get
+ * arround that we log additional information which allows us to get the
+ * original (cmin, cmax) pair during visibility checks.
+ *
+ * To facilitate all this we need our own visibility routine, as the normal
+ * ones are optimized for different usecases. We also need the code to use out
+ * special snapshots automatically whenever SnapshotNow behaviour is expected
+ * (specifying our snapshot everywhere would be far to invasive).
+ *
+ * To replace the normal SnapshotNows snapshots use the SetupDecodingSnapshots
+ * RevertFromDecodingSnapshots functions.
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/snapbuild.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/heapam_xlog.h"
+#include "access/rmgr.h"
+#include "access/transam.h"
+#include "access/xact.h"
+#include "access/xlogreader.h"
+
+#include "catalog/catalog.h"
+#include "catalog/pg_control.h"
+#include "catalog/pg_class.h"
+#include "catalog/pg_tablespace.h"
+
+#include "miscadmin.h"
+
+#include "replication/reorderbuffer.h"
+#include "replication/snapbuild.h"
+#include "replication/walsender_private.h"
+
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/relmapper.h"
+#include "utils/snapshot.h"
+#include "utils/syscache.h"
+#include "utils/tqual.h"
+
+#include "storage/block.h" /* debugging output */
+#include "storage/standby.h"
+#include "storage/sinval.h"
+
+/* transaction state manipulation functions */
+static void SnapBuildEndTxn(Snapstate *snapstate, TransactionId xid);
+
+static void SnapBuildAbortTxn(Snapstate *state, TransactionId xid, int nsubxacts,
+							  TransactionId *subxacts);
+
+static void SnapBuildCommitTxn(Snapstate *snapstate, ReorderBuffer *reorder,
+							   XLogRecPtr lsn, TransactionId xid,
+							   int nsubxacts, TransactionId *subxacts);
+
+/* ->running manipulation */
+static bool SnapBuildTxnIsRunning(Snapstate *snapstate, TransactionId xid);
+
+/* ->committed manipulation */
+static void SnapBuildPurgeCommittedTxn(Snapstate *snapstate);
+
+/* snapshot building/manipulation/distribution functions */
+/* XXX */
+static Snapshot SnapBuildBuildSnapshot(Snapstate *snapstate, TransactionId xid);
+
+static void	SnapBuildFreeSnapshot(Snapshot snap);
+
+static void SnapBuildSnapIncRefcount(Snapshot snap);
+
+static void SnapBuildDistributeSnapshotNow(Snapstate *snapstate, ReorderBuffer *reorder, XLogRecPtr lsn);
+
+/*
+ * Lookup a table via its current relfilenode.
+ *
+ * This requires that some snapshot in which that relfilenode is actually
+ * visible to be set up.
+ *
+ * The result of this function needs to be released from the syscache.
+ */
+HeapTuple
+LookupTableByRelFileNode(RelFileNode *relfilenode)
+{
+	Oid spc;
+	HeapTuple tuple;
+	Oid heaprel;
+
+	/*
+	 * relations in the default tablespace are stored with a reltablespace = 0
+	 * for some reason.
+	 */
+	spc = relfilenode->spcNode == DEFAULTTABLESPACE_OID ?
+		InvalidOid : relfilenode->spcNode;
+
+	tuple = SearchSysCache2(RELFILENODE,
+							spc,
+							relfilenode->relNode);
+
+	if (!HeapTupleIsValid(tuple))
+	{
+		if (relfilenode->spcNode == GLOBALTABLESPACE_OID)
+		{
+			heaprel = RelationMapFilenodeToOid(relfilenode->relNode, true);
+		}
+		else
+		{
+			heaprel = RelationMapFilenodeToOid(relfilenode->relNode, false);
+		}
+
+		if (heaprel != InvalidOid)
+		{
+			tuple = SearchSysCache1(RELOID,
+									heaprel);
+		}
+	}
+	return tuple;
+}
+
+/*
+ * Does this relation carry catalog information? Important for knowing whether
+ * a transaction made changes to the catalog, in which case it need to be
+ * included in snapshots.
+ *
+ * Requires that an appropriate timetravel snapshot is set up!
+ */
+bool
+SnapBuildHasCatalogChanges(Snapstate *snapstate, TransactionId xid, RelFileNode *relfilenode)
+{
+	HeapTuple table;
+	Form_pg_class class_form;
+	bool ret;
+
+	if (relfilenode->spcNode == GLOBALTABLESPACE_OID)
+		return true;
+
+
+	table = LookupTableByRelFileNode(relfilenode);
+
+	/*
+	 * tables in the default tablespace are stored in pg_class with 0 as their
+	 * reltablespace
+	 */
+	if (!HeapTupleIsValid(table))
+	{
+		elog(FATAL, "failed pg_class lookup for %u:%u",
+			 relfilenode->spcNode, relfilenode->relNode);
+		return false;
+	}
+
+	class_form = (Form_pg_class) GETSTRUCT(table);
+	ret = IsSystemClass(class_form);
+
+	ReleaseSysCache(table);
+	return ret;
+}
+
+/*
+ * Allocate a new snapshot builder.
+ */
+Snapstate *
+AllocateSnapshotBuilder(ReorderBuffer *reorder)
+{
+	Snapstate *snapstate = malloc(sizeof(Snapstate));
+
+	snapstate->state = SNAPBUILD_START;
+
+	snapstate->nrrunning = 0;
+	snapstate->nrrunning_initial = 0;
+	snapstate->running = NULL;
+
+	snapstate->nrcommitted = 0;
+	snapstate->nrcommitted_space = 128; /* arbitrary number */
+	snapstate->committed = malloc(snapstate->nrcommitted_space * sizeof(TransactionId));
+	snapstate->transactions_after = InvalidXLogRecPtr;
+
+	if (!snapstate->committed)
+		elog(ERROR, "could not allocate memory for snapstate->committed");
+
+	snapstate->snapshot = NULL;
+
+	return snapstate;
+}
+
+/*
+ * Freesnapshot builder.
+ */
+void
+FreeSnapshotBuilder(Snapstate *snapstate)
+{
+	if (snapstate->snapshot)
+		SnapBuildFreeSnapshot(snapstate->snapshot);
+
+	if (snapstate->committed)
+		free(snapstate->committed);
+
+	if (snapstate->running)
+		free(snapstate->running);
+
+	free(snapstate);
+}
+
+/*
+ * Free an unreferenced snapshot that has previously been built by us.
+ */
+static void
+SnapBuildFreeSnapshot(Snapshot snap)
+{
+	/* make sure we don't get passed an external snapshot */
+	Assert(snap->satisfies == HeapTupleSatisfiesMVCCDuringDecoding);
+
+	/* make sure nobody modified our snapshot */
+	Assert(snap->curcid == FirstCommandId);
+	Assert(!snap->suboverflowed);
+	Assert(!snap->takenDuringRecovery);
+	Assert(!snap->regd_count);
+
+	/* slightly more likely, so its checked even without casserts */
+	if (snap->copied)
+		elog(ERROR, "we don't deal with copied snapshots here.");
+
+	if (snap->active_count)
+		elog(ERROR, "freeing active snapshot");
+
+	free(snap);
+}
+
+/*
+ * Increase refcount of a snapshot.
+ *
+ * This is used when handing out a snapshot to some external resource or when
+ * adding a Snapshot as snapstate->snapshot.
+ */
+static void
+SnapBuildSnapIncRefcount(Snapshot snap)
+{
+	snap->active_count++;
+}
+
+/*
+ * Decrease refcount of a snapshot and free if the refcount reaches zero.
+ *
+ * Externally visible so external resources that have been handed an IncRef'ed
+ * Snapshot can free it easily.
+ */
+void
+SnapBuildSnapDecRefcount(Snapshot snap)
+{
+	/* make sure we don't get passed an external snapshot */
+	Assert(snap->satisfies == HeapTupleSatisfiesMVCCDuringDecoding);
+
+	/* make sure nobody modified our snapshot */
+	Assert(snap->curcid == FirstCommandId);
+	Assert(!snap->suboverflowed);
+	Assert(!snap->takenDuringRecovery);
+	Assert(!snap->regd_count);
+
+	Assert(snap->active_count);
+
+	/* slightly more likely, so its checked even without casserts */
+	if (snap->copied)
+		elog(ERROR, "we don't deal with copied snapshots here.");
+
+	snap->active_count--;
+	if (!snap->active_count)
+		SnapBuildFreeSnapshot(snap);
+}
+
+/*
+ * Build a new snapshot, based on currently committed, catalog modifying
+ * transactions.
+ *
+ * In-Progress transaction with catalog access are *not* allowed to modify
+ * these snapshots, they have to copy them and fill in appropriate ->curcid and
+ * ->subxip/subxcnt values.
+ */
+static Snapshot
+SnapBuildBuildSnapshot(Snapstate *snapstate, TransactionId xid)
+{
+	Snapshot snapshot = malloc(sizeof(SnapshotData) +
+							   sizeof(TransactionId) * snapstate->nrcommitted +
+							   sizeof(TransactionId) * 1 /* toplevel xid */);
+
+	snapshot->satisfies = HeapTupleSatisfiesMVCCDuringDecoding;
+	/*
+	 * we copy all currently in progress transaction to ->xip, all
+	 * transactions added to the transaction that committed during running -
+	 * which thus need to be considered visible in SnapshotNow semantics - get
+	 * copied to ->subxip.
+	 *
+	 * XXX: Do we want extra fields for those two instead?
+	 */
+	snapshot->xmin = snapstate->xmin;
+	snapshot->xmax = snapstate->xmax;
+
+	/* store all transaction to be treated as committed */
+	snapshot->xip = (TransactionId *) ((char *) snapshot + sizeof(SnapshotData));
+	snapshot->xcnt = snapstate->nrcommitted;
+	memcpy(snapshot->xip, snapstate->committed,
+	       snapstate->nrcommitted * sizeof(TransactionId));
+	/* sort so we can bsearch() */
+	qsort(snapshot->xip, snapshot->xcnt, sizeof(TransactionId), xidComparator);
+
+
+	snapshot->subxcnt = 0;
+	snapshot->subxip = NULL;
+
+	snapshot->suboverflowed = false;
+	snapshot->takenDuringRecovery = false;
+	snapshot->copied = false;
+	snapshot->curcid = FirstCommandId;
+	snapshot->active_count = 0;
+	snapshot->regd_count = 0;
+
+	return snapshot;
+}
+
+/*
+ * Handle the effects of a single heap change, appropriate to the current state
+ * of the snapshot builder.
+ */
+static SnapBuildAction
+SnapBuildProcessChange(ReorderBuffer *reorder, Snapstate *snapstate,
+					   TransactionId xid, XLogRecordBuffer *buf,
+					   RelFileNode *relfilenode)
+{
+	SnapBuildAction ret = SNAPBUILD_SKIP;
+
+	/*
+	 * We can't handle data in transactions if we haven't built a snapshot yet,
+	 * so don't store them.
+	 */
+	if (snapstate->state < SNAPBUILD_FULL_SNAPSHOT)
+		;
+	/*
+	 * No point in keeping track of changes in transactions that we don't have
+	 * enough information about to decode.
+	 */
+	else if (snapstate->state < SNAPBUILD_CONSISTENT &&
+			 SnapBuildTxnIsRunning(snapstate, xid))
+		;
+	else
+	{
+		bool old_tx = ReorderBufferIsXidKnown(reorder, xid);
+
+		ret = SNAPBUILD_DECODE;
+
+		if (!old_tx || !ReorderBufferXidHasBaseSnapshot(reorder, xid))
+		{
+			if (!snapstate->snapshot) {
+				snapstate->snapshot = SnapBuildBuildSnapshot(snapstate, xid);
+				/* refcount of the snapshot builder */
+				SnapBuildSnapIncRefcount(snapstate->snapshot);
+			}
+
+			/* refcount of the transaction */
+			SnapBuildSnapIncRefcount(snapstate->snapshot);
+			ReorderBufferAddBaseSnapshot(reorder, xid,
+									  InvalidXLogRecPtr,
+									  snapstate->snapshot);
+		}
+	}
+
+	return ret;
+}
+
+/*
+ * Process a single xlog record.
+ */
+SnapBuildAction
+SnapBuildDecodeCallback(ReorderBuffer *reorder, Snapstate *snapstate,
+						XLogRecordBuffer *buf)
+{
+	XLogRecord *r = &buf->record;
+	uint8 info = r->xl_info & ~XLR_INFO_MASK;
+	TransactionId xid = buf->record.xl_xid;
+
+	SnapBuildAction ret = SNAPBUILD_SKIP;
+
+#if DEBUG_ME_LOUDLY
+	{
+		StringInfoData s;
+
+		initStringInfo(&s);
+		RmgrTable[r->xl_rmid].rm_desc(&s,
+									  r->xl_info,
+									  buf->record_data);
+
+		/* don't bother emitting empty description */
+		if (s.len > 0)
+			elog(LOG, "xlog redo %u: %s", xid, s.data);
+	}
+#endif
+
+	/*
+	 * Only search for an initial starting point if we haven't build a full
+	 * snapshot yet
+	 */
+	if (snapstate->state <= SNAPBUILD_CONSISTENT)
+	{
+		/*
+		 * Build snapshot incrementally using information about the currently
+		 * running transactions. As soon as all of those have finished
+		 */
+		if (r->xl_rmid == RM_STANDBY_ID &&
+			info == XLOG_RUNNING_XACTS)
+		{
+			xl_running_xacts *running = (xl_running_xacts *) buf->record_data;
+
+			snapstate->state = SNAPBUILD_FULL_SNAPSHOT;
+
+			/*
+			 * inrease shared memory state, so vacuum can work
+			 * on tuples we prevent from being purged.
+			 */
+			IncreaseLogicalXminForSlot(buf->origptr, running->oldestRunningXid);
+
+			if (running->xcnt == 0)
+			{
+				/*
+				 * might have already started to incrementally assemble
+				 * transactions.
+				 */
+				snapstate->transactions_after = buf->origptr;
+
+				snapstate->xmin_running = InvalidTransactionId;
+				snapstate->xmax_running = InvalidTransactionId;
+
+				/*
+				 * FIXME: abort everything we have stored about running
+				 * transactions, relevant e.g. after a crash.
+				 */
+				snapstate->state = SNAPBUILD_CONSISTENT;
+			}
+			/* first encounter of a xl_running_xacts record */
+			else if (!snapstate->nrrunning)
+			{
+				/*
+				 * We only care about toplevel xids as those are the ones we
+				 * definitely see in the wal stream. As snapbuild.c tracks
+				 * committed instead of running transactions we don't need to
+				 * know anything about uncommitted subtransactions.
+				 */
+				snapstate->xmin = running->oldestRunningXid;
+				TransactionIdRetreat(snapstate->xmin);
+				snapstate->xmax = running->latestCompletedXid;
+				TransactionIdAdvance(snapstate->xmax);
+
+				snapstate->nrrunning = running->xcnt;
+				snapstate->nrrunning_initial = running->xcnt;
+
+				snapstate->running = malloc(snapstate->nrrunning
+											* sizeof(TransactionId));
+
+				memcpy(snapstate->running, running->xids,
+					   snapstate->nrrunning_initial * sizeof(TransactionId));
+
+				/* sort so we can do a binary search */
+				qsort(snapstate->running, snapstate->nrrunning_initial,
+					  sizeof(TransactionId), xidComparator);
+
+				snapstate->xmin_running = snapstate->running[0];
+				snapstate->xmax_running = snapstate->running[running->xcnt - 1];
+
+				/* makes comparisons cheaper later */
+				TransactionIdRetreat(snapstate->xmin_running);
+				TransactionIdAdvance(snapstate->xmax_running);
+
+				snapstate->state = SNAPBUILD_FULL_SNAPSHOT;
+			}
+
+			elog(LOG, "found initial snapshot (via running xacts). Done: %i",
+				 snapstate->state == SNAPBUILD_CONSISTENT);
+		}
+		else if (r->xl_rmid == RM_XLOG_ID &&
+				 (info == XLOG_CHECKPOINT_SHUTDOWN || info == XLOG_CHECKPOINT_ONLINE))
+		{
+			/* FIXME: Check whether there is a valid state dumped to disk */
+		}
+	}
+
+	if (snapstate->state == SNAPBUILD_START)
+		return SNAPBUILD_SKIP;
+
+	/*
+	 * This switch is - partially due to PGs indentation rules - rather deep
+	 * and large. Maybe break it into separate functions?
+	 */
+	switch (r->xl_rmid)
+	{
+		case RM_XLOG_ID:
+			{
+				switch (info)
+				{
+					case XLOG_CHECKPOINT_SHUTDOWN:
+#ifdef NOT_YET
+						{
+							/*
+							 * FIXME: abort everything but prepared xacts, we
+							 * don't track prepared xacts though so far.  It
+							 * might be neccesary to do this to handle subtxn
+							 * ids that haven't been assigned to a toplevel xid
+							 * after a crash.
+							 */
+							for ( /* FIXME */ )
+							{
+							}
+						}
+#endif
+					case XLOG_CHECKPOINT_ONLINE:
+						{
+							/*
+							 * FIXME: dump state to disk so we can restart
+							 * from here later
+							 */
+							break;
+						}
+				}
+				break;
+			}
+		case RM_STANDBY_ID:
+			{
+				switch (info)
+				{
+					case XLOG_RUNNING_XACTS:
+						{
+							xl_running_xacts *running =
+								(xl_running_xacts *) buf->record_data;
+
+							/*
+							 * update range of interesting xids. We don't
+							 * increase ->xmax because once we are in a
+							 * consistent state we can do that ourselves and
+							 * much more efficiently so because we only need to
+							 * do it for catalog transactions.
+							 */
+							snapstate->xmin = running->oldestRunningXid;
+							TransactionIdRetreat(snapstate->xmin);
+
+							/*
+							 * Remove transactions we don't need to keep track
+							 * off anymore.
+							 */
+							SnapBuildPurgeCommittedTxn(snapstate);
+
+							/*
+							 * inrease shared memory state, so vacuum can work
+							 * on tuples we prevent from being purged.
+							 */
+							IncreaseLogicalXminForSlot(buf->origptr, running->oldestRunningXid);
+
+							break;
+						}
+					case XLOG_STANDBY_LOCK:
+						break;
+				}
+				break;
+			}
+		case RM_XACT_ID:
+			{
+				switch (info)
+				{
+					case XLOG_XACT_COMMIT:
+						{
+							xl_xact_commit *xlrec =
+								(xl_xact_commit *) buf->record_data;
+
+							ret = SNAPBUILD_DECODE;
+
+							/*
+							 * Queue cache invalidation messages.
+							 */
+							if (xlrec->nmsgs)
+							{
+								TransactionId *subxacts;
+								SharedInvalidationMessage *inval_msgs;
+
+								/* subxid array follows relfilenodes */
+								subxacts = (TransactionId *)
+									&(xlrec->xnodes[xlrec->nrels]);
+								/* invalidation messages follow subxids */
+								inval_msgs = (SharedInvalidationMessage *)
+									&(subxacts[xlrec->nsubxacts]);
+
+								/*
+								 * no need to check
+								 * XactCompletionRelcacheInitFileInval, we will
+								 * process the sinval messages that the
+								 * relmapper change has generated.
+								 */
+								ReorderBufferAddInvalidations(reorder, xid,
+														   InvalidXLogRecPtr,
+								                           xlrec->nmsgs,
+														   inval_msgs);
+
+								/*
+								 * Let everyone know that this transaction
+								 * modified the catalog. We need this at commit
+								 * time.
+								 */
+								ReorderBufferXidSetTimetravel(reorder, xid);
+
+							}
+
+							SnapBuildCommitTxn(snapstate, reorder,
+											   buf->origptr, xid,
+											   xlrec->nsubxacts,
+											   (TransactionId *) &xlrec->xnodes);
+							break;
+						}
+					case XLOG_XACT_COMMIT_COMPACT:
+						{
+							xl_xact_commit_compact *xlrec =
+								(xl_xact_commit_compact *) buf->record_data;
+
+							ret = SNAPBUILD_DECODE;
+
+							SnapBuildCommitTxn(snapstate, reorder,
+											   buf->origptr, xid,
+											   xlrec->nsubxacts,
+											   xlrec->subxacts);
+							break;
+						}
+					case XLOG_XACT_COMMIT_PREPARED:
+						{
+							xl_xact_commit_prepared *xlrec =
+								(xl_xact_commit_prepared *) buf->record_data;
+
+							/* FIXME: check for invalidation messages! */
+
+							SnapBuildCommitTxn(snapstate, reorder,
+											   buf->origptr, xlrec->xid,
+											   xlrec->crec.nsubxacts,
+											   (TransactionId *) &xlrec->crec.xnodes);
+
+							ret = SNAPBUILD_DECODE;
+							break;
+						}
+					case XLOG_XACT_ABORT:
+						{
+							xl_xact_abort *xlrec =
+								(xl_xact_abort *) buf->record_data;
+
+							SnapBuildAbortTxn(snapstate, xid, xlrec->nsubxacts,
+											  (TransactionId *) &(xlrec->xnodes[xlrec->nrels]));
+							ret = SNAPBUILD_DECODE;
+							break;
+						}
+					case XLOG_XACT_ABORT_PREPARED:
+						{
+							xl_xact_abort_prepared *xlrec =
+								(xl_xact_abort_prepared *) buf->record_data;
+							xl_xact_abort *arec = &xlrec->arec;
+
+							SnapBuildAbortTxn(snapstate, xlrec->xid,
+											  arec->nsubxacts,
+											  (TransactionId *) &(arec->xnodes[arec->nrels]));
+							ret = SNAPBUILD_DECODE;
+							break;
+						}
+					case XLOG_XACT_ASSIGNMENT:
+						break;
+					case XLOG_XACT_PREPARE:
+						/*
+						 * XXX: We could take note of all in-progress prepared
+						 * xacts so we can use shutdown checkpoints to abort
+						 * in-progress transactions...
+						 */
+					default:
+						break;
+						;
+				}
+				break;
+			}
+		case RM_HEAP_ID:
+			{
+				switch (info & XLOG_HEAP_OPMASK)
+				{
+					case XLOG_HEAP_INPLACE:
+						{
+							xl_heap_inplace *xlrec =
+								(xl_heap_inplace *) buf->record_data;
+
+							ret = SnapBuildProcessChange(reorder, snapstate,
+														 xid, buf,
+														 &xlrec->target.node);
+
+							/*
+							 * inplace records happen in catalog modifying
+							 * txn's
+							 */
+							ReorderBufferXidSetTimetravel(reorder, xid);
+
+							break;
+						}
+					/*
+					 * we only ever read changes, so row level locks
+					 * aren't interesting
+					 */
+					case XLOG_HEAP_LOCK:
+						break;
+
+					case XLOG_HEAP_INSERT:
+						{
+							xl_heap_insert *xlrec =
+								(xl_heap_insert *) buf->record_data;
+
+							ret = SnapBuildProcessChange(reorder, snapstate,
+														 xid, buf,
+														 &xlrec->target.node);
+							break;
+						}
+					/* HEAP(_HOT)?_UPDATE use the same data layout */
+					case XLOG_HEAP_UPDATE:
+					case XLOG_HEAP_HOT_UPDATE:
+						{
+							xl_heap_update *xlrec =
+								(xl_heap_update *) buf->record_data;
+
+							ret = SnapBuildProcessChange(reorder, snapstate,
+														 xid, buf,
+														 &xlrec->target.node);
+							break;
+						}
+					case XLOG_HEAP_DELETE:
+						{
+							xl_heap_delete *xlrec =
+								(xl_heap_delete *) buf->record_data;
+
+							ret = SnapBuildProcessChange(reorder, snapstate,
+														 xid, buf,
+														 &xlrec->target.node);
+							break;
+						}
+					default:
+						;
+				}
+				break;
+			}
+		case RM_HEAP2_ID:
+			{
+				switch (info)
+				{
+					case XLOG_HEAP2_MULTI_INSERT:
+						{
+							xl_heap_multi_insert *xlrec =
+								(xl_heap_multi_insert *) buf->record_data;
+
+							ret = SnapBuildProcessChange(reorder, snapstate, xid,
+														 buf, &xlrec->node);
+							break;
+						}
+					case XLOG_HEAP2_NEW_CID:
+						{
+							CommandId cid;
+
+							xl_heap_new_cid *xlrec =
+								(xl_heap_new_cid *) buf->record_data;
+#if 0
+							elog(WARNING, "found new cid in xid %u: relfilenode %u/%u/%u: tid: (%u, %u) cmin: %u, cmax: %u, combo: %u",
+								 xlrec->top_xid,
+								 xlrec->target.node.dbNode, xlrec->target.node.spcNode,	xlrec->target.node.relNode,
+								 BlockIdGetBlockNumber(&xlrec->target.tid.ip_blkid), xlrec->target.tid.ip_posid,
+								 xlrec->cmin, xlrec->cmax, xlrec->combocid);
+#endif
+							/* we only log new_cid's if a catalog tuple was modified */
+							ReorderBufferXidSetTimetravel(reorder, xid);
+
+							if (!ReorderBufferXidHasBaseSnapshot(reorder, xid))
+							{
+								if (!snapstate->snapshot) {
+									snapstate->snapshot = SnapBuildBuildSnapshot(snapstate, xid);
+									/* refcount of the snapshot builder */
+									SnapBuildSnapIncRefcount(snapstate->snapshot);
+								}
+
+								/* refcount of the transaction */
+								SnapBuildSnapIncRefcount(snapstate->snapshot);
+
+								ReorderBufferAddBaseSnapshot(reorder, xid,
+														  InvalidXLogRecPtr,
+														  snapstate->snapshot);
+							}
+
+							ReorderBufferAddNewTupleCids(reorder, xlrec->top_xid, buf->origptr,
+													  xlrec->target.node, xlrec->target.tid,
+													  xlrec->cmin, xlrec->cmax, xlrec->combocid);
+
+							/* figure out new command id */
+							if (xlrec->cmin != InvalidCommandId && xlrec->cmax != InvalidCommandId)
+								cid = Max(xlrec->cmin, xlrec->cmax);
+							else if (xlrec->cmax != InvalidCommandId)
+								cid = xlrec->cmax;
+							else if (xlrec->cmin != InvalidCommandId)
+								cid = xlrec->cmin;
+							else
+							{
+								cid = InvalidCommandId; /* silence compiler */
+								elog(ERROR, "broken arrow, no cid?");
+							}
+							/*
+							 * FIXME: potential race condition here: if
+							 * multiple snapshots were running & generating
+							 * changes in the same transaction on the source
+							 * side this could be problematic.  But this cannot
+							 * happen for system catalogs, right?
+							 */
+							ReorderBufferAddNewCommandId(reorder, xid, buf->origptr,
+													  cid + 1);
+						}
+					default:
+						;
+				}
+			}
+			break;
+	}
+
+	return ret;
+}
+
+
+/*
+ * check whether `xid` is currently running
+ */
+static bool
+SnapBuildTxnIsRunning(Snapstate *snapstate, TransactionId xid)
+{
+	if (snapstate->nrrunning &&
+		NormalTransactionIdFollows(xid, snapstate->xmin_running) &&
+		NormalTransactionIdPrecedes(xid, snapstate->xmax_running))
+	{
+		TransactionId *search =
+			bsearch(&xid, snapstate->running, snapstate->nrrunning_initial,
+					sizeof(TransactionId), xidComparator);
+
+		if (search != NULL)
+		{
+			Assert(*search == xid);
+			return true;
+		}
+	}
+
+	return false;
+}
+
+/*
+ * FIXME: Analogous struct to the private one in reorderbuffer.c.
+ *
+ * Maybe introduce reorderbuffer_internal.h?
+ */
+typedef struct ReorderBufferTXNByIdEnt
+{
+	TransactionId xid;
+	ReorderBufferTXN *txn;
+}  ReorderBufferTXNByIdEnt;
+
+/*
+ * Add a new SnapshotNow to all transactions we're decoding that currently are
+ * in-progress so they can see new catalog contents made by the transaction
+ * that just committed.
+ */
+static void
+SnapBuildDistributeSnapshotNow(Snapstate *snapstate, ReorderBuffer *reorder, XLogRecPtr lsn)
+{
+	HASH_SEQ_STATUS status;
+	ReorderBufferTXNByIdEnt* ent;
+	elog(DEBUG1, "distributing snapshots to all running transactions");
+
+	hash_seq_init(&status, reorder->by_txn);
+
+	/*
+	 * FIXME: were providing snapshots the txn that committed just now...
+	 *
+	 * XXX: If we would handle XLOG_ASSIGNMENT records we could avoid handing
+	 * out snapshots to transactions that we recognize as being subtransactions.
+	 */
+	while ((ent = (ReorderBufferTXNByIdEnt*) hash_seq_search(&status)) != NULL)
+	{
+		if (ReorderBufferXidHasBaseSnapshot(reorder, ent->xid))
+		{
+			SnapBuildSnapIncRefcount(snapstate->snapshot);
+			ReorderBufferAddBaseSnapshot(reorder, ent->xid, lsn, snapstate->snapshot);
+		}
+	}
+}
+
+/*
+ * Keep track of a new catalog changing transaction that has committed
+ */
+static void
+SnapBuildAddCommittedTxn(Snapstate *snapstate, TransactionId xid)
+{
+	if (snapstate->nrcommitted == snapstate->nrcommitted_space)
+	{
+		elog(WARNING, "increasing space for committed transactions");
+
+		snapstate->nrcommitted_space *= 2;
+		snapstate->committed = realloc(snapstate->committed,
+									   snapstate->nrcommitted_space * sizeof(TransactionId));
+		if (!snapstate->committed)
+			elog(ERROR, "couldn't enlarge space for committed transactions");
+	}
+	snapstate->committed[snapstate->nrcommitted++] = xid;
+}
+
+/*
+ * Remove all transactions we treat as committed that are smaller than
+ * ->xmin. Those won't ever get checked via the ->commited array anyway.
+ */
+static void
+SnapBuildPurgeCommittedTxn(Snapstate *snapstate)
+{
+	int off;
+	TransactionId *workspace;
+	int surviving_xids = 0;
+
+	/* XXX: Neater algorithm? */
+	workspace = malloc(snapstate->nrcommitted * sizeof(TransactionId));
+
+	if (!workspace)
+		elog(ERROR, "could not allocate memory for workspace during xmin purging");
+
+	for (off = 0; off < snapstate->nrcommitted; off++)
+	{
+		if (snapstate->committed[off] > snapstate->xmin)
+			workspace[surviving_xids++] = snapstate->committed[off];
+	}
+
+	memcpy(snapstate->committed, workspace,
+		   surviving_xids * sizeof(TransactionId));
+
+	snapstate->nrcommitted = surviving_xids;
+	free(workspace);
+}
+
+/*
+ * Common logic for SnapBuildAbortTxn and SnapBuildCommitTxn dealing with
+ * keeping track of the amount of running transactions.
+ */
+static void
+SnapBuildEndTxn(Snapstate *snapstate, TransactionId xid)
+{
+	if (snapstate->state == SNAPBUILD_CONSISTENT)
+		return;
+
+	if (SnapBuildTxnIsRunning(snapstate, xid))
+	{
+		if (!--snapstate->nrrunning)
+		{
+			/*
+			 * none of the originally running transaction is running
+			 * anymore. Due to that our incrementaly built snapshot now is
+			 * complete.
+			 */
+			elog(LOG, "found consistent point due to SnapBuildEndTxn + running: %u", xid);
+			snapstate->state = SNAPBUILD_CONSISTENT;
+		}
+	}
+}
+
+/* Abort a transaction, throw away all state we kept */
+static void
+SnapBuildAbortTxn(Snapstate *snapstate, TransactionId xid, int nsubxacts, TransactionId *subxacts)
+{
+	int i;
+
+	for (i = 0; i < nsubxacts; i++)
+	{
+		TransactionId subxid = subxacts[i];
+		SnapBuildEndTxn(snapstate, subxid);
+	}
+
+	SnapBuildEndTxn(snapstate, xid);
+}
+
+/* Handle everything that needs to be done when a transaction commits */
+static void
+SnapBuildCommitTxn(Snapstate *snapstate, ReorderBuffer *reorder,
+				   XLogRecPtr lsn, TransactionId xid,
+				   int nsubxacts, TransactionId *subxacts)
+{
+	int nxact;
+
+	bool forced_timetravel = false;
+	bool sub_does_timetravel = false;
+	bool top_does_timetravel = false;
+
+	/*
+	 * If we couldn't observe every change of a transaction because it was
+	 * already running at the point we started to observe we have to assume it
+	 * made catalog changes.
+	 *
+	 * This has the positive benefit that we afterwards have enough information
+	 * to build an exportable snapshot thats usable by pg_dump et al.
+	 */
+	if (snapstate->state < SNAPBUILD_CONSISTENT)
+	{
+		if (XLByteLT(snapstate->transactions_after, lsn))
+			snapstate->transactions_after = lsn;
+
+		if (SnapBuildTxnIsRunning(snapstate, xid))
+		{
+			elog(LOG, "forced to assume catalog changes for xid %u because it was running to early", xid);
+			SnapBuildAddCommittedTxn(snapstate, xid);
+			forced_timetravel = true;
+		}
+	}
+
+	for (nxact = 0; nxact < nsubxacts; nxact++)
+	{
+		TransactionId subxid = subxacts[nxact];
+
+		/*
+		 * make sure txn is not tracked in running txn's anymore, switch
+		 * state
+		 */
+		SnapBuildEndTxn(snapstate, subxid);
+
+		if (forced_timetravel)
+		{
+			SnapBuildAddCommittedTxn(snapstate, subxid);
+		}
+		/*
+		 * add subtransaction to base snapshot, we don't distinguish after
+		 * that
+		 */
+		else if (ReorderBufferXidDoesTimetravel(reorder, subxid))
+		{
+			sub_does_timetravel = true;
+
+			elog(DEBUG1, "found subtransaction %u:%u with catalog changes.",
+				 xid, subxid);
+
+			SnapBuildAddCommittedTxn(snapstate, subxid);
+		}
+
+
+		if (forced_timetravel && sub_does_timetravel &&
+			NormalTransactionIdFollows(subxid, snapstate->xmax))
+		{
+			snapstate->xmax = subxid;
+			TransactionIdAdvance(snapstate->xmax);
+		}
+	}
+
+	/*
+	 * make sure txn is not tracked in running txn's anymore, switch state
+	 */
+	SnapBuildEndTxn(snapstate, xid);
+
+	if (forced_timetravel)
+	{
+		elog(DEBUG1, "forced transaction %u to do timetravel.", xid);
+
+		SnapBuildAddCommittedTxn(snapstate, xid);
+	}
+	/* add toplevel transaction to base snapshot */
+	else if (ReorderBufferXidDoesTimetravel(reorder, xid))
+	{
+		elog(DEBUG1, "found top level transaction %u, with catalog changes!", xid);
+
+		top_does_timetravel = true;
+		SnapBuildAddCommittedTxn(snapstate, xid);
+	}
+	else if (sub_does_timetravel)
+	{
+		/* mark toplevel txn as timetravel as well */
+		SnapBuildAddCommittedTxn(snapstate, xid);
+	}
+
+	if (forced_timetravel || top_does_timetravel || sub_does_timetravel)
+	{
+		if (!TransactionIdIsValid(snapstate->xmax) ||
+			NormalTransactionIdFollows(xid, snapstate->xmax))
+		{
+			snapstate->xmax = xid;
+			TransactionIdAdvance(snapstate->xmax);
+		}
+
+		if (snapstate->state < SNAPBUILD_FULL_SNAPSHOT)
+			return;
+
+		if (snapstate->snapshot) {
+			/* refcount of the transaction */
+			SnapBuildSnapDecRefcount(snapstate->snapshot);
+		}
+
+		snapstate->snapshot = SnapBuildBuildSnapshot(snapstate, xid);
+
+		/* refcount of the snapshot builder */
+		SnapBuildSnapIncRefcount(snapstate->snapshot);
+
+		/* add a new SnapshotNow to all currently running transactions */
+		SnapBuildDistributeSnapshotNow(snapstate, reorder, lsn);
+	}
+}
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index b6cfdac..07f1cdb 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -76,9 +76,11 @@ Node *replication_parse_result;
 %token K_NOWAIT
 %token K_WAL
 %token K_START_REPLICATION
+%token K_INIT_LOGICAL_REPLICATION
+%token K_START_LOGICAL_REPLICATION
 
 %type <node>	command
-%type <node>	base_backup start_replication identify_system
+%type <node>	base_backup start_replication identify_system start_logical_replication init_logical_replication
 %type <list>	base_backup_opt_list
 %type <defelt>	base_backup_opt
 %%
@@ -97,6 +99,8 @@ command:
 			identify_system
 			| base_backup
 			| start_replication
+			| init_logical_replication
+			| start_logical_replication
 			;
 
 /*
@@ -166,6 +170,32 @@ start_replication:
 					$$ = (Node *) cmd;
 				}
 			;
+
+/* FIXME: don't use SCONST */
+init_logical_replication:
+			K_INIT_LOGICAL_REPLICATION SCONST
+				{
+					InitLogicalReplicationCmd *cmd;
+					cmd = makeNode(InitLogicalReplicationCmd);
+					cmd->plugin = $2;
+
+					$$ = (Node *) cmd;
+				}
+			;
+
+/* FIXME: don't use SCONST */
+start_logical_replication:
+			K_START_LOGICAL_REPLICATION SCONST RECPTR
+				{
+					StartLogicalReplicationCmd *cmd;
+					cmd = makeNode(StartLogicalReplicationCmd);
+					cmd->name = $2;
+					cmd->startpoint = $3;
+
+					$$ = (Node *) cmd;
+				}
+			;
+
 %%
 
 #include "repl_scanner.c"
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 51f381d..58f7972 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -64,6 +64,8 @@ NOWAIT			{ return K_NOWAIT; }
 PROGRESS			{ return K_PROGRESS; }
 WAL			{ return K_WAL; }
 START_REPLICATION	{ return K_START_REPLICATION; }
+INIT_LOGICAL_REPLICATION	{ return K_INIT_LOGICAL_REPLICATION; }
+START_LOGICAL_REPLICATION	{ return K_START_LOGICAL_REPLICATION; }
 ","				{ return ','; }
 ";"				{ return ';'; }
 
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 6452c34..4713847 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -52,6 +52,7 @@
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
 #include "replication/walsender_private.h"
+#include "replication/logicalfuncs.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
 #include "storage/pmsignal.h"
@@ -83,6 +84,9 @@ WalSndCtlData *WalSndCtl = NULL;
 /* My slot in the shared memory array */
 WalSnd	   *MyWalSnd = NULL;
 
+/* My slot for logical rep in the shared memory array */
+LogicalWalSnd *MyLogicalWalSnd = NULL;
+
 /* Global state */
 bool		am_walsender = false;		/* Am I a walsender process ? */
 bool		am_cascading_walsender = false;		/* Am I cascading WAL to
@@ -92,6 +96,7 @@ static bool	replication_started = false; /* Started streaming yet? */
 
 /* User-settable parameters for walsender */
 int			max_wal_senders = 0;	/* the maximum number of concurrent walsenders */
+int			max_logical_slots = 0;	/* the maximum number of logical slots */
 int			wal_sender_timeout = 60 * 1000;	/* maximum time to send one
 												 * WAL data message */
 /*
@@ -129,18 +134,30 @@ static bool	ping_sent = false;
 static volatile sig_atomic_t got_SIGHUP = false;
 volatile sig_atomic_t walsender_ready_to_stop = false;
 
+/* XXX reader */
+static MemoryContext decoding_ctx = NULL;
+static MemoryContext old_decoding_ctx = NULL;
+
+static XLogReaderState *logical_reader = NULL;
+
+
 /* Signal handlers */
 static void WalSndSigHupHandler(SIGNAL_ARGS);
 static void WalSndXLogSendHandler(SIGNAL_ARGS);
 static void WalSndLastCycleHandler(SIGNAL_ARGS);
 
 /* Prototypes for private functions */
-static void WalSndLoop(void) __attribute__((noreturn));
+typedef void (*WalSndSendData)(bool *);
+static void WalSndLoop(WalSndSendData send_data) __attribute__((noreturn));
 static void InitWalSenderSlot(void);
 static void WalSndKill(int code, Datum arg);
-static void XLogSend(bool *caughtup);
+static void XLogSendPhysical(bool *caughtup);
+static void XLogSendLogical(bool *caughtup);
 static void IdentifySystem(void);
 static void StartReplication(StartReplicationCmd *cmd);
+static void InitLogicalReplication(InitLogicalReplicationCmd *cmd);
+static void StartLogicalReplication(StartLogicalReplicationCmd *cmd);
+static void ComputeLogicalXmin(void);
 static void ProcessStandbyMessage(void);
 static void ProcessStandbyReplyMessage(void);
 static void ProcessStandbyHSFeedbackMessage(void);
@@ -192,6 +209,58 @@ WalSndErrorCleanup()
 		proc_exit(0);
 }
 
+extern void
+IncreaseLogicalXminForSlot(XLogRecPtr lsn, TransactionId xmin)
+{
+	Assert(MyLogicalWalSnd != NULL);
+
+	SpinLockAcquire(&MyLogicalWalSnd->mutex);
+	/*
+	 * Only increase if the previous value has been applied...
+	 */
+	if (!TransactionIdIsValid(MyLogicalWalSnd->candidate_xmin))
+	{
+		MyLogicalWalSnd->candidate_xmin_after = lsn;
+		MyLogicalWalSnd->candidate_xmin = xmin;
+		elog(LOG, "got new xmin %u at %lu", xmin, lsn);
+	}
+	SpinLockRelease(&MyLogicalWalSnd->mutex);
+}
+
+static void
+ComputeLogicalXmin(void)
+{
+	int slot;
+	TransactionId xmin = InvalidTransactionId;
+	LogicalWalSnd *logical_base;
+	LogicalWalSnd *walsnd;
+
+	logical_base = (LogicalWalSnd*)&WalSndCtl->walsnds[max_wal_senders];
+
+	LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+
+	for (slot = 0; slot < max_logical_slots; slot++)
+	{
+		walsnd = &logical_base[slot];
+
+		SpinLockAcquire(&walsnd->mutex);
+		if (walsnd->in_use &&
+			TransactionIdIsValid(walsnd->xmin) && (
+				!TransactionIdIsValid(xmin) ||
+				TransactionIdPrecedes(walsnd->xmin, xmin))
+			)
+		{
+			xmin = walsnd->xmin;
+		}
+		SpinLockRelease(&walsnd->mutex);
+	}
+	WalSndCtl->logical_xmin = xmin;
+	LWLockRelease(ProcArrayLock);
+
+	elog(LOG, "computed new xmin: %u", xmin);
+
+}
+
 /*
  * IDENTIFY_SYSTEM
  */
@@ -376,7 +445,362 @@ StartReplication(StartReplicationCmd *cmd)
 	SyncRepInitConfig();
 
 	/* Main loop of walsender */
-	WalSndLoop();
+	WalSndLoop(XLogSendPhysical);
+}
+
+static void
+InitLogicalReplication(InitLogicalReplicationCmd *cmd)
+{
+	int slot;
+	LogicalWalSnd *logical_base;
+	LogicalWalSnd *walsnd;
+	char *slot_name;
+	StringInfoData buf;
+	char		xpos[MAXFNAMELEN];
+
+	elog(WARNING, "Initiating logical rep");
+
+	logical_base = (LogicalWalSnd*)&WalSndCtl->walsnds[max_wal_senders];
+
+	Assert(!MyLogicalWalSnd);
+
+	for (slot = 0; slot < max_logical_slots; slot++)
+	{
+		walsnd = &logical_base[slot];
+
+		SpinLockAcquire(&walsnd->mutex);
+		if (!walsnd->in_use)
+		{
+			Assert(!walsnd->active);
+			/* NOT releasing the lock yet */
+			break;
+		}
+		SpinLockRelease(&walsnd->mutex);
+		walsnd = NULL;
+	}
+
+	if (!walsnd)
+	{
+		elog(ERROR, "couldn't find free logical slot. free one or increase max_logical_slots");
+	}
+
+	/* so we get reset on exit/failure */
+	MyLogicalWalSnd = walsnd;
+	MyLogicalWalSnd->last_required_checkpoint = GetRedoRecPtr();
+
+	walsnd->in_use = true;
+	/* mark slot as active till we build the base snapshot */
+	walsnd->active = true;
+	walsnd->xmin = InvalidTransactionId;
+
+	strcpy(NameStr(walsnd->plugin), cmd->plugin);
+
+	/* FIXME: permanent name allocation scheme */
+	slot_name = NameStr(walsnd->name);
+	sprintf(slot_name, "id-%d", slot);
+
+	/* release spinlock, so this slot can be examined  */
+	SpinLockRelease(&walsnd->mutex);
+
+	/*
+	 * FIXME: think about race conditions here...
+	 *
+	 * We need to do this *after* releasing the spinlock, otherwise
+	 * GetOldestXmin will deadlock with ourselves.
+	 */
+	walsnd->xmin = GetOldestXmin(true, true);
+	ComputeLogicalXmin();
+
+	decoding_ctx = AllocSetContextCreate(TopMemoryContext,
+										 "decoding context",
+										 ALLOCSET_DEFAULT_MINSIZE,
+										 ALLOCSET_DEFAULT_INITSIZE,
+										 ALLOCSET_DEFAULT_MAXSIZE);
+	old_decoding_ctx = MemoryContextSwitchTo(decoding_ctx);
+	TopTransactionContext = decoding_ctx;
+
+	logical_reader = initial_snapshot_reader();
+
+	logical_reader->startptr = MyLogicalWalSnd->last_required_checkpoint;
+
+	for (;;)
+	{
+		ResetLatch(&MyWalSnd->latch);
+
+		/*
+		 * Emergency bailout if postmaster has died.  This is to avoid the
+		 * necessity for manual cleanup of all postmaster children.
+		 */
+		if (!PostmasterIsAlive())
+			exit(1);
+
+		/* Process any requests or signals received recently */
+		if (got_SIGHUP)
+		{
+			got_SIGHUP = false;
+			ProcessConfigFile(PGC_SIGHUP);
+		}
+
+		logical_reader->endptr = GetFlushRecPtr();
+
+		/* continue building initial snapshot */
+		XLogReaderRead(logical_reader);
+
+		if (logical_reader->needs_input || !initial_snapshot_ready(logical_reader))
+		{
+			long		sleeptime = 10000;		/* 10 s */
+			int			wakeEvents;
+
+			wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH | WL_TIMEOUT;
+
+			/* Sleep until something happens or we time out */
+			ImmediateInterruptOK = true;
+			CHECK_FOR_INTERRUPTS();
+			WaitLatch(&MyWalSnd->latch, wakeEvents,
+					  sleeptime);
+			ImmediateInterruptOK = false;
+		}
+		else
+			break;
+	}
+
+	walsnd->confirmed_flush = logical_reader->curptr;
+
+	snprintf(xpos, sizeof(xpos), "%X/%X",
+			 (uint32) (logical_reader->curptr >> 32), (uint32) logical_reader->curptr);
+
+	pq_beginmessage(&buf, 'T');
+	pq_sendint(&buf, 4, 2);		/* 4 fields */
+
+	/* first field */
+	pq_sendstring(&buf, "replication_id");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	pq_sendstring(&buf, "consistent_point");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	pq_sendstring(&buf, "snapshot_name");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	pq_sendstring(&buf, "plugin");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	pq_endmessage(&buf);
+
+	/* Send a DataRow message */
+	pq_beginmessage(&buf, 'D');
+	pq_sendint(&buf, 4, 2);		/* # of columns */
+
+	pq_sendint(&buf, strlen(slot_name), 4); /* col1 len */
+	pq_sendbytes(&buf, slot_name, strlen(slot_name));
+
+	pq_sendint(&buf, strlen(xpos), 4); /* col2 len */
+	pq_sendbytes(&buf, xpos, strlen(xpos));
+
+	pq_sendint(&buf, strlen("0xDEADBEEF"), 4); /* col3 len */
+	pq_sendbytes(&buf, "0xDEADBEEF", strlen("0xDEADBEEF"));
+
+	pq_sendint(&buf, strlen(cmd->plugin), 4); /* col4 len */
+	pq_sendbytes(&buf, cmd->plugin, strlen(cmd->plugin));
+
+	pq_endmessage(&buf);
+
+	SpinLockAcquire(&walsnd->mutex);
+	walsnd->active = false;
+	MyLogicalWalSnd = NULL;
+	SpinLockRelease(&walsnd->mutex);
+
+	MemoryContextSwitchTo(old_decoding_ctx);
+	TopTransactionContext = NULL;
+}
+
+
+
+static void
+StartLogicalReplication(StartLogicalReplicationCmd *cmd)
+{
+	StringInfoData buf;
+
+	int slot;
+	LogicalWalSnd *logical_base;
+	logical_base = (LogicalWalSnd*)&WalSndCtl->walsnds[max_wal_senders];
+
+	elog(WARNING, "Starting logical replication");
+
+	Assert(!MyLogicalWalSnd);
+
+	for (slot = 0; slot < max_logical_slots; slot++)
+	{
+		LogicalWalSnd   *walsnd = &logical_base[slot];
+
+		SpinLockAcquire(&walsnd->mutex);
+		if (walsnd->in_use && strcmp(cmd->name, NameStr(walsnd->name)) == 0)
+		{
+			MyLogicalWalSnd = walsnd;
+			/* NOT releasing the lock yet */
+			break;
+		}
+		SpinLockRelease(&walsnd->mutex);
+	}
+
+	if (!MyLogicalWalSnd)
+		elog(ERROR, "couldn't find logical slot for \"%s\"",
+		     cmd->name);
+
+	if (MyLogicalWalSnd->active)
+	{
+		SpinLockRelease(&MyLogicalWalSnd->mutex);
+		elog(ERROR, "slot already active");
+	}
+
+	MyLogicalWalSnd->active = true;
+	SpinLockRelease(&MyLogicalWalSnd->mutex);
+
+	MarkPostmasterChildWalSender();
+	SendPostmasterSignal(PMSIGNAL_ADVANCE_STATE_MACHINE);
+	replication_started = true;
+
+	if (am_cascading_walsender && !RecoveryInProgress())
+	{
+		ereport(LOG,
+		   (errmsg("terminating walsender process to force cascaded standby "
+				   "to update timeline and reconnect")));
+		walsender_ready_to_stop = true;
+	}
+
+	WalSndSetState(WALSNDSTATE_CATCHUP);
+
+	/* Send a CopyBothResponse message, and start streaming */
+	pq_beginmessage(&buf, 'W');
+	pq_sendbyte(&buf, 0);
+	pq_sendint(&buf, 0, 2);
+	pq_endmessage(&buf);
+	pq_flush();
+
+	/*
+	 * Initialize position to the received one, then the xlog records begin to
+	 * be shipped from that position
+	 */
+	sentPtr = MyLogicalWalSnd->last_required_checkpoint;
+
+	/* Also update the start position status in shared memory */
+	{
+		/* use volatile pointer to prevent code rearrangement */
+		volatile WalSnd *walsnd = MyWalSnd;
+
+		SpinLockAcquire(&walsnd->mutex);
+		walsnd->sentPtr = MyLogicalWalSnd->last_required_checkpoint;
+		SpinLockRelease(&walsnd->mutex);
+	}
+
+	SyncRepInitConfig();
+
+	logical_reader = normal_snapshot_reader(NameStr(MyLogicalWalSnd->plugin),
+											cmd->startpoint);
+
+	logical_reader->startptr = MyLogicalWalSnd->last_required_checkpoint;
+	logical_reader->curptr = logical_reader->startptr;
+	logical_reader->endptr = GetFlushRecPtr();
+
+	/* Main loop of walsender */
+	WalSndLoop(XLogSendLogical);
+}
+
+/*
+ * Prepare a write into a StringInfo.
+ *
+ * Don't do anything lasting in here, its quite possible that nothing will done
+ * with the data.
+ */
+void
+WalSndPrepareWrite(StringInfo out, XLogRecPtr lsn)
+{
+	pq_sendbyte(out, 'w');
+	pq_sendint64(out, lsn);	/* dataStart */
+	pq_sendint64(out, lsn);	/* walEnd */
+	/* XXX: gather that value later just as its done in XLogSendPhysical */
+	pq_sendint64(out, 0 /*GetCurrentIntegerTimestamp() */);/* sendtime */
+}
+
+/*
+ * Actually write out data previously prepared by WalSndPrepareWrite out to the
+ * network, take as long as needed but process replies from the other side
+ * during that.
+ */
+void
+WalSndWriteData(StringInfo data)
+{
+	long		sleeptime = 10000;		/* 10 s */
+	int			wakeEvents;
+
+	pq_putmessage_noblock('d', data->data, data->len);
+
+	for (;;)
+	{
+		if (!pq_is_send_pending())
+			return;
+
+		/*
+		 * Emergency bailout if postmaster has died.  This is to avoid the
+		 * necessity for manual cleanup of all postmaster children.
+		 */
+		if (!PostmasterIsAlive())
+			exit(1);
+
+		/* Process any requests or signals received recently */
+		if (got_SIGHUP)
+		{
+			got_SIGHUP = false;
+			ProcessConfigFile(PGC_SIGHUP);
+			SyncRepInitConfig();
+		}
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* Check for input from the client */
+		ProcessRepliesIfAny();
+
+		/* Clear any already-pending wakeups */
+		ResetLatch(&MyWalSnd->latch);
+
+		/* Try to flush pending output to the client */
+		if (pq_flush_if_writable() != 0)
+			break;
+
+		if (!pq_is_send_pending())
+			return;
+
+		/* FIXME: wal_sender_timeout integration */
+		wakeEvents = WL_LATCH_SET | WL_POSTMASTER_DEATH |
+			WL_SOCKET_WRITEABLE | WL_SOCKET_READABLE | WL_TIMEOUT;
+
+		ImmediateInterruptOK = true;
+		CHECK_FOR_INTERRUPTS();
+		WaitLatchOrSocket(&MyWalSnd->latch, wakeEvents,
+						  MyProcPort->sock, sleeptime);
+		ImmediateInterruptOK = false;
+	}
+	SetLatch(&MyWalSnd->latch);
 }
 
 /*
@@ -421,6 +845,14 @@ exec_replication_command(const char *cmd_string)
 			StartReplication((StartReplicationCmd *) cmd_node);
 			break;
 
+		case T_InitLogicalReplicationCmd:
+			InitLogicalReplication((InitLogicalReplicationCmd *) cmd_node);
+			break;
+
+		case T_StartLogicalReplicationCmd:
+			StartLogicalReplication((StartLogicalReplicationCmd *) cmd_node);
+			break;
+
 		case T_BaseBackupCmd:
 			SendBaseBackup((BaseBackupCmd *) cmd_node);
 			break;
@@ -588,6 +1020,37 @@ ProcessStandbyReplyMessage(void)
 		SpinLockRelease(&walsnd->mutex);
 	}
 
+	/*
+	 * Do an unlocked check for candidate_xmin first.
+	 */
+	if (MyLogicalWalSnd &&
+		TransactionIdIsValid(MyLogicalWalSnd->candidate_xmin))
+	{
+		bool updated_xmin = false;
+
+		/* use volatile pointer to prevent code rearrangement */
+		volatile LogicalWalSnd *walsnd = MyLogicalWalSnd;
+
+		SpinLockAcquire(&walsnd->mutex);
+
+		/* if were past the location required for bumping xmin, do so */
+		if (TransactionIdIsValid(walsnd->candidate_xmin) &&
+			flushPtr != InvalidXLogRecPtr &&
+			XLByteLE(walsnd->candidate_xmin_after, flushPtr)
+			)
+		{
+			walsnd->xmin = walsnd->candidate_xmin;
+			walsnd->candidate_xmin = InvalidTransactionId;
+			walsnd->candidate_xmin_after = InvalidXLogRecPtr;
+			updated_xmin = true;
+		}
+
+		SpinLockRelease(&walsnd->mutex);
+
+		if (updated_xmin)
+			ComputeLogicalXmin();
+	}
+
 	if (!am_cascading_walsender)
 		SyncRepReleaseWaiters();
 }
@@ -669,7 +1132,7 @@ ProcessStandbyHSFeedbackMessage(void)
 
 /* Main loop of walsender process that streams the WAL over Copy messages. */
 static void
-WalSndLoop(void)
+WalSndLoop(WalSndSendData send_data)
 {
 	bool		caughtup = false;
 
@@ -713,12 +1176,12 @@ WalSndLoop(void)
 
 		/*
 		 * If we don't have any pending data in the output buffer, try to send
-		 * some more.  If there is some, we don't bother to call XLogSend
+		 * some more.  If there is some, we don't bother to call send_data
 		 * again until we've flushed it ... but we'd better assume we are not
 		 * caught up.
 		 */
 		if (!pq_is_send_pending())
-			XLogSend(&caughtup);
+			send_data(&caughtup);
 		else
 			caughtup = false;
 
@@ -754,7 +1217,7 @@ WalSndLoop(void)
 			if (walsender_ready_to_stop)
 			{
 				/* ... let's just be real sure we're caught up ... */
-				XLogSend(&caughtup);
+				send_data(&caughtup);
 				if (caughtup && !pq_is_send_pending())
 				{
 					/* Inform the standby that XLOG streaming is done */
@@ -769,7 +1232,7 @@ WalSndLoop(void)
 		/*
 		 * We don't block if not caught up, unless there is unsent data
 		 * pending in which case we'd better block until the socket is
-		 * write-ready.  This test is only needed for the case where XLogSend
+		 * write-ready.  This test is only needed for the case where send_data
 		 * loaded a subset of the available data but then pq_flush_if_writable
 		 * flushed it all --- we should immediately try to send more.
 		 */
@@ -917,6 +1380,13 @@ WalSndKill(int code, Datum arg)
 	 * for this.
 	 */
 	MyWalSnd->pid = 0;
+
+	/* LOCK? */
+	if(MyLogicalWalSnd && MyLogicalWalSnd->active)
+	{
+		MyLogicalWalSnd->active = false;
+	}
+
 	DisownLatch(&MyWalSnd->latch);
 
 	/* WalSnd struct isn't mine anymore */
@@ -1068,6 +1538,8 @@ retry:
 }
 
 /*
+ * Send out the WAL in its normal physical/stored form.
+ *
  * Read up to MAX_SEND_SIZE bytes of WAL that's been flushed to disk,
  * but not yet sent to the client, and buffer it in the libpq output
  * buffer.
@@ -1076,7 +1548,7 @@ retry:
  * *caughtup is set to false.
  */
 static void
-XLogSend(bool *caughtup)
+XLogSendPhysical(bool *caughtup)
 {
 	XLogRecPtr	SendRqstPtr;
 	XLogRecPtr	startptr;
@@ -1210,6 +1682,65 @@ XLogSend(bool *caughtup)
 }
 
 /*
+ * Send out the WAL after it being decoded into a logical format by the output
+ * plugin specified in INIT_LOGICAL_DECODING
+ */
+static void
+XLogSendLogical(bool *caughtup)
+{
+	XLogRecPtr	endptr;
+	XLogRecPtr	curptr;
+
+	if (decoding_ctx == NULL)
+	{
+		decoding_ctx = AllocSetContextCreate(TopMemoryContext,
+											 "decoding context",
+											 ALLOCSET_DEFAULT_MINSIZE,
+											 ALLOCSET_DEFAULT_INITSIZE,
+											 ALLOCSET_DEFAULT_MAXSIZE);
+	}
+
+	logical_reader->endptr = logical_reader->curptr;
+	curptr = logical_reader->curptr;
+
+	/*
+	 * read at most MAX_SEND_SIZE of wal. We chunk the reading only to allow
+	 * reading keepalives and such inbetween.
+	 */
+	XLByteAdvance(logical_reader->endptr, MAX_SEND_SIZE);
+
+	/* only read up to already flushed wal */
+	endptr = GetFlushRecPtr();
+	if (XLByteLT(endptr, logical_reader->endptr))
+		logical_reader->endptr = endptr;
+
+	old_decoding_ctx = MemoryContextSwitchTo(decoding_ctx);
+	TopTransactionContext = decoding_ctx;
+
+	/* continue building initial snapshot */
+	XLogReaderRead(logical_reader);
+
+	MemoryContextSwitchTo(old_decoding_ctx);
+	TopTransactionContext = NULL;
+
+	if (curptr == logical_reader->curptr ||
+		logical_reader->curptr == endptr)
+		*caughtup = true;
+	else
+		*caughtup = false;
+
+	/* Update shared memory status */
+	{
+		/* use volatile pointer to prevent code rearrangement */
+		volatile WalSnd *walsnd = MyWalSnd;
+
+		SpinLockAcquire(&walsnd->mutex);
+		walsnd->sentPtr = logical_reader->curptr;
+		SpinLockRelease(&walsnd->mutex);
+	}
+}
+
+/*
  * Request walsenders to reload the currently-open WAL file
  */
 void
@@ -1301,6 +1832,8 @@ WalSndShmemSize(void)
 	size = offsetof(WalSndCtlData, walsnds);
 	size = add_size(size, mul_size(max_wal_senders, sizeof(WalSnd)));
 
+	size = add_size(size, mul_size(max_logical_slots, sizeof(LogicalWalSnd)));
+
 	return size;
 }
 
@@ -1329,6 +1862,21 @@ WalSndShmemInit(void)
 			SpinLockInit(&walsnd->mutex);
 			InitSharedLatch(&walsnd->latch);
 		}
+
+		WalSndCtl->logical_xmin = InvalidTransactionId;
+
+		if (max_logical_slots > 0)
+		{
+			LogicalWalSnd *logical_base;
+			logical_base = (LogicalWalSnd*)&WalSndCtl->walsnds[max_wal_senders];
+
+			for (i = 0; i < max_logical_slots; i++)
+			{
+				LogicalWalSnd   *walsnd = &logical_base[i];
+				walsnd->xmin = InvalidTransactionId;
+				SpinLockInit(&walsnd->mutex);
+			}
+		}
 	}
 }
 
diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index a98358d..da9e7e5 100644
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -51,6 +51,8 @@
 #include "access/xact.h"
 #include "access/twophase.h"
 #include "miscadmin.h"
+#include "replication/walsender.h"
+#include "replication/walsender_private.h"
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "storage/spin.h"
@@ -1156,6 +1158,14 @@ GetOldestXmin(bool allDbs, bool ignoreVacuum)
 		}
 	}
 
+	if (max_logical_slots > 0 &&
+		TransactionIdIsValid(WalSndCtl->logical_xmin) &&
+		TransactionIdPrecedes(WalSndCtl->logical_xmin, result))
+	{
+		result = WalSndCtl->logical_xmin;
+	}
+
+
 	if (RecoveryInProgress())
 	{
 		/*
@@ -1442,10 +1452,17 @@ GetSnapshotData(Snapshot snapshot)
 			suboverflowed = true;
 	}
 
+	/* FIXME: comment & concurrency */
+	if (TransactionIdIsValid(WalSndCtl->logical_xmin) &&
+		TransactionIdPrecedes(WalSndCtl->logical_xmin, xmin))
+		xmin = WalSndCtl->logical_xmin;
+
 	if (!TransactionIdIsValid(MyPgXact->xmin))
 		MyPgXact->xmin = TransactionXmin = xmin;
+
 	LWLockRelease(ProcArrayLock);
 
+
 	/*
 	 * Update globalxmin to include actual process xids.  This is a slightly
 	 * different way of computing it than GetOldestXmin uses, but should give
@@ -1695,6 +1712,12 @@ GetRunningTransactionData(void)
 		}
 	}
 
+	/*
+	 * Its important *not* to track decoding tasks here because snapbuild.c
+	 * uses ->oldestRunningXid to manage its xmin. If it were to be included
+	 * here the initial value could never increase.
+	 */
+
 	CurrentRunningXacts->xcnt = count - subcount;
 	CurrentRunningXacts->subxcnt = subcount;
 	CurrentRunningXacts->subxid_overflow = suboverflowed;
diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c
index 0cab243..6cc112d 100644
--- a/src/backend/storage/ipc/standby.c
+++ b/src/backend/storage/ipc/standby.c
@@ -804,7 +804,13 @@ standby_desc_running_xacts(StringInfo buf, xl_running_xacts *xlrec)
 			appendStringInfo(buf, " %u", xlrec->xids[i]);
 	}
 
-	if (xlrec->subxid_overflow)
+	if (xlrec->subxcnt > 0)
+	{
+		appendStringInfo(buf, "; %d subxacts:", xlrec->subxcnt);
+		for (i = 0; i < xlrec->subxcnt; i++)
+			appendStringInfo(buf, " %u", xlrec->xids[xlrec->xcnt + i]);
+	}
+	else if (xlrec->subxid_overflow)
 		appendStringInfo(buf, "; subxid ovf");
 }
 
diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c
index e26bf0b..7839a14 100644
--- a/src/backend/utils/cache/inval.c
+++ b/src/backend/utils/cache/inval.c
@@ -475,7 +475,7 @@ RegisterRelcacheInvalidation(Oid dbId, Oid relId)
  * Only the local caches are flushed; this does not transmit the message
  * to other backends.
  */
-static void
+void
 LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg)
 {
 	if (msg->id >= 0)
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 8c9ebe0..7bd2c27 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -506,9 +506,10 @@ RelationBuildTupleDesc(Relation relation)
 	heap_close(pg_attribute_desc, AccessShareLock);
 
 	if (need != 0)
+	{
 		elog(ERROR, "catalog is missing %d attribute(s) for relid %u",
 			 need, RelationGetRelid(relation));
-
+	}
 	/*
 	 * The attcacheoff values we read from pg_attribute should all be -1
 	 * ("unknown").  Verify this if assert checking is on.	They will be
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 745e7be..19a03cef 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -2030,6 +2030,17 @@ static struct config_int ConfigureNamesInt[] =
 	},
 
 	{
+		/* see max_connections */
+		{"max_logical_slots", PGC_POSTMASTER, REPLICATION_SENDING,
+			gettext_noop("Sets the maximum number of simultaneously defined WAL decoding slots."),
+			NULL
+		},
+		&max_logical_slots,
+		0, 0, MAX_BACKENDS /*?*/,
+		NULL, NULL, NULL
+	},
+
+	{
 		{"wal_sender_timeout", PGC_SIGHUP, REPLICATION_SENDING,
 			gettext_noop("Sets the maximum time to wait for WAL replication."),
 			NULL,
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index f64d52d..e24c712 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -64,6 +64,8 @@
 #include "access/xact.h"
 #include "storage/bufmgr.h"
 #include "storage/procarray.h"
+#include "utils/builtins.h"
+#include "utils/combocid.h"
 #include "utils/tqual.h"
 
 
@@ -73,6 +75,8 @@ SnapshotData SnapshotSelfData = {HeapTupleSatisfiesSelf};
 SnapshotData SnapshotAnyData = {HeapTupleSatisfiesAny};
 SnapshotData SnapshotToastData = {HeapTupleSatisfiesToast};
 
+static Snapshot SnapshotNowDecoding;
+
 /* local functions */
 static bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 
@@ -1407,3 +1411,248 @@ XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot)
 
 	return false;
 }
+
+static bool
+TransactionIdInArray(TransactionId xid, TransactionId *xip, Size num)
+{
+	return bsearch(&xid, xip, num,
+	               sizeof(TransactionId), xidComparator) != NULL;
+}
+
+static HTAB *tuplecid_data = NULL;
+
+/*
+ * See the comments for HeapTupleSatisfiesMVCC for the semantics this function
+ * obeys.
+ *
+ * Only usable on tuples from catalog tables!
+ *
+ * We don't need to support HEAP_MOVED_(IN|OFF) for now because we only support
+ * reading catalog pages which couldn't have been created in an older version.
+ *
+ * We don't set any hint bits in here as it seems unlikely to be beneficial as
+ * those should already be set by normal access and it seems to be too
+ * dangerous to do so as the semantics of doing so during timetravel are more
+ * complicated than when dealing "only" with the present.
+ */
+bool
+HeapTupleSatisfiesMVCCDuringDecoding(HeapTuple htup, Snapshot snapshot,
+                                     Buffer buffer)
+{
+	HeapTupleHeader tuple = htup->t_data;
+/*#define DEBUG_ME*/
+	TransactionId xmin = HeapTupleHeaderGetXmin(tuple);
+	TransactionId xmax = HeapTupleHeaderGetXmax(tuple);
+
+	Assert(ItemPointerIsValid(&htup->t_self));
+	Assert(htup->t_tableOid != InvalidOid);
+
+	/* transaction aborted */
+	if (tuple->t_infomask & HEAP_XMIN_INVALID)
+	{
+		Assert(!TransactionIdDidCommit(xmin));
+		goto invisible;
+	}
+    /* check if its one of our txids, toplevel is also in there */
+	else if (TransactionIdInArray(xmin, snapshot->subxip, snapshot->subxcnt))
+	{
+		CommandId cmin = HeapTupleHeaderGetRawCommandId(tuple);
+		CommandId cmax = InvalidCommandId;
+
+		/*
+		 * if another transaction deleted this tuple or if cmin/cmax is stored
+		 * in a combocid we need to to lookup the actual values externally.
+		 */
+		if ((!(tuple->t_infomask & HEAP_XMAX_INVALID) &&
+			 !TransactionIdInArray(xmax, snapshot->subxip, snapshot->subxcnt)) ||
+			tuple->t_infomask & HEAP_COMBOCID
+			)
+		{
+			bool resolved;
+
+			resolved = ResolveCminCmaxDuringDecoding(tuplecid_data, htup,
+													 buffer, &cmin, &cmax);
+
+			if (!resolved)
+				elog(ERROR, "could not resolve cmin/cmax of catalog tuple");
+		}
+
+#ifdef DEBUG_ME
+		elog(LOG, "curcid: %u cmin: %u; invisible: %u", snapshot->curcid, cmin,
+			 cmin >= snapshot->curcid);
+#endif
+		if (cmin >= snapshot->curcid)
+			goto invisible;	/* inserted after scan started */
+	}
+	/* normal transaction state counts */
+	else if (TransactionIdPrecedes(xmin, snapshot->xmin))
+	{
+		Assert(!(tuple->t_infomask & HEAP_XMIN_COMMITTED &&
+				 !TransactionIdDidCommit(xmin)));
+
+		if (!(tuple->t_infomask & HEAP_XMIN_COMMITTED) &&
+			!TransactionIdDidCommit(xmin))
+			goto invisible;
+	}
+	/* beyond our xmax horizon, i.e. invisible */
+	else if (TransactionIdFollows(xmin, snapshot->xmax))
+	{
+		goto invisible;
+	}
+	/* check if we know the transaction has committed */
+	else if(TransactionIdInArray(xmin, snapshot->xip, snapshot->xcnt))
+	{
+	}
+	else
+	{
+		goto invisible;
+	}
+
+	/* at this point we know xmin is visible */
+
+	/* why should those be in catalog tables? */
+	Assert(!(tuple->t_infomask & HEAP_XMAX_IS_MULTI));
+
+	/* xid invalid or aborted */
+	if (tuple->t_infomask & HEAP_XMAX_INVALID)
+		goto visible;
+	/* locked tuples are always visible */
+	else if (tuple->t_infomask & HEAP_IS_LOCKED)
+		goto visible;
+    /* check if its one of our txids, toplevel is also in there */
+	else if (TransactionIdInArray(xmax, snapshot->subxip, snapshot->subxcnt))
+	{
+		CommandId cmin;
+		CommandId cmax = HeapTupleHeaderGetRawCommandId(tuple);
+
+		/* Lookup actual cmin/cmax values */
+		if (tuple->t_infomask & HEAP_COMBOCID){
+#ifdef DEBUG_ME
+			CommandId combocid = cmax;
+#endif
+			bool resolved;
+
+			resolved = ResolveCminCmaxDuringDecoding(tuplecid_data, htup,
+													 buffer, &cmin, &cmax);
+
+			if (!resolved)
+			{
+				elog(FATAL, "could not resolve combocid to cmax");
+				goto invisible;
+			}
+
+
+#ifdef DEBUG_ME
+			elog(LOG, "converting combocid %u to cmax %u (cmin %u)",
+				 combocid, cmax, cmin);
+#endif
+		}
+#ifdef DEBUG_ME
+		elog(LOG, "curcid: %u, cmax %u, visible: %u", snapshot->curcid, cmax,
+			 cmax >= snapshot->curcid);
+#endif
+		if (cmax >= snapshot->curcid)
+			goto visible;	/* deleted after scan started */
+		else
+			goto invisible;	/* deleted before scan started */
+	}
+	/* we cannot possibly see the deleting transaction */
+	else if (TransactionIdFollows(xmax, snapshot->xmax))
+	{
+		goto visible;
+	}
+	/* normal transaction state is valid */
+	else if (TransactionIdPrecedes(xmax, snapshot->xmin))
+	{
+		Assert(!(tuple->t_infomask & HEAP_XMAX_COMMITTED &&
+				 !TransactionIdDidCommit(xmax)));
+
+		if (tuple->t_infomask & HEAP_XMAX_COMMITTED)
+			goto invisible;
+
+		if (TransactionIdDidCommit(xmax))
+			goto invisible;
+		else
+			goto visible;
+	}
+	/* do we know that the deleting txn is valid? */
+	else if (TransactionIdInArray(xmax, snapshot->xip, snapshot->xcnt))
+	{
+		goto invisible;
+	}
+	else
+	{
+		goto visible;
+	}
+visible:
+#ifdef DEBUG_ME
+	if (xmin > FirstNormalTransactionId)
+		elog(DEBUG1, "visible tuple with xmin: %u, xmax: %u, cmin %u, snapmin: %u, snapmax: %u, snapcid: %u, owncnt: %u top: %u, combo: %u, (%u, %u)",
+			 xmin, xmax, HeapTupleHeaderGetRawCommandId(tuple),
+			 snapshot->xmin, snapshot->xmax, snapshot->curcid,
+			 snapshot->subxcnt, snapshot->subxcnt ? snapshot->subxip[0] : 0,
+			 !!(tuple->t_infomask & HEAP_COMBOCID),
+			 BlockIdGetBlockNumber(&htup->t_self.ip_blkid), htup->t_self.ip_posid);
+#endif
+	return true;
+
+invisible:
+#ifdef DEBUG_ME
+	if (xmin > FirstNormalTransactionId)
+		elog(DEBUG1, "invisible tuple with xmin: %u, xmax: %u, cmin %u, snapmin: %u, snapmax: %u, snapcid: %u, owncnt: %u top: %u, combo: %u, (%u, %u)",
+			 xmin, xmax, HeapTupleHeaderGetRawCommandId(tuple),
+			 snapshot->xmin, snapshot->xmax, snapshot->curcid,
+			 snapshot->subxcnt, snapshot->subxcnt ? snapshot->subxip[0] : 0,
+			 !!(tuple->t_infomask & HEAP_COMBOCID),
+			 BlockIdGetBlockNumber(&htup->t_self.ip_blkid), htup->t_self.ip_posid);
+#endif
+	return false;
+}
+
+static bool
+FailsSatisfies(HeapTuple htup, Snapshot snapshot, Buffer buffer)
+{
+	elog(ERROR, "Normal snapshots cannot be used during timetravel access.");
+	return false;
+}
+
+static bool
+RedirectSatisfiesNow(HeapTuple htup, Snapshot snapshot, Buffer buffer)
+{
+	Assert(SnapshotNowDecoding != NULL);
+	return HeapTupleSatisfiesMVCCDuringDecoding(htup, SnapshotNowDecoding,
+	                                            buffer);
+}
+
+void
+SetupDecodingSnapshots(Snapshot snapshot_now, HTAB *tuplecids)
+{
+	/* prevent recursively setting up decoding snapshots */
+	Assert(SnapshotNowData.satisfies != RedirectSatisfiesNow);
+
+	SnapshotNowData.satisfies = RedirectSatisfiesNow;
+	/* make sure normal snapshots aren't used*/
+	SnapshotSelfData.satisfies = FailsSatisfies;
+	SnapshotAnyData.satisfies = FailsSatisfies;
+	SnapshotToastData.satisfies = FailsSatisfies;
+
+	/* setup the timetravel snapshot */
+	SnapshotNowDecoding = snapshot_now;
+
+	/* setup (cmin, cmax) lookup hash */
+	tuplecid_data = tuplecids;
+}
+
+
+void
+RevertFromDecodingSnapshots(void)
+{
+	SnapshotNowDecoding = NULL;
+	tuplecid_data = NULL;
+
+	/* rally to restore sanity and/or boredom */
+	SnapshotNowData.satisfies = HeapTupleSatisfiesNow;
+	SnapshotSelfData.satisfies = HeapTupleSatisfiesSelf;
+	SnapshotAnyData.satisfies = HeapTupleSatisfiesAny;
+	SnapshotToastData.satisfies = HeapTupleSatisfiesToast;
+}
diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c
index 129c4d0..10080d0 100644
--- a/src/bin/pg_controldata/pg_controldata.c
+++ b/src/bin/pg_controldata/pg_controldata.c
@@ -77,6 +77,8 @@ wal_level_str(WalLevel wal_level)
 			return "archive";
 		case WAL_LEVEL_HOT_STANDBY:
 			return "hot_standby";
+		case WAL_LEVEL_LOGICAL:
+			return "logical";
 	}
 	return _("unrecognized wal_level");
 }
diff --git a/src/include/access/heapam_xlog.h b/src/include/access/heapam_xlog.h
index 8ec710e..1405317 100644
--- a/src/include/access/heapam_xlog.h
+++ b/src/include/access/heapam_xlog.h
@@ -54,6 +54,7 @@
 #define XLOG_HEAP2_CLEANUP_INFO 0x30
 #define XLOG_HEAP2_VISIBLE		0x40
 #define XLOG_HEAP2_MULTI_INSERT 0x50
+#define XLOG_HEAP2_NEW_CID 0x60
 
 /*
  * All what we need to find changed tuple
@@ -238,6 +239,28 @@ typedef struct xl_heap_visible
 
 #define SizeOfHeapVisible (offsetof(xl_heap_visible, cutoff_xid) + sizeof(TransactionId))
 
+typedef struct xl_heap_new_cid
+{
+	/*
+	 * store toplevel xid so we don't have to merge cids from different
+	 * transactions
+	 */
+	TransactionId top_xid;
+	CommandId cmin;
+	CommandId cmax;
+	/*
+	 * don't really need the combocid but the padding makes it free and its
+	 * useful for debugging.
+	 */
+	CommandId combocid;
+	/*
+	 * Store the relfilenode/ctid pair to facilitate lookups.
+	 */
+	xl_heaptid target;
+} xl_heap_new_cid;
+
+#define SizeOfHeapNewCid (offsetof(xl_heap_new_cid, target) + SizeOfHeapTid)
+
 extern void HeapTupleHeaderAdvanceLatestRemovedXid(HeapTupleHeader tuple,
 									   TransactionId *latestRemovedXid);
 
diff --git a/src/include/access/transam.h b/src/include/access/transam.h
index 228f6a1..915b2cd 100644
--- a/src/include/access/transam.h
+++ b/src/include/access/transam.h
@@ -63,6 +63,11 @@
 	(AssertMacro(TransactionIdIsNormal(id1) && TransactionIdIsNormal(id2)), \
 	(int32) ((id1) - (id2)) < 0)
 
+/* compare two XIDs already known to be normal; this is a macro for speed */
+#define NormalTransactionIdFollows(id1, id2) \
+	(AssertMacro(TransactionIdIsNormal(id1) && TransactionIdIsNormal(id2)), \
+	(int32) ((id1) - (id2)) > 0)
+
 /* ----------
  *		Object ID (OID) zero is InvalidOid.
  *
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 32c2e40..ae4f849 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -196,7 +196,8 @@ typedef enum WalLevel
 {
 	WAL_LEVEL_MINIMAL = 0,
 	WAL_LEVEL_ARCHIVE,
-	WAL_LEVEL_HOT_STANDBY
+	WAL_LEVEL_HOT_STANDBY,
+	WAL_LEVEL_LOGICAL
 } WalLevel;
 extern int	wal_level;
 
diff --git a/src/include/catalog/index.h b/src/include/catalog/index.h
index 298641b..8848fd2 100644
--- a/src/include/catalog/index.h
+++ b/src/include/catalog/index.h
@@ -103,4 +103,8 @@ extern bool ReindexIsProcessingHeap(Oid heapOid);
 extern bool ReindexIsProcessingIndex(Oid indexOid);
 extern Oid	IndexGetRelation(Oid indexId, bool missing_ok);
 
+extern void relationFindPrimaryKey(Relation pkrel, Oid *indexOid,
+                                   int16 *nratts, int16 *attnums, Oid *atttypids,
+                                   Oid *opclasses);
+
 #endif   /* INDEX_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 438a1d9..223849b 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -407,6 +407,8 @@ typedef enum NodeTag
 	T_IdentifySystemCmd,
 	T_BaseBackupCmd,
 	T_StartReplicationCmd,
+	T_InitLogicalReplicationCmd,
+	T_StartLogicalReplicationCmd,
 
 	/*
 	 * TAGS FOR RANDOM OTHER STUFF
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index 236a36d..91a4986 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -49,4 +49,26 @@ typedef struct StartReplicationCmd
 	XLogRecPtr	startpoint;
 } StartReplicationCmd;
 
+
+/* ----------------------
+ *		INIT_LOGICAL_REPLICATION command
+ * ----------------------
+ */
+typedef struct InitLogicalReplicationCmd
+{
+	NodeTag		type;
+	char       *plugin;
+} InitLogicalReplicationCmd;
+
+/* ----------------------
+ *		START_LOGICAL_REPLICATION command
+ * ----------------------
+ */
+typedef struct StartLogicalReplicationCmd
+{
+	NodeTag		type;
+	char       *name;
+	XLogRecPtr	startpoint;
+} StartLogicalReplicationCmd;
+
 #endif   /* REPLNODES_H */
diff --git a/src/include/replication/decode.h b/src/include/replication/decode.h
new file mode 100644
index 0000000..1caa98d
--- /dev/null
+++ b/src/include/replication/decode.h
@@ -0,0 +1,21 @@
+/*-------------------------------------------------------------------------
+ * decode.h
+ *     PostgreSQL WAL to logical transformation
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef DECODE_H
+#define DECODE_H
+
+#include "access/xlogreader.h"
+#include "replication/reorderbuffer.h"
+#include "replication/logicalfuncs.h"
+
+void DecodeRecordIntoReorderBuffer(XLogReaderState *reader,
+								   ReaderApplyState* state,
+								   XLogRecordBuffer* buf);
+
+#endif
diff --git a/src/include/replication/logicalfuncs.h b/src/include/replication/logicalfuncs.h
new file mode 100644
index 0000000..db78797
--- /dev/null
+++ b/src/include/replication/logicalfuncs.h
@@ -0,0 +1,44 @@
+/*-------------------------------------------------------------------------
+ * decode.h
+ *     PostgreSQL WAL to logical transformation
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef LOGICALFUNCS_H
+#define LOGICALFUNCS_H
+
+#include "access/xlogreader.h"
+#include "replication/reorderbuffer.h"
+#include "replication/output_plugin.h"
+
+typedef struct ReaderApplyState
+{
+	struct ReorderBuffer *reorderbuffer;
+	bool stop_after_consistent;
+	struct Snapstate *snapstate;
+
+	LogicalDecodeInitCB init_cb;
+	LogicalDecodeBeginCB begin_cb;
+	LogicalDecodeChangeCB change_cb;
+	LogicalDecodeCommitCB commit_cb;
+	LogicalDecodeCleanupCB cleanup_cb;
+
+	StringInfo out;
+	void *user_private;
+
+
+} ReaderApplyState;
+
+XLogReaderState *
+initial_snapshot_reader(void);
+
+XLogReaderState *
+normal_snapshot_reader(char *plugin, XLogRecPtr valid_after);
+
+bool
+initial_snapshot_ready(XLogReaderState *);
+
+#endif
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
new file mode 100644
index 0000000..27d5982
--- /dev/null
+++ b/src/include/replication/output_plugin.h
@@ -0,0 +1,76 @@
+/*-------------------------------------------------------------------------
+ * output_plugin.h
+ *     PostgreSQL Logical Decode Plugin Interface
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef OUTPUT_PLUGIN_H
+#define OUTPUT_PLUGIN_H
+
+#include "lib/stringinfo.h"
+
+#include "replication/reorderbuffer.h"
+
+/*
+ * Callback that gets called in a user-defined plugin.
+ * 'private' can be set to some private data.
+ *
+ * Gets looked up in the library symbol pg_decode_init.
+ */
+typedef void (*LogicalDecodeInitCB) (
+	void **private
+	);
+
+/*
+ * Gets called for every BEGIN of a successful transaction.
+ *
+ * Return "true" if the message in "out" should get sent, false otherwise.
+ *
+ * Gets looked up in the library symbol pg_decode_begin_txn.
+ */
+typedef bool (*LogicalDecodeBeginCB) (
+	void *private,
+	StringInfo out,
+	ReorderBufferTXN *txn);
+
+/*
+ * Gets called for every change in a successful transaction.
+ *
+ * Return "true" if the message in "out" should get sent, false otherwise.
+ *
+ * Gets looked up in the library symbol pg_decode_change.
+ */
+typedef bool (*LogicalDecodeChangeCB) (
+	void *private,
+	StringInfo out,
+	ReorderBufferTXN *txn,
+	Oid tableoid,
+	ReorderBufferChange *change
+	);
+
+/*
+ * Gets called for every COMMIT of a successful transaction.
+ *
+ * Return "true" if the message in "out" should get sent, false otherwise.
+ *
+ * Gets looked up in the library symbol pg_decode_commit_txn.
+ */
+typedef bool (*LogicalDecodeCommitCB) (
+	void *private,
+	StringInfo out,
+	ReorderBufferTXN *txn,
+	XLogRecPtr commit_lsn);
+
+/*
+ * Gets called to cleanup the state of an output plugin
+ *
+ * Gets looked up in the library symbol pg_decode_cleanup.
+ */
+typedef void (*LogicalDecodeCleanupCB) (
+	void *private
+	);
+
+#endif
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
new file mode 100644
index 0000000..a79dd79
--- /dev/null
+++ b/src/include/replication/reorderbuffer.h
@@ -0,0 +1,284 @@
+/*
+ * reorderbuffer.h
+ *
+ * PostgreSQL logical replay "cache" management
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/replication/reorderbuffer.h
+ */
+#ifndef REORDERBUFFER_H
+#define REORDERBUFFER_H
+
+#include "access/htup_details.h"
+#include "utils/hsearch.h"
+
+#include "lib/ilist.h"
+
+#include "storage/sinval.h"
+
+#include "utils/snapshot.h"
+
+
+typedef struct ReorderBuffer ReorderBuffer;
+
+enum ReorderBufferChangeType
+{
+	REORDER_BUFFER_CHANGE_INSERT,
+	REORDER_BUFFER_CHANGE_UPDATE,
+	REORDER_BUFFER_CHANGE_DELETE
+};
+
+typedef struct ReorderBufferTupleBuf
+{
+	/* position in preallocated list */
+	slist_node node;
+
+	HeapTupleData tuple;
+	HeapTupleHeaderData header;
+	char		data[MaxHeapTupleSize];
+}	ReorderBufferTupleBuf;
+
+typedef struct ReorderBufferChange
+{
+	XLogRecPtr	lsn;
+
+	union {
+		enum ReorderBufferChangeType action;
+		/* do not leak internal enum values to the outside */
+		int action_internal;
+	};
+
+	RelFileNode relnode;
+
+	union
+	{
+		ReorderBufferTupleBuf *newtuple;
+		Snapshot	snapshot;
+		CommandId	command_id;
+		struct {
+			RelFileNode node;
+			ItemPointerData tid;
+			CommandId	cmin;
+			CommandId	cmax;
+			CommandId	combocid;
+		} tuplecid;
+	};
+
+	ReorderBufferTupleBuf *oldtuple;
+
+	/*
+	 * While in use this is how a change is linked into a transactions,
+	 * otherwise its the preallocated list.
+	 */
+	dlist_node node;
+} ReorderBufferChange;
+
+typedef struct ReorderBufferTXN
+{
+	TransactionId xid;
+
+	XLogRecPtr	lsn;
+
+	/* did the TX have catalog changes */
+	bool does_timetravel;
+
+	bool has_base_snapshot;
+
+	/*
+	 * How many ReorderBufferChange's do we have in this txn.
+	 *
+	 * Subtransactions are *not* included.
+	 */
+	Size		nentries;
+
+	/*
+	 * How many of the above entries are stored in memory in contrast to being
+	 * spilled to disk.
+	 */
+	Size		nentries_mem;
+
+	/*
+	 * List of actual changes, those include new Snapshots and new CommandIds
+	 */
+	dlist_head changes;
+
+	/*
+	 * List of cmin/cmax pairs for catalog tuples
+	 */
+	dlist_head tuplecids;
+
+	/*
+	 * Numer of stored cmin/cmax pairs. Used to create the tuplecid_hash with
+	 * the correct size.
+	 */
+	Size       ntuplecids;
+
+	/*
+	 * On-demand built hash for looking up the above values.
+	 */
+	HTAB	   *tuplecid_hash;
+
+	/*
+	 * non-hierarchical list of subtransactions that are *not* aborted
+	 */
+	dlist_head subtxns;
+	Size nsubtxns;
+
+	/*
+	 * our position in a list of subtransactions while the TXN is in use.
+	 * Otherwise its the position in the list of preallocated transactions.
+	 */
+	dlist_node node;
+
+	/*
+	 * Number of stored cache invalidations.
+	 */
+	Size ninvalidations;
+
+	/*
+	 * Stored cache invalidations. This is not a linked list because we get all
+	 * the invalidations at once.
+	 */
+	SharedInvalidationMessage *invalidations;
+
+} ReorderBufferTXN;
+
+
+/* XXX: were currently passing the originating subtxn. Not sure thats necessary */
+typedef void (*ReorderBufferApplyChangeCB) (
+	ReorderBuffer *cache,
+	ReorderBufferTXN *txn,
+	ReorderBufferChange *change);
+
+typedef void (*ReorderBufferBeginCB) (
+	ReorderBuffer *cache,
+	ReorderBufferTXN *txn);
+
+typedef void (*ReorderBufferCommitCB) (
+	ReorderBuffer *cache,
+	ReorderBufferTXN *txn,
+	XLogRecPtr commit_lsn);
+
+/*
+ * max number of concurrent top-level transactions or transaction where we
+ * don't know if they are top-level can be calculated by:
+ * (max_connections + max_prepared_xactx + ?)  * PGPROC_MAX_CACHED_SUBXIDS
+ */
+struct ReorderBuffer
+{
+	/*
+	 * Should snapshots for decoding be collected. If many catalog changes
+	 * happen this can be considerably expensive.
+	 */
+	bool		build_snapshots;
+
+	TransactionId last_txn;
+	ReorderBufferTXN *last_txn_cache;
+	HTAB	   *by_txn;
+
+	ReorderBufferBeginCB begin;
+	ReorderBufferApplyChangeCB apply_change;
+	ReorderBufferCommitCB commit;
+
+	void	   *private_data;
+
+	MemoryContext context;
+
+	/*
+	 * we don't want to repeatedly (de-)allocated those structs, so cache them
+	 * for reusage.
+	 */
+	dlist_head cached_transactions;
+	Size		nr_cached_transactions;
+
+	dlist_head cached_changes;
+	Size		nr_cached_changes;
+
+	slist_head cached_tuplebufs;
+	Size		nr_cached_tuplebufs;
+};
+
+
+ReorderBuffer *ReorderBufferAllocate(void);
+
+void ReorderBufferFree(ReorderBuffer *);
+
+ReorderBufferTupleBuf *ReorderBufferGetTupleBuf(ReorderBuffer *);
+
+void ReorderBufferReturnTupleBuf(ReorderBuffer *cache, ReorderBufferTupleBuf * tuple);
+
+/*
+ * Returns a (potentically preallocated) change struct. Its lifetime is managed
+ * by the reorderbuffer module.
+ *
+ * If not added to a transaction with ReorderBufferAddChange it needs to be
+ * returned via ReorderBufferReturnChange
+ *
+ * FIXME: better name
+ */
+ReorderBufferChange *
+			ReorderBufferGetChange(ReorderBuffer *);
+
+/*
+ * Return an unused ReorderBufferChange struct
+ */
+void ReorderBufferReturnChange(ReorderBuffer *, ReorderBufferChange *);
+
+
+/*
+ * record the transaction as in-progress if not already done, add the current
+ * change.
+ *
+ * We have a one-entry cache for lookin up the current ReorderBufferTXN so we
+ * don't need to do a full hash-lookup if the same xid is used
+ * sequentially. Them being used multiple times that way is rather frequent.
+ */
+void ReorderBufferAddChange(ReorderBuffer *, TransactionId, XLogRecPtr lsn, ReorderBufferChange *);
+
+/*
+ *
+ */
+void ReorderBufferCommit(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+
+void ReorderBufferCommitChild(ReorderBuffer *, TransactionId, TransactionId, XLogRecPtr lsn);
+
+void ReorderBufferAbort(ReorderBuffer *, TransactionId, XLogRecPtr lsn);
+
+/*
+ * if lsn == InvalidXLogRecPtr this is the first snap for the transaction
+ *
+ * most callers don't need snapshot.h, so we use struct SnapshotData instead
+ */
+void ReorderBufferAddBaseSnapshot(ReorderBuffer *, TransactionId, XLogRecPtr lsn, struct SnapshotData *snap);
+
+/*
+ * Will only be called for command ids > 1
+ */
+void ReorderBufferAddNewCommandId(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+								  CommandId cid);
+
+void ReorderBufferAddNewTupleCids(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+								  RelFileNode node, ItemPointerData pt,
+								  CommandId cmin, CommandId cmax, CommandId combocid);
+
+void ReorderBufferAddInvalidations(ReorderBuffer *, TransactionId, XLogRecPtr lsn,
+								   Size nmsgs, SharedInvalidationMessage* msgs);
+
+bool ReorderBufferIsXidKnown(ReorderBuffer *cache, TransactionId xid);
+
+/*
+ * Announce that tx does timetravel. Relevant for the whole toplevel/subtxn
+ * tree.
+ */
+void ReorderBufferXidSetTimetravel(ReorderBuffer *cache, TransactionId xid);
+
+/*
+ * Does the transaction indicated by 'xid' do timetravel?
+ */
+bool ReorderBufferXidDoesTimetravel(ReorderBuffer *cache, TransactionId xid);
+
+bool ReorderBufferXidHasBaseSnapshot(ReorderBuffer *cache, TransactionId xid);
+
+#endif
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
new file mode 100644
index 0000000..e8c5fcb
--- /dev/null
+++ b/src/include/replication/snapbuild.h
@@ -0,0 +1,128 @@
+/*-------------------------------------------------------------------------
+ *
+ * snapbuild.h
+ *	  Exports from replication/logical/snapbuild.c.
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ * src/include/replication/snapbuild.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SNAPBUILD_H
+#define SNAPBUILD_H
+
+#include "replication/reorderbuffer.h"
+
+#include "utils/hsearch.h"
+#include "utils/snapshot.h"
+#include "access/htup.h"
+
+typedef enum
+{
+	/*
+	 * Initial state, we can't do much yet.
+	 */
+	SNAPBUILD_START,
+
+	/*
+	 * We have collected enough information to decode tuples in transactions
+	 * that started after this.
+	 *
+	 * Once we reached this we start to collect changes. We cannot apply them
+	 * yet because the might be based on transactions that were still running
+	 * when we reached them yet.
+	 */
+	SNAPBUILD_FULL_SNAPSHOT,
+
+	/*
+	 * Found a point after hitting built_full_snapshot where all transactions
+	 * that were running at that point finished. Till we reach that we hold
+	 * off calling any commit callbacks.
+	 */
+	SNAPBUILD_CONSISTENT
+}	SnapBuildState;
+
+typedef enum
+{
+	SNAPBUILD_SKIP,
+	SNAPBUILD_DECODE
+}	SnapBuildAction;
+
+typedef struct Snapstate
+{
+	SnapBuildState state;
+
+	/* all transactions smaller than this have committed/aborted */
+	TransactionId xmin;
+
+	/* all transactions bigger than this are uncommitted */
+	TransactionId xmax;
+
+	/*
+	 * All transactions in this window have to be checked via the running
+	 * array. This will only be used initially till we are past xmax_running.
+	 *
+	 * Note that we initially assume treat already running transactions to
+	 * have catalog modifications because we don't have enough information
+	 * about them to properly judge that.
+	 */
+	TransactionId xmin_running;
+	TransactionId xmax_running;
+
+	/*
+	 * array of running transactions.
+	 *
+	 * Kept in xidComparator order so it can be searched with bsearch().
+	 */
+	TransactionId *running;
+	/* how many transactions are still running */
+	size_t		nrrunning;
+
+	/*
+	 * we need to keep track of the amount of tracked transactions separately
+	 * from nrrunning_space as nrunning_initial gives the range of valid xids
+	 * in the array so bsearch() can work.
+	 */
+	size_t		nrrunning_initial;
+
+	XLogRecPtr transactions_after;
+
+	/*
+	 * Transactions which could have catalog changes that committed between
+	 * xmin and xmax
+	 */
+	size_t		nrcommitted;
+	size_t		nrcommitted_space;
+	/*
+	 * Array of committed transactions that have modified the catalog.
+	 *
+	 * As this array is frequently modified we do *not* keep it in
+	 * xidComparator order. Instead we sort the array when building &
+	 * distributing a snapshot.
+	 */
+	TransactionId *committed;
+
+	/*
+	 * Snapshot thats valid to see all committed transactions that see catalog
+	 * modifications.
+	 */
+	Snapshot snapshot;
+}	Snapstate;
+
+extern Snapstate *AllocateSnapshotBuilder(ReorderBuffer *cache);
+
+extern void	FreeSnapshotBuilder(Snapstate *cache);
+
+struct XLogRecordBuffer;
+
+extern SnapBuildAction SnapBuildDecodeCallback(ReorderBuffer *cache, Snapstate *snapstate, struct XLogRecordBuffer *buf);
+
+extern HeapTuple LookupTableByRelFileNode(RelFileNode *r);
+
+extern bool SnapBuildHasCatalogChanges(Snapstate *snapstate, TransactionId xid,
+                                       RelFileNode *relfilenode);
+
+extern void SnapBuildSnapDecRefcount(Snapshot snap);
+
+#endif   /* SNAPBUILD_H */
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index df8e951..b2d9434 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -24,6 +24,7 @@ extern bool wake_wal_senders;
 
 /* user-settable parameters */
 extern int	max_wal_senders;
+extern int	max_logical_slots;
 extern int	wal_sender_timeout;
 
 extern void InitWalSender(void);
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 66234cd..c712659 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -66,6 +66,28 @@ typedef struct WalSnd
 
 extern WalSnd *MyWalSnd;
 
+typedef struct
+{
+	TransactionId xmin;
+	NameData      name;
+	NameData      plugin;
+
+	XLogRecPtr	  last_required_checkpoint;
+	XLogRecPtr	  confirmed_flush;
+
+	TransactionId candidate_xmin;
+	XLogRecPtr	  candidate_xmin_after;
+
+	/* is this slot defined */
+	bool          in_use;
+	/* is somebody streaming out changes for this slot */
+	bool          active;
+	slock_t		mutex;
+} LogicalWalSnd;
+
+extern LogicalWalSnd *MyLogicalWalSnd;
+
+
 /* There is one WalSndCtl struct for the whole database cluster */
 typedef struct
 {
@@ -88,12 +110,14 @@ typedef struct
 	 */
 	bool		sync_standbys_defined;
 
+	TransactionId logical_xmin;
+
 	WalSnd		walsnds[1];		/* VARIABLE LENGTH ARRAY */
+	LogicalWalSnd logical_walsnds[1];		/* VARIABLE LENGTH ARRAY */
 } WalSndCtlData;
 
 extern WalSndCtlData *WalSndCtl;
 
-
 extern void WalSndSetState(WalSndState state);
 extern void XLogRead(char *buf, XLogRecPtr startptr, Size count);
 
@@ -109,4 +133,12 @@ extern void replication_scanner_finish(void);
 
 extern Node *replication_parse_result;
 
+/* change logical xmin */
+extern void IncreaseLogicalXminForSlot(XLogRecPtr lsn, TransactionId xmin);
+
+/* logical wal sender data gathering functions */
+extern void WalSndWriteData(StringInfo data);
+extern void WalSndPrepareWrite(StringInfo out, XLogRecPtr lsn);
+
+
 #endif   /* _WALSENDER_PRIVATE_H */
diff --git a/src/include/storage/itemptr.h b/src/include/storage/itemptr.h
index 331812b..fc5f86d 100644
--- a/src/include/storage/itemptr.h
+++ b/src/include/storage/itemptr.h
@@ -116,6 +116,9 @@ typedef ItemPointerData *ItemPointer;
 /*
  * ItemPointerCopy
  *		Copies the contents of one disk item pointer to another.
+ *
+ * Should there ever be padding in an ItemPointer this would need to be handled
+ * differently as its used in hashes.
  */
 #define ItemPointerCopy(fromPointer, toPointer) \
 ( \
diff --git a/src/include/storage/sinval.h b/src/include/storage/sinval.h
index bcf2c81..1b68bab 100644
--- a/src/include/storage/sinval.h
+++ b/src/include/storage/sinval.h
@@ -136,4 +136,6 @@ extern void ProcessCommittedInvalidationMessages(SharedInvalidationMessage *msgs
 									 int nmsgs, bool RelcacheInitFileInval,
 									 Oid dbid, Oid tsid);
 
+extern void LocalExecuteInvalidationMessage(SharedInvalidationMessage *msg);
+
 #endif   /* SINVAL_H */
diff --git a/src/include/utils/tqual.h b/src/include/utils/tqual.h
index b129ae9..2e9a7d8 100644
--- a/src/include/utils/tqual.h
+++ b/src/include/utils/tqual.h
@@ -39,7 +39,8 @@ extern PGDLLIMPORT SnapshotData SnapshotToastData;
 
 /* This macro encodes the knowledge of which snapshots are MVCC-safe */
 #define IsMVCCSnapshot(snapshot)  \
-	((snapshot)->satisfies == HeapTupleSatisfiesMVCC)
+	((snapshot)->satisfies == HeapTupleSatisfiesMVCC || \
+	 (snapshot)->satisfies == HeapTupleSatisfiesMVCCDuringDecoding)
 
 /*
  * HeapTupleSatisfiesVisibility
@@ -89,4 +90,32 @@ extern bool HeapTupleIsSurelyDead(HeapTuple htup,
 extern void HeapTupleSetHintBits(HeapTupleHeader tuple, Buffer buffer,
 					 uint16 infomask, TransactionId xid);
 
+/*
+ * Special "satisfies" routines used during decoding xlog from a different
+ * point of lsn. Also used for timetravel SnapshotNow's.
+ */
+extern bool HeapTupleSatisfiesMVCCDuringDecoding(HeapTuple htup,
+                                                 Snapshot snapshot, Buffer buffer);
+
+/*
+ * install the 'snapshot_now' snapshot as a timetravelling snapshot replacing
+ * the normal SnapshotNow behaviour. This snapshot needs to have been created
+ * by snapbuild.c otherwise you will see crashes!
+ *
+ * FIXME: We need something resembling the real SnapshotNow to handle things
+ * like enum lookups from indices correctly.
+ */
+extern void SetupDecodingSnapshots(Snapshot snapshot_now, HTAB *tuplecids);
+extern void RevertFromDecodingSnapshots(void);
+
+/*
+ * resolve combocids and overwritten cmin values
+ *
+ * To avoid leaking to much knowledge about the reorderbuffer this is
+ * implemented in reorderbuffer.c not tqual.c.
+ */
+extern bool ResolveCminCmaxDuringDecoding(HTAB *tuplecid_data, HeapTuple htup,
+										  Buffer buffer,
+										  CommandId *cmin, CommandId *cmax);
+
 #endif   /* TQUAL_H */