From 0aa5eb65e4e8eb6a14654e4f877ad8d1541d6049 Mon Sep 17 00:00:00 2001
From: Petr Jelinek <pjmodos@pjmodos.net>
Date: Sat, 4 Jun 2016 17:57:09 +0200
Subject: [PATCH 3/6] Define logical replication protocol and output plugin

---
 doc/src/sgml/protocol.sgml                         | 712 +++++++++++++++++++++
 src/Makefile                                       |   1 +
 src/backend/replication/logical/Makefile           |   4 +-
 src/backend/replication/logical/proto.c            | 641 +++++++++++++++++++
 src/backend/replication/pgoutput/Makefile          |  33 +
 src/backend/replication/pgoutput/pgoutput.c        | 445 +++++++++++++
 src/backend/replication/pgoutput/pgoutput_config.c | 188 ++++++
 src/include/replication/logicalproto.h             |  76 +++
 src/include/replication/pgoutput.h                 |  32 +
 9 files changed, 2130 insertions(+), 2 deletions(-)
 create mode 100644 src/backend/replication/logical/proto.c
 create mode 100644 src/backend/replication/pgoutput/Makefile
 create mode 100644 src/backend/replication/pgoutput/pgoutput.c
 create mode 100644 src/backend/replication/pgoutput/pgoutput_config.c
 create mode 100644 src/include/replication/logicalproto.h
 create mode 100644 src/include/replication/pgoutput.h

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 8e701aa..6dd0e55 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2104,6 +2104,136 @@ The commands accepted in walsender mode are:
 
 </sect1>
 
+<sect1 id="protocol-logical-replication">
+ <title>Logical Streaming Replication Protocol</title>
+
+ <para>
+  This section describes the logical replication protocol which is the message
+  flow started by the <literal>START_REPLICATION</literal>
+  <literal>SLOT</literal> <replaceable class="parameter">slot_name</>
+  <literal>LOGICAL</literal> replication command.
+ </para>
+
+ <para>
+  The logical streaming replication protocol builds on the primitives of
+  physical streaming replication protocol.
+ </para>
+
+ <sect2 id="protocol-logical-replication-params">
+  <title>Logical Streaming Replication Parameters</title>
+
+  <para>
+   The logical replication  <literal>START_REPLICATION</literal> command
+   accepts following parameters:
+
+   <variablelist>
+    <varlistentry>
+     <term>
+      proto_version
+     </term>
+     <listitem>
+      <para>
+       Protocol version. Currently only version <literal>1</literal> is
+       supported.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term>
+      encoding
+     </term>
+     <listitem>
+      <para>
+       Name of the encoding. This currently correspond to the encoding of the
+       server.
+      </para>
+     </listitem>
+    </varlistentry>
+
+    <varlistentry>
+     <term>
+      publication_names
+     </term>
+     <listitem>
+      <para>
+       Comma separated list of publication names for which to subscribe
+       (receive changes). See
+       <xref linkend="logical-replication-publication"> for more info.
+      </para>
+     </listitem>
+    </varlistentry>
+   </variablelist>
+
+  </para>
+ </sect2>
+
+ <sect2 id="protocol-logical-messages">
+  <title>Logical Replication Protocol Messages</title>
+
+  <para>
+   The individual protocol messages are discussed in the following
+   sub-sections. Individual messages are describer in
+   <xref linkend="protocol-logicalrep-message-formats"> section.
+  </para>
+
+  <para>
+   All top-level protocol messages begin with a message type byte.
+   While represented in code as a character, this is a signed byte with no
+   associated encoding.
+  </para>
+
+  <para>
+   Since the streaming replication protocol supplies a message length there
+   is no need for top-level protocol messages to embed a length in their
+   header.
+  </para>
+
+ </sect2>
+
+ <sect2 id="protocol-logical-messages-flow">
+  <title>Logical Replication Protocol Message Flow</title>
+
+  <para>
+   With the exception of the <literal>START_REPLICATION</literal> command and
+   the replay progress messages, all information flows only from the backend
+   to the frontend.
+  </para>
+
+  <para>
+   The logical replication protocol sends individual transactions one by one.
+   This means that all messages between two Begin and Commit messages all
+   belong to the same transaction.
+  </para>
+
+  <para>
+   Every sent transaction contains zero or more DML messages (Insert,
+   Update, Delete) and in case of cascaded setup it can also contain Origin
+   messages. The origin message indicated that the transaction originated on
+   different replication node. Since replication node in the scope of logical
+   replication protocol can be pretty much anything, the only identifier
+   is the origin name. It's downstream responsibility to handle this as
+   needed (if needed). The Origin message is always sent before any DML
+   messages in the transaction.
+  </para>
+
+  <para>
+   Every DML message contains arbitraty relation id, which can be mapped to
+   an id in the Relation messages. The Relation describe the schema of the
+   given relation. The Relation message is sent for a given relation either
+   because it's the first time we send DML message for given relation in
+   current session or because the relation definition has changed since the
+   last Relation message was sent for it. The protocol assumes that the client
+   is capable of caching the metadata for as many relations as needed.
+  </para>
+
+  <para>
+  </para>
+
+ </sect2>
+
+</sect1>
+
 <sect1 id="protocol-message-types">
 <title>Message Data Types</title>
 
@@ -5112,6 +5242,588 @@ not line breaks.
 
 </sect1>
 
