[PATCH 08/16] Introduce the ApplyCache module which can reassemble transactions from a stream of interspersed changes

From: Andres Freund <andres@anarazel.de>

The individual changes need to be identified by an xid. The xid can be a
subtransaction or a toplevel one, at commit those can be reintegrated by doing
a k-way mergesort between the individual transaction.

Callbacks for apply_begin, apply_change and apply_commit are provided to
retrieve complete transactions.

Missing:
- spill-to-disk
- correct subtransaction merge, current behaviour is simple/wrong
- DDL handling (?)
- resource usage controls
---
 src/backend/replication/Makefile             |    2 +
 src/backend/replication/logical/Makefile     |   19 ++
 src/backend/replication/logical/applycache.c |  380 ++++++++++++++++++++++++++
 src/include/replication/applycache.h         |  185 +++++++++++++
 4 files changed, 586 insertions(+)
 create mode 100644 src/backend/replication/logical/Makefile
 create mode 100644 src/backend/replication/logical/applycache.c
 create mode 100644 src/include/replication/applycache.h

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..2eadab8
--- /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
+
+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..b73b0ba
--- /dev/null
+++ b/src/backend/replication/logical/applycache.c
@@ -0,0 +1,380 @@
+/*-------------------------------------------------------------------------
+ *
+ * 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 "utils/ilist.h"
+#include "utils/memutils.h"
+#include "utils/relcache.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);
+
+
+ApplyCache*
+ApplyCacheAllocate(void)
+{
+	ApplyCache* cache = (ApplyCache*)malloc(sizeof(ApplyCache));
+	HASHCTL         hash_ctl;
+
+	if (!cache)
+		elog(ERROR, "Could not allocate the ApplyCache");
+
+	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);
+	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)
+{
+	if (change->newtuple)
+		ApplyCacheReturnTupleBuf(cache, change->newtuple);
+	if (change->oldtuple)
+		ApplyCacheReturnTupleBuf(cache, change->oldtuple);
+
+	if (change->table)
+		heap_freetuple(change->table);
+
+	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)
+{
+	ApplyCacheTXNByIdEnt* ent;
+	bool found;
+
+	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);
+	}
+
+	return ent->txn;
+}
+
+void
+ApplyCacheAddChange(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn,
+                    ApplyCacheChange* change)
+{
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, true);
+	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);
+
+	/*
+	 * No need to do anything if that subtxn didn't contain any changes
+	 */
+	if (!subtxn)
+		return;
+
+	subtxn->lsn = lsn;
+
+	txn = ApplyCacheTXNByXid(cache, xid, true);
+
+	ilist_d_push_back(&txn->subtxns, &subtxn->node);
+}
+
+void
+ApplyCacheCommit(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, false);
+	ilist_d_node* cur_change, *next_change;
+	ilist_d_node* cur_txn, *next_txn;
+	bool found;
+
+	if (!txn)
+		return;
+
+	txn->lsn = lsn;
+
+	cache->begin(cache, txn);
+
+	/*
+	 * FIXME:
+	 * do a k-way mergesort of all changes ordered by xid
+	 *
+	 * For now we just iterate through all subtransactions and then through the
+	 * main txn. But thats *WRONG*.
+	 *
+	 * The best way to do is probably to model the current heads of all TXNs as
+	 * a heap and always remove from the smallest lsn till thats not the case
+	 * anymore.
+	 */
+	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);
+			cache->apply_change(cache, txn, subtxn, 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);
+		cache->apply_change(cache, txn, NULL, change);
+
+		ApplyCacheReturnChange(cache, change);
+	}
+
+	cache->commit(cache, txn);
+
+	/* now remove reference from cache */
+	hash_search(cache->by_txn,
+	            (void *)&xid,
+	            HASH_REMOVE,
+	            &found);
+	Assert(found);
+
+	ApplyCacheReturnTXN(cache, txn);
+}
+
+void
+ApplyCacheAbort(ApplyCache* cache, TransactionId xid, XLogRecPtr lsn)
+{
+	ilist_d_node* cur_change, *next_change;
+	ilist_d_node* cur_txn, *next_txn;
+	ApplyCacheTXN* txn = ApplyCacheTXNByXid(cache, xid, false);
+	bool found;
+
+	/* no changes in this commit */
+	if (!txn)
+		return;
+
+	/* iterate through all subtransactions and free memory */
+	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 *)&xid,
+	            HASH_REMOVE,
+	            &found);
+	Assert(found);
+
+	ApplyCacheReturnTXN(cache, txn);
+}
diff --git a/src/include/replication/applycache.h b/src/include/replication/applycache.h
new file mode 100644
index 0000000..4ceba63
--- /dev/null
+++ b/src/include/replication/applycache.h
@@ -0,0 +1,185 @@
+/*
+ * 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.h"
+#include "utils/hsearch.h"
+#include "utils/ilist.h"
+
+typedef struct ApplyCache ApplyCache;
+
+enum ApplyCacheChangeType
+{
+	APPLY_CACHE_CHANGE_INSERT,
+	APPLY_CACHE_CHANGE_UPDATE,
+	APPLY_CACHE_CHANGE_DELETE
+};
+
+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;
+
+	ApplyCacheTupleBuf* newtuple;
+
+	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;
+} 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
+{
+	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);
+
+#endif
-- 
1.7.10.rc3.3.g19a6c.dirty