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

diff --git a/src/backend/replication/Makefile b/src/backend/replication/Makefile
index 9d9ec87..ae7f6b1 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..4e56769
--- /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 = applycache.o decode.o snapbuild.o logicalfuncs.o
+
+include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/applycache.c b/src/backend/replication/logical/applycache.c
new file mode 100644
index 0000000..1e08371
--- /dev/null
+++ b/src/backend/replication/logical/applycache.c
@@ -0,0 +1,574 @@
+/*-------------------------------------------------------------------------
+ *
+ * applycache.c
+ *
+ * PostgreSQL logical replay "cache" management
+ *
+ *
+ * Portions Copyright (c) 2012, PostgreSQL Global Development Group
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/applycache.c
+ *
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/xact.h"
+#include "catalog/pg_class.h"
+#include "catalog/pg_control.h"
+#include "replication/applycache.h"
+
+#include "lib/simpleheap.h"
+
+#include "utils/ilist.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;
+
+typedef struct ApplyCacheTXNByIdEnt
+{
+	TransactionId xid;
+	ApplyCacheTXN* txn;
+} ApplyCacheTXNByIdEnt;
+
+static ApplyCacheTXN* ApplyCacheGetTXN(ApplyCache *cache);
+static void ApplyCacheReturnTXN(ApplyCache *cache, ApplyCacheTXN* txn);
+
+static ApplyCacheTXN* ApplyCacheTXNByXid(ApplyCache*, TransactionId xid,
+                                         bool create, bool* is_new);
+
+
+ApplyCache*
+ApplyCacheAllocate(void)
+{
+	ApplyCache* cache = (ApplyCache*)malloc(sizeof(ApplyCache));
+	HASHCTL         hash_ctl;
+
+	if (!cache)
+		elog(ERROR, "Could not allocate the ApplyCache");
+
+	cache->build_snapshots = true;
+
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+
+	cache->context = AllocSetContextCreate(TopMemoryContext,
+	                                       "ApplyCache",
+	                                       ALLOCSET_DEFAULT_MINSIZE,
+	                                       ALLOCSET_DEFAULT_INITSIZE,
+	                                       ALLOCSET_DEFAULT_MAXSIZE);
+
+	hash_ctl.keysize = sizeof(TransactionId);
+	hash_ctl.entrysize = sizeof(ApplyCacheTXNByIdEnt);
+	hash_ctl.hash = tag_hash;
+	hash_ctl.hcxt = cache->context;
+
+	cache->by_txn = hash_create("ApplyCacheByXid", 1000, &hash_ctl,
+	                            HASH_ELEM | HASH_FUNCTION | HASH_CONTEXT);
+
+	cache->nr_cached_transactions = 0;
+	cache->nr_cached_changes = 0;
+	cache->nr_cached_tuplebufs = 0;
+
+	ilist_d_init(&cache->cached_transactions);
+	ilist_d_init(&cache->cached_changes);
+	ilist_s_init(&cache->cached_tuplebufs);
+
+	return cache;
+}
+
+void ApplyCacheFree(ApplyCache* 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);
+}
+
+static ApplyCacheTXN* ApplyCacheGetTXN(ApplyCache *cache)
+{
+	ApplyCacheTXN* txn;
+
+	if (cache->nr_cached_transactions)
+	{
+		cache->nr_cached_transactions--;
+		txn = ilist_container(ApplyCacheTXN, node,
+		                      ilist_d_pop_front(&cache->cached_transactions));
+	}
+	else
+	{
+		txn = (ApplyCacheTXN*)
+			malloc(sizeof(ApplyCacheTXN));
+
+		if (!txn)
+			elog(ERROR, "Could not allocate a ApplyCacheTXN struct");
+	}
+
+	memset(txn, 0, sizeof(ApplyCacheTXN));
+	ilist_d_init(&txn->changes);
+	ilist_d_init(&txn->subtxns);
+	ilist_d_init(&txn->snapshots);
+	ilist_d_init(&txn->commandids);
+
+	return txn;
+}
+
+void ApplyCacheReturnTXN(ApplyCache *cache, ApplyCacheTXN* txn)
+{
+	if(cache->nr_cached_transactions < max_cached_transactions){
+		cache->nr_cached_transactions++;
+		ilist_d_push_front(&cache->cached_transactions, &txn->node);
+	}
+	else{
+		free(txn);
+	}
+}
+
+ApplyCacheChange*
+ApplyCacheGetChange(ApplyCache* cache)
+{
+	ApplyCacheChange* change;
+
+	if (cache->nr_cached_changes)
+	{
+		cache->nr_cached_changes--;
+		change = ilist_container(ApplyCacheChange, node,
+		                         ilist_d_pop_front(&cache->cached_changes));
+	}
+	else
+	{
+		change = (ApplyCacheChange*)malloc(sizeof(ApplyCacheChange));
+
+		if (!change)
+			elog(ERROR, "Could not allocate a ApplyCacheChange struct");
+	}
+
+
+	memset(change, 0, sizeof(ApplyCacheChange));
+	return change;
+}
+
+void
+ApplyCacheReturnChange(ApplyCache* cache, ApplyCacheChange* change)
+{
+	switch(change->action){
+		case APPLY_CACHE_CHANGE_INSERT:
+		case APPLY_CACHE_CHANGE_UPDATE:
+		case APPLY_CACHE_CHANGE_DELETE:
+			if (change->newtuple)
+			{
+				ApplyCacheReturnTupleBuf(cache, change->newtuple);
+				change->newtuple = NULL;
+			}
+
+			if (change->oldtuple)
+			{
+				ApplyCacheReturnTupleBuf(cache, change->oldtuple);
+				change->oldtuple = NULL;
+			}
+
+			if (change->table)
+			{
+				heap_freetuple(change->table);
+				change->table = NULL;
+			}
+			break;
+		case APPLY_CACHE_CHANGE_SNAPSHOT:
+			if (change->snapshot)
+			{
+				/* FIXME: free snapshot */
+				change->snapshot = NULL;
+			}
+		case APPLY_CACHE_CHANGE_COMMAND_ID:
+			break;
+	}
+
+	if(cache->nr_cached_changes < max_cached_changes){
+		cache->nr_cached_changes++;
+		ilist_d_push_front(&cache->cached_changes, &change->node);
+	}
+	else{
+		free(change);
+	}
+}
+
+ApplyCacheTupleBuf*
+ApplyCacheGetTupleBuf(ApplyCache* cache)
+{
+	ApplyCacheTupleBuf* tuple;
+
+	if (cache->nr_cached_tuplebufs)
+	{
+		cache->nr_cached_tuplebufs--;
+		tuple = ilist_container(ApplyCacheTupleBuf, node,
+		                        ilist_s_pop_front(&cache->cached_tuplebufs));
+	}
+	else
+	{
+		tuple =
+			(ApplyCacheTupleBuf*)malloc(sizeof(ApplyCacheTupleBuf));
+
+		if (!tuple)
+			elog(ERROR, "Could not allocate a ApplyCacheTupleBuf struct");
+	}
+
+	return tuple;
+}
+
+void
+ApplyCacheReturnTupleBuf(ApplyCache* cache, ApplyCacheTupleBuf* tuple)
+{
+	if(cache->nr_cached_tuplebufs < max_cached_tuplebufs){
+		cache->nr_cached_tuplebufs++;
+		ilist_s_push_front(&cache->cached_tuplebufs, &tuple->node);
+	}
+	else{
+		free(tuple);
+	}
+}
+
+
+static
+ApplyCacheTXN*
+ApplyCacheTXNByXid(ApplyCache* cache, TransactionId xid, bool create, bool* is_new)
+{
+	ApplyCacheTXNByIdEnt* ent;
+	bool found;
+
+	/* FIXME: add one entry fast-path cache */
+
+	ent = (ApplyCacheTXNByIdEnt*)
+		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 = ApplyCacheGetTXN(cache);
+		ent->txn->xid = xid;
+	}
+
+	if (is_new)
+		*is_new = !found;
+
+	return ent->txn;
+}
+
+void
+ApplyCacheAddChange(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn,
+                    ApplyCacheChange* change)
+{
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, true, NULL);
+	txn->lsn = lsn;
+	ilist_d_push_back(&txn->changes, &change->node);
+}
+
+
+void
+ApplyCacheCommitChild(ApplyCache* cache, TransactionId xid,
+                      TransactionId subxid, XLogRecPtr lsn)
+{
+	ApplyCacheTXN* txn;
+	ApplyCacheTXN* subtxn;
+
+	subtxn = ApplyCacheTXNByXid(cache, subxid, false, NULL);
+
+	/*
+	 * No need to do anything if that subtxn didn't contain any changes
+	 */
+	if (!subtxn)
+		return;
+
+	subtxn->lsn = lsn;
+
+	txn = ApplyCacheTXNByXid(cache, xid, true, NULL);
+
+	ilist_d_push_back(&txn->subtxns, &subtxn->node);
+}
+
+typedef struct ApplyCacheIterTXNState
+{
+	simpleheap *heap;
+} ApplyCacheIterTXNState;
+
+static int
+ApplyCacheIterCompare(simpleheap_kv* a, simpleheap_kv* b)
+{
+	ApplyCacheChange *change_a = ilist_container(ApplyCacheChange, node, a->key);
+	ApplyCacheChange *change_b = ilist_container(ApplyCacheChange, node, b->key);
+
+	if (change_a->lsn < change_b->lsn)
+		return -1;
+
+	else if (change_a->lsn == change_b->lsn)
+		return 0;
+
+	return 1;
+}
+
+static ApplyCacheIterTXNState*
+ApplyCacheIterTXNInit(ApplyCache* cache, ApplyCacheTXN* txn);
+
+static ApplyCacheChange*
+ApplyCacheIterTXNNext(ApplyCache* cache, ApplyCacheIterTXNState* state);
+
+static void
+ApplyCacheIterTXNFinish(ApplyCache* cache, ApplyCacheIterTXNState* state);
+
+
+
+static ApplyCacheIterTXNState*
+ApplyCacheIterTXNInit(ApplyCache* cache, ApplyCacheTXN* txn)
+{
+	size_t nr_txns = 0; /* main txn */
+	ApplyCacheIterTXNState *state;
+	ilist_d_node* cur_txn_i;
+	ApplyCacheTXN *cur_txn;
+	ApplyCacheChange *cur_change;
+
+	if (!ilist_d_is_empty(&txn->changes))
+		nr_txns++;
+
+	/* count how large our heap must be */
+	ilist_d_foreach(cur_txn_i, &txn->subtxns)
+	{
+		cur_txn = ilist_container(ApplyCacheTXN, node, cur_txn_i);
+
+		if (!ilist_d_is_empty(&cur_txn->changes))
+			nr_txns++;
+	}
+
+	/* allocate array for our heap */
+	state = palloc0(sizeof(ApplyCacheIterTXNState));
+
+	state->heap = simpleheap_allocate(nr_txns);
+	state->heap->compare = ApplyCacheIterCompare;
+
+	/* fill array with elements, heap condition not yet fullfilled */
+	if (!ilist_d_is_empty(&txn->changes))
+	{
+		cur_change = ilist_d_front_unchecked(ApplyCacheChange, node, &txn->changes);
+
+		simpleheap_add_unordered(state->heap, &cur_change->node, txn);
+	}
+
+	ilist_d_foreach(cur_txn_i, &txn->subtxns)
+	{
+		cur_txn = ilist_container(ApplyCacheTXN, node, cur_txn_i);
+
+		if (!ilist_d_is_empty(&cur_txn->changes))
+		{
+			cur_change = ilist_d_front_unchecked(ApplyCacheChange, 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;
+}
+
+static ApplyCacheChange*
+ApplyCacheIterTXNNext(ApplyCache* cache, ApplyCacheIterTXNState* state)
+{
+	ApplyCacheTXN *txn = NULL;
+	ApplyCacheChange *change;
+	simpleheap_kv *kv;
+
+	/*
+	 * Do a k-way merge between transactions/subtransactions to extract changes
+	 * merged by the lsn of their change. 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.
+	 */
+
+	/* nothing there anymore */
+	if (state->heap->size == 0)
+		return NULL;
+
+	kv = simpleheap_first(state->heap);
+
+	change = ilist_container(ApplyCacheChange, node, kv->key);
+
+	txn = (ApplyCacheTXN*)kv->value;
+
+	if (!ilist_d_has_next(&txn->changes, &change->node))
+	{
+		simpleheap_remove_first(state->heap);
+	}
+	else
+	{
+		simpleheap_change_key(state->heap, change->node.next);
+	}
+	return change;
+}
+
+static void
+ApplyCacheIterTXNFinish(ApplyCache* cache, ApplyCacheIterTXNState* state)
+{
+	simpleheap_free(state->heap);
+	pfree(state);
+}
+
+
+static void
+ApplyCacheCleanupTXN(ApplyCache* cache, ApplyCacheTXN* txn)
+{
+	bool found;
+	ilist_d_node* cur_change, *next_change;
+	ilist_d_node* cur_txn, *next_txn;
+
+	/* cleanup transactions & changes */
+	ilist_d_foreach_modify (cur_txn, next_txn, &txn->subtxns)
+	{
+		ApplyCacheTXN* subtxn = ilist_container(ApplyCacheTXN, node, cur_txn);
+
+		ilist_d_foreach_modify (cur_change, next_change, &subtxn->changes)
+		{
+			ApplyCacheChange* change =
+				ilist_container(ApplyCacheChange, node, cur_change);
+
+			ApplyCacheReturnChange(cache, change);
+		}
+		ApplyCacheReturnTXN(cache, subtxn);
+	}
+
+	ilist_d_foreach_modify (cur_change, next_change, &txn->changes)
+	{
+		ApplyCacheChange* change =
+			ilist_container(ApplyCacheChange, node, cur_change);
+
+		ApplyCacheReturnChange(cache, change);
+	}
+
+	/* now remove reference from cache */
+	hash_search(cache->by_txn,
+	            (void *)&txn->xid,
+	            HASH_REMOVE,
+	            &found);
+	Assert(found);
+
+	ApplyCacheReturnTXN(cache, txn);
+}
+void
+ApplyCacheCommit(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, false, NULL);
+	ApplyCacheIterTXNState* iterstate;
+	ApplyCacheChange* change;
+	CommandId command_id;
+	Snapshot snapshot_mvcc = NULL;
+
+	if (!txn)
+		return;
+
+	txn->lsn = lsn;
+
+	cache->begin(cache, txn);
+
+	PG_TRY();
+	{
+		iterstate = ApplyCacheIterTXNInit(cache, txn);
+		while((change = ApplyCacheIterTXNNext(cache, iterstate)))
+		{
+			switch(change->action){
+				case APPLY_CACHE_CHANGE_INSERT:
+				case APPLY_CACHE_CHANGE_UPDATE:
+				case APPLY_CACHE_CHANGE_DELETE:
+					Assert(snapshot_mvcc != NULL);
+					cache->apply_change(cache, txn, txn /*FIXME*/, change);
+					break;
+				case APPLY_CACHE_CHANGE_SNAPSHOT:
+					/*
+					 * the first snapshot seen in a transaction is its mvcc
+					 * snapshot
+					 */
+					if (!snapshot_mvcc)
+						snapshot_mvcc = change->snapshot;
+					SetupDecodingSnapshots(change->snapshot);
+					break;
+				case APPLY_CACHE_CHANGE_COMMAND_ID:
+					/* FIXME */
+					command_id = change->command_id;
+					break;
+			}
+		}
+
+		ApplyCacheIterTXNFinish(cache, iterstate);
+
+		cache->commit(cache, txn);
+
+		ApplyCacheCleanupTXN(cache, txn);
+		RevertFromDecodingSnapshots();
+	}
+	PG_CATCH();
+	{
+		RevertFromDecodingSnapshots();
+		PG_RE_THROW();
+	}
+	PG_END_TRY();
+}
+
+void
+ApplyCacheAbort(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, false, NULL);
+
+	/* no changes in this commit */
+	if (!txn)
+		return;
+
+	ApplyCacheCleanupTXN(cache, txn);
+}
+
+bool
+ApplyCacheIsXidKnown(ApplyCache* cache, TransactionId xid)
+{
+	bool is_new;
+	/* FIXME: for efficiency reasons we create the xid here, that doesn't seem
+	 * like a good idea though */
+	ApplyCacheTXNByXid(cache, xid, true, &is_new);
+
+	/* no changes in this commit */
+	return !is_new;
+}
+
+void
+ApplyCacheAddBaseSnapshot(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn, Snapshot snap)
+{
+	ApplyCacheChange *change = ApplyCacheGetChange(cache);
+	change->snapshot = snap;
+	change->action = APPLY_CACHE_CHANGE_SNAPSHOT;
+
+	ApplyCacheAddChange(cache, xid, lsn, change);
+}
diff --git a/src/backend/replication/logical/decode.c b/src/backend/replication/logical/decode.c
new file mode 100644
index 0000000..244dd7b
--- /dev/null
+++ b/src/backend/replication/logical/decode.c
@@ -0,0 +1,366 @@
+/*-------------------------------------------------------------------------
+ *
+ * decode.c
+ *
+ * Decodes wal records from an xlogreader.h callback into an applycache.
+ *
+ * Portions Copyright (c) 2010-2012, PostgreSQL Global Development Group
+ *
+ * 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 struct.
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/decode.c
+ *
+ */
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/transam.h"
+#include "access/xlog_internal.h"
+#include "access/xact.h"
+#include "access/heapam_xlog.h"
+
+#include "catalog/pg_control.h"
+
+#include "replication/applycache.h"
+#include "replication/decode.h"
+#include "replication/snapbuild.h"
+
+#include "utils/memutils.h"
+#include "utils/syscache.h"
+#include "utils/lsyscache.h"
+
+static void DecodeXLogTuple(char* data, Size len, ApplyCacheTupleBuf* tuple);
+
+static void DecodeInsert(ApplyCache *cache, XLogRecordBuffer* buf);
+
+static void DecodeUpdate(ApplyCache *cache, XLogRecordBuffer* buf);
+
+static void DecodeDelete(ApplyCache *cache, XLogRecordBuffer* buf);
+
+static void DecodeNewpage(ApplyCache *cache, XLogRecordBuffer* buf);
+static void DecodeMultiInsert(ApplyCache *cache, XLogRecordBuffer* buf);
+
+static void DecodeCommit(ApplyCache* cache, XLogRecordBuffer* buf, TransactionId xid,
+	                     TransactionId *sub_xids, int nsubxacts);
+
+
+void DecodeRecordIntoApplyCache(ReaderApplyState *state, XLogRecordBuffer* buf)
+{
+	XLogRecord* r = &buf->record;
+	uint8 info = r->xl_info & ~XLR_INFO_MASK;
+	ApplyCache *cache = state->apply_cache;
+	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(cache);
+	}
+
+	/*
+	 * 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 ApplyCache before we add tuples using them
+	 */
+	action = SnapBuildCallback(cache, state->snapstate, buf);
+
+	if (action == SNAPBUILD_SKIP)
+		return;
+
+	switch (r->xl_rmid)
+	{
+		case RM_HEAP_ID:
+		{
+			info &= XLOG_HEAP_OPMASK;
+			switch (info)
+			{
+				case XLOG_HEAP_INSERT:
+					DecodeInsert(cache, 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(cache, buf);
+					break;
+
+				case XLOG_HEAP_NEWPAGE:
+					DecodeNewpage(cache, buf);
+					break;
+
+				case XLOG_HEAP_DELETE:
+					DecodeDelete(cache, buf);
+					break;
+				default:
+					break;
+			}
+			break;
+		}
+		case RM_HEAP2_ID:
+		{
+			info &= XLOG_HEAP_OPMASK;
+			switch (info)
+			{
+				case XLOG_HEAP2_MULTI_INSERT:
+					DecodeMultiInsert(cache, 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: this is not really allowed if there is no subtransactions */
+					sub_xids = (TransactionId *) &(xlrec->xnodes[xlrec->nrels]);
+					DecodeCommit(cache, 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(cache, 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(cache, buf, r->xl_xid, xlrec->subxacts,
+					             xlrec->nsubxacts);
+					break;
+				}
+				case XLOG_XACT_ABORT:
+				case XLOG_XACT_ABORT_PREPARED:
+				{
+					TransactionId *sub_xids;
+					xl_xact_abort *xlrec = (xl_xact_abort*)buf->record_data;
+					int i;
+
+					/* FIXME: this is not really allowed if there is no subtransaction */
+					sub_xids = (TransactionId *) &(xlrec->xnodes[xlrec->nrels]);
+
+					for(i = 0; i < xlrec->nsubxacts; i++)
+					{
+						ApplyCacheAbort(cache, *sub_xids, buf->origptr);
+						sub_xids += 1;
+					}
+
+					/* TODO: check that this also contains not-yet-aborted subtxns */
+					ApplyCacheAbort(cache, r->xl_xid, buf->origptr);
+
+					elog(WARNING, "ABORT %u", 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(ApplyCache* cache, XLogRecordBuffer* buf, TransactionId xid,
+             TransactionId *sub_xids, int nsubxacts)
+{
+	int i;
+
+	for (i = 0; i < nsubxacts; i++)
+	{
+		ApplyCacheCommitChild(cache, xid, *sub_xids, buf->origptr);
+		sub_xids++;
+	}
+
+	/* replay actions of all transaction + subtransactions in order */
+	ApplyCacheCommit(cache, xid, buf->origptr);
+}
+
+static void DecodeInsert(ApplyCache *cache, XLogRecordBuffer* buf)
+{
+	XLogRecord* r = &buf->record;
+	xl_heap_insert *xlrec = (xl_heap_insert *) buf->record_data;
+
+	ApplyCacheChange* change;
+
+	if (r->xl_info & XLR_BKP_BLOCK_1
+	    && r->xl_len < (SizeOfHeapUpdate + SizeOfHeapHeader))
+	{
+		elog(FATAL, "huh, no tuple data on wal_level = logical?");
+	}
+
+	change = ApplyCacheGetChange(cache);
+	change->action = APPLY_CACHE_CHANGE_INSERT;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	change->newtuple = ApplyCacheGetTupleBuf(cache);
+
+	DecodeXLogTuple((char*)xlrec + SizeOfHeapInsert,
+	                r->xl_len - SizeOfHeapInsert,
+	                change->newtuple);
+
+	ApplyCacheAddChange(cache, r->xl_xid, buf->origptr, change);
+}
+
+static void
+DecodeUpdate(ApplyCache *cache, XLogRecordBuffer* buf)
+{
+	XLogRecord* r = &buf->record;
+	xl_heap_update *xlrec = (xl_heap_update *) buf->record_data;
+
+
+	ApplyCacheChange* change;
+
+	if ((r->xl_info & XLR_BKP_BLOCK_1 || r->xl_info & XLR_BKP_BLOCK_2) &&
+	    (r->xl_len < (SizeOfHeapUpdate + SizeOfHeapHeader)))
+	{
+		elog(FATAL, "huh, no tuple data on wal_level = logical?");
+	}
+
+	change = ApplyCacheGetChange(cache);
+	change->action = APPLY_CACHE_CHANGE_UPDATE;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	/* FIXME: need to save the old tuple as well if we want primary key changes to work. */
+	change->newtuple = ApplyCacheGetTupleBuf(cache);
+
+	DecodeXLogTuple((char*)xlrec + SizeOfHeapUpdate,
+	                r->xl_len - SizeOfHeapUpdate,
+	                change->newtuple);
+
+	ApplyCacheAddChange(cache, r->xl_xid, buf->origptr, change);
+}
+
+static void DecodeDelete(ApplyCache *cache, XLogRecordBuffer* buf)
+{
+	XLogRecord* r = &buf->record;
+
+	xl_heap_delete *xlrec = (xl_heap_delete *) buf->record_data;
+
+	ApplyCacheChange* change;
+
+	change = ApplyCacheGetChange(cache);
+	change->action = APPLY_CACHE_CHANGE_DELETE;
+
+	memcpy(&change->relnode, &xlrec->target.node, sizeof(RelFileNode));
+
+	if (r->xl_len <= (SizeOfHeapDelete + SizeOfHeapHeader))
+	{
+		elog(FATAL, "huh, no primary key for a delete on wal_level = logical?");
+	}
+
+	change->oldtuple = ApplyCacheGetTupleBuf(cache);
+
+	DecodeXLogTuple((char*)xlrec + SizeOfHeapDelete,
+	                r->xl_len - SizeOfHeapDelete,
+	                change->oldtuple);
+
+	ApplyCacheAddChange(cache, r->xl_xid, buf->origptr, change);
+}
+
+
+static void
+DecodeNewpage(ApplyCache *cache, XLogRecordBuffer* buf)
+{
+	elog(WARNING, "skipping XLOG_HEAP_NEWPAGE record because we are too dumb");
+}
+
+static void
+DecodeMultiInsert(ApplyCache *cache, XLogRecordBuffer* buf)
+{
+	elog(WARNING, "skipping XLOG_HEAP2_MULTI_INSERT record because we are too dumb");
+}
+
+
+static void DecodeXLogTuple(char* data, Size len, ApplyCacheTupleBuf* 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);
+
+	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..035c48a
--- /dev/null
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -0,0 +1,237 @@
+/*-------------------------------------------------------------------------
+ *
+ * logicalfuncs.c
+ *
+ *     Support functions for using xlog decoding
+ *
+ * NOTE:
+ *     Nothing in here should be sued for anythign but debugging!
+ *
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/snapbuild.c
+ *
+ */
+
+#include "postgres.h"
+
+#include "access/xlogreader.h"
+
+#include "catalog/pg_class.h"
+#include "catalog/pg_type.h"
+
+#include "replication/applycache.h"
+#include "replication/decode.h"
+#include "replication/walreceiver.h"
+/*FIXME: XLogRead*/
+#include "replication/walsender_private.h"
+
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/syscache.h"
+#include "utils/typcache.h"
+
+
+
+/* We don't need no header */
+extern Datum
+decode_xlog(PG_FUNCTION_ARGS);
+
+
+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;
+	DecodeRecordIntoApplyCache(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);
+	}
+}
+
+static
+void decode_begin_txn(ApplyCache* cache, ApplyCacheTXN* txn)
+{
+	elog(WARNING, "BEGIN");
+}
+
+static void
+decode_commit_txn(ApplyCache* cache, ApplyCacheTXN* txn)
+{
+	elog(WARNING, "COMMIT");
+}
+
+/* don't want to include that header */
+extern HeapTuple
+LookupTableByRelFileNode(RelFileNode* r);
+
+
+/* This is is just for demonstration, don't ever use this code for anything real! */
+static void
+decode_change(ApplyCache* cache, ApplyCacheTXN* txn, ApplyCacheTXN* subtxn, ApplyCacheChange* change)
+{
+	InvalidateSystemCaches();
+
+	if (change->action == APPLY_CACHE_CHANGE_INSERT)
+	{
+		StringInfoData s;
+		HeapTuple table = LookupTableByRelFileNode(&change->relnode);
+		Form_pg_class class_form;
+		HeapTuple	typeTuple;
+		Form_pg_type pt;
+		TupleDesc	tupdesc;
+		int			i;
+
+		if (!table)
+		{
+			elog(LOG, "couldn't lookup %u", change->relnode.relNode);
+			return;
+		}
+
+		class_form = (Form_pg_class) GETSTRUCT(table);
+
+		initStringInfo(&s);
+
+		tupdesc = lookup_rowtype_tupdesc(class_form->reltype, -1);
+
+		for (i = 0; i < tupdesc->natts; i++)
+		{
+			Oid			typid, typoutput;
+			bool		typisvarlena;
+			Datum		origval, val;
+			char        *outputstr;
+			bool        isnull;
+			if (tupdesc->attrs[i]->attisdropped)
+				continue;
+			if (tupdesc->attrs[i]->attnum < 0)
+				continue;
+
+			typid = tupdesc->attrs[i]->atttypid;
+
+			typeTuple = SearchSysCache1(TYPEOID, ObjectIdGetDatum(typid));
+			if (!HeapTupleIsValid(typeTuple))
+				elog(ERROR, "cache lookup failed for type %u", typid);
+			pt = (Form_pg_type) GETSTRUCT(typeTuple);
+
+			appendStringInfo(&s, " %s[%s]",
+			                 NameStr(tupdesc->attrs[i]->attname),
+			                 NameStr(pt->typname));
+
+			getTypeOutputInfo(typid,
+			                  &typoutput, &typisvarlena);
+
+			ReleaseSysCache(typeTuple);
+
+			origval = heap_getattr(&change->newtuple->tuple, i + 1, tupdesc, &isnull);
+
+			if (typisvarlena && !isnull)
+				val = PointerGetDatum(PG_DETOAST_DATUM(origval));
+			else
+				val = origval;
+
+			outputstr = OidOutputFunctionCall(typoutput, val);
+
+			appendStringInfo(&s, ":%s", isnull ? "(null)" : outputstr);
+		}
+		ReleaseTupleDesc(tupdesc);
+
+		elog(WARNING, "tuple is:%s", s.data);
+	}
+}
+
+/* test the xlog decoding infrastructure from lsn, to lsn */
+Datum
+decode_xlog(PG_FUNCTION_ARGS)
+{
+	char* start = PG_GETARG_CSTRING(0);
+	char* end = PG_GETARG_CSTRING(1);
+
+	ApplyCache *apply_cache;
+	XLogReaderState *xlogreader_state = XLogReaderAllocate();
+	ReaderApplyState *apply_state;
+
+	XLogRecPtr startpoint;
+	XLogRecPtr endpoint;
+
+	uint32		hi,
+				lo;
+
+	if (sscanf(start, "%X/%X",
+	           &hi, &lo) != 2)
+		elog(ERROR, "unparseable xlog pos");
+	startpoint = ((uint64) hi) << 32 | lo;
+
+	elog(LOG, "starting to parse at %X/%X", hi, lo);
+
+	if (sscanf(end, "%X/%X",
+	           &hi, &lo) != 2)
+		elog(ERROR, "unparseable xlog pos");
+	endpoint = ((uint64) hi) << 32 | lo;
+
+	elog(LOG, "end parse at %X/%X", hi, lo);
+
+	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");
+
+	xlogreader_state->startptr = startpoint;
+	xlogreader_state->curptr = startpoint;
+	xlogreader_state->endptr = endpoint;
+
+	apply_state = (ReaderApplyState*)xlogreader_state->private_data;
+
+	/*
+	 * allocate an ApplyCache that will apply data using lowlevel calls
+	 * without type conversion et al. This requires binary compatibility
+	 * between both systems.
+	 * XXX: This would be the place too hook different apply methods, like
+	 * producing sql and applying it.
+	 */
+	apply_cache = ApplyCacheAllocate();
+	apply_cache->begin = decode_begin_txn;
+	apply_cache->apply_change = decode_change;
+	apply_cache->commit = decode_commit_txn;
+
+	apply_state->apply_cache = apply_cache;
+
+	XLogReaderRead(xlogreader_state);
+
+	PG_RETURN_BOOL(true);
+}
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
new file mode 100644
index 0000000..05b176d
--- /dev/null
+++ b/src/backend/replication/logical/snapbuild.c
@@ -0,0 +1,1045 @@
+/*-------------------------------------------------------------------------
+ *
+ * snapbuild.c
+ *
+ *     Support for building timetravel snapshots based on the contents of the
+ *     wal
+ *
+ * NOTE:
+ *     This is complex, in-progress and underdocumented.
+ *
+ *     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 user of an ApplyCache.
+ *
+ *     The percentage of transactions modifying the catalog should be fairly
+ *     small, so instead of keeping track of all running transactions an
+ *     treating everything inside (xmin, xmax) thats not running as commited we
+ *     do the contrary. That, and other implementation details, neccisate using
+ *     our own ->satisfies visibility routine.
+ *     In contrast to a class SnapshotNow which doesn't need any data this
+ *     module provides something that *behaves* like a SnapshotNow would have
+ *     back then (minus some races). Minus some minor things a SnapshotNow
+ *     behaves like a SnapshotMVCC taken exactly in the moment the SnapshotNow
+ *     was used. Because of that we simply model our timetravel-SnapshotNow's
+ *     as mvcc Snapshots.
+ *
+ *     To replace the normal handling of SnapshotNow snapshots use the
+ *     SetupDecodingSnapshots/RevertFromDecodingSnapshots functions. Be careful
+ *     to handle errors properly, otherwise the rest of the session will have
+ *     very strange behaviour.
+ *
+ * Portions Copyright (c) 1996-2012, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/snapbuild.c
+ *
+ */
+
+#include "postgres.h"
+
+#include "access/heapam_xlog.h"
+#include "access/rmgr.h"
+#include "access/transam.h"
+#include "access/xlogreader.h"
+#include "access/xact.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/applycache.h"
+#include "replication/snapbuild.h"
+
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/snapshot.h"
+#include "utils/syscache.h"
+#include "utils/tqual.h"
+
+#include "storage/standby.h"
+
+typedef struct SnapstateTxnEnt
+{
+	TransactionId xid;
+	bool does_timetravel;
+} SnapstateTxnEnt;
+
+
+static bool
+SnapBuildHasCatalogChanges(Snapstate* snapstate, TransactionId xid, RelFileNode* relfilenode);
+
+/* 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, TransactionId xid, int nsubxacts,
+                   TransactionId* subxacts);
+
+/* ->running manipulation */
+static bool
+SnapBuildTxnRunning(Snapstate* snapstate, TransactionId xid);
+
+static void
+SnapBuildReserveRunning(Snapstate *snapstate, Size count);
+
+static void
+SnapBuildSortRunning(Snapstate *snapstate);
+
+static void
+SnapBuildAddRunningTxn(Snapstate *snapstate, TransactionId xid);
+
+
+/* ->committed manipulation */
+static void
+SnapBuildPurgeCommittedTxn(Snapstate* snapstate);
+
+static void
+SnapBuildCommitTxn(Snapstate* snapstate, TransactionId xid, int nsubxacts,
+                   TransactionId* subxacts);
+
+
+/* snapshot building/manipulation/distribution functions */
+static void
+SnapBuildDistributeSnapshotNow(Snapstate* snapstate, TransactionId xid);
+
+static Snapshot
+SnapBuildBuildSnapshot(Snapstate *snapstate, TransactionId xid);
+
+
+HeapTuple
+LookupTableByRelFileNode(RelFileNode* relfilenode)
+{
+	Oid spc;
+
+	InvalidateSystemCaches();
+
+	/*
+	 * relations in the default tablespace are stored with a reltablespace = 0
+	 * for some reason.
+	 */
+	spc = relfilenode->spcNode == DEFAULTTABLESPACE_OID ?
+		0 : relfilenode->spcNode;
+
+	return SearchSysCacheCopy2(RELFILENODE,
+	                           spc,
+	                           relfilenode->relNode);
+}
+
+Snapstate*
+AllocateSnapshotBuilder(ApplyCache *applycache)
+{
+	Snapstate *snapstate = malloc(sizeof(Snapstate));
+	HASHCTL hash_ctl;
+
+	snapstate->state = SNAPBUILD_START;
+	snapstate->valid_after = InvalidTransactionId;
+
+	snapstate->nrrunning = 0;
+	snapstate->nrrunning_initial = 0;
+	snapstate->nrrunning_space = 0;
+	snapstate->running = NULL;
+
+	snapstate->nrcommitted = 0;
+	snapstate->nrcommitted_space = 128;
+	snapstate->committed = malloc(snapstate->nrcommitted_space * sizeof(TransactionId));
+	if (!snapstate->committed)
+		elog(ERROR, "could not allocate memory for snapstate->committed");
+
+	memset(&hash_ctl, 0, sizeof(hash_ctl));
+	hash_ctl.keysize = sizeof(TransactionId);
+	hash_ctl.entrysize = sizeof(SnapstateTxnEnt);
+	hash_ctl.hash = tag_hash;
+	hash_ctl.hcxt = TopMemoryContext;
+
+	snapstate->by_txn = hash_create("SnapstateByXid", 1000, &hash_ctl,
+	                            HASH_ELEM | HASH_FUNCTION);
+
+	elog(LOG, "allocating snapshotbuilder");
+	return snapstate;
+}
+
+void
+FreeSnapshotBuilder(Snapstate* snapstate)
+{
+	hash_destroy(snapstate->by_txn);
+	free(snapstate);
+}
+
+SnapBuildAction
+SnapBuildCallback(ApplyCache *applycache, Snapstate* snapstate, XLogRecordBuffer* buf)
+{
+	XLogRecord* r = &buf->record;
+	uint8 info = r->xl_info & ~XLR_INFO_MASK;
+	TransactionId xid = buf->record.xl_xid;
+
+	/*  relfilenode with the table changes have happened in */
+	bool found_changes = false;
+
+	RelFileNode *relfilenode;
+	SnapBuildAction ret = SNAPBUILD_SKIP;
+
+	{
+		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);
+	}
+
+	if (snapstate->state <= SNAPBUILD_FULL_SNAPSHOT)
+	{
+		if (r->xl_rmid == RM_STANDBY_ID &&
+		   info == XLOG_RUNNING_XACTS)
+		{
+			xl_running_xacts *running = (xl_running_xacts*)buf->record_data;
+
+			if (!running->subxid_overflow)
+			{
+				snapstate->state = SNAPBUILD_FULL_SNAPSHOT;
+
+
+				snapstate->xmin = running->oldestRunningXid;
+				TransactionIdRetreat(snapstate->xmin);
+				snapstate->xmax = running->latestCompletedXid;
+
+				snapstate->nrrunning = running->xcnt;
+				snapstate->nrrunning_initial = running->xcnt;
+				snapstate->nrrunning_space = running->xcnt;
+
+				SnapBuildReserveRunning(snapstate, snapstate->nrrunning_space);
+
+				memcpy(snapstate->running, running->xids,
+				       snapstate->nrrunning_initial * sizeof(TransactionId));
+
+				/* sort so we can do a binary search */
+				SnapBuildSortRunning(snapstate);
+
+				if (running->xcnt)
+				{
+					snapstate->xmin_running = snapstate->running[0];
+					snapstate->xmax_running = snapstate->running[running->xcnt - 1];
+				}
+				else
+				{
+					snapstate->xmin_running = InvalidTransactionId;
+					snapstate->xmax_running = InvalidTransactionId;
+					/* FIXME: abort everything considered running */
+					snapstate->state = SNAPBUILD_CONSISTENT;
+				}
+				elog(LOG, "built initial snapshot (via running xacts). Done: %i",
+				     snapstate->state == SNAPBUILD_CONSISTENT);
+			}
+			else if (TransactionIdIsValid(snapstate->valid_after))
+			{
+				if (NormalTransactionIdPrecedes(snapstate->valid_after, running->oldestRunningXid))
+				{
+					snapstate->state = SNAPBUILD_FULL_SNAPSHOT;
+					snapstate->xmin_running = InvalidTransactionId;
+					snapstate->xmax_running = InvalidTransactionId;
+					/* FIXME: copy all transactions we have seen starting to ->running */
+				}
+			}
+			else
+			{
+				snapstate->state = SNAPBUILD_INITIAL_POINT;
+
+				snapstate->valid_after = running->nextXid;
+				elog(INFO, "starting to build snapshot, valid_after xid: %u",
+				     snapstate->valid_after);
+			}
+		}
+		/* we know nothing has been in progress at this point... */
+		else if (r->xl_rmid == RM_XLOG_ID &&
+		        info == XLOG_CHECKPOINT_SHUTDOWN)
+		{
+			CheckPoint* checkpoint = (CheckPoint*)buf->record_data;
+
+			snapstate->xmin = checkpoint->nextXid;
+			snapstate->xmax = checkpoint->nextXid;
+
+			snapstate->nrrunning = 0;
+			snapstate->nrrunning_initial = 0;
+			snapstate->nrrunning_space = 0;
+			free(snapstate->running);
+			snapstate->running = NULL;
+
+			snapstate->state = SNAPBUILD_CONSISTENT;
+
+			elog(LOG, "built initial snapshot (via shutdown)!!!!");
+			/*FIXME: cleanup state */
+		}
+		else if(r->xl_rmid == RM_XLOG_ID &&
+		        info == XLOG_CHECKPOINT_ONLINE)
+		{
+			/* FIXME: Check whether there is a valid state dumped to disk */
+		}
+	}
+
+	if (snapstate->state == SNAPBUILD_START)
+		return SNAPBUILD_SKIP;
+
+	switch (r->xl_rmid)
+	{
+		case RM_XLOG_ID:
+		{
+			switch (info)
+			{
+				case XLOG_CHECKPOINT_SHUTDOWN:
+				{
+					CheckPoint* checkpoint = (CheckPoint*)buf->record_data;
+
+					/*
+					 * we know nothing can be running anymore, normal
+					 * transaction state is sufficient
+					 */
+
+					/* no need to have any transaction state anymore */
+#ifdef NOT_YES
+					for (/*FIXME*/)
+					{
+						SnapBuildAbortTxn(snapstate, xid);
+					}
+#endif
+					snapstate->xmin = checkpoint->nextXid;
+					TransactionIdRetreat(snapstate->xmin);
+					snapstate->xmax = checkpoint->nextXid;
+
+					free(snapstate->running);
+					snapstate->running = NULL;
+					snapstate->nrrunning = 0;
+					snapstate->nrrunning_initial = 0;
+					snapstate->nrrunning_space = 0;
+
+					/*FIXME: cleanup state */
+
+
+					ret = SNAPBUILD_DECODE;
+
+					break;
+				}
+				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;
+					snapstate->xmin = running->oldestRunningXid;
+					TransactionIdRetreat(snapstate->xmin);
+					snapstate->xmax = running->latestCompletedXid;
+					TransactionIdAdvance(snapstate->xmax);
+
+					SnapBuildPurgeCommittedTxn(snapstate);
+
+					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;
+
+					SnapBuildCommitTxn(snapstate, xid, xlrec->nsubxacts,
+					                   (TransactionId*)&xlrec->xnodes);
+					ret = SNAPBUILD_DECODE;
+
+					break;
+				}
+				case XLOG_XACT_COMMIT_COMPACT:
+				{
+					xl_xact_commit_compact* xlrec =
+						(xl_xact_commit_compact*)buf->record_data;
+
+					SnapBuildCommitTxn(snapstate, xid, xlrec->nsubxacts,
+					                   xlrec->subxacts);
+					ret = SNAPBUILD_DECODE;
+					break;
+				}
+				case XLOG_XACT_COMMIT_PREPARED:
+				{
+					xl_xact_commit_prepared* xlrec =
+						(xl_xact_commit_prepared*)buf->record_data;
+
+					SnapBuildCommitTxn(snapstate, 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);
+					ret = SNAPBUILD_DECODE;
+
+				}
+				case XLOG_XACT_ABORT_PREPARED:
+				{
+					xl_xact_abort_prepared* xlrec =
+						(xl_xact_abort_prepared*)buf->record_data;
+
+					SnapBuildAbortTxn(snapstate, xid, xlrec->arec.nsubxacts,
+					                  (TransactionId*)&xlrec->arec.xnodes);
+					ret = SNAPBUILD_DECODE;
+				}
+				case XLOG_XACT_ASSIGNMENT:
+				case XLOG_XACT_PREPARE: /* boring? */
+				default:
+					break;
+					;
+			}
+			break;
+		}
+		case RM_HEAP_ID:
+		{
+			switch (info & XLOG_HEAP_OPMASK)
+			{
+				/* XXX: this only happens for "irrelevant" changes? Ignore for now */
+				case XLOG_HEAP_INPLACE:
+				{
+					xl_heap_inplace *xlrec = (xl_heap_inplace*)buf->record_data;
+					relfilenode = &xlrec->target.node;
+					found_changes = false; /* <----- LOOK */
+					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;
+					relfilenode = &xlrec->target.node;
+					found_changes = true;
+					break;
+				}
+				case XLOG_HEAP_UPDATE:
+				case XLOG_HEAP_HOT_UPDATE:
+				{
+					xl_heap_update *xlrec = (xl_heap_update*)buf->record_data;
+					relfilenode = &xlrec->target.node;
+					found_changes = true;
+					break;
+				}
+				case XLOG_HEAP_DELETE:
+				{
+					xl_heap_delete *xlrec = (xl_heap_delete*)buf->record_data;
+					relfilenode = &xlrec->target.node;
+					found_changes = true;
+					break;
+				}
+				default:
+					;
+			}
+			break;
+		}
+		case RM_HEAP2_ID:
+		{
+			/* some HEAP2 things don't necessarily happen in a transaction? */
+			if (!TransactionIdIsValid(xid))
+				break;
+
+			switch (info)
+			{
+				case XLOG_HEAP2_MULTI_INSERT:
+				{
+					xl_heap_multi_insert *xlrec =
+						(xl_heap_multi_insert*)buf->record_data;
+
+					relfilenode = &xlrec->node;
+
+					found_changes = true;
+
+					/*
+					 * we only decode the first tuple as all the following ones
+					 * will have the same cmin (and no cmax)
+					 */
+					break;
+				}
+				default:
+					;
+			}
+		}
+		break;
+	}
+
+
+
+	if (found_changes)
+	{
+		/*
+		 * we unfortunately cannot access the catalog of other databases, so
+		 * don't think about changes in them
+		 */
+		if (relfilenode->dbNode != MyDatabaseId)
+			;
+		/*
+		 * we need to keep track of new transactions while we didn't know what
+		 * was already running. Only actual data changes are relevant, so its
+		 * fine to track them here.
+		 */
+		else if (snapstate->state < SNAPBUILD_FULL_SNAPSHOT)
+			SnapBuildAddRunningTxn(snapstate, xid);
+		/*
+		 * 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 &&
+		         SnapBuildTxnRunning(snapstate, xid))
+			;
+		else
+		{
+			bool does_timetravel;
+			bool old_tx = ApplyCacheIsXidKnown(applycache, xid);
+			bool found;
+			SnapstateTxnEnt *ent;
+
+			Assert(TransactionIdIsNormal(xid));
+			Assert(!SnapBuildTxnRunning(snapstate, xid));
+
+
+
+			ent = hash_search(snapstate->by_txn,
+			                  (void *)&xid,
+			                  HASH_FIND,
+			                  &found);
+
+			/* FIXME: For now skip transactions with catalog changes entirely */
+			if (ent && ent->does_timetravel)
+				does_timetravel = true;
+			else
+				does_timetravel = SnapBuildHasCatalogChanges(snapstate, xid, relfilenode);
+
+			/*
+			 * we don't add catalog changes to the applycache, we could use
+			 * them to queue local cache inval messages for catalog tables if
+			 * the relmapper would map from relfilenode to relid with correct
+			 * visibility rules.
+			 */
+			if (!does_timetravel)
+				ret = SNAPBUILD_DECODE;
+
+			elog(LOG, "found changes in xid %u (known: %u), timetravel: %i",
+			     xid, old_tx, does_timetravel);
+
+			/*
+			 * FIXME: At this point we have might have a problem if somebody
+			 * would CLUSTER, REINDEX or similar a system table inside a
+			 * transaction and *also* does other catalog modifications because
+			 * we can only build proper snapshots to look at the catalog after
+			 * we have reached the commit record because only then we know the
+			 * subxids of a toplevel txid. Because we wouldn't notice the
+			 * changed system table relfilenodes we wouldn't see the any of
+			 * those catalog changes.
+			 *
+			 * So we need to forbid that.
+			 */
+
+			if (!old_tx)
+			{
+				/* update global snapshot information */
+				if (does_timetravel)
+				{
+					ent = hash_search(snapstate->by_txn,
+					                  (void *)&xid,
+					                  HASH_FIND|HASH_ENTER,
+					                  &found);
+
+					elog(LOG, "found catalog change in tx %u without changes, did we know it: %u",
+					     xid, found);
+
+					ent->does_timetravel = true;
+
+				}
+				else
+				{
+					elog(LOG, "adding initial snapshot to xid %u", xid);
+				}
+
+				/* add initial snapshot*/
+				{
+					Snapshot snap = SnapBuildBuildSnapshot(snapstate, xid);
+
+					elog(LOG, "adding base snap");
+					ApplyCacheAddBaseSnapshot(applycache, xid,
+					                          InvalidXLogRecPtr,
+					                          snap);
+				}
+
+			}
+			/* update already distributed snapshots */
+			else if (does_timetravel && old_tx)
+			{
+				/*
+				 * check whether we already know the xid as a catalog modifying
+				 * one
+				 */
+				SnapstateTxnEnt *ent =
+					hash_search(snapstate->by_txn,
+					            (void *)&xid,
+					            HASH_FIND|HASH_ENTER,
+				            &found);
+
+				elog(LOG, "found catalog change in tx %u with changes, did we know it: %u",
+				     xid, found);
+
+				ent->does_timetravel = true;
+
+				/* FIXME: add a new CommandId to the applycache's ->changes queue */
+			}
+		}
+	}
+
+	return ret;
+}
+
+
+/* Does this relation carry catalog information */
+static bool
+SnapBuildHasCatalogChanges(Snapstate* snapstate, TransactionId xid, RelFileNode* relfilenode)
+{
+	/* FIXME: build snapshot for transaction */
+	HeapTuple table;
+	Form_pg_class class_form;
+
+	Snapshot snap = SnapBuildBuildSnapshot(snapstate, xid);
+
+	if (relfilenode->spcNode == GLOBALTABLESPACE_OID)
+		return true;
+
+	SetupDecodingSnapshots(snap);
+
+	InvalidateSystemCaches();
+
+	table = LookupTableByRelFileNode(relfilenode);
+
+	RevertFromDecodingSnapshots();
+	InvalidateSystemCaches();
+
+	/*
+	 * tables in the default tablespace are stored in pg_class with 0 as their
+	 * reltablespace
+	 */
+	if (!HeapTupleIsValid(table))
+	{
+		if (relfilenode->relNode >= FirstNormalObjectId)
+		{
+			elog(WARNING, "failed pg_class lookup for %u:%u with a oid in >= FirstNormalObjectId",
+			     relfilenode->spcNode, relfilenode->relNode);
+		}
+		return true;
+	}
+
+	class_form = (Form_pg_class) GETSTRUCT(table);
+
+	return IsSystemClass(class_form);
+}
+
+/* build a new snapshot, based on currently committed transactions */
+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 fileds 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);
+
+	/* store toplevel xid */
+	/*
+	 * FIXME: subtransaction handling currently needs to be done in
+	 * applycache. Yuck.
+	 */
+	snapshot->subxip = (TransactionId*)(
+		(char*)snapshot
+		+ sizeof(SnapshotData) /* offset to ->xip's data */
+		+ sizeof(TransactionId) * snapstate->nrcommitted /* data */
+		);
+
+	snapshot->subxcnt = 1;
+	snapshot->subxip[0] = xid;
+
+	snapshot->suboverflowed = false;
+	snapshot->takenDuringRecovery = false;
+	snapshot->copied = false;
+	snapshot->curcid = 0;
+	snapshot->active_count = 0;
+	snapshot->regd_count = 0;
+
+	return snapshot;
+}
+
+/* check whether `xid` is currently running */
+static bool
+SnapBuildTxnRunning(Snapstate* snapstate, TransactionId xid)
+{
+	if (snapstate->nrrunning &&
+	    NormalTransactionIdFollows(xid, snapstate->xmin_running) &&
+	    NormalTransactionIdPrecedes(xid, snapstate->xmax_running))
+	{
+		TransactionId* xid =
+			bsearch(&xid, snapstate->running, snapstate->nrrunning_initial,
+			        sizeof(TransactionId), xidComparator);
+
+		if (xid != NULL)
+			return true;
+	}
+
+	return false;
+}
+
+/*
+ * add a new SnapshotNow to all transactions were decoding that are currently
+ * in-progress so they can see new catalog contents.
+ */
+static void
+SnapBuildDistributeSnapshotNow(Snapstate* snapstate, TransactionId xid)
+{
+	/* FIXME: implement */
+}
+
+/*
+ * 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;
+
+	/* FIXME: 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);
+}
+
+/*
+ * makes sure we have enough space for at least `count` additional txn's,
+ * reallocates if necessary
+ */
+static void
+SnapBuildReserveRunning(Snapstate *snapstate, Size count)
+{
+	const Size reserve = 100;
+
+	if (snapstate->nrrunning_initial + count < snapstate->nrrunning_space)
+		return;
+
+	if (snapstate->running)
+	{
+		snapstate->nrrunning_space += count + reserve;
+		snapstate->running =
+			realloc(snapstate->running,
+			        snapstate->nrrunning_space *
+			        sizeof(TransactionId));
+		if (!snapstate->running)
+			elog(ERROR, "could not reallocate ->running");
+	}
+	else
+	{
+		snapstate->nrrunning_space = count + reserve;
+		snapstate->running = malloc(snapstate->nrrunning_space
+		                            * sizeof(TransactionId));
+	}
+}
+
+/*
+ * To allow binary search in the set of running transactions, sort them with
+ * xidComparator.
+ */
+static void
+SnapBuildSortRunning(Snapstate *snapstate)
+{
+	qsort(snapstate->running, snapstate->nrrunning_initial,
+	      sizeof(TransactionId), xidComparator);
+}
+
+/*
+ * Add transaction to the set of currently runnign transactions.
+ */
+static void
+SnapBuildAddRunningTxn(Snapstate *snapstate, TransactionId xid)
+{
+	Assert(snapstate->state == SNAPBUILD_INITIAL_POINT &&
+	       TransactionIdIsValid(snapstate->valid_after));
+
+	/*
+	 * we only need those running txn's if were switching state due to reaching
+	 * the xmin horizon. Transactions before we reached that are not
+	 * interesting.
+	 */
+	if (NormalTransactionIdPrecedes(xid, snapstate->valid_after) )
+		return;
+
+	if (SnapBuildTxnRunning(snapstate, xid))
+		return;
+
+	Assert(!TransactionIdPrecedesOrEquals(xid, snapstate->xmin_running));
+
+	if (TransactionIdFollowsOrEquals(xid, snapstate->xmax_running))
+		snapstate->xmax_running = xid;
+
+	SnapBuildReserveRunning(snapstate, 1);
+
+	/* FIXME: inefficient insertion logic, should at least be insertion sort */
+	snapstate->running[snapstate->nrrunning_initial++] = xid;
+	snapstate->nrrunning++;
+	SnapBuildSortRunning(snapstate);
+}
+
+/*
+ * 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 (SnapBuildTxnRunning(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)
+{
+	bool found;
+	int i;
+
+	for(i = 0; i < nsubxacts; i++)
+	{
+		TransactionId subxid = subxacts[i];
+		SnapBuildEndTxn(snapstate, subxid);
+
+		hash_search(snapstate->by_txn,
+		            (void *)&subxid,
+		            HASH_REMOVE,
+		            &found);
+
+	}
+
+	SnapBuildEndTxn(snapstate, xid);
+
+	hash_search(snapstate->by_txn,
+	            (void *)&xid,
+	            HASH_REMOVE,
+	            &found);
+}
+
+/* Handle everything that needs to be done when a transaction commits */
+static void
+SnapBuildCommitTxn(Snapstate* snapstate, TransactionId xid, int nsubxacts,
+                   TransactionId* subxacts)
+{
+	int off;
+	bool found;
+	bool forced_timetravel = false;
+	bool sub_does_timetravel = false;
+	SnapstateTxnEnt *ent;
+
+	/*
+	 * 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.
+	 */
+	if (snapstate->state < SNAPBUILD_CONSISTENT && SnapBuildTxnRunning(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(off = 0; off < nsubxacts; off++)
+	{
+		TransactionId subxid = subxacts[off];
+
+		SnapBuildEndTxn(snapstate, subxid);
+
+		ent = hash_search(snapstate->by_txn,
+		                  (void *)&subxid,
+		                  HASH_FIND,
+		                  &found);
+
+		if (forced_timetravel)
+		{
+			SnapBuildAddCommittedTxn(snapstate, subxid);
+		}
+		/* add subtransaction to base snapshot, we don't distinguish after that */
+		else if (found && ent->does_timetravel)
+		{
+			sub_does_timetravel = true;
+
+			elog(WARNING, "found subtransaction %u:%u with catalog changes",
+			     xid, subxid);
+
+			SnapBuildAddCommittedTxn(snapstate, subxid);
+		}
+
+		/* make sure its not tracked in running txn's anymore, switch state */
+		SnapBuildEndTxn(snapstate, subxid);
+
+		if (found)
+		{
+			hash_search(snapstate->by_txn,
+			            (void *)&xid,
+			            HASH_REMOVE,
+			            &found);
+			Assert(found);
+		}
+
+		if (NormalTransactionIdFollows(subxid, snapstate->xmax))
+		{
+			snapstate->xmax = subxid;
+			TransactionIdAdvance(snapstate->xmax);
+		}
+	}
+
+	/* make sure its not tracked in running txn's anymore, switch state */
+	SnapBuildEndTxn(snapstate, xid);
+
+	ent =
+		hash_search(snapstate->by_txn,
+		            (void *)&xid,
+		            HASH_FIND,
+		            &found);
+
+	/* add toplevel transaction to base snapshot */
+	if (found && ent->does_timetravel)
+	{
+		elog(DEBUG1, "found top level transaction %u, with catalog changes !!!!", xid);
+		SnapBuildAddCommittedTxn(snapstate, xid);
+	}
+
+	if ((found && ent->does_timetravel) || sub_does_timetravel || forced_timetravel)
+	{
+		elog(DEBUG1, "found transaction %u, with catalog changes !!!!", xid);
+
+		/* add a new SnapshotNow to all currently running transactions */
+		SnapBuildDistributeSnapshotNow(snapstate, xid);
+	}
+
+	if (found)
+	{
+		/* now we don't need the contents anymore, remove */
+		hash_search(snapstate->by_txn,
+		            (void *)&xid,
+		            HASH_REMOVE,
+		            &found);
+		Assert(found);
+	}
+
+	if (NormalTransactionIdFollows(xid, snapstate->xmax))
+	{
+		snapstate->xmax = xid;
+		TransactionIdAdvance(snapstate->xmax);
+	}
+}
diff --git a/src/backend/utils/time/tqual.c b/src/backend/utils/time/tqual.c
index b531db5..25af26a 100644
--- a/src/backend/utils/time/tqual.c
+++ b/src/backend/utils/time/tqual.c
@@ -65,6 +65,7 @@
 #include "storage/bufmgr.h"
 #include "storage/procarray.h"
 #include "utils/tqual.h"
+#include "utils/builtins.h"
 
 
 /* Static variables representing various special snapshot semantics */
@@ -73,6 +74,8 @@ SnapshotData SnapshotSelfData = {HeapTupleSatisfiesSelf};
 SnapshotData SnapshotAnyData = {HeapTupleSatisfiesAny};
 SnapshotData SnapshotToastData = {HeapTupleSatisfiesToast};
 
+static Snapshot SnapshotNowDecoding;
+
 /* local functions */
 static bool XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot);
 
@@ -1375,3 +1378,161 @@ 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;
+}
+
+
+/*
+ * 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.
+ *
+ * Basically we record all transactions that are in progress when the
+ * transaction starts and treat them as in-progress for the duration of the
+ * snapshot, everything below xmin is comitted, everything above xmax is
+ * in-progress, and everything thats not in our in-progress array is committed
+ * as well.
+ */
+bool
+HeapTupleSatisfiesMVCCDuringDecoding(HeapTupleHeader tuple, Snapshot snapshot,
+                                     Buffer buffer)
+{
+	TransactionId xmin = HeapTupleHeaderGetXmin(tuple);
+	TransactionId xmax = HeapTupleHeaderGetXmax(tuple);
+
+	/*
+	 * FIXME: The not yet existing decoding infrastructure will need to force
+	 * the xmin to stay lower than what they are currently decoding.
+	 */
+	bool fixme_xmin_horizon = false;
+
+	if (fixme_xmin_horizon && tuple->t_infomask & HEAP_XMIN_INVALID)
+	{
+		return false;
+	}
+	/* normal transaction state counts */
+	else if (TransactionIdPrecedes(xmin, snapshot->xmin))
+	{
+		if (!TransactionIdDidCommit(xmin))
+			return false;
+	}
+	/* beyond our xmax horizon, i.e. invisible */
+	else if (TransactionIdFollows(xmin, snapshot->xmax))
+	{
+		return false;
+	}
+    /* check if its one of our txids, toplevel is also in there */
+	else if (TransactionIdInArray(xmin, snapshot->subxip, snapshot->subxcnt))
+	{
+		CommandId cmin = HeapTupleHeaderGetRawCommandId(tuple);
+		/* no support for that yet */
+		if (tuple->t_infomask & HEAP_COMBOCID){
+			elog(WARNING, "combocids not yet supported");
+			return false;
+		}
+		if (cmin >= snapshot->curcid)
+			return false;	/* inserted after scan started */
+	}
+	/* check if we know the transaction has committed */
+	else if(TransactionIdInArray(xmin, snapshot->xip, snapshot->xcnt))
+	{
+	}
+	else
+	{
+		return false;
+	}
+
+	/* at this point we know xmin is visible */
+
+	/* why should those be in catalog tables? */
+	Assert(!(tuple->t_infomask & HEAP_XMAX_IS_MULTI));
+
+	if (tuple->t_infomask & HEAP_XMAX_INVALID)	/* xid invalid or aborted */
+		return true;
+
+	if (tuple->t_infomask & HEAP_IS_LOCKED)
+		return true;
+
+	/* we cannot possibly see the deleting transaction */
+	if (TransactionIdFollows(xmax, snapshot->xmax))
+	{
+		return true;
+	}
+	/* normal transaction state is valid */
+	else if (TransactionIdPrecedes(xmax, snapshot->xmin))
+	{
+		return !TransactionIdDidCommit(xmax);
+	}
+    /* check if its one of our txids, toplevel is also in there */
+	else if (TransactionIdInArray(xmax, snapshot->subxip, snapshot->subxcnt))
+	{
+		CommandId cmax = HeapTupleHeaderGetRawCommandId(tuple);
+		/* no support for that yet */
+		if (tuple->t_infomask & HEAP_COMBOCID){
+			elog(WARNING, "combocids not yet supported");
+			return true;
+		}
+
+		if (cmax >= snapshot->curcid)
+			return true;	/* deleted after scan started */
+		else
+			return false;	/* deleted before scan started */
+	}
+	/* do we know that the deleting txn is valid? */
+	else if (TransactionIdInArray(xmax, snapshot->xip, snapshot->xcnt))
+	{
+		return false;
+	}
+	else
+	{
+		return true;
+	}
+}
+
+static bool
+FailsSatisfies(HeapTupleHeader tuple, Snapshot snapshot, Buffer buffer)
+{
+	elog(ERROR, "should not be called after SetupDecodingSnapshots!");
+	return false;
+}
+
+static bool
+RedirectSatisfiesNow(HeapTupleHeader tuple, Snapshot snapshot, Buffer buffer)
+{
+	Assert(SnapshotNowDecoding != NULL);
+	return HeapTupleSatisfiesMVCCDuringDecoding(tuple, SnapshotNowDecoding,
+	                                            buffer);
+}
+
+void
+SetupDecodingSnapshots(Snapshot snapshot_now)
+{
+	SnapshotNowData.satisfies = RedirectSatisfiesNow;
+	SnapshotSelfData.satisfies = FailsSatisfies;
+	SnapshotAnyData.satisfies = FailsSatisfies;
+	SnapshotToastData.satisfies = FailsSatisfies;
+
+	SnapshotNowDecoding = snapshot_now;
+}
+
+
+void
+RevertFromDecodingSnapshots(void)
+{
+	SnapshotNowDecoding = NULL;
+
+	SnapshotNowData.satisfies = HeapTupleSatisfiesNow;
+	SnapshotSelfData.satisfies = HeapTupleSatisfiesSelf;
+	SnapshotAnyData.satisfies = HeapTupleSatisfiesAny;
+	SnapshotToastData.satisfies = HeapTupleSatisfiesToast;
+
+}
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/catalog/pg_proc.h b/src/include/catalog/pg_proc.h
index d88248a..b5b886b 100644
--- a/src/include/catalog/pg_proc.h
+++ b/src/include/catalog/pg_proc.h
@@ -4655,6 +4655,9 @@ DESCR("SP-GiST support for suffix tree over text");
 DATA(insert OID = 4031 (  spg_text_leaf_consistent	PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 16 "2281 2281" _null_ _null_ _null_ _null_  spg_text_leaf_consistent _null_ _null_ _null_ ));
 DESCR("SP-GiST support for suffix tree over text");
 
+DATA(insert OID = 4033 (  decode_xlog	PGNSP PGUID 12 1  0 0 0 f f f f t f i 2 0 16 "2275 2275" _null_ _null_ _null_ _null_ decode_xlog _null_ _null_ _null_ ));
+DESCR("decode xlog");
+
 DATA(insert OID = 3469 (  spg_range_quad_config	PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 2278 "2281 2281" _null_ _null_ _null_ _null_  spg_range_quad_config _null_ _null_ _null_ ));
 DESCR("SP-GiST support for quad tree over range");
 DATA(insert OID = 3470 (  spg_range_quad_choose	PGNSP PGUID 12 1 0 0 0 f f f f t f i 2 0 2278 "2281 2281" _null_ _null_ _null_ _null_  spg_range_quad_choose _null_ _null_ _null_ ));
diff --git a/src/include/replication/applycache.h b/src/include/replication/applycache.h
new file mode 100644
index 0000000..f101eeb
--- /dev/null
+++ b/src/include/replication/applycache.h
@@ -0,0 +1,239 @@
+/*
+ * applycache.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/applycache.h
+ */
+#ifndef APPLYCACHE_H
+#define APPLYCACHE_H
+
+#include "access/htup_details.h"
+#include "utils/hsearch.h"
+#include "utils/ilist.h"
+#include "utils/snapshot.h"
+
+typedef struct ApplyCache ApplyCache;
+
+enum ApplyCacheChangeType
+{
+	APPLY_CACHE_CHANGE_INSERT,
+	APPLY_CACHE_CHANGE_UPDATE,
+	APPLY_CACHE_CHANGE_DELETE,
+	/*
+	 * for efficiency and simplicity reasons we keep those in the same list,
+	 * thats somewhat annoying because switch()es warn if those aren't
+	 * handled... Make those private values?
+	 */
+	APPLY_CACHE_CHANGE_SNAPSHOT,
+	APPLY_CACHE_CHANGE_COMMAND_ID
+};
+
+typedef struct ApplyCacheTupleBuf
+{
+	/* position in preallocated list */
+	ilist_s_node node;
+
+	HeapTupleData tuple;
+	HeapTupleHeaderData header;
+	char data[MaxHeapTupleSize];
+} ApplyCacheTupleBuf;
+
+typedef struct ApplyCacheChange
+{
+	XLogRecPtr lsn;
+	enum ApplyCacheChangeType action;
+
+	RelFileNode relnode;
+
+	union {
+		ApplyCacheTupleBuf* newtuple;
+		Snapshot snapshot;
+		CommandId command_id;
+	};
+	ApplyCacheTupleBuf* oldtuple;
+
+
+	HeapTuple table;
+
+	/*
+	 * While in use this is how a change is linked into a transactions,
+	 * otherwise its the preallocated list.
+	*/
+	ilist_d_node node;
+} ApplyCacheChange;
+
+typedef struct ApplyCacheTXN
+{
+	TransactionId xid;
+
+	XLogRecPtr lsn;
+
+	/*
+	 * How many ApplyCacheChange'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
+	 */
+	ilist_d_head changes;
+
+	/*
+	 * non-hierarchical list of subtransactions that are *not* aborted
+	 */
+	ilist_d_head subtxns;
+
+	/*
+	 * our position in a list of subtransactions while the TXN is in
+	 * use. Otherwise its the position in the list of preallocated
+	 * transactions.
+	 */
+	ilist_d_node node;
+
+	/*
+	 * List of (lsn, command_id).
+	 *
+	 * Everytime a catalog change happens this list gets appended with the
+	 * current commandid. This is used to be able to construct proper
+	 * Snapshot's for decoding.
+	 */
+	ilist_d_head commandids;
+
+	/*
+	 * List of (lsn, Snapshot) pairs.
+	 *
+	 * The first record always is the (InvalidXLogRecPtr, SnapshotAtStart)
+	 * pair. Everytime *another* transaction commits this gets appended with a
+	 * new Snapshot that has enough information to make SnapshotNow lookups.
+	 */
+	ilist_d_head snapshots;
+} ApplyCacheTXN;
+
+
+/* XXX: were currently passing the originating subtxn. Not sure thats necessary */
+typedef void (*ApplyCacheApplyChangeCB)(ApplyCache* cache, ApplyCacheTXN* txn, ApplyCacheTXN* subtxn, ApplyCacheChange* change);
+typedef void (*ApplyCacheBeginCB)(ApplyCache* cache, ApplyCacheTXN* txn);
+typedef void (*ApplyCacheCommitCB)(ApplyCache* cache, ApplyCacheTXN* txn);
+
+/*
+ * 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 ApplyCache
+{
+	/*
+	 * Should snapshots for decoding be collected. If many catalog changes
+	 * happen this can be considerably expensive.
+	 */
+	bool build_snapshots;
+
+	TransactionId last_txn;
+	ApplyCacheTXN *last_txn_cache;
+	HTAB *by_txn;
+
+	ApplyCacheBeginCB begin;
+	ApplyCacheApplyChangeCB apply_change;
+	ApplyCacheCommitCB commit;
+
+	void* private_data;
+
+	MemoryContext context;
+
+	/*
+	 * we don't want to repeatedly (de-)allocated those structs, so cache them for reusage.
+	 */
+	ilist_d_head cached_transactions;
+	size_t nr_cached_transactions;
+
+	ilist_d_head cached_changes;
+	size_t nr_cached_changes;
+
+	ilist_s_head cached_tuplebufs;
+	size_t nr_cached_tuplebufs;
+};
+
+
+ApplyCache*
+ApplyCacheAllocate(void);
+
+void
+ApplyCacheFree(ApplyCache*);
+
+ApplyCacheTupleBuf*
+ApplyCacheGetTupleBuf(ApplyCache*);
+
+void
+ApplyCacheReturnTupleBuf(ApplyCache* cache, ApplyCacheTupleBuf* tuple);
+
+/*
+ * Returns a (potentically preallocated) change struct. Its lifetime is managed
+ * by the applycache module.
+ *
+ * If not added to a transaction with ApplyCacheAddChange it needs to be
+ * returned via ApplyCacheReturnChange
+ *
+ * FIXME: better name
+ */
+ApplyCacheChange*
+ApplyCacheGetChange(ApplyCache*);
+
+/*
+ * Return an unused ApplyCacheChange struct
+ */
+void
+ApplyCacheReturnChange(ApplyCache*, ApplyCacheChange*);
+
+
+/*
+ * 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 ApplyCacheTXN 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
+ApplyCacheAddChange(ApplyCache*, TransactionId, XLogRecPtr lsn, ApplyCacheChange*);
+
+/*
+ *
+ */
+void
+ApplyCacheCommit(ApplyCache*, TransactionId, XLogRecPtr lsn);
+
+void
+ApplyCacheCommitChild(ApplyCache*, TransactionId, TransactionId, XLogRecPtr lsn);
+
+void
+ApplyCacheAbort(ApplyCache*, TransactionId, XLogRecPtr lsn);
+
+typedef struct SnapshotData* Snapshot;
+
+/*
+ * if lsn == InvalidXLogRecPtr this is the first snap for the transaction
+ */
+void
+ApplyCacheAddBaseSnapshot(ApplyCache*, TransactionId, XLogRecPtr lsn, Snapshot snap);
+
+/*
+ * Will only be called for command ids > 1
+ */
+void
+ApplyCacheAddNewCommandId(ApplyCache*, TransactionId, XLogRecPtr lsn, CommandId cid);
+
+bool
+ApplyCacheIsXidKnown(ApplyCache* cache, TransactionId xid);
+#endif
diff --git a/src/include/replication/decode.h b/src/include/replication/decode.h
new file mode 100644
index 0000000..86312d1
--- /dev/null
+++ b/src/include/replication/decode.h
@@ -0,0 +1,26 @@
+/*-------------------------------------------------------------------------
+ * 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/applycache.h"
+
+struct Snapstate;
+
+typedef struct ReaderApplyState
+{
+	ApplyCache *apply_cache;
+	struct Snapstate *snapstate;
+} ReaderApplyState;
+
+void DecodeRecordIntoApplyCache(ReaderApplyState* state, XLogRecordBuffer* buf);
+
+#endif
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
new file mode 100644
index 0000000..ed92e75
--- /dev/null
+++ b/src/include/replication/snapbuild.h
@@ -0,0 +1,119 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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/applycache.h"
+
+#include "utils/hsearch.h"
+#include "utils/snapshot.h"
+#include "access/htup.h"
+
+typedef enum
+{
+	SNAPBUILD_START,
+	/*
+	 * found initial visibility information.
+	 *
+	 * Thats either: XLOG_RUNNING_XACTS or XLOG_CHECKPOINT_SHUTDOWN
+	 */
+	SNAPBUILD_INITIAL_POINT,
+	/*
+	 * 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;
+
+	/* sorted array of running transactions, can be searched with bsearch() */
+	TransactionId* running;
+	/* how many running transactions remain */
+	size_t nrrunning;
+	/* how much free space do we have to add more running txn's */
+	size_t nrrunning_space;
+	/*
+	 * 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;
+
+	TransactionId valid_after;
+
+	/*
+	 * Running (sub-)transactions with catalog changes. This will be used to
+	 * fill the committed array with a transactions xid and all it subxids
+	 * at commit.
+	 */
+	HTAB *by_txn;
+
+	/*
+	 * Transactions which could have catalog changes that committed between
+	 * xmin and xmax
+	 */
+	size_t nrcommitted;
+	size_t nrcommitted_space;
+	TransactionId* committed;
+
+	/* contains all catalog modifying txns */
+} Snapstate;
+
+extern Snapstate*
+AllocateSnapshotBuilder(ApplyCache *cache);
+
+extern void
+FreeSnapshotBuilder(Snapstate *cache);
+
+extern SnapBuildAction
+SnapBuildCallback(ApplyCache *cache, Snapstate* snapstate, XLogRecordBuffer* buf);
+
+extern HeapTuple
+LookupTableByRelFileNode(RelFileNode* r);
+
+#endif /* SNAPBUILD_H */
diff --git a/src/include/utils/tqual.h b/src/include/utils/tqual.h
index ff74f86..6c9b261 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,22 @@ extern bool HeapTupleIsSurelyDead(HeapTupleHeader tuple,
 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(HeapTupleHeader tuple,
+                                                 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);
+extern void RevertFromDecodingSnapshots(void);
+
 #endif   /* TQUAL_H */