+<sect1 id="protocol-logicalrep-message-formats">
+<title>Logical Replication Message Formats</title>
+
+<para>
+This section describes the detailed format of each logical replication message.
+These messages are returned either by replication slot SQL interface or by
+WalSender. In case of WalSender they are encapsulated inside the replication
+protocol WAL messages as described in <xref linkend="protocol-replication">
+section and generaly obey same message flow as physical replication.
+</para>
+
+<variablelist>
+
+<varlistentry>
+<term>
+Begin
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('B')
+</term>
+<listitem>
+<para>
+                Identifies the message as a begin message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                The final LSN of the transaction.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                Commit timestamp of the transaction.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Xid of the transaction.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Commit
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('C')
+</term>
+<listitem>
+<para>
+                Identifies the message as a commit message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                The LSN of the commit.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                The end LSN of the transaction.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                Commit timestamp of the transaction.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Origin
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('O')
+</term>
+<listitem>
+<para>
+                Identifies the message as an origin message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int64
+</term>
+<listitem>
+<para>
+                The LSN of the commit on the origin server.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int8
+</term>
+<listitem>
+<para>
+                Length of the origin name (including the NULL-termination
+                character).
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        String
+</term>
+<listitem>
+<para>
+                Name of the origin.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Relation
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('R')
+</term>
+<listitem>
+<para>
+                Identifies the message as an relation message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Id of the relation.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int8
+</term>
+<listitem>
+<para>
+                Length of the namespace name (including the NULL-termination
+                character).
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        String
+</term>
+<listitem>
+<para>
+                Namespace.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int8
+</term>
+<listitem>
+<para>
+                Length of the relation name (including the NULL-termination
+                character).
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        String
+</term>
+<listitem>
+<para>
+                Relation name.
+</para>
+</listitem>
+</varlistentry>
+</variablelist>
+
+</para>
+
+<para>
+This message is always followed by Attributes message.
+</para>
+
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Attrinutes
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('A')
+</term>
+<listitem>
+<para>
+                Identifies the message as an attributes message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int16
+</term>
+<listitem>
+<para>
+                Number of columns.
+</para>
+</listitem>
+</varlistentry>
+</variablelist>
+        Next, the following submessage appears for each column:
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('C')
+</term>
+<listitem>
+<para>
+                Start of column block.
+</para>
+</listitem>
+</varlistentry><varlistentry>
+<term>
+        Int8
+</term>
+<listitem>
+<para>
+                Flags for the column. Currently can be either 0 for no flags
+                or one which marks the column as part of the key.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int8
+</term>
+<listitem>
+<para>
+                Length of column name (including the NULL-termination
+                character).
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        String
+</term>
+<listitem>
+<para>
+                Name of the column.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Insert
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('I')
+</term>
+<listitem>
+<para>
+                Identifies the message as an insert message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Id of the relation corresponding to the id in the relation
+                message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Byte1('N')
+</term>
+<listitem>
+<para>
+                Identifies the following TupleData message as a new tuple.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Update
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('U')
+</term>
+<listitem>
+<para>
+                Identifies the message as an update message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Id of the relation corresponding to the id in the relation
+                message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Byte1('O')
+</term>
+<listitem>
+<para>
+                Identifies the following TupleData message as a old tuple.
+                This field is optional and is only present if table in which
+                the update happened has REPLICA IDENTITY set to FULL or when
+                the REPLICA IDENTITY index values have changed.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Byte1('N')
+</term>
+<listitem>
+<para>
+                Identifies the following TupleData message as a new tuple.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+Delete
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('D')
+</term>
+<listitem>
+<para>
+                Identifies the message as a delete message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Id of the relation corresponding to the id in the relation
+                message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Byte1('O')
+</term>
+<listitem>
+<para>
+                Identifies the following TupleData message as the old tuple
+                (deleted tuple).
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+<varlistentry>
+<term>
+TupleData
+</term>
+<listitem>
+<para>
+
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('T')
+</term>
+<listitem>
+<para>
+                Identifies the message as an tuple data message.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int16
+</term>
+<listitem>
+<para>
+                Number of columns.
+</para>
+</listitem>
+</varlistentry>
+</variablelist>
+        Next, one of the following submessages appears for each column:
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('n')
+</term>
+<listitem>
+<para>
+                Idenfifies the data as NULL value.
+</para>
+</listitem>
+</varlistentry>
+</variablelist>
+        Or
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('u')
+</term>
+<listitem>
+<para>
+                Idenfifies unchanged TOASTed value (the actual value is not
+                sent).
+</para>
+</listitem>
+</varlistentry>
+</variablelist>
+        Or
+<variablelist>
+<varlistentry>
+<term>
+        Byte1('t')
+</term>
+<listitem>
+<para>
+                Idenfifies the data as text formatted value.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        Int32
+</term>
+<listitem>
+<para>
+                Length of the column value.
+</para>
+</listitem>
+</varlistentry>
+<varlistentry>
+<term>
+        String
+</term>
+<listitem>
+<para>
+                The text value.
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+</para>
+</listitem>
+</varlistentry>
+
+</variablelist>
+
+</sect1>
+
 <sect1 id="protocol-changes">
 <title>Summary of Changes since Protocol 2.0</title>
 
diff --git a/src/Makefile b/src/Makefile
index b526be7..9723a1e 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -22,6 +22,7 @@ SUBDIRS = \
 	include \
 	interfaces \
 	backend/replication/libpqwalreceiver \
+	backend/replication/pgoutput \
 	fe_utils \
 	bin \
 	pl \
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index e4b093b..438811e 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -14,7 +14,7 @@ include $(top_builddir)/src/Makefile.global
 
 override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
-OBJS = decode.o logical.o logicalfuncs.o message.o origin.o publication.o \
-	   reorderbuffer.o snapbuild.o subscription.o
+OBJS = decode.o logical.o logicalfuncs.o message.o origin.o proto.o \
+	   publication.o reorderbuffer.o snapbuild.o subscription.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
new file mode 100644
index 0000000..2b82495
--- /dev/null
+++ b/src/backend/replication/logical/proto.c
@@ -0,0 +1,641 @@
+/*-------------------------------------------------------------------------
+ *
+ * proto.c
+ * 		logical replication protocol functions
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		src/backend/replication/logical/proto.c
+ *
+ * TODO
+ *		unaligned access
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "miscadmin.h"
+
+#include "access/htup_details.h"
+#include "access/heapam.h"
+
+#include "access/sysattr.h"
+#include "access/tuptoaster.h"
+#include "access/xact.h"
+
+#include "catalog/catversion.h"
+#include "catalog/index.h"
+
+#include "catalog/namespace.h"
+#include "catalog/pg_class.h"
+#include "catalog/pg_database.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/pg_type.h"
+
+#include "commands/dbcommands.h"
+
+#include "executor/spi.h"
+
+#include "libpq/pqformat.h"
+
+#include "mb/pg_wchar.h"
+
+#include "nodes/makefuncs.h"
+
+#include "replication/logicalproto.h"
+#include "replication/reorderbuffer.h"
+
+#include "utils/builtins.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+#include "utils/timestamp.h"
+#include "utils/typcache.h"
+
+#define IS_REPLICA_IDENTITY	1
+
+static void logicalrep_write_attrs(StringInfo out, Relation rel);
+static void logicalrep_write_tuple(StringInfo out, Relation rel,
+								   HeapTuple tuple);
+
+static void logicalrep_read_attrs(StringInfo in, char ***attrnames,
+								  int *nattrnames);
+static void logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple);
+
+
+/*
+ * Given a List of strings, return it as single comma separated
+ * string, quoting identifiers as needed.
+ *
+ * This is essentially the reverse of SplitIdentifierString.
+ *
+ * The caller should free the result.
+ */
+static char *
+stringlist_to_identifierstr(List *strings)
+{
+	ListCell *lc;
+	StringInfoData res;
+	bool first = true;
+
+	initStringInfo(&res);
+
+	foreach (lc, strings)
+	{
+		if (first)
+			first = false;
+		else
+			appendStringInfoChar(&res, ',');
+
+		appendStringInfoString(&res, quote_identifier(strVal(lfirst(lc))));
+	}
+
+	return res.data;
+}
+
+/*
+ * Build string of options for logical replication plugin.
+ */
+char *
+logicalrep_build_options(List *publications)
+{
+	StringInfoData	options;
+	char		   *publicationstr;
+
+	initStringInfo(&options);
+	appendStringInfo(&options, "proto_version '%u'", LOGICALREP_PROTO_VERSION_NUM);
+	appendStringInfo(&options, ", encoding %s",
+					 quote_literal_cstr(GetDatabaseEncodingName()));
+	appendStringInfo(&options, ", pg_version '%u'", PG_VERSION_NUM);
+	publicationstr = stringlist_to_identifierstr(publications);
+	appendStringInfo(&options, ", publication_names %s",
+					 quote_literal_cstr(publicationstr));
+	pfree(publicationstr);
+
+	return options.data;
+}
+
+/*`
+ * Write BEGIN to the output stream.
+ */
+void
+logicalrep_write_begin(StringInfo out, ReorderBufferTXN *txn)
+{
+	pq_sendbyte(out, 'B');		/* BEGIN */
+
+	/* fixed fields */
+	pq_sendint64(out, txn->final_lsn);
+	pq_sendint64(out, txn->commit_time);
+	pq_sendint(out, txn->xid, 4);
+}
+
+/*
+ * Read transaction BEGIN from the stream.
+ */
+void
+logicalrep_read_begin(StringInfo in, XLogRecPtr *remote_lsn,
+					  TimestampTz *committime, TransactionId *remote_xid)
+{
+	/* read fields */
+	*remote_lsn = pq_getmsgint64(in);
+	Assert(*remote_lsn != InvalidXLogRecPtr);
+	*committime = pq_getmsgint64(in);
+	*remote_xid = pq_getmsgint(in, 4);
+}
+
+
+/*
+ * Write COMMIT to the output stream.
+ */
+void
+logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn,
+						XLogRecPtr commit_lsn)
+{
+	pq_sendbyte(out, 'C');		/* sending COMMIT */
+
+	/* send fixed fields */
+	pq_sendint64(out, commit_lsn);
+	pq_sendint64(out, txn->end_lsn);
+	pq_sendint64(out, txn->commit_time);
+}
+
+/*
+ * Read transaction COMMIT from the stream.
+ */
+void
+logicalrep_read_commit(StringInfo in, XLogRecPtr *commit_lsn,
+					   XLogRecPtr *end_lsn, TimestampTz *committime)
+{
+	/* read fields */
+	*commit_lsn = pq_getmsgint64(in);
+	*end_lsn = pq_getmsgint64(in);
+	*committime = pq_getmsgint64(in);
+}
+
+/*
+ * Write ORIGIN to the output stream.
+ */
+void
+logicalrep_write_origin(StringInfo out, const char *origin,
+						XLogRecPtr origin_lsn)
+{
+	uint8	len;
+
+	Assert(strlen(origin) < 255);
+
+	pq_sendbyte(out, 'O');		/* ORIGIN */
+
+	/* fixed fields */
+	pq_sendint64(out, origin_lsn);
+
+	/* origin */
+	len = strlen(origin) + 1;
+	pq_sendbyte(out, len);
+	pq_sendbytes(out, origin, len);
+}
+
+
+/*
+ * Read ORIGIN from the output stream.
+ */
+char *
+logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn)
+{
+	uint8	len;
+
+	/* fixed fields */
+	*origin_lsn = pq_getmsgint64(in);
+
+	/* origin */
+	len = pq_getmsgbyte(in);
+
+	return pnstrdup(pq_getmsgbytes(in, len), len);
+}
+
+
+/*
+ * Write INSERT to the output stream.
+ */
+void
+logicalrep_write_insert(StringInfo out,	Relation rel, HeapTuple newtuple)
+{
+	pq_sendbyte(out, 'I');		/* action INSERT */
+
+	/* use Oid as relation identifier */
+	pq_sendint(out, RelationGetRelid(rel), 4);
+
+	pq_sendbyte(out, 'N');		/* new tuple follows */
+	logicalrep_write_tuple(out, rel, newtuple);
+}
+
+/*
+ * Read INSERT from stream.
+ *
+ * Fills the new tuple.
+ */
+LogicalRepRelId
+logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup)
+{
+	char		action;
+	LogicalRepRelId		relid;
+
+	/* read the relation id */
+	relid = pq_getmsgint(in, 4);
+
+	action = pq_getmsgbyte(in);
+	if (action != 'N')
+		elog(ERROR, "expected new tuple but got %d",
+			 action);
+
+	logicalrep_read_tuple(in, newtup);
+
+	return relid;
+}
+
+/*
+ * Write UPDATE to the output stream.
+ */
+void
+logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple,
+					   HeapTuple newtuple)
+{
+	pq_sendbyte(out, 'U');		/* action UPDATE */
+
+	/* use Oid as relation identifier */
+	pq_sendint(out, RelationGetRelid(rel), 4);
+
+	if (oldtuple != NULL)
+	{
+		pq_sendbyte(out, 'O');	/* old tuple follows */
+		logicalrep_write_tuple(out, rel, oldtuple);
+	}
+
+	pq_sendbyte(out, 'N');		/* new tuple follows */
+	logicalrep_write_tuple(out, rel, newtuple);
+}
+
+/*
+ * Read UPDATE from stream.
+ */
+LogicalRepRelId
+logicalrep_read_update(StringInfo in, bool *hasoldtup,
+					   LogicalRepTupleData *oldtup,
+					   LogicalRepTupleData *newtup)
+{
+	char		action;
+	LogicalRepRelId		relid;
+
+	/* read the relation id */
+	relid = pq_getmsgint(in, 4);
+
+	/* read and verify action */
+	action = pq_getmsgbyte(in);
+	if (action != 'O' && action != 'N')
+		elog(ERROR, "expected action 'N' or 'O', got %c",
+			 action);
+
+	/* check for old tuple */
+	if (action == 'O')
+	{
+		logicalrep_read_tuple(in, oldtup);
+		*hasoldtup = true;
+		action = pq_getmsgbyte(in);
+	}
+	else
+		*hasoldtup = false;
+
+	/* check for new  tuple */
+	if (action != 'N')
+		elog(ERROR, "expected action 'N', got %c",
+			 action);
+
+	logicalrep_read_tuple(in, newtup);
+
+	return relid;
+}
+
+/*
+ * Write DELETE to the output stream.
+ */
+void
+logicalrep_write_delete(StringInfo out, Relation rel, HeapTuple oldtuple)
+{
+	pq_sendbyte(out, 'D');		/* action DELETE */
+
+	/* use Oid as relation identifier */
+	pq_sendint(out, RelationGetRelid(rel), 4);
+
+	pq_sendbyte(out, 'O');	/* old tuple follows */
+	logicalrep_write_tuple(out, rel, oldtuple);
+}
+
+/*
+ * Read DELETE from stream.
+ *
+ * Fills the old tuple.
+ */
+LogicalRepRelId
+logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup)
+{
+	char		action;
+	LogicalRepRelId		relid;
+
+	/* read the relation id */
+	relid = pq_getmsgint(in, 4);
+
+	/* read and verify action */
+	action = pq_getmsgbyte(in);
+	if (action != 'O')
+		elog(ERROR, "expected action 'O', got %c", action);
+
+	logicalrep_read_tuple(in, oldtup);
+
+	return relid;
+}
+
+/*
+ * Write relation description to the output stream.
+ */
+void
+logicalrep_write_rel(StringInfo out, Relation rel)
+{
+	char	   *nspname;
+	uint8		nspnamelen;
+	const char *relname;
+	uint8		relnamelen;
+
+	pq_sendbyte(out, 'R');		/* sending RELATION */
+
+	/* use Oid as relation identifier */
+	pq_sendint(out, RelationGetRelid(rel), 4);
+
+	nspname = get_namespace_name(RelationGetNamespace(rel));
+	if (nspname == NULL)
+		elog(ERROR, "cache lookup failed for namespace %u",
+			 rel->rd_rel->relnamespace);
+	nspnamelen = strlen(nspname) + 1;
+
+	relname = RelationGetRelationName(rel);
+	relnamelen = strlen(relname) + 1;
+
+	pq_sendbyte(out, nspnamelen);		/* schema name length */
+	pq_sendbytes(out, nspname, nspnamelen);
+
+	pq_sendbyte(out, relnamelen);		/* table name length */
+	pq_sendbytes(out, relname, relnamelen);
+
+	/* send the attribute info */
+	logicalrep_write_attrs(out, rel);
+
+	pfree(nspname);
+}
+
+/*
+ * Read schema.relation from stream and return as LogicalRepRelation opened in
+ * lockmode.
+ */
+LogicalRepRelation *
+logicalrep_read_rel(StringInfo in)
+{
+	LogicalRepRelation	*rel = palloc(sizeof(LogicalRepRelation));
+	int			len;
+
+	rel->remoteid = pq_getmsgint(in, 4);
+
+	/* Read relation from stream */
+	len = pq_getmsgbyte(in);
+	rel->nspname = (char *) pq_getmsgbytes(in, len);
+
+	len = pq_getmsgbyte(in);
+	rel->relname = (char *) pq_getmsgbytes(in, len);
+
+	/* Get attribute description */
+	logicalrep_read_attrs(in, &rel->attnames, &rel->natts);
+
+	return rel;
+}
+
+/*
+ * Write a tuple to the outputstream, in the most efficient format possible.
+ */
+static void
+logicalrep_write_tuple(StringInfo out, Relation rel, HeapTuple tuple)
+{
+	TupleDesc	desc;
+	Datum		values[MaxTupleAttributeNumber];
+	bool		isnull[MaxTupleAttributeNumber];
+	int			i;
+	uint16		nliveatts = 0;
+
+	desc = RelationGetDescr(rel);
+
+	pq_sendbyte(out, 'T');			/* sending TUPLE */
+
+	for (i = 0; i < desc->natts; i++)
+	{
+		if (desc->attrs[i]->attisdropped)
+			continue;
+		nliveatts++;
+	}
+	pq_sendint(out, nliveatts, 2);
+
+	/* try to allocate enough memory from the get go */
+	enlargeStringInfo(out, tuple->t_len +
+					  nliveatts * (1 + 4));
+
+	heap_deform_tuple(tuple, desc, values, isnull);
+
+	/* Write the values */
+	for (i = 0; i < desc->natts; i++)
+	{
+		HeapTuple	typtup;
+		Form_pg_type typclass;
+		Form_pg_attribute att = desc->attrs[i];
+		char   	   *outputstr;
+		int			len;
+
+		/* skip dropped columns */
+		if (att->attisdropped)
+			continue;
+
+		if (isnull[i])
+		{
+			pq_sendbyte(out, 'n');	/* null column */
+			continue;
+		}
+		else if (att->attlen == -1 && VARATT_IS_EXTERNAL_ONDISK(values[i]))
+		{
+			pq_sendbyte(out, 'u');	/* unchanged toast column */
+			continue;
+		}
+
+		typtup = SearchSysCache1(TYPEOID, ObjectIdGetDatum(att->atttypid));
+		if (!HeapTupleIsValid(typtup))
+			elog(ERROR, "cache lookup failed for type %u", att->atttypid);
+		typclass = (Form_pg_type) GETSTRUCT(typtup);
+
+		pq_sendbyte(out, 't');	/* 'text' data follows */
+
+		outputstr =	OidOutputFunctionCall(typclass->typoutput, values[i]);
+		len = strlen(outputstr) + 1;	/* null terminated */
+		pq_sendint(out, len, 4);		/* length */
+		appendBinaryStringInfo(out, outputstr, len); /* data */
+
+		pfree(outputstr);
+
+		ReleaseSysCache(typtup);
+	}
+}
+
+/*
+ * Read tuple in remote format from stream.
+ *
+ * The returned tuple points into the input stringinfo.
+ */
+static void
+logicalrep_read_tuple(StringInfo in, LogicalRepTupleData *tuple)
+{
+	int			i;
+	int			natts;
+	char		action;
+
+	/* Check that the action is what we expect. */
+	action = pq_getmsgbyte(in);
+	if (action != 'T')
+		elog(ERROR, "expected TUPLE, got %c", action);
+
+	/* Get of attributes. */
+	natts = pq_getmsgint(in, 2);
+
+	memset(tuple->changed, 0, sizeof(tuple->changed));
+
+	/* Read the data */
+	for (i = 0; i < natts; i++)
+	{
+		char		kind;
+		int			len;
+
+		kind = pq_getmsgbyte(in);
+
+		switch (kind)
+		{
+			case 'n': /* null */
+				tuple->values[i] = NULL;
+				tuple->changed[i] = true;
+				break;
+			case 'u': /* unchanged column */
+				tuple->values[i] = (char *) 0xdeadbeef; /* make bad usage more obvious */
+				break;
+			case 't': /* text formatted value */
+				{
+					tuple->changed[i] = true;
+
+					len = pq_getmsgint(in, 4); /* read length */
+
+					/* and data */
+					tuple->values[i] = (char *) pq_getmsgbytes(in, len);
+				}
+				break;
+			default:
+				elog(ERROR, "unknown data representation type '%c'", kind);
+		}
+	}
+}
+
+
+/*
+ * Write relation attributes to the outputstream.
+ */
+static void
+logicalrep_write_attrs(StringInfo out, Relation rel)
+{
+	TupleDesc	desc;
+	int			i;
+	uint16		nliveatts = 0;
+	Bitmapset  *idattrs;
+
+	desc = RelationGetDescr(rel);
+
+	pq_sendbyte(out, 'A');			/* sending ATTRS */
+
+	/* send number of live attributes */
+	for (i = 0; i < desc->natts; i++)
+	{
+		if (desc->attrs[i]->attisdropped)
+			continue;
+		nliveatts++;
+	}
+	pq_sendint(out, nliveatts, 2);
+
+	/* fetch bitmap of REPLICATION IDENTITY attributes */
+	idattrs = RelationGetIndexAttrBitmap(rel, INDEX_ATTR_BITMAP_IDENTITY_KEY);
+
+	/* send the attributes */
+	for (i = 0; i < desc->natts; i++)
+	{
+		Form_pg_attribute att = desc->attrs[i];
+		uint8			flags = 0;
+		uint8			len;
+		const char	   *attname;
+
+		if (att->attisdropped)
+			continue;
+
+		if (bms_is_member(att->attnum - FirstLowInvalidHeapAttributeNumber,
+						  idattrs))
+			flags |= IS_REPLICA_IDENTITY;
+
+		pq_sendbyte(out, 'C');		/* column definition follows */
+		pq_sendbyte(out, flags);
+
+		attname = NameStr(att->attname);
+		len = strlen(attname) + 1;
+		pq_sendbyte(out, len);
+		pq_sendbytes(out, attname, len); /* data */
+	}
+
+	bms_free(idattrs);
+}
+
+
+/*
+ * Read relation attribute names from the outputstream.
+ */
+static void
+logicalrep_read_attrs(StringInfo in, char ***attrnames, int *nattrnames)
+{
+	int			i;
+	uint16		nattrs;
+	char	  **attrs;
+	char		blocktype;
+
+	blocktype = pq_getmsgbyte(in);
+	if (blocktype != 'A')
+		elog(ERROR, "expected ATTRS, got %c", blocktype);
+
+	nattrs = pq_getmsgint(in, 2);
+	attrs = palloc(nattrs * sizeof(char *));
+
+	/* read the attributes */
+	for (i = 0; i < nattrs; i++)
+	{
+		uint8			len;
+
+		blocktype = pq_getmsgbyte(in);		/* column definition follows */
+		if (blocktype != 'C')
+			elog(ERROR, "expected COLUMN, got %c", blocktype);
+
+		/* We ignore flags atm. */
+		(void) pq_getmsgbyte(in);
+
+		/* attribute name */
+		len = pq_getmsgbyte(in);
+		/* the string is NULL terminated */
+		attrs[i] = (char *) pq_getmsgbytes(in, len);
+	}
+
+	*attrnames = attrs;
+	*nattrnames = nattrs;
+}
diff --git a/src/backend/replication/pgoutput/Makefile b/src/backend/replication/pgoutput/Makefile
new file mode 100644
index 0000000..f25ed20
--- /dev/null
+++ b/src/backend/replication/pgoutput/Makefile
@@ -0,0 +1,33 @@
+#-------------------------------------------------------------------------
+#
+# Makefile--
+#    Makefile for src/backend/replication/pgoutput
+#
+# IDENTIFICATION
+#    src/backend/replication/pgoutput
+#
+#-------------------------------------------------------------------------
+
+subdir = src/backend/replication/pgoutput
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+
+override CPPFLAGS := -I$(srcdir) -I$(libpq_srcdir) $(CPPFLAGS)
+
+OBJS = pgoutput.o pgoutput_config.o $(WIN32RES)
+SHLIB_LINK = $(libpq)
+PGFILEDESC = "pgoutput - standard logical replication output plugin"
+NAME = pgoutput
+
+all: all-shared-lib
+
+include $(top_srcdir)/src/Makefile.shlib
+
+install: all installdirs install-lib
+
+installdirs: installdirs-lib
+
+uninstall: uninstall-lib
+
+clean distclean maintainer-clean: clean-lib
+	rm -f $(OBJS)
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
new file mode 100644
index 0000000..d74c7e9
--- /dev/null
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -0,0 +1,445 @@
+/*-------------------------------------------------------------------------
+ *
+ * pgoutput.c
+ *		  Logical Replication output plugin
+ *
+ * Copyright (c) 2012-2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		  pgoutput.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "access/xact.h"
+
+#include "mb/pg_wchar.h"
+
+#include "replication/logical.h"
+#include "replication/logicalproto.h"
+#include "replication/origin.h"
+#include "replication/pgoutput.h"
+#include "replication/publication.h"
+
+#include "utils/builtins.h"
+#include "utils/inval.h"
+#include "utils/memutils.h"
+
+PG_MODULE_MAGIC;
+
+extern void		_PG_output_plugin_init(OutputPluginCallbacks *cb);
+
+static void pgoutput_startup(LogicalDecodingContext * ctx,
+							  OutputPluginOptions *opt, bool is_init);
+static void pgoutput_shutdown(LogicalDecodingContext * ctx);
+static void pgoutput_begin_txn(LogicalDecodingContext *ctx,
+					ReorderBufferTXN *txn);
+static void pgoutput_commit_txn(LogicalDecodingContext *ctx,
+					 ReorderBufferTXN *txn, XLogRecPtr commit_lsn);
+static void pgoutput_change(LogicalDecodingContext *ctx,
+				 ReorderBufferTXN *txn, Relation rel,
+				 ReorderBufferChange *change);
+static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
+						RepOriginId origin_id);
+
+/* Entry in the map used to remember which relation schemas we sent. */
+typedef struct RelSchemaSyncEntry
+{
+	Oid		relid;			/* relation oid */
+	bool	schema_sent;	/* did we send the schema? */
+} RelSchemaSyncEntry;
+
+/* Map used to remember which relation schemas we sent. */
+static HTAB *RelSchemaSyncCache = NULL;
+
+static void rel_schema_sync_cache_cb(Datum arg, Oid relid);
+
+static void init_rel_schema_sync_cache(MemoryContext decoding_context);
+static void destroy_rel_schema_sync_cache(void);
+static RelSchemaSyncEntry *get_rel_schema_sync_entry(Oid relid);
+static void rel_schema_sync_cache_cb(Datum arg, Oid relid);
+
+/*
+ * Specify output plugin callbacks
+ */
+void
+_PG_output_plugin_init(OutputPluginCallbacks *cb)
+{
+	AssertVariableIsOfType(&_PG_output_plugin_init, LogicalOutputPluginInit);
+
+	cb->startup_cb = pgoutput_startup;
+	cb->begin_cb = pgoutput_begin_txn;
+	cb->change_cb = pgoutput_change;
+	cb->commit_cb = pgoutput_commit_txn;
+	cb->filter_by_origin_cb = pgoutput_origin_filter;
+	cb->shutdown_cb = pgoutput_shutdown;
+}
+
+/*
+ * Initialize this plugin
+ */
+static void
+pgoutput_startup(LogicalDecodingContext * ctx, OutputPluginOptions *opt,
+				  bool is_init)
+{
+	PGOutputData   *data = palloc0(sizeof(PGOutputData));
+	int				client_encoding;
+
+	/* Create our memory context for private allocations. */
+	data->context = AllocSetContextCreate(ctx->context,
+										  "logical replication output context",
+										  ALLOCSET_DEFAULT_MINSIZE,
+										  ALLOCSET_DEFAULT_INITSIZE,
+										  ALLOCSET_DEFAULT_MAXSIZE);
+
+	ctx->output_plugin_private = data;
+
+	/*
+	 * This is replication start and not slot initialization.
+	 *
+	 * Parse and validate options passed by the client.
+	 */
+	if (!is_init)
+	{
+		/* We can only do binary */
+		if (opt->output_type != OUTPUT_PLUGIN_BINARY_OUTPUT)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("only binary mode is supported for logical replication protocol")));
+
+		/* Parse the params and ERROR if we see any we don't recognise */
+		pgoutput_process_parameters(ctx->output_plugin_options, data);
+
+		/* Check if we support requested protol */
+		if (data->protocol_version > LOGICALREP_PROTO_VERSION_NUM)
+			ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("client sent protocol_version=%d but we only support protocol %d or lower",
+					 data->protocol_version, LOGICALREP_PROTO_VERSION_NUM)));
+
+		if (data->protocol_version < LOGICALREP_PROTO_MIN_VERSION_NUM)
+			ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("client sent protocol_version=%d but we only support protocol %d or higher",
+				 	data->protocol_version, LOGICALREP_PROTO_MIN_VERSION_NUM)));
+
+		/* Check for encoding match */
+		if (data->client_encoding == NULL ||
+			strlen(data->client_encoding) == 0)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("encoding parameter missing")));
+
+		client_encoding = pg_char_to_encoding(data->client_encoding);
+
+		if (client_encoding == -1)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("unrecognised encoding name %s passed to expected_encoding",
+							data->client_encoding)));
+
+		if (client_encoding != GetDatabaseEncoding())
+			ereport(ERROR,
+					(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+					 errmsg("encoding conversion for logical replication is not supported yet"),
+						 errdetail("encoding %s must be unset or match server_encoding %s",
+							 data->client_encoding, GetDatabaseEncodingName())));
+
+		if (list_length(data->publication_names) < 1)
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("publication_names parameter missing")));
+
+		/* Initialize relation schema cache. */
+		init_rel_schema_sync_cache(CacheMemoryContext);
+	}
+}
+
+/*
+ * BEGIN callback
+ */
+static void
+pgoutput_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn)
+{
+	bool	send_replication_origin = txn->origin_id != InvalidRepOriginId;
+
+	OutputPluginPrepareWrite(ctx, !send_replication_origin);
+	logicalrep_write_begin(ctx->out, txn);
+
+	if (send_replication_origin)
+	{
+		char *origin;
+
+		/* Message boundary */
+		OutputPluginWrite(ctx, false);
+		OutputPluginPrepareWrite(ctx, true);
+
+		/*
+		 * XXX: which behaviour we want here?
+		 *
+		 * Alternatives:
+		 *  - don't send origin message if origin name not found
+		 *    (that's what we do now)
+		 *  - throw error - that will break replication, not good
+		 *  - send some special "unknown" origin
+		 */
+		if (replorigin_by_oid(txn->origin_id, true, &origin))
+			logicalrep_write_origin(ctx->out, origin, txn->origin_lsn);
+	}
+
+	OutputPluginWrite(ctx, true);
+}
+
+/*
+ * COMMIT callback
+ */
+static void
+pgoutput_commit_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
+					 XLogRecPtr commit_lsn)
+{
+	OutputPluginPrepareWrite(ctx, true);
+	logicalrep_write_commit(ctx->out, txn, commit_lsn);
+	OutputPluginWrite(ctx, true);
+}
+
+/*
+ * Convert ReorderBufferChange to PublicationChangeType
+ */
+static PublicationChangeType
+get_publication_change_type(ReorderBufferChange *change)
+{
+	switch (change->action)
+	{
+		case REORDER_BUFFER_CHANGE_INSERT:
+			return PublicationChangeInsert;
+		case REORDER_BUFFER_CHANGE_UPDATE:
+			return PublicationChangeUpdate;
+		case REORDER_BUFFER_CHANGE_DELETE:
+			return PublicationChangeDelete;
+		default:
+			elog(ERROR, "unexpected action %d", change->action);
+	}
+}
+
+/*
+ * Sends the decoded DML over wire.
+ */
+static void
+pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
+				Relation relation, ReorderBufferChange *change)
+{
+	PGOutputData	   *data = (PGOutputData *) ctx->output_plugin_private;
+	MemoryContext		old;
+	RelSchemaSyncEntry *relentry = NULL;
+
+	/* First check the table filter */
+	if (!publication_change_is_replicated(relation,
+										  get_publication_change_type(change),
+										  data->publication_names))
+		return;
+
+	/* Avoid leaking memory by using and resetting our own context */
+	old = MemoryContextSwitchTo(data->context);
+
+	/*
+	 * Write the relation schema if the current schema haven't been sent yet.
+	 */
+	relentry = get_rel_schema_sync_entry(RelationGetRelid(relation));
+	if (!relentry->schema_sent)
+	{
+		OutputPluginPrepareWrite(ctx, false);
+		logicalrep_write_rel(ctx->out, relation);
+		OutputPluginWrite(ctx, false);
+		relentry->schema_sent = true;
+	}
+
+	/* Send the data */
+	switch (change->action)
+	{
+		case REORDER_BUFFER_CHANGE_INSERT:
+			OutputPluginPrepareWrite(ctx, true);
+			logicalrep_write_insert(ctx->out, relation,
+									&change->data.tp.newtuple->tuple);
+			OutputPluginWrite(ctx, true);
+			break;
+		case REORDER_BUFFER_CHANGE_UPDATE:
+			{
+				HeapTuple oldtuple = change->data.tp.oldtuple ?
+					&change->data.tp.oldtuple->tuple : NULL;
+
+				OutputPluginPrepareWrite(ctx, true);
+				logicalrep_write_update(ctx->out, relation, oldtuple,
+										&change->data.tp.newtuple->tuple);
+				OutputPluginWrite(ctx, true);
+				break;
+			}
+		case REORDER_BUFFER_CHANGE_DELETE:
+			if (change->data.tp.oldtuple)
+			{
+				OutputPluginPrepareWrite(ctx, true);
+				logicalrep_write_delete(ctx->out, relation,
+										&change->data.tp.oldtuple->tuple);
+				OutputPluginWrite(ctx, true);
+			}
+			else
+				elog(DEBUG1, "didn't send DELETE change because of missing oldtuple");
+			break;
+		default:
+			Assert(false);
+	}
+
+	/* Cleanup */
+	MemoryContextSwitchTo(old);
+	MemoryContextReset(data->context);
+}
+
+/*
+ * Currently we always forward.
+ */
+static bool
+pgoutput_origin_filter(LogicalDecodingContext *ctx,
+						RepOriginId origin_id)
+{
+	return false;
+}
+
+/*
+ * Shutdown the output plugin.
+ *
+ * Note, we don't need to clean the data->context as it's child context
+ * of the ctx->context so it will be cleaned up by logical decoding machinery.
+ */
+static void
+pgoutput_shutdown(LogicalDecodingContext * ctx)
+{
+	destroy_rel_schema_sync_cache();
+}
+
+
+/*
+ * Initialize the relation schema sync cache for a decoding session.
+ *
+ * The hash table is destoyed at the end of a decoding session. While
+ * relcache invalidations still exist and will still be invoked, they
+ * will just see the null hash table global and take no action.
+ */
+static void
+init_rel_schema_sync_cache(MemoryContext cachectx)
+{
+	HASHCTL	ctl;
+	int		hash_flags;
+	MemoryContext old_ctxt;
+
+	if (RelSchemaSyncCache != NULL)
+		return;
+
+	/* Make a new hash table for the cache */
+	hash_flags = HASH_ELEM | HASH_CONTEXT;
+
+	MemSet(&ctl, 0, sizeof(ctl));
+	ctl.keysize = sizeof(Oid);
+	ctl.entrysize = sizeof(struct RelSchemaSyncEntry);
+	ctl.hcxt = cachectx;
+
+	hash_flags |= HASH_BLOBS;
+
+	old_ctxt = MemoryContextSwitchTo(cachectx);
+	RelSchemaSyncCache = hash_create("logical replication output relation cache",
+									 128, &ctl, hash_flags);
+	(void) MemoryContextSwitchTo(old_ctxt);
+
+	Assert(RelSchemaSyncCache != NULL);
+
+	CacheRegisterRelcacheCallback(rel_schema_sync_cache_cb, (Datum) 0);
+}
+
+/*
+ * Remove all the entries from our relation cache.
+ */
+static void
+destroy_rel_schema_sync_cache(void)
+{
+	HASH_SEQ_STATUS		status;
+	RelSchemaSyncEntry *entry;
+
+	if (RelSchemaSyncCache == NULL)
+		return;
+
+	hash_seq_init(&status, RelSchemaSyncCache);
+
+	while ((entry = (RelSchemaSyncEntry *) hash_seq_search(&status)) != NULL)
+	{
+		if (hash_search(RelSchemaSyncCache,	(void *) &entry->relid,
+						HASH_REMOVE, NULL) == NULL)
+			elog(ERROR, "hash table corrupted");
+	}
+
+	RelSchemaSyncCache = NULL;
+}
+
+/*
+ * Find or create entry in the relation schema cache.
+ */
+static RelSchemaSyncEntry *
+get_rel_schema_sync_entry(Oid relid)
+{
+	RelSchemaSyncEntry *entry;
+	bool				found;
+	MemoryContext		oldctx;
+
+	Assert(RelSchemaSyncCache != NULL);
+
+	/* Find cached function info, creating if not found */
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	entry = (RelSchemaSyncEntry *) hash_search(RelSchemaSyncCache,
+											   (void *) &relid,
+											   HASH_ENTER, &found);
+	Assert(entry != NULL);
+	(void) MemoryContextSwitchTo(oldctx);
+
+	/* Not found means schema wasn't sent */
+	if (!found)
+		entry->schema_sent = false;
+
+	return entry;
+}
+
+/*
+ * Relcache invalidation callback
+ */
+static void
+rel_schema_sync_cache_cb(Datum arg, Oid relid)
+{
+	RelSchemaSyncEntry *entry;
+
+	/*
+	 * We can get here if the plugin was used in SQL interface as the
+	 * RelSchemaSyncCache is detroyed when the decoding finishes, but there
+	 * is no way to unregister the relcache invalidation callback.
+	 */
+	if (RelSchemaSyncCache == NULL)
+		return;
+
+	/*
+	 * Nobody keeps pointers to entries in this hash table around outside
+	 * logical decoding callback calls - but invalidation events can come in
+	 * *during* a callback if we access the relcache in the callback. Because
+	 * of that we must mark the cache entry as invalid but not remove it from
+	 * the hash while it could still be referenced, then prune it at a later
+	 * safe point.
+	 *
+	 * Getting invalidations for relations that aren't in the table is
+	 * entirely normal, since there's no way to unregister for an
+	 * invalidation event. So we don't care if it's found or not.
+	 */
+	entry = (RelSchemaSyncEntry *) hash_search(RelSchemaSyncCache, &relid,
+											   HASH_FIND, NULL);
+
+	/*
+	 * Reset schema sent status as the relation definition may have
+	 * changed.
+	 */
+	if (entry != NULL)
+		entry->schema_sent = false;
+}
diff --git a/src/backend/replication/pgoutput/pgoutput_config.c b/src/backend/replication/pgoutput/pgoutput_config.c
new file mode 100644
index 0000000..335b971
--- /dev/null
+++ b/src/backend/replication/pgoutput/pgoutput_config.c
@@ -0,0 +1,188 @@
+/*-------------------------------------------------------------------------
+ *
+ * pgoutput_param.c
+ *		  Logical Replication output plugin parameter parsing
+ *
+ * Copyright (c) 2012-2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		  pgoutput_param.c
+ *
+ *-------------------------------------------------------------------------
+ */
+#include "postgres.h"
+
+#include "catalog/catversion.h"
+
+#include "mb/pg_wchar.h"
+
+#include "nodes/makefuncs.h"
+
+#include "replication/pgoutput.h"
+
+#include "utils/builtins.h"
+#include "utils/int8.h"
+
+
+typedef enum OutputParamType
+{
+	OUTPUT_PARAM_TYPE_UNDEFINED,
+	OUTPUT_PARAM_TYPE_UINT32,
+	OUTPUT_PARAM_TYPE_STRING
+} OutputParamType;
+
+/* param parsing */
+static int get_param_key(const char * const param_name);
+static Datum get_param_value(DefElem *elem, OutputParamType type,
+							 bool null_ok);
+static uint32 parse_param_uint32(DefElem *elem);
+
+enum {
+	PARAM_UNRECOGNISED,
+	PARAM_PROTOCOL_VERSION,
+	PARAM_ENCODING,
+	PARAM_PG_VERSION,
+	PARAM_PUBLICATION_NAMES,
+} OutputPluginParamKey;
+
+typedef struct {
+	const char * const paramname;
+	int	paramkey;
+} OutputPluginParam;
+
+/* Oh, if only C had switch on strings */
+static OutputPluginParam param_lookup[] = {
+	{"proto_version", PARAM_PROTOCOL_VERSION},
+	{"encoding", PARAM_ENCODING},
+	{"pg_version", PARAM_PG_VERSION},
+	{"publication_names", PARAM_PUBLICATION_NAMES},
+	{NULL, PARAM_UNRECOGNISED}
+};
+
+
+/*
+ * Read parameters sent by client at startup and store recognised
+ * ones in the parameters PGOutputData.
+ *
+ * The data must have all client-supplied parameter fields zeroed,
+ * such as by memset or palloc0, since values not supplied
+ * by the client are not set.
+ */
+void
+pgoutput_process_parameters(List *options, PGOutputData *data)
+{
+	ListCell	*lc;
+
+	/* Examine all the other params in the message. */
+	foreach(lc, options)
+	{
+		DefElem    *elem = lfirst(lc);
+		Datum		val;
+
+		Assert(elem->arg == NULL || IsA(elem->arg, String));
+
+		/* Check each param, whether or not we recognise it */
+		switch(get_param_key(elem->defname))
+		{
+			case PARAM_PROTOCOL_VERSION:
+				val = get_param_value(elem, OUTPUT_PARAM_TYPE_UINT32, false);
+				data->protocol_version = DatumGetUInt32(val);
+				break;
+
+			case PARAM_ENCODING:
+				val = get_param_value(elem, OUTPUT_PARAM_TYPE_STRING, false);
+				data->client_encoding = DatumGetCString(val);
+				break;
+
+			case PARAM_PG_VERSION:
+				val = get_param_value(elem, OUTPUT_PARAM_TYPE_UINT32, false);
+				data->client_pg_version = DatumGetUInt32(val);
+				break;
+
+			case PARAM_PUBLICATION_NAMES:
+				val = get_param_value(elem, OUTPUT_PARAM_TYPE_STRING, false);
+				if (!SplitIdentifierString(DatumGetCString(val), ',',
+										   &data->publication_names))
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_NAME),
+							 errmsg("invalid publication name syntax")));
+
+				break;
+
+			default:
+				ereport(ERROR,
+						(errmsg("Unrecognised pgoutput parameter %s",
+								elem->defname)));
+				break;
+		}
+	}
+}
+
+/*
+ * Look up a param name to find the enum value for the
+ * param, or PARAM_UNRECOGNISED if not found.
+ */
+static int
+get_param_key(const char * const param_name)
+{
+	OutputPluginParam *param = &param_lookup[0];
+
+	do {
+		if (strcmp(param->paramname, param_name) == 0)
+			return param->paramkey;
+		param++;
+	} while (param->paramname != NULL);
+
+	return PARAM_UNRECOGNISED;
+}
+
+/*
+ * Parse parameter as given type and return the value as Datum.
+ */
+static Datum
+get_param_value(DefElem *elem, OutputParamType type, bool null_ok)
+{
+	/* Check for NULL value */
+	if (elem->arg == NULL || strVal(elem->arg) == NULL)
+	{
+		if (null_ok)
+			return (Datum) 0;
+		else
+			ereport(ERROR,
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					 errmsg("parameter \"%s\" cannot be NULL", elem->defname)));
+	}
+
+	switch (type)
+	{
+		case OUTPUT_PARAM_TYPE_UINT32:
+			return UInt32GetDatum(parse_param_uint32(elem));
+		case OUTPUT_PARAM_TYPE_STRING:
+			return CStringGetDatum(pstrdup(strVal(elem->arg)));
+		default:
+			elog(ERROR, "unknown parameter type %d", type);
+	}
+}
+
+/*
+ * Parse string DefElem as uint32.
+ */
+static uint32
+parse_param_uint32(DefElem *elem)
+{
+	int64		res;
+
+	if (!scanint8(strVal(elem->arg), true, &res))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("could not parse integer value \"%s\" for parameter \"%s\"",
+						strVal(elem->arg), elem->defname)));
+
+	if (res > PG_UINT32_MAX || res < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("value \"%s\" out of range for parameter \"%s\"",
+						strVal(elem->arg), elem->defname)));
+
+	return (uint32) res;
+}
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
new file mode 100644
index 0000000..b69d015
--- /dev/null
+++ b/src/include/replication/logicalproto.h
@@ -0,0 +1,76 @@
+/*-------------------------------------------------------------------------
+ *
+ * logicalproto.h
+ *		logical replication protocol
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		src/include/replication/logicalproto.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef LOGICAL_PROTO_H
+#define LOGICAL_PROTO_H
+
+#include "replication/reorderbuffer.h"
+#include "utils/rel.h"
+
+/*
+ * Protocol capabilities
+ *
+ * LOGICAL_PROTO_VERSION_NUM is our native protocol and the greatest version
+ * we can support. PGLOGICAL_PROTO_MIN_VERSION_NUM is the oldest version we
+ * have backwards compatibility for. The client requests protocol version at
+ * connect time.
+ */
+#define LOGICALREP_PROTO_MIN_VERSION_NUM 1
+#define LOGICALREP_PROTO_VERSION_NUM 1
+
+/* Tuple coming via logical replication. */
+typedef struct LogicalRepTupleData
+{
+        char   *values[MaxTupleAttributeNumber];	/* value in out function format or NULL if values is NULL */
+        bool    changed[MaxTupleAttributeNumber];	/* marker for changed/unchanged values */
+} LogicalRepTupleData;
+
+typedef uint32	LogicalRepRelId;
+
+/* Relation information */
+typedef struct LogicalRepRelation
+{
+	/* Info coming from the remote side. */
+	LogicalRepRelId		remoteid;	/* unique id of the relation */
+	char       *nspname;			/* schema name */
+	char       *relname;			/* relation name */
+    int         natts;				/* number of columns */
+    char      **attnames;			/* column names */
+} LogicalRepRelation;
+
+extern char *logicalrep_build_options(List *publications);
+extern void logicalrep_write_begin(StringInfo out,  ReorderBufferTXN *txn);
+extern void logicalrep_read_begin(StringInfo in, XLogRecPtr *remote_lsn,
+					  TimestampTz *committime, TransactionId *remote_xid);
+extern void logicalrep_write_commit(StringInfo out, ReorderBufferTXN *txn,
+						XLogRecPtr commit_lsn);
+extern void logicalrep_read_commit(StringInfo in, XLogRecPtr *commit_lsn,
+					   XLogRecPtr *end_lsn, TimestampTz *committime);
+extern void logicalrep_write_origin(StringInfo out, const char *origin,
+						XLogRecPtr origin_lsn);
+extern char *logicalrep_read_origin(StringInfo in, XLogRecPtr *origin_lsn);
+extern void logicalrep_write_insert(StringInfo out, Relation rel,
+							 HeapTuple newtuple);
+extern LogicalRepRelId logicalrep_read_insert(StringInfo in, LogicalRepTupleData *newtup);
+extern void logicalrep_write_update(StringInfo out, Relation rel, HeapTuple oldtuple,
+					   HeapTuple newtuple);
+extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *hasoldtup,
+					   LogicalRepTupleData *oldtup,
+					   LogicalRepTupleData *newtup);
+extern void logicalrep_write_delete(StringInfo out, Relation rel,
+							 HeapTuple oldtuple);
+extern LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup);
+extern void logicalrep_write_rel(StringInfo out, Relation rel);
+
+extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
+
+#endif /* LOGICALREP_PROTO_H */
diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
new file mode 100644
index 0000000..8cd82e7
--- /dev/null
+++ b/src/include/replication/pgoutput.h
@@ -0,0 +1,32 @@
+/*-------------------------------------------------------------------------
+ *
+ * pgoutput.h
+ *		Logical Replication output plugin
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		pgoutput.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PGOUTPUT_H
+#define PGOUTPUT_H
+
+
+typedef struct PGOutputData
+{
+	MemoryContext	context;			/* pricate memory context for transient
+										 * allocations */
+
+	/* client info */
+	uint32			protocol_version;
+	const char	   *client_encoding;
+	uint32			client_pg_version;
+
+	List		   *publication_names;
+} PGOutputData;
+
+extern void pgoutput_process_parameters(List *options, PGOutputData *data);
+
+#endif /* PGOUTPUT_H */
-- 
2.7.4

