From f3a729fad9e2a54e0efd83c57562791ab49b4317 Mon Sep 17 00:00:00 2001
From: Petr Jelinek <pjmodos@pjmodos.net>
Date: Wed, 13 Jul 2016 20:00:06 +0200
Subject: [PATCH 5/6] Add logical replication workers

---
 doc/src/sgml/catalogs.sgml                         |    8 +-
 doc/src/sgml/filelist.sgml                         |    1 +
 doc/src/sgml/logical-replication.sgml              |  361 +++++
 doc/src/sgml/postgres.sgml                         |    1 +
 doc/src/sgml/reference.sgml                        |    6 +
 src/backend/commands/subscriptioncmds.c            |  175 ++-
 src/backend/executor/nodeModifyTable.c             |    6 +-
 src/backend/postmaster/bgworker.c                  |    6 +-
 src/backend/postmaster/postmaster.c                |   41 +
 .../libpqwalreceiver/libpqwalreceiver.c            |   28 +-
 src/backend/replication/logical/Makefile           |    5 +-
 src/backend/replication/logical/apply.c            | 1435 ++++++++++++++++++++
 src/backend/replication/logical/launcher.c         |  542 ++++++++
 src/backend/storage/ipc/ipci.c                     |    3 +
 src/backend/storage/lmgr/lwlocknames.txt           |    1 +
 src/backend/utils/misc/guc.c                       |   22 +
 src/include/executor/nodeModifyTable.h             |   20 +
 src/include/postmaster/bgworker_internals.h        |    2 +
 src/include/replication/logicalworker.h            |   41 +
 src/include/replication/walreceiver.h              |    4 +
 src/test/perl/PostgresNode.pm                      |   10 +-
 src/test/subscription/.gitignore                   |    2 +
 src/test/subscription/Makefile                     |   20 +
 src/test/subscription/README                       |   16 +
 src/test/subscription/t/001_rep_changes.pl         |   89 ++
 src/test/subscription/t/002_types.pl               |  509 +++++++
 26 files changed, 3338 insertions(+), 16 deletions(-)
 create mode 100644 doc/src/sgml/logical-replication.sgml
 create mode 100644 src/backend/replication/logical/apply.c
 create mode 100644 src/backend/replication/logical/launcher.c
 create mode 100644 src/include/replication/logicalworker.h
 create mode 100644 src/test/subscription/.gitignore
 create mode 100644 src/test/subscription/Makefile
 create mode 100644 src/test/subscription/README
 create mode 100644 src/test/subscription/t/001_rep_changes.pl
 create mode 100644 src/test/subscription/t/002_types.pl

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 84211c1..5951716 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -5134,7 +5134,8 @@
 
   <para>
    The <structname>pg_publication</structname> catalog contains
-   all publications created in the database.
+   all publications created in the database.  For more on publications
+   see <xref linkend="logical-replication-publication">.
   </para>
 
   <table>
@@ -6051,7 +6052,8 @@
 
   <para>
    The <structname>pg_subscription</structname> catalog contains
-   all existing logical replication subscriptions.
+   all existing logical replication subscriptions.  For more information
+   about logical replication see <xref linkend="logical-replication">.
   </para>
 
   <para>
@@ -6120,7 +6122,7 @@
       <entry><type>text[]</type></entry>
       <entry></entry>
       <entry>Array of subscribed publication names. For more on publications
-       see <xref linkend="publications">.
+       see <xref linkend="logical-replication-publication">.
       </entry>
      </row>
     </tbody>
diff --git a/doc/src/sgml/filelist.sgml b/doc/src/sgml/filelist.sgml
index 4383711..7067a21 100644
--- a/doc/src/sgml/filelist.sgml
+++ b/doc/src/sgml/filelist.sgml
@@ -49,6 +49,7 @@
 <!ENTITY config        SYSTEM "config.sgml">
 <!ENTITY user-manag    SYSTEM "user-manag.sgml">
 <!ENTITY wal           SYSTEM "wal.sgml">
+<!ENTITY logical-replication    SYSTEM "logical-replication.sgml">
 
 <!-- programmer's guide -->
 <!ENTITY bgworker   SYSTEM "bgworker.sgml">
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
new file mode 100644
index 0000000..3179add
--- /dev/null
+++ b/doc/src/sgml/logical-replication.sgml
@@ -0,0 +1,361 @@
+<!-- doc/src/sgml/logical-replication.sgml -->
+
+<chapter id="logical-replication">
+
+  <title>Logical Replication</title>
+  <para>
+    Logical Replication is a method of replicating data objects and their
+    changes, based upon their Primary Keys (or Replication Identity). We
+    use the term Logical in contrast to Physical replication which
+    uses exact block addresses and byte-by-byte replication.
+    PostgreSQL supports both mechanisms concurrently, see
+    <xref linkend="high-availability">. Logical Replication allows
+    fine-grained control over both data replication and security.
+  </para>
+  <para>
+    Logical Replication uses a Publish and Subscribe model with one or
+    more Subscribers subscribing to one or more Publications on a
+    Provider node. Subscribers pull data from the Publications they
+    subscribe to and may subsequently re-publish data to allow
+    cascading replication or more complex configurations.
+  </para>
+  <para>
+    Logical replication typically starts with snapshot of the data on
+    the Provider database. Once that is done, the changes on Provider
+    are sent to Subscriber as they occur in real-time. The Subscriber
+    applies the data in same order as the Provider so that the
+    transactional consistency is guaranteed for the Publications within
+    single Subscription. This method of data replication is sometimes
+    referred to as transactional replication.
+  </para>
+  <para>
+    The typical use-cases for logical replication are:
+  </para>
+  <itemizedlist>
+    <listitem>
+      <para>
+        Sending incremental changes in a single database or a subset of
+        a database to Subscribers as they occur.
+      </para>
+    </listitem>
+    <listitem>
+      <para>
+        Firing triggers for individual changes as they are incoming to
+        subscriber.
+      </para>
+    </listitem>
+    <listitem>
+      <para>
+        Consolidating multiple databases into a single one (for example
+        for analytical purposes).
+      </para>
+    </listitem>
+    <listitem>
+      <para>
+        Replicating between different major versions of the PostgreSQL
+      </para>
+    </listitem>
+    <listitem>
+      <para>
+        Giving access to the replicated data to different groups of
+        users.
+      </para>
+    </listitem>
+    <listitem>
+      <para>
+        Sharing subset of the database between multiple databases.
+      </para>
+    </listitem>
+  </itemizedlist>
+  <para>
+    The Subscriber database behaves in a same way as any other
+    PostgreSQL instance and can be used as a Provider for other
+    databases by defining its own Publications. When the Subscriber is
+    treated as read-only by application, there will be no conflicts from
+    a single Subscription. On the other hand if there are other writes
+    done either by application or other Subscribers to the same set of
+    tables conflicts can arise.
+  </para>
+
+<sect1 id="logical-replication-publication">
+  <title>Publication</title>
+  <para>
+    A Publication object can be defined on any master node, owned by one
+    user. A Publication is a set of changes generated from a group of
+    tables, and might also be described as a Change Set or Replication Set.
+    Each Publication exists in only one database.
+  </para>
+  <para>
+    Publications are different from table schema and do not affect
+    how the table is accessed. Each table can be added to multiple
+    Publications if needed.  Publications may include both tables
+    and materialized views. Objects must be added explicitly, except
+    when a Publication is created for "ALL TABLES". There is no
+    default name for a Publication which specifies all tables.
+  </para>
+  <para>
+    The Publication is different from table schema, it does not affect
+    how the table is accessed and each table can be added to multiple
+    Publications as needed. All tables in the database can be added to
+    single Publication. Since the logical replication requires REPLICA
+    IDENTITY index to be present on a table for replication of UPDATEs
+    and DELETEs, the table without such index can't be added to
+    Publication which replicates UPDATEs and DELETEs.
+  </para>
+  <para>
+    Publications can choose to limit the changes they produce to show
+    any combination of INSERT, UPDATE, DELETE and TRUNCATE in a similar
+    way to the way triggers are fired by particular event types.
+  </para>
+  <para>
+    All tables added to the Publication must be accessible via SELECT
+    privilege for the user owning the Publication. Usage on the
+    Publication can be GRANTed to other users.
+  </para>
+  <para>
+    The definition of a Publication object will be included within
+    pg_dump.
+  </para>
+  <para>
+    Every Publication can have multiple Subscribers.
+  </para>
+  <para>
+    Publication is created using the <xref linkend="sql-createpublication">
+    command and may be later altered or dropped using corresponding commands.
+  </para>
+  <para>
+    The individual tables can be added and removed dynamically using
+    <xref linkend="sql-alterpublication">. Both the ADD TABLE and DROP
+    TABLE operations are transactional so the table will start or stop
+    replicating at the correct snapshot once the transaction has committed.
+  </para>
+</sect1>
+<sect1 id="logical-replication-subscription">
+  <title>Subscription</title>
+  <para>
+    A Subscription is the downstream side of the Logical Replication. The
+    node where Subscription is defined is referred to as a Subscriber.
+    Subscription defines the connection to another database and set of
+    Publications (one or more) to which it wants to be subscribed.
+  </para>
+  <para>
+    The Subscriber database behaves in a same way as any other
+    PostgreSQL instance and can be used as a Provider for other
+    databases by defining its own Publications.
+  </para>
+  <para>
+    A Subscriber may have multiple Subscriptions if desired. It is
+    possible to define multiple Subscriptions between single
+    Provider-Subscriber pair, provided that each Publications can only
+    be subscribed to from one Subscriber.
+  </para>
+  <para>
+    Each Subscription will receive changes via one replication slot (see
+    <xref linkend="streaming-replication-slots">). Additional temporary
+    replication slots may be required for the initial data synchronizations
+    of pre-existing table data.
+  </para>
+  <para>
+    Subscriptions are not dumped by pg_dump by default, but can be
+    requested using --subscriptions parameter.
+  </para>
+  <para>
+    The Subscription is added using <xref linkend="sql-createsubscription">
+    and can be stopped/resumed at any time using
+    <xref linkend="sql-altersubscription"> command or removed using
+    <xref linkend="sql-dropsubscription">.
+  </para>
+  <para>
+    When subscription is dropped and recreated the synchronization
+    information is lost. This means that the data has to be
+    resynchronized afterwards.
+  </para>
+</sect1>
+<sect1 id="logical-replication-conflicts">
+  <title>Conflicts</title>
+  <para>
+    Conflicts happen when the replicated changes is breaking any
+    specified constraints (with the exception of foreign keys which are
+    not checked). Currently conflicts are not resolved automatically and
+    cause replication to be stopped with an error until the conflict is
+    manually resolved.
+  </para>
+</sect1>
+<sect1 id="logical-replication-architecture">
+  <title>Architecture</title>
+  <para>
+    Logical replication starts by copying a snapshot of the data on
+    the Provider database. Once that is done, the changes on Provider
+    are sent to Subscriber as they occur in real-time. The Subscriber
+    applies the data in the order in which commits were made on the
+    Provider so that transactional consistency is guaranteed for the
+    Publications within any single Subscription.
+  </para>
+  <para>
+    The Logical Replication is built on the similar architecture as the
+    physical streaming replication
+    (see <xref linkend="streaming-replication">). It is implemented by
+    WalSender and the Apply processes. The WalSender starts the logical
+    decoding (described in <xref linkend="logicaldecoding">) of the WAL and
+    loads the standard logical decoding plugin (pgoutput). The plugin
+    transforms the changes read from WAL to the logical replication protocol
+    (see <xref linkend="protocol-logical-replication">) and filters the data
+    according to Publication specifications. The data are then continuously
+    transferred using the streaming replication protocol to the Apply worker
+    which maps them to the local tables and applies the individual changes as
+    they are received in exact transactional order.
+  </para>
+  <para>
+    The Apply process on Subscriber database always runs with
+    session_replication_role set to replica, which produces the normal effects
+    on triggers and constraints.
+  </para>
+  <sect2 id="logical-replication-snapshot">
+    <title>Initial snapshot</title>
+    <para>
+      The initial snapshot is taken when the replication slot for
+      Subscription is created. The existing data at that snapshot are
+      then sent over using the streaming replication protocol between
+      WalSender and Apply processes in similar way the changes are sent.
+      Once the initial data are copied, the Apply enters catch up phase
+      where it replays the changes which happened on the Provider while
+      the initial snapshot was being copied. Once the replication catches
+      up the Apply switches to normal replication streaming mode and
+      replicates transactions as they happen.
+    </para>
+  </sect2>
+  <sect2 id="logical-replication-table-resync">
+    <title>Individual table resynchronization</title>
+    <para>
+      The table can be resynchronized at any point during the normal
+      replication operation. When the table resynchronization is
+      requested a parallel instance of special kind of the Apply process
+      is started which registers its own temporary replication slot and
+      does new snapshot. Then it works same way as the initial snapshot
+      <xref linkend="logical-replication-snapshot"> with the exception that
+      it only does data copy of single table and once the catchup phase is
+      finished the control of the replication of the table is given back to
+      the main Apply process.
+    </para>
+  </sect2>
+</sect1>
+<sect1 id="logical-replication-monitoring">
+  <title>Monitoring</title>
+  <para>
+    pg_stat_replication
+  </para>
+  <para>
+    pg_stat_subscription
+  </para>
+</sect1>
+<sect1 id="logical-replication-security">
+  <title>Security</title>
+  <para>
+    Replication connection can occur in the same way as physical streaming
+    replication. It requires access to be specifically given using
+    pg_hba.conf. The role used for the replication must have
+    <literal>REPLICATION</literal> privilege <command>GRANTED</command>.
+    This gives a role access to both logical and physical replication.
+  </para>
+  <para>
+    In addition, logical replication can be accessed with the
+    <literal>SUBSCRIPTION</literal> privilege. This allows you to create
+    roles which can pull data from Publications yet cannot request
+    physical replication.
+  </para>
+  <para>
+    To create or subscribe to a Publication the user must have the
+    REPLICATION role, the SUBSCRIPTION role or be a superuser.
+  </para>
+  <para>
+    <literal>SELECT</literal> privilege is required when the user
+    adds a table to a Publication.
+    To subscribe to a Publication, user must be owner or have USAGE
+    privileges granted to the Publication.
+  </para>
+  <para>
+    To create a Subscription the user must have the
+    REPLICATION role, the SUBSCRIPTION role or be a superuser.
+    The Subscription Apply process will run in local database
+    with the privileges of the owner of the Subscription. In practice this
+    means that the owner of the Subscription must have <literal>INSERT</>,
+    <literal>UPDATE</>, <literal>DELETE</> and <literal>TRUNCATE</>
+    privileges on Subscriber to the tables that are being replicated by the
+    Subscription, or be superuser, though this is not recommended.
+  </para>
+  <para>
+    In particular, note that privileges are not re-checked as each change
+    record is read from the Provider, nor are they re-checked for each change
+    when applied. Security is checked once at startup. Concurrent REVOKEs
+    of privilege will interrupt logical replication if they have a material
+    affect on the security of the change stream.
+  </para>
+</sect1>
+<sect1 id="logical-replication-gucs">
+  <title>Logical replication related configuration parameters</title>
+  <para>
+    The Logical Replication requires several configuration options to be
+    set.
+  </para>
+  <para>
+    On the provider side the <varname>wal_level</> must be set to
+    <literal>logical</>, <varname>max_replication_slots</> has to be set to
+    at least number of Subscriptions expected to connect with some reserve
+    for table synchronization as well. And <varname>max_wal_senders</>
+    should be set to at least same as <varname>max_replication_slots</> plus
+    the number of physical replicas that are connected at the same time.
+  </para>
+  <para>
+    The Subscriber also requires the <varname>max_replication_slots</> to
+    be set. In this case it should be set to at least the number of
+    Subscriptions that will be added to the Subscriber. The
+    <varname>max_logical_replication_workers</> has to be set to at least
+    the number of Subscriptions again with some reserve for the table
+    synchronization. Additionally the <varname>max_worker_processes</> may
+    need to be adjusted to accommodate for replication workers at least
+    (<varname>max_logical_replication_workers</> + <literal>1</>). Please
+    note that some extensions and parallel queries also take worker slots
+    from <varname>max_worker_processes</>.
+  </para>
+</sect1>
+<sect1 id="logical-replication-quick-setup">
+  <title>Quick setup</title>
+  <para>
+    First set the configuration options in the postgresql.conf:
+<programlisting>
+wal_level = logical
+max_worker_processes = 10 # one per subscription + one per instance needed on subscriber
+max_logical_replication_workers = 10 # one per subscription + one per instance needed on subscriber
+max_replication_slots = 10 # one per subscription needed both provider and subscriber
+max_wal_senders = 10 # one per subscription needed on provider
+</programlisting>
+  </para>
+  <para>
+    The pg_hba.conf needs to be adjusted to allow replication (the
+    values here depend on your actual network configuration and user you
+    want to use for connecting):
+<programlisting>
+host    replication     repuser     0.0.0.0/0       md5
+</programlisting>
+  </para>
+  <para>
+    Then on Provider database:
+<programlisting>
+CREATE PUBLICATION mypub;
+ALTER PUBLICATION mypub ADD TABLE users, departments;
+</programlisting>
+  </para>
+  <para>
+    And on Subscriber database:
+<programlisting>
+CREATE SUBSCRIPTION mysub WITH CONNECTION <quote>dbname=foo host=bar user=repuser</quote> PUBLICATION mypub;
+</programlisting>
+  </para>
+  <para>
+    The above will start the replication process which synchronizes the
+    initial table contents of <literal>users</literal> and
+    <literal>departments</literal> tables and then starts replicating
+    incremental changes to those tables.
+  </para>
+</sect1>
+</chapter>
diff --git a/doc/src/sgml/postgres.sgml b/doc/src/sgml/postgres.sgml
index 0346d36..1c94015 100644
--- a/doc/src/sgml/postgres.sgml
+++ b/doc/src/sgml/postgres.sgml
@@ -159,6 +159,7 @@
   &monitoring;
   &diskusage;
   &wal;
+  &logical-replication;
   &regress;
 
  </part>
diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml
index 8acdff1..34007d3 100644
--- a/doc/src/sgml/reference.sgml
+++ b/doc/src/sgml/reference.sgml
@@ -54,11 +54,13 @@
    &alterOperatorClass;
    &alterOperatorFamily;
    &alterPolicy;
+   &alterPublication;
    &alterRole;
    &alterRule;
    &alterSchema;
    &alterSequence;
    &alterServer;
+   &alterSubscription;
    &alterSystem;
    &alterTable;
    &alterTableSpace;
@@ -100,11 +102,13 @@
    &createOperatorClass;
    &createOperatorFamily;
    &createPolicy;
+   &createPublication;
    &createRole;
    &createRule;
    &createSchema;
    &createSequence;
    &createServer;
+   &createSubscription;
    &createTable;
    &createTableAs;
    &createTableSpace;
@@ -144,11 +148,13 @@
    &dropOperatorFamily;
    &dropOwned;
    &dropPolicy;
+   &dropPublication;
    &dropRole;
    &dropRule;
    &dropSchema;
    &dropSequence;
    &dropServer;
+   &dropSubscription;
    &dropTable;
    &dropTableSpace;
    &dropTSConfig;
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 54d66d5..43e2853 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -29,13 +29,19 @@
 #include "catalog/pg_subscription.h"
 
 #include "commands/defrem.h"
+#include "commands/replicationcmds.h"
 
 #include "executor/spi.h"
 
 #include "nodes/makefuncs.h"
 
+#include "replication/logical.h"
+#include "replication/logicalproto.h"
+#include "replication/logicalworker.h"
+#include "replication/origin.h"
 #include "replication/reorderbuffer.h"
-#include "commands/replicationcmds.h"
+#include "replication/logicalworker.h"
+#include "replication/walreceiver.h"
 
 #include "utils/array.h"
 #include "utils/builtins.h"
@@ -164,7 +170,12 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 	bool		enabled_given;
 	bool		enabled;
 	char	   *conninfo;
+	char	   *slotname;
 	List	   *publications;
+	WalReceiverConnHandle  *wrchandle = NULL;
+	WalReceiverConnAPI	   *wrcapi = NULL;
+	walrcvconn_init_fn		walrcvconn_init;
+	XLogRecPtr	lsn;
 
 	check_replication_permissions();
 
@@ -184,6 +195,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 	/* Parse and check options. */
 	parse_subscription_options(stmt->options, &enabled_given, &enabled,
 							   &conninfo, &publications);
+	slotname = stmt->subname;
 
 	/* TODO: improve error messages here. */
 	if (conninfo == NULL)
@@ -202,7 +214,7 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 
 	values[Anum_pg_subscription_dbid - 1] = ObjectIdGetDatum(MyDatabaseId);
 	values[Anum_pg_subscription_subname - 1] =
-		DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
+		DirectFunctionCall1(namein, CStringGetDatum(slotname));
 	values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
 	values[Anum_pg_subscription_subconninfo - 1] =
 		CStringGetTextDatum(conninfo);
@@ -218,13 +230,47 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 	CatalogUpdateIndexes(rel, tup);
 	heap_freetuple(tup);
 
-	ObjectAddressSet(myself, SubscriptionRelationId, suboid);
+	/*
+	 * Now that the catalog update is done, try to reserve slot at the
+	 * provider node using replication connection.
+	 */
+	wrcapi = palloc0(sizeof(WalReceiverConnAPI));
+
+	walrcvconn_init = (walrcvconn_init_fn)
+		load_external_function("libpqwalreceiver",
+							   "_PG_walreceirver_conn_init", false, NULL);
+
+	if (walrcvconn_init == NULL)
+		elog(ERROR, "libpqwalreceiver does not declare _PG_walreceirver_conn_init symbol");
+
+	wrchandle = walrcvconn_init(wrcapi);
+	if (wrcapi->connect == NULL ||
+		wrcapi->create_slot == NULL)
+		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
+
+	wrcapi->connect(wrchandle, conninfo, true, stmt->subname);
+	wrcapi->create_slot(wrchandle, slotname, true, &lsn);
+	ereport(NOTICE,
+			(errmsg("created replication slot \"%s\" on provider",
+					slotname)));
+	/*
+	 * Setup replication origin tracking.
+	 * TODO: do this only when it does not already exist?
+	 */
+	replorigin_create(slotname);
+
+	/* And we are done with the remote side. */
+	wrcapi->disconnect(wrchandle);
 
 	heap_close(rel, RowExclusiveLock);
 
 	/* Make the changes visible. */
 	CommandCounterIncrement();
 
+	ApplyLauncherWakeupOnCommit();
+
+	ObjectAddressSet(myself, SubscriptionRelationId, subid);
+
 	return myself;
 }
 
@@ -302,6 +348,11 @@ AlterSubscription(AlterSubscriptionStmt *stmt)
 	heap_freetuple(tup);
 	heap_close(rel, RowExclusiveLock);
 
+	/* Make the changes visible. */
+	CommandCounterIncrement();
+
+	ApplyLauncherWakeupOnCommit();
+
 	return myself;
 }
 
@@ -313,19 +364,135 @@ DropSubscriptionById(Oid subid)
 {
 	Relation	rel;
 	HeapTuple	tup;
+	Datum		datum;
+	bool		isnull;
+	char	   *subname;
+	char	   *conninfo;
+	char	   *slotname;
+	TupleDesc	tupdesc;
+	RepOriginId	originid;
+	MemoryContext			tmpctx,
+							oldctx;
+	WalReceiverConnHandle  *wrchandle = NULL;
+	WalReceiverConnAPI	   *wrcapi = NULL;
+	walrcvconn_init_fn		walrcvconn_init;
 
 	check_replication_permissions();
 
 	rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
 
+	if (GetTopTransactionIdIfAny() != InvalidTransactionId)
+		ereport(ERROR,
+				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+				 errmsg("DROP SUBSCRIPTION must be first action in transaction")));
+
 	tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
 
 	if (!HeapTupleIsValid(tup))
 		elog(ERROR, "cache lookup failed for subscription %u", subid);
 
+	tupdesc = RelationGetDescr(rel);
+
+	/*
+	 * Create temporary memory context to keep copy of subscription
+	 * info needed later in the execution.
+	 */
+	tmpctx = AllocSetContextCreate(TopMemoryContext,
+										  "DropSubscription Ctx",
+										  ALLOCSET_DEFAULT_MINSIZE,
+										  ALLOCSET_DEFAULT_INITSIZE,
+										  ALLOCSET_DEFAULT_MAXSIZE);
+	oldctx = MemoryContextSwitchTo(tmpctx);
+
+	/* Get subname */
+	datum = heap_getattr(tup, Anum_pg_subscription_subname, tupdesc,
+						 &isnull);
+	Assert(!isnull);
+	subname = pstrdup(NameStr(*DatumGetName(datum)));
+
+	/* Get conninfo */
+	datum = heap_getattr(tup, Anum_pg_subscription_subconninfo, tupdesc,
+						 &isnull);
+	Assert(!isnull);
+	conninfo = pstrdup(TextDatumGetCString(datum));
+
+	/* Get slotname */
+	datum = heap_getattr(tup, Anum_pg_subscription_subslotname, tupdesc,
+						 &isnull);
+	Assert(!isnull);
+	slotname = pstrdup(NameStr(*DatumGetName(datum)));
+
+	MemoryContextSwitchTo(oldctx);
+
+	/* Remove the tuple from catalog. */
 	simple_heap_delete(rel, &tup->t_self);
 
 	ReleaseSysCache(tup);
 
-	heap_close(rel, RowExclusiveLock);
+	heap_close(rel, NoLock);
+
+	originid = replorigin_by_name(slotname, true);
+	if (originid != InvalidRepOriginId)
+		replorigin_drop(originid);
+
+	/* Commit the transaction to make the change visible to laucnher. */
+	PopActiveSnapshot();
+	CommitTransactionCommand();
+
+	/* Signal the launcher so that it kills the apply proccess. */
+	ApplyLauncherWakeup();
+
+	StartTransactionCommand();
+
+	/*
+	 * Now that the catalog update is done, try to reserve slot at the
+	 * provider node using replication connection.
+	 */
+	wrcapi = palloc0(sizeof(WalReceiverConnAPI));
+
+	walrcvconn_init = (walrcvconn_init_fn)
+		load_external_function("libpqwalreceiver",
+							   "_PG_walreceirver_conn_init", false, NULL);
+
+	if (walrcvconn_init == NULL)
+		elog(ERROR, "libpqwalreceiver does not declare _PG_walreceirver_conn_init symbol");
+
+	wrchandle = walrcvconn_init(wrcapi);
+	if (wrcapi->connect == NULL ||
+		wrcapi->drop_slot == NULL)
+		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
+
+	/*
+	 * We must ignore error as that would make it impossible to drop
+	 * subscription when provider is down.
+	 */
+	oldctx = CurrentMemoryContext;
+	PG_TRY();
+	{
+		wrcapi->connect(wrchandle, conninfo, true, subname);
+		wrcapi->drop_slot(wrchandle, slotname);
+		ereport(NOTICE,
+				(errmsg("dropped replication slot \"%s\" on provider",
+						slotname)));
+		wrcapi->disconnect(wrchandle);
+	}
+	PG_CATCH();
+	{
+		MemoryContext	ectx;
+		ErrorData	   *edata;
+
+		ectx = MemoryContextSwitchTo(oldctx);
+		/* Save error info */
+		edata = CopyErrorData();
+		MemoryContextSwitchTo(ectx);
+		FlushErrorState();
+
+		ereport(WARNING,
+				(errmsg("there was problem dropping the replication slot "
+						"\"%s\" on provider", slotname),
+				 errdetail("The error was: %s", edata->message),
+				 errhint("You may have to drop it manually")));
+		FreeErrorData(edata);
+	}
+	PG_END_TRY();
 }
diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c
index af7b26c..907623b 100644
--- a/src/backend/executor/nodeModifyTable.c
+++ b/src/backend/executor/nodeModifyTable.c
@@ -232,7 +232,7 @@ ExecCheckTIDVisible(EState *estate,
  *		Returns RETURNING result if any, otherwise NULL.
  * ----------------------------------------------------------------
  */
-static TupleTableSlot *
+TupleTableSlot *
 ExecInsert(ModifyTableState *mtstate,
 		   TupleTableSlot *slot,
 		   TupleTableSlot *planSlot,
@@ -536,7 +536,7 @@ ExecInsert(ModifyTableState *mtstate,
  *		Returns RETURNING result if any, otherwise NULL.
  * ----------------------------------------------------------------
  */
-static TupleTableSlot *
+TupleTableSlot *
 ExecDelete(ItemPointer tupleid,
 		   HeapTuple oldtuple,
 		   TupleTableSlot *planSlot,
@@ -794,7 +794,7 @@ ldelete:;
  *		Returns RETURNING result if any, otherwise NULL.
  * ----------------------------------------------------------------
  */
-static TupleTableSlot *
+TupleTableSlot *
 ExecUpdate(ItemPointer tupleid,
 		   HeapTuple oldtuple,
 		   TupleTableSlot *slot,
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 699c934..fc998cd 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -93,6 +93,9 @@ struct BackgroundWorkerHandle
 
 static BackgroundWorkerArray *BackgroundWorkerData;
 
+/* Enables registration of internal background workers. */
+bool internal_bgworker_registration_in_progress = false;
+
 /*
  * Calculate shared memory needed.
  */
@@ -745,7 +748,8 @@ RegisterBackgroundWorker(BackgroundWorker *worker)
 		ereport(DEBUG1,
 		 (errmsg("registering background worker \"%s\"", worker->bgw_name)));
 
-	if (!process_shared_preload_libraries_in_progress)
+	if (!process_shared_preload_libraries_in_progress &&
+		!internal_bgworker_registration_in_progress)
 	{
 		if (!IsUnderPostmaster)
 			ereport(LOG,
diff --git a/src/backend/postmaster/postmaster.c b/src/backend/postmaster/postmaster.c
index f5c8e9d..2211532 100644
--- a/src/backend/postmaster/postmaster.c
+++ b/src/backend/postmaster/postmaster.c
@@ -113,6 +113,7 @@
 #include "postmaster/pgarch.h"
 #include "postmaster/postmaster.h"
 #include "postmaster/syslogger.h"
+#include "replication/logicalworker.h"
 #include "replication/walsender.h"
 #include "storage/fd.h"
 #include "storage/ipc.h"
@@ -416,6 +417,7 @@ static void maybe_start_bgworker(void);
 static bool CreateOptsFile(int argc, char *argv[], char *fullprogname);
 static pid_t StartChildProcess(AuxProcType type);
 static void StartAutovacuumWorker(void);
+static void register_internal_bgworkers(void);
 static void InitPostmasterDeathWatchHandle(void);
 
 /*
@@ -925,6 +927,12 @@ PostmasterMain(int argc, char *argv[])
 #endif
 
 	/*
+	 * Register internal bgworkers before we give external modules chance
+	 * to do the same.
+	 */
+	register_internal_bgworkers();
+
+	/*
 	 * process any libraries that should be preloaded at postmaster start
 	 */
 	process_shared_preload_libraries();
@@ -5641,6 +5649,39 @@ assign_backendlist_entry(RegisteredBgWorker *rw)
 }
 
 /*
+ * Register internal background workers.
+ *
+ * This is here mainly because the permanent bgworkers are normally allowed
+ * to be registered only when share preload libraries are loaded which does
+ * not work for the internal ones.
+ */
+static void
+register_internal_bgworkers(void)
+{
+	internal_bgworker_registration_in_progress = true;
+
+	/* Register the logical replication worker launcher if appropriate. */
+	if (!IsBinaryUpgrade && max_logical_replication_workers > 0)
+	{
+		BackgroundWorker bgw;
+
+		bgw.bgw_flags =	BGWORKER_SHMEM_ACCESS |
+			BGWORKER_BACKEND_DATABASE_CONNECTION;
+		bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+		bgw.bgw_main = ApplyLauncherMain;
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "logical replication launcher");
+		bgw.bgw_restart_time = 5;
+		bgw.bgw_notify_pid = 0;
+		bgw.bgw_main_arg = (Datum) 0;
+
+		RegisterBackgroundWorker(&bgw);
+	}
+
+	internal_bgworker_registration_in_progress = false;
+}
+
+/*
  * If the time is right, start one background worker.
  *
  * As a side effect, the bgworker control variables are set or reset whenever
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index f28a792..4c4d441 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -60,6 +60,7 @@ static void libpqrcv_readtimelinehistoryfile(WalReceiverConnHandle *handle,
 static char *libpqrcv_create_slot(WalReceiverConnHandle *handle,
 								  char *slotname, bool logical,
 								  XLogRecPtr *lsn);
+static void libpqrcv_drop_slot(WalReceiverConnHandle *handle, char *slotname);
 static bool libpqrcv_startstreaming_physical(WalReceiverConnHandle *handle,
 								 TimeLineID tli, XLogRecPtr startpoint,
 								 char *slotname);
@@ -96,6 +97,7 @@ _PG_walreceirver_conn_init(WalReceiverConnAPI *wrcapi)
 	wrcapi->identify_system = libpqrcv_identify_system;
 	wrcapi->readtimelinehistoryfile = libpqrcv_readtimelinehistoryfile;
 	wrcapi->create_slot = libpqrcv_create_slot;
+	wrcapi->drop_slot = libpqrcv_drop_slot;
 	wrcapi->startstreaming_physical = libpqrcv_startstreaming_physical;
 	wrcapi->startstreaming_logical = libpqrcv_startstreaming_logical;
 	wrcapi->endstreaming = libpqrcv_endstreaming;
@@ -274,7 +276,7 @@ libpqrcv_create_slot(WalReceiverConnHandle *handle, char *slotname,
 
 	if (PQresultStatus(res) != PGRES_TUPLES_OK)
 	{
-		elog(FATAL, "could not crate replication slot \"%s\": %s\n",
+		elog(ERROR, "could not crate replication slot \"%s\": %s\n",
 			 slotname, PQerrorMessage(handle->streamConn));
 	}
 
@@ -287,6 +289,28 @@ libpqrcv_create_slot(WalReceiverConnHandle *handle, char *slotname,
 	return snapshot;
 }
 
+/*
+ * Drop replication slot.
+ */
+static void
+libpqrcv_drop_slot(WalReceiverConnHandle *handle, char *slotname)
+{
+	PGresult	   *res;
+	char			cmd[256];
+
+	snprintf(cmd, sizeof(cmd),
+			 "DROP_REPLICATION_SLOT \"%s\"", slotname);
+
+	res = libpqrcv_PQexec(handle, cmd);
+
+	if (PQresultStatus(res) != PGRES_COMMAND_OK)
+	{
+		elog(ERROR, "could not drop replication slot \"%s\": %s\n",
+			 slotname, PQerrorMessage(handle->streamConn));
+	}
+
+	PQclear(res);
+}
 
 /*
  * Start streaming WAL data from given startpoint and timeline.
@@ -353,7 +377,7 @@ libpqrcv_startstreaming_logical(WalReceiverConnHandle *handle,
 					 (uint32) (startpoint >> 32),
 					 (uint32) startpoint);
 
-	/* Send options */
+	/* Add options */
 	if (options)
 		appendStringInfo(&cmd, "( %s )", options);
 
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 438811e..ab6e11e 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -14,7 +14,8 @@ include $(top_builddir)/src/Makefile.global
 
 override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
-OBJS = decode.o logical.o logicalfuncs.o message.o origin.o proto.o \
-	   publication.o reorderbuffer.o snapbuild.o subscription.o
+OBJS = apply.o decode.o launcher.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/apply.c b/src/backend/replication/logical/apply.c
new file mode 100644
index 0000000..eb7af19
--- /dev/null
+++ b/src/backend/replication/logical/apply.c
@@ -0,0 +1,1435 @@
+/*-------------------------------------------------------------------------
+ * apply.c
+ *	   PostgreSQL logical replication
+ *
+ * Copyright (c) 2012-2016, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/apply.c
+ *
+ * NOTES
+ *	  This file contains the worker which applies logical changes as they come
+ *	  from remote logical replication stream.
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "pgstat.h"
+#include "funcapi.h"
+
+#include "access/xact.h"
+#include "access/xlog_internal.h"
+
+#include "catalog/namespace.h"
+
+#include "commands/trigger.h"
+
+#include "executor/executor.h"
+#include "executor/nodeModifyTable.h"
+
+#include "libpq/pqformat.h"
+#include "libpq/pqsignal.h"
+
+#include "mb/pg_wchar.h"
+
+#include "optimizer/planner.h"
+
+#include "parser/parse_relation.h"
+
+#include "postmaster/bgworker.h"
+#include "postmaster/postmaster.h"
+
+#include "replication/decode.h"
+#include "replication/logical.h"
+#include "replication/logicalproto.h"
+#include "replication/logicalworker.h"
+#include "replication/reorderbuffer.h"
+#include "replication/origin.h"
+#include "replication/snapbuild.h"
+#include "replication/subscription.h"
+#include "replication/walreceiver.h"
+
+#include "rewrite/rewriteHandler.h"
+
+#include "storage/bufmgr.h"
+#include "storage/ipc.h"
+#include "storage/lmgr.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/guc.h"
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/timeout.h"
+#include "utils/tqual.h"
+#include "utils/syscache.h"
+
+typedef struct FlushPosition
+{
+	dlist_node node;
+	XLogRecPtr local_end;
+	XLogRecPtr remote_end;
+} FlushPosition;
+
+static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);
+
+static MemoryContext	ApplyContext;
+static bool				in_remote_transaction = false;
+
+static Subscription	   *MySubscription = NULL;
+static bool				got_SIGTERM = false;
+
+typedef struct LogicalRepRelMapEntry {
+	LogicalRepRelation	remoterel;		/* key is remoterel.remoteid */
+
+	/* Mapping to local relation, filled as needed. */
+    Oid					reloid;			/* local relation id */
+    Relation			rel;			/* relcache entry */
+	int                *attmap;			/* map of remote attributes to
+										 * local ones */
+	AttInMetadata	   *attin;			/* cached info used in type
+										 * conversion */
+} LogicalRepRelMapEntry;
+
+static HTAB *LogicalRepRelMap = NULL;
+
+/* filled by libpqreceiver when loaded */
+static WalReceiverConnAPI *wrcapi = NULL;
+static WalReceiverConnHandle *wrchandle = NULL;
+
+static void send_feedback(XLogRecPtr recvpos, int64 now, bool force);
+void pglogical_apply_main(Datum main_arg);
+
+static bool tuple_find_by_replidx(Relation rel, LockTupleMode lockmode,
+					  TupleTableSlot *searchslot, TupleTableSlot *slot);
+
+
+
+/*
+ * Relcache invalidation callback for our relation map cache.
+ */
+static void
+logicalreprelmap_invalidate_cb(Datum arg, Oid reloid)
+{
+	LogicalRepRelMapEntry  *entry;
+
+	/* Just to be sure. */
+	if (LogicalRepRelMap == NULL)
+		return;
+
+	if (reloid != InvalidOid)
+	{
+		HASH_SEQ_STATUS status;
+
+		hash_seq_init(&status, LogicalRepRelMap);
+
+		/* TODO, use inverse lookup hastable? */
+		while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+		{
+			if (entry->reloid == reloid)
+				entry->reloid = InvalidOid;
+		}
+	}
+	else
+	{
+		/* invalidate all cache entries */
+		HASH_SEQ_STATUS status;
+
+		hash_seq_init(&status, LogicalRepRelMap);
+
+		while ((entry = (LogicalRepRelMapEntry *) hash_seq_search(&status)) != NULL)
+			entry->reloid = InvalidOid;
+	}
+}
+
+/*
+ * Initialize the relation map cache.
+ */
+static void
+remoterelmap_init(void)
+{
+	HASHCTL		ctl;
+
+	/* Make sure we've initialized CacheMemoryContext. */
+	if (CacheMemoryContext == NULL)
+		CreateCacheMemoryContext();
+
+	/* Initialize the hash table. */
+	MemSet(&ctl, 0, sizeof(ctl));
+	ctl.keysize = sizeof(uint32);
+	ctl.entrysize = sizeof(LogicalRepRelMapEntry);
+	ctl.hcxt = CacheMemoryContext;
+
+	LogicalRepRelMap = hash_create("logicalrep relation map cache", 128, &ctl,
+								   HASH_ELEM | HASH_CONTEXT);
+
+	/* Watch for invalidation events. */
+	CacheRegisterRelcacheCallback(logicalreprelmap_invalidate_cb,
+								  (Datum) 0);
+}
+
+/*
+ * Free the entry of a relation map cache.
+ */
+static void
+remoterelmap_free_entry(LogicalRepRelMapEntry *entry)
+{
+	LogicalRepRelation *remoterel;
+
+	remoterel = &entry->remoterel;
+
+	pfree(remoterel->nspname);
+	pfree(remoterel->relname);
+
+	if (remoterel->natts > 0)
+	{
+		int	i;
+
+		for (i = 0; i < remoterel->natts; i++)
+			pfree(remoterel->attnames[i]);
+
+		pfree(remoterel->attnames);
+	}
+
+	if (entry->attmap)
+		pfree(entry->attmap);
+
+	remoterel->natts = 0;
+	entry->reloid = InvalidOid;
+	entry->rel = NULL;
+}
+
+/*
+ * Add new entry or update existing entry in the relation map cache.
+ *
+ * Called when new relation mapping is sent by the provider to update
+ * our expected view of incoming data from said provider.
+ */
+static void
+remoterelmap_update(LogicalRepRelation *remoterel)
+{
+	MemoryContext			oldctx;
+	LogicalRepRelMapEntry  *entry;
+	bool					found;
+	int						i;
+
+	if (LogicalRepRelMap == NULL)
+		remoterelmap_init();
+
+	/*
+	 * HASH_ENTER returns the existing entry if present or creates a new one.
+	 */
+	entry = hash_search(LogicalRepRelMap, (void *) &remoterel->remoteid,
+						HASH_ENTER, &found);
+
+	if (found)
+		remoterelmap_free_entry(entry);
+
+	/* Make cached copy of the data */
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	entry->remoterel.remoteid = remoterel->remoteid;
+	entry->remoterel.nspname = pstrdup(remoterel->nspname);
+	entry->remoterel.relname = pstrdup(remoterel->relname);
+	entry->remoterel.natts = remoterel->natts;
+	entry->remoterel.attnames = palloc(remoterel->natts * sizeof(char*));
+	for (i = 0; i < remoterel->natts; i++)
+		entry->remoterel.attnames[i] = pstrdup(remoterel->attnames[i]);
+	entry->attmap = palloc(remoterel->natts * sizeof(int));
+	entry->reloid = InvalidOid;
+	MemoryContextSwitchTo(oldctx);
+}
+
+/*
+ * Find attribute index in TupleDesc struct by attribute name.
+ */
+static int
+tupdesc_get_att_by_name(TupleDesc desc, const char *attname)
+{
+	int		i;
+
+	for (i = 0; i < desc->natts; i++)
+	{
+		Form_pg_attribute att = desc->attrs[i];
+
+		if (strcmp(NameStr(att->attname), attname) == 0)
+			return i;
+	}
+
+	elog(ERROR, "unknown column name %s", attname);
+}
+
+/*
+ * Open the local relation associated with the remote one.
+ */
+static LogicalRepRelMapEntry *
+logicalreprel_open(uint32 remoteid, LOCKMODE lockmode)
+{
+	LogicalRepRelMapEntry  *entry;
+	bool		found;
+
+	if (LogicalRepRelMap == NULL)
+		remoterelmap_init();
+
+	/* Search for existing entry. */
+	entry = hash_search(LogicalRepRelMap, (void *) &remoteid,
+						HASH_FIND, &found);
+
+	if (!found)
+		elog(ERROR, "cache lookup failed for remote relation %u",
+			 remoteid);
+
+	/* Need to update the local cache? */
+	if (!OidIsValid(entry->reloid))
+	{
+		Oid			nspid;
+		Oid			relid;
+		int			i;
+		TupleDesc	desc;
+		LogicalRepRelation *remoterel;
+
+		remoterel = &entry->remoterel;
+
+		nspid = LookupExplicitNamespace(remoterel->nspname, false);
+		relid = get_relname_relid(remoterel->relname, nspid);
+		entry->rel = heap_open(relid, lockmode);
+
+		desc = RelationGetDescr(entry->rel);
+		for (i = 0; i < remoterel->natts; i++)
+			entry->attmap[i] = tupdesc_get_att_by_name(desc,
+													   remoterel->attnames[i]);
+
+		entry->reloid = RelationGetRelid(entry->rel);
+	}
+	else
+		entry->rel = heap_open(entry->reloid, lockmode);
+
+	return entry;
+}
+
+/*
+ * Close the previously opened logical relation.
+ */
+static void
+logicalreprel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
+{
+	heap_close(rel->rel, lockmode);
+	rel->rel = NULL;
+}
+
+
+/*
+ * Make sure that we started local transaction.
+ *
+ * Also switches to ApplyContext as necessary.
+ */
+static bool
+ensure_transaction(void)
+{
+	if (IsTransactionState())
+	{
+		if (CurrentMemoryContext != ApplyContext)
+			MemoryContextSwitchTo(ApplyContext);
+		return false;
+	}
+
+	StartTransactionCommand();
+	MemoryContextSwitchTo(ApplyContext);
+	return true;
+}
+
+
+/*
+ * Executor state preparation for evaluation of constraint expressions,
+ * indexes and triggers.
+ *
+ * This is based on similar code in copy.c
+ */
+static EState *
+create_estate_for_relation(LogicalRepRelMapEntry *rel)
+{
+	EState	   *estate;
+	ResultRelInfo *resultRelInfo;
+	RangeTblEntry *rte;
+
+	estate = CreateExecutorState();
+
+	rte = makeNode(RangeTblEntry);
+	rte->rtekind = RTE_RELATION;
+	rte->relid = RelationGetRelid(rel->rel);
+	rte->relkind = rel->rel->rd_rel->relkind;
+	estate->es_range_table = list_make1(rte);
+
+	resultRelInfo = makeNode(ResultRelInfo);
+	InitResultRelInfo(resultRelInfo, rel->rel, 1, 0);
+
+	estate->es_result_relations = resultRelInfo;
+	estate->es_num_result_relations = 1;
+	estate->es_result_relation_info = resultRelInfo;
+
+	/* Triggers might need a slot */
+	if (resultRelInfo->ri_TrigDesc)
+		estate->es_trig_tuple_slot = ExecInitExtraTupleSlot(estate);
+
+	return estate;
+}
+
+/*
+ * Check if the local attribute is present in relation definition used
+ * by upstream and hence updated by the replication.
+ */
+static bool
+physatt_in_attmap(LogicalRepRelMapEntry *rel, int attid)
+{
+	AttrNumber	i;
+
+	/* Fast path for tables that are same on upstream and downstream. */
+	if (attid < rel->remoterel.natts && rel->attmap[attid] == attid)
+		return true;
+
+	/* Try to find the attribute in the map. */
+	for (i = 0; i < rel->remoterel.natts; i++)
+		if (rel->attmap[i] == attid)
+			return true;
+
+	return false;
+}
+
+/*
+ * Executes default values for columns for which we can't map to remote
+ * relation columns.
+ *
+ * This allows us to support tables which have more columns on the downstream
+ * than on the upsttream.
+ */
+static void
+FillSlotDefaults(LogicalRepRelMapEntry *rel, EState *estate,
+				 TupleTableSlot *slot)
+{
+	TupleDesc	desc = RelationGetDescr(rel->rel);
+	AttrNumber	num_phys_attrs = desc->natts;
+	int			i;
+	AttrNumber	attnum,
+				num_defaults = 0;
+	int		   *defmap;
+	ExprState **defexprs;
+	ExprContext *econtext;
+
+	econtext = GetPerTupleExprContext(estate);
+
+	/* We got all the data via replication, no need to evaluate anything. */
+	if (num_phys_attrs == rel->remoterel.natts)
+		return;
+
+	defmap = (int *) palloc(num_phys_attrs * sizeof(int));
+	defexprs = (ExprState **) palloc(num_phys_attrs * sizeof(ExprState *));
+
+	for (attnum = 0; attnum < num_phys_attrs; attnum++)
+	{
+		Expr	   *defexpr;
+
+		if (desc->attrs[attnum]->attisdropped)
+			continue;
+
+		if (physatt_in_attmap(rel, attnum))
+			continue;
+
+		defexpr = (Expr *) build_column_default(rel->rel, attnum + 1);
+
+		if (defexpr != NULL)
+		{
+			/* Run the expression through planner */
+			defexpr = expression_planner(defexpr);
+
+			/* Initialize executable expression in copycontext */
+			defexprs[num_defaults] = ExecInitExpr(defexpr, NULL);
+			defmap[num_defaults] = attnum;
+			num_defaults++;
+		}
+
+	}
+
+	for (i = 0; i < num_defaults; i++)
+		slot->tts_values[defmap[i]] =
+			ExecEvalExpr(defexprs[i], econtext, &slot->tts_isnull[defmap[i]],
+						 NULL);
+}
+
+/*
+ * Store data in C string form into slot.
+ * This is similar to BuildTupleFromCStrings but TupleTableSlot fits our
+ * use better.
+ */
+static void
+SlotStoreCStrings(TupleTableSlot *slot, char **values)
+{
+	int		natts = slot->tts_tupleDescriptor->natts;
+	int		i;
+
+	ExecClearTuple(slot);
+
+	/* Call the "in" function for each non-dropped attribute */
+	for (i = 0; i < natts; i++)
+	{
+		Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i];
+
+		if (!att->attisdropped && values[i] != NULL)
+		{
+			Oid typinput;
+			Oid typioparam;
+
+			getTypeInputInfo(att->atttypid, &typinput, &typioparam);
+			slot->tts_values[i] = OidInputFunctionCall(typinput, values[i],
+													   typioparam,
+													   att->atttypmod);
+			slot->tts_isnull[i] = false;
+		}
+		else
+		{
+			/* We assign NULL for both NULL values and dropped attributes. */
+			slot->tts_values[i] = (Datum) 0;
+			slot->tts_isnull[i] = true;
+		}
+	}
+
+	ExecStoreVirtualTuple(slot);
+}
+
+/*
+ * Modify slot with user data provided as C strigs.
+ * This is somewhat similar to heap_modify_tuple but also calls the type
+ * input fuction on the user data as the input is the rext representation
+ * of the types.
+ */
+static void
+SlotModifyCStrings(TupleTableSlot *slot, char **values, bool *replaces)
+{
+	int		natts = slot->tts_tupleDescriptor->natts;
+	int		i;
+
+	slot_getallattrs(slot);
+	ExecClearTuple(slot);
+
+	/* Call the "in" function for each replaced attribute */
+	for (i = 0; i < natts; i++)
+	{
+		Form_pg_attribute att = slot->tts_tupleDescriptor->attrs[i];
+
+		if (!replaces[i])
+			continue;
+
+		if (values[i] != NULL)
+		{
+			Oid typinput;
+			Oid typioparam;
+
+			getTypeInputInfo(att->atttypid, &typinput, &typioparam);
+			slot->tts_values[i] = OidInputFunctionCall(typinput, values[i],
+													   typioparam,
+													   att->atttypmod);
+			slot->tts_isnull[i] = false;
+		}
+		else
+		{
+			slot->tts_values[i] = (Datum) 0;
+			slot->tts_isnull[i] = true;
+		}
+	}
+
+	ExecStoreVirtualTuple(slot);
+}
+
+/*
+ * Handle BEGIN message.
+ */
+static void
+handle_begin(StringInfo s)
+{
+	XLogRecPtr		commit_lsn;
+	TimestampTz		commit_time;
+	TransactionId	remote_xid;
+
+	logicalrep_read_begin(s, &commit_lsn, &commit_time, &remote_xid);
+
+	replorigin_session_origin_timestamp = commit_time;
+	replorigin_session_origin_lsn = commit_lsn;
+
+	in_remote_transaction = true;
+
+	pgstat_report_activity(STATE_RUNNING, NULL);
+}
+
+/*
+ * Handle COMMIT message.
+ *
+ * TODO, support tracking of multiple origins
+ */
+static void
+handle_commit(StringInfo s)
+{
+	XLogRecPtr		commit_lsn;
+	XLogRecPtr		end_lsn;
+	TimestampTz		commit_time;
+
+	logicalrep_read_commit(s, &commit_lsn, &end_lsn, &commit_time);
+
+	Assert(commit_lsn == replorigin_session_origin_lsn);
+	Assert(commit_time == replorigin_session_origin_timestamp);
+
+	if (IsTransactionState())
+	{
+		FlushPosition *flushpos;
+
+		CommitTransactionCommand();
+		MemoryContextSwitchTo(CacheMemoryContext);
+
+		/* Track commit lsn  */
+		flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
+		flushpos->local_end = XactLastCommitEnd;
+		flushpos->remote_end = end_lsn;
+
+		dlist_push_tail(&lsn_mapping, &flushpos->node);
+		MemoryContextSwitchTo(ApplyContext);
+	}
+
+	in_remote_transaction = false;
+
+	pgstat_report_activity(STATE_IDLE, NULL);
+}
+
+/*
+ * Handle ORIGIN message.
+ *
+ * TODO, support tracking of multiple origins
+ */
+static void
+handle_origin(StringInfo s)
+{
+	/*
+	 * ORIGIN message can only come inside remote transaction and before
+	 * any actual writes.
+	 */
+	if (!in_remote_transaction || IsTransactionState())
+		elog(ERROR, "ORIGIN message sent out of order");
+}
+
+/*
+ * Handle RELATION message.
+ *
+ * Note we don't do validation against local schema here. The validation is
+ * posponed until first change for given relation comes.
+ */
+static void
+handle_relation(StringInfo s)
+{
+	LogicalRepRelation  *rel;
+
+	rel = logicalrep_read_rel(s);
+	remoterelmap_update(rel);
+}
+
+
+/*
+ * Handle INSERT message.
+ */
+static void
+handle_insert(StringInfo s)
+{
+	LogicalRepRelMapEntry *rel;
+	LogicalRepTupleData	newtup;
+	LogicalRepRelId		relid;
+	EState			   *estate;
+	TupleTableSlot	   *remoteslot;
+	MemoryContext		oldctx;
+
+	ensure_transaction();
+
+	relid = logicalrep_read_insert(s, &newtup);
+	rel = logicalreprel_open(relid, RowExclusiveLock);
+
+	/* Initialize the executor state. */
+	estate = create_estate_for_relation(rel);
+	remoteslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(remoteslot, RelationGetDescr(rel->rel));
+
+	/* Process and store remote tuple in the slot */
+	oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+	SlotStoreCStrings(remoteslot, newtup.values);
+	FillSlotDefaults(rel, estate, remoteslot);
+	MemoryContextSwitchTo(oldctx);
+
+	PushActiveSnapshot(GetTransactionSnapshot());
+	ExecOpenIndices(estate->es_result_relation_info, false);
+
+	ExecInsert(NULL, /* mtstate is only used for onconflict handling which we don't support atm */
+			   remoteslot,
+			   remoteslot,
+			   NIL,
+			   ONCONFLICT_NONE,
+			   estate,
+			   false);
+
+	/* Cleanup. */
+	ExecCloseIndices(estate->es_result_relation_info);
+	PopActiveSnapshot();
+	ExecResetTupleTable(estate->es_tupleTable, false);
+	FreeExecutorState(estate);
+
+	logicalreprel_close(rel, NoLock);
+
+	CommandCounterIncrement();
+}
+
+/*
+ * Handle UPDATE message.
+ *
+ * TODO: FDW support
+ */
+static void
+handle_update(StringInfo s)
+{
+	LogicalRepRelMapEntry *rel;
+	LogicalRepRelId		relid;
+	EState			   *estate;
+	EPQState			epqstate;
+	LogicalRepTupleData	oldtup;
+	LogicalRepTupleData	newtup;
+	bool				hasoldtup;
+	TupleTableSlot	   *localslot;
+	TupleTableSlot	   *remoteslot;
+	bool				found;
+	MemoryContext		oldctx;
+
+	ensure_transaction();
+
+	relid = logicalrep_read_update(s, &hasoldtup, &oldtup,
+								   &newtup);
+	rel = logicalreprel_open(relid, RowExclusiveLock);
+
+	/* Initialize the executor state. */
+	estate = create_estate_for_relation(rel);
+	remoteslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(remoteslot, RelationGetDescr(rel->rel));
+	localslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(localslot, RelationGetDescr(rel->rel));
+	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
+
+	PushActiveSnapshot(GetTransactionSnapshot());
+	ExecOpenIndices(estate->es_result_relation_info, false);
+
+	/* Find the tuple using the replica identity index. */
+	oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+	SlotStoreCStrings(remoteslot, hasoldtup ? oldtup.values : newtup.values);
+	MemoryContextSwitchTo(oldctx);
+	found = tuple_find_by_replidx(rel->rel, LockTupleExclusive,
+								  remoteslot, localslot);
+	ExecClearTuple(remoteslot);
+
+	/*
+	 * Tuple found.
+	 *
+	 * Note this will fail if there are other conflicting unique indexes.
+	 */
+	if (found)
+	{
+		/* Process and store remote tuple in the slot */
+		oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+		ExecStoreTuple(localslot->tts_tuple, remoteslot, InvalidBuffer, false);
+		SlotModifyCStrings(remoteslot, newtup.values, newtup.changed);
+		MemoryContextSwitchTo(oldctx);
+
+		EvalPlanQualSetSlot(&epqstate, remoteslot);
+
+		ExecUpdate(&localslot->tts_tuple->t_self,
+				   localslot->tts_tuple,
+				   remoteslot,
+				   localslot,
+				   &epqstate,
+				   estate,
+				   false);
+	}
+	else
+	{
+		/*
+		 * The tuple to be updated could not be found.
+		 *
+		 * TODO what to do here?
+		 */
+	}
+
+	/* Cleanup. */
+	ExecCloseIndices(estate->es_result_relation_info);
+	PopActiveSnapshot();
+	EvalPlanQualEnd(&epqstate);
+	ExecResetTupleTable(estate->es_tupleTable, false);
+	FreeExecutorState(estate);
+
+	logicalreprel_close(rel, NoLock);
+
+	CommandCounterIncrement();
+}
+
+/*
+ * Handle DELETE message.
+ *
+ * TODO: FDW support
+ */
+static void
+handle_delete(StringInfo s)
+{
+	LogicalRepRelMapEntry *rel;
+	LogicalRepTupleData	oldtup;
+	LogicalRepRelId		relid;
+	EState			   *estate;
+	EPQState			epqstate;
+	TupleTableSlot	   *remoteslot;
+	TupleTableSlot	   *localslot;
+	bool				found;
+	MemoryContext		oldctx;
+
+	ensure_transaction();
+
+	relid = logicalrep_read_delete(s, &oldtup);
+	rel = logicalreprel_open(relid, RowExclusiveLock);
+
+	/* Initialize the executor state. */
+	estate = create_estate_for_relation(rel);
+	remoteslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(remoteslot, RelationGetDescr(rel->rel));
+	localslot = ExecInitExtraTupleSlot(estate);
+	ExecSetSlotDescriptor(localslot, RelationGetDescr(rel->rel));
+	EvalPlanQualInit(&epqstate, estate, NULL, NIL, -1);
+
+	PushActiveSnapshot(GetTransactionSnapshot());
+	ExecOpenIndices(estate->es_result_relation_info, false);
+
+	/* Find the tuple using the replica identity index. */
+	oldctx = MemoryContextSwitchTo(GetPerTupleMemoryContext(estate));
+	SlotStoreCStrings(remoteslot, oldtup.values);
+	MemoryContextSwitchTo(oldctx);
+	found = tuple_find_by_replidx(rel->rel, LockTupleExclusive,
+								  remoteslot, localslot);
+	/* If found delete it. */
+	if (found)
+	{
+		EvalPlanQualSetSlot(&epqstate, localslot);
+		ExecDelete(&localslot->tts_tuple->t_self,
+				   localslot->tts_tuple,
+				   localslot,
+				   &epqstate,
+				   estate,
+				   false);
+	}
+	else
+	{
+		/* The tuple to be deleted could not be found.*/
+	}
+
+	/* Cleanup. */
+	ExecCloseIndices(estate->es_result_relation_info);
+	PopActiveSnapshot();
+	EvalPlanQualEnd(&epqstate);
+	ExecResetTupleTable(estate->es_tupleTable, false);
+	FreeExecutorState(estate);
+
+	logicalreprel_close(rel, NoLock);
+
+	CommandCounterIncrement();
+}
+
+
+/*
+ * Logical replication protocol message dispatcher.
+ */
+static void
+handle_message(StringInfo s)
+{
+	char action = pq_getmsgbyte(s);
+
+	switch (action)
+	{
+		/* BEGIN */
+		case 'B':
+			handle_begin(s);
+			break;
+		/* COMMIT */
+		case 'C':
+			handle_commit(s);
+			break;
+		/* INSERT */
+		case 'I':
+			handle_insert(s);
+			break;
+		/* UPDATE */
+		case 'U':
+			handle_update(s);
+			break;
+		/* DELETE */
+		case 'D':
+			handle_delete(s);
+			break;
+		/* RELATION */
+		case 'R':
+			handle_relation(s);
+			break;
+		/* ORIGIN */
+		case 'O':
+			handle_origin(s);
+			break;
+		default:
+			elog(ERROR, "unknown action of type %c", action);
+	}
+}
+
+/*
+ * Figure out which write/flush positions to report to the walsender process.
+ *
+ * We can't simply report back the last LSN the walsender sent us because the
+ * local transaction might not yet be flushed to disk locally. Instead we
+ * build a list that associates local with remote LSNs for every commit. When
+ * reporting back the flush position to the sender we iterate that list and
+ * check which entries on it are already locally flushed. Those we can report
+ * as having been flushed.
+ *
+ * Returns true if there's no outstanding transactions that need to be
+ * flushed.
+ */
+static bool
+get_flush_position(XLogRecPtr *write, XLogRecPtr *flush)
+{
+	dlist_mutable_iter iter;
+	XLogRecPtr	local_flush = GetFlushRecPtr();
+
+	*write = InvalidXLogRecPtr;
+	*flush = InvalidXLogRecPtr;
+
+	dlist_foreach_modify(iter, &lsn_mapping)
+	{
+		FlushPosition *pos =
+			dlist_container(FlushPosition, node, iter.cur);
+
+		*write = pos->remote_end;
+
+		if (pos->local_end <= local_flush)
+		{
+			*flush = pos->remote_end;
+			dlist_delete(iter.cur);
+			pfree(pos);
+		}
+		else
+		{
+			/*
+			 * Don't want to uselessly iterate over the rest of the list which
+			 * could potentially be long. Instead get the last element and
+			 * grab the write position from there.
+			 */
+			pos = dlist_tail_element(FlushPosition, node,
+									 &lsn_mapping);
+			*write = pos->remote_end;
+			return false;
+		}
+	}
+
+	return dlist_is_empty(&lsn_mapping);
+}
+
+
+/*
+ * Apply main loop.
+ */
+static void
+ApplyLoop(void)
+{
+	XLogRecPtr	last_received = InvalidXLogRecPtr;
+
+	/* Init the ApplyContext which we use for easier cleanup. */
+	ApplyContext = AllocSetContextCreate(TopMemoryContext,
+										 "ApplyContext",
+										 ALLOCSET_DEFAULT_MINSIZE,
+										 ALLOCSET_DEFAULT_INITSIZE,
+										 ALLOCSET_DEFAULT_MAXSIZE);
+
+	/* mark as idle, before starting to loop */
+	pgstat_report_activity(STATE_IDLE, NULL);
+
+	while (!got_SIGTERM)
+	{
+		pgsocket	fd = PGINVALID_SOCKET;
+		int			rc;
+		int			len;
+		char	   *buf = NULL;
+		bool		endofstream = false;
+
+		CHECK_FOR_INTERRUPTS();
+
+		MemoryContextSwitchTo(ApplyContext);
+
+		len = wrcapi->receive(wrchandle, &buf, &fd);
+
+		if (len != 0)
+		{
+			/* Process the data */
+			for (;;)
+			{
+				CHECK_FOR_INTERRUPTS();
+
+				if (len == 0)
+				{
+					break;
+				}
+				else if (len < 0)
+				{
+					elog(NOTICE, "data stream from provider has ended");
+					endofstream = true;
+					break;
+				}
+				else
+				{
+					int c;
+					StringInfoData s;
+
+					/* Ensure we are reading the data into our memory context. */
+					MemoryContextSwitchTo(ApplyContext);
+
+					initStringInfo(&s);
+					s.data = buf;
+					s.len = len;
+					s.maxlen = -1;
+
+					c = pq_getmsgbyte(&s);
+
+					if (c == 'w')
+					{
+						XLogRecPtr	start_lsn;
+						XLogRecPtr	end_lsn;
+
+						start_lsn = pq_getmsgint64(&s);
+						end_lsn = pq_getmsgint64(&s);
+						pq_getmsgint64(&s); /* sendTime */
+
+						if (last_received < start_lsn)
+							last_received = start_lsn;
+
+						if (last_received < end_lsn)
+							last_received = end_lsn;
+
+						handle_message(&s);
+					}
+					else if (c == 'k')
+					{
+						XLogRecPtr endpos;
+						bool reply_requested;
+
+						endpos = pq_getmsgint64(&s);
+						/* timestamp = */ pq_getmsgint64(&s);
+						reply_requested = pq_getmsgbyte(&s);
+
+						send_feedback(endpos,
+									  GetCurrentTimestamp(),
+									  reply_requested);
+					}
+					/* other message types are purposefully ignored */
+				}
+
+				len = wrcapi->receive(wrchandle, &buf, &fd);
+			}
+		}
+
+		/* confirm all writes at once */
+		send_feedback(last_received, GetCurrentTimestamp(), false);
+
+		/* Cleanup the memory. */
+		MemoryContextResetAndDeleteChildren(ApplyContext);
+		MemoryContextSwitchTo(TopMemoryContext);
+
+		/* Check if we need to exit the streaming loop. */
+		if (endofstream)
+			break;
+
+		/*
+		 * Wait for more data or latch.
+		 */
+		rc = WaitLatchOrSocket(&MyProc->procLatch,
+							   WL_SOCKET_READABLE | WL_LATCH_SET |
+							   WL_TIMEOUT | WL_POSTMASTER_DEATH,
+							   fd, 1000L);
+
+		/* Emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+		ResetLatch(&MyProc->procLatch);
+	}
+}
+
+/*
+ * Send a Standby Status Update message to server.
+ *
+ * 'recvpos' is the latest LSN we've received data to, force is set if we need
+ * to send a response to avoid timeouts.
+ */
+static void
+send_feedback(XLogRecPtr recvpos, int64 now, bool force)
+{
+	static StringInfo	reply_message = NULL;
+
+	static XLogRecPtr last_recvpos = InvalidXLogRecPtr;
+	static XLogRecPtr last_writepos = InvalidXLogRecPtr;
+	static XLogRecPtr last_flushpos = InvalidXLogRecPtr;
+
+	XLogRecPtr writepos;
+	XLogRecPtr flushpos;
+
+	/* It's legal to not pass a recvpos */
+	if (recvpos < last_recvpos)
+		recvpos = last_recvpos;
+
+	if (get_flush_position(&writepos, &flushpos))
+	{
+		/*
+		 * No outstanding transactions to flush, we can report the latest
+		 * received position. This is important for synchronous replication.
+		 */
+		flushpos = writepos = recvpos;
+	}
+
+	if (writepos < last_writepos)
+		writepos = last_writepos;
+
+	if (flushpos < last_flushpos)
+		flushpos = last_flushpos;
+
+	/* if we've already reported everything we're good */
+	if (!force &&
+		writepos == last_writepos &&
+		flushpos == last_flushpos)
+		return;
+
+	if (!reply_message)
+	{
+		MemoryContext	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+		reply_message = makeStringInfo();
+		MemoryContextSwitchTo(oldctx);
+	}
+	else
+		resetStringInfo(reply_message);
+
+	pq_sendbyte(reply_message, 'r');
+	pq_sendint64(reply_message, recvpos);		/* write */
+	pq_sendint64(reply_message, flushpos);		/* flush */
+	pq_sendint64(reply_message, writepos);		/* apply */
+	pq_sendint64(reply_message, now);			/* sendTime */
+	pq_sendbyte(reply_message, false);			/* replyRequested */
+
+	elog(DEBUG2, "sending feedback (force %d) to recv %X/%X, write %X/%X, flush %X/%X",
+		 force,
+		 (uint32) (recvpos >> 32), (uint32) recvpos,
+		 (uint32) (writepos >> 32), (uint32) writepos,
+		 (uint32) (flushpos >> 32), (uint32) flushpos
+		);
+
+	wrcapi->send(wrchandle, reply_message->data, reply_message->len);
+
+	if (recvpos > last_recvpos)
+		last_recvpos = recvpos;
+	if (writepos > last_writepos)
+		last_writepos = writepos;
+	if (flushpos > last_flushpos)
+		last_flushpos = flushpos;
+}
+
+/* SIGTERM: set flag to exit at next convenient time */
+static void
+LogicalWorkerSigTermHandler(SIGNAL_ARGS)
+{
+	got_SIGTERM = true;
+}
+
+/* Logical Replication Apply worker entry point */
+void
+ApplyWorkerMain(Datum main_arg)
+{
+	int				worker_slot = DatumGetObjectId(main_arg);
+	MemoryContext	oldctx;
+	RepOriginId		originid;
+	XLogRecPtr		origin_startpos;
+	char		   *options;
+	walrcvconn_init_fn walrcvconn_init;
+
+	/* Attach to slot */
+	logicalrep_worker_attach(worker_slot);
+
+	/* Setup signal handling */
+	pqsignal(SIGTERM, LogicalWorkerSigTermHandler);
+	BackgroundWorkerUnblockSignals();
+
+	/* Make it easy to identify our processes. */
+	SetConfigOption("application_name", MyBgworkerEntry->bgw_name,
+					PGC_USERSET, PGC_S_SESSION);
+
+	/* Load the libpq-specific functions */
+	wrcapi = palloc0(sizeof(WalReceiverConnAPI));
+
+	walrcvconn_init = (walrcvconn_init_fn)
+		load_external_function("libpqwalreceiver",
+							   "_PG_walreceirver_conn_init", false, NULL);
+
+	if (walrcvconn_init == NULL)
+		elog(ERROR, "libpqwalreceiver does not declare _PG_walreceirver_conn_init symbol");
+
+	wrchandle = walrcvconn_init(wrcapi);
+	if (wrcapi->connect == NULL ||
+		wrcapi->startstreaming_logical == NULL ||
+		wrcapi->identify_system == NULL ||
+		wrcapi->receive == NULL || wrcapi->send == NULL ||
+		wrcapi->disconnect == NULL)
+		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
+
+	Assert(CurrentResourceOwner == NULL);
+	CurrentResourceOwner = ResourceOwnerCreate(NULL,
+											   "logical replication apply");
+
+	/* Setup synchronous commit according to the user's wishes */
+/*	SetConfigOption("synchronous_commit",
+					logical_apply_synchronous_commit,
+					PGC_BACKEND, PGC_S_OVERRIDE);
+*/
+	/* Run as replica session replication role. */
+	SetConfigOption("session_replication_role", "replica",
+					PGC_SUSET, PGC_S_OVERRIDE);
+
+	/* Connect to our database. */
+	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
+											  InvalidOid);
+
+	StartTransactionCommand();
+
+	/* Load the subscription into persistent memory context. */
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	MySubscription = GetSubscription(MyLogicalRepWorker->subid);
+	MemoryContextSwitchTo(oldctx);
+
+	elog(LOG, "logical replication apply for subscription %s started",
+		 MySubscription->name);
+
+	/* Setup replication origin tracking. */
+	originid = replorigin_by_name(MySubscription->slotname, true);
+	if (!OidIsValid(originid))
+		originid = replorigin_create(MySubscription->slotname);
+	replorigin_session_setup(originid);
+	replorigin_session_origin = originid;
+	origin_startpos = replorigin_session_get_progress(false);
+
+	CommitTransactionCommand();
+
+	/* Connect to the origin and start the replication. */
+	elog(DEBUG1, "connecting to provider using connection string %s",
+		 MySubscription->conninfo);
+	wrcapi->connect(wrchandle, MySubscription->conninfo, true,
+					MySubscription->name);
+
+	/* Build option string for the plugin. */
+	options = logicalrep_build_options(MySubscription->publications);
+
+	/* Start streaming from the slot. */
+	wrcapi->startstreaming_logical(wrchandle, origin_startpos,
+								   MySubscription->slotname, options);
+
+	/* Run the main loop. */
+	ApplyLoop();
+
+	wrcapi->disconnect(wrchandle);
+
+	/* We should only get here if we received sigTERM */
+	proc_exit(0);
+}
+
+/*
+ * Setup a ScanKey for a search in the relation 'rel' for a tuple 'key' that
+ * is setup to match 'rel' (*NOT* idxrel!).
+ *
+ * Returns whether any column contains NULLs.
+ *
+ * This is not generic routine, it expects the idxrel to be replication
+ * identity of a rel and meet all limitations associated with that.
+ */
+static bool
+build_replindex_scan_key(ScanKey skey, Relation rel, Relation idxrel,
+						 TupleTableSlot *searchslot)
+{
+	int			attoff;
+	bool		isnull;
+	Datum		indclassDatum;
+	oidvector  *opclass;
+	int2vector *indkey = &idxrel->rd_index->indkey;
+	bool		hasnulls = false;
+
+	Assert(RelationGetReplicaIndex(rel) == RelationGetRelid(idxrel));
+
+	indclassDatum = SysCacheGetAttr(INDEXRELID, idxrel->rd_indextuple,
+									Anum_pg_index_indclass, &isnull);
+	Assert(!isnull);
+	opclass = (oidvector *) DatumGetPointer(indclassDatum);
+
+	/* Build scankey for every attribute in the index. */
+	for (attoff = 0; attoff < RelationGetNumberOfAttributes(idxrel); attoff++)
+	{
+		Oid			operator;
+		Oid			opfamily;
+		RegProcedure regop;
+		int			pkattno = attoff + 1;
+		int			mainattno = indkey->values[attoff];
+		Oid			atttype = attnumTypeId(rel, mainattno);
+		Oid			optype = get_opclass_input_type(opclass->values[attoff]);
+
+		/*
+		 * Load the operator info, we need this to get the equality operator
+		 * function for the scankey.
+		 */
+		opfamily = get_opclass_family(opclass->values[attoff]);
+
+		operator = get_opfamily_member(opfamily, optype,
+									   optype,
+									   BTEqualStrategyNumber);
+
+		if (!OidIsValid(operator))
+			elog(ERROR,
+				 "could not lookup equality operator for type %u, optype %u in opfamily %u",
+				 atttype, optype, opfamily);
+
+		regop = get_opcode(operator);
+
+		/* Initialize the scankey. */
+		ScanKeyInit(&skey[attoff],
+					pkattno,
+					BTEqualStrategyNumber,
+					regop,
+					searchslot->tts_values[mainattno - 1]);
+
+		/* Check for null value. */
+		if (searchslot->tts_isnull[mainattno - 1])
+		{
+			hasnulls = true;
+			skey[attoff].sk_flags |= SK_ISNULL;
+		}
+	}
+
+	return hasnulls;
+}
+
+/*
+ * Search the relation 'rel' for tuple using the replication index.
+ *
+ * If a matching tuple is found lock it with lockmode, fill the slot with its
+ * contents and return true, return false is returned otherwise.
+ */
+static bool
+tuple_find_by_replidx(Relation rel, LockTupleMode lockmode,
+					  TupleTableSlot *searchslot, TupleTableSlot *slot)
+{
+	HeapTuple		scantuple;
+	ScanKeyData		skey[INDEX_MAX_KEYS];
+	IndexScanDesc	scan;
+	SnapshotData	snap;
+	TransactionId	xwait;
+	Oid				idxoid;
+	Relation		idxrel;
+	bool			found;
+
+	/* Open REPLICA IDENTITY index.*/
+	idxoid = RelationGetReplicaIndex(rel);
+	if (!OidIsValid(idxoid))
+	{
+		elog(ERROR, "could not find configured replica identity for table \"%s\"",
+			 RelationGetRelationName(rel));
+		return false;
+	}
+	idxrel = index_open(idxoid, RowExclusiveLock);
+
+	/* Start an index scan. */
+	InitDirtySnapshot(snap);
+	scan = index_beginscan(rel, idxrel, &snap,
+						   RelationGetNumberOfAttributes(idxrel),
+						   0);
+
+	/* Build scan key. */
+	build_replindex_scan_key(skey, rel, idxrel, searchslot);
+
+retry:
+	found = false;
+
+	index_rescan(scan, skey, RelationGetNumberOfAttributes(idxrel), NULL, 0);
+
+	/* Try to find the tuple */
+	if ((scantuple = index_getnext(scan, ForwardScanDirection)) != NULL)
+	{
+		found = true;
+		ExecStoreTuple(scantuple, slot, InvalidBuffer, false);
+		ExecMaterializeSlot(slot);
+
+		xwait = TransactionIdIsValid(snap.xmin) ?
+			snap.xmin : snap.xmax;
+
+		/*
+		 * If the tuple is locked, wait for locking transaction to finish
+		 * and retry.
+		 */
+		if (TransactionIdIsValid(xwait))
+		{
+			XactLockTableWait(xwait, NULL, NULL, XLTW_None);
+			goto retry;
+		}
+	}
+
+	/* Found tuple, try to lock it in the lockmode. */
+	if (found)
+	{
+		Buffer buf;
+		HeapUpdateFailureData hufd;
+		HTSU_Result res;
+		HeapTupleData locktup;
+
+		ItemPointerCopy(&slot->tts_tuple->t_self, &locktup.t_self);
+
+		PushActiveSnapshot(GetLatestSnapshot());
+
+		res = heap_lock_tuple(rel, &locktup, GetCurrentCommandId(false),
+							  lockmode,
+							  false /* wait */,
+							  false /* don't follow updates */,
+							  &buf, &hufd);
+		/* the tuple slot already has the buffer pinned */
+		ReleaseBuffer(buf);
+
+		PopActiveSnapshot();
+
+		switch (res)
+		{
+			case HeapTupleMayBeUpdated:
+				break;
+			case HeapTupleUpdated:
+				/* XXX: Improve handling here */
+				ereport(LOG,
+						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+						 errmsg("concurrent update, retrying")));
+				goto retry;
+			case HeapTupleInvisible:
+				elog(ERROR, "attempted to lock invisible tuple");
+			default:
+				elog(ERROR, "unexpected heap_lock_tuple status: %u", res);
+				break;
+		}
+	}
+
+	index_endscan(scan);
+
+	/* Don't release lock until commit. */
+	index_close(idxrel, NoLock);
+
+	return found;
+}
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
new file mode 100644
index 0000000..385260e
--- /dev/null
+++ b/src/backend/replication/logical/launcher.c
@@ -0,0 +1,542 @@
+/*-------------------------------------------------------------------------
+ * launcher.c
+ *	   PostgreSQL logical replication apply launcher process
+ *
+ * Copyright (c) 2012-2016, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/launcher.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "miscadmin.h"
+#include "pgstat.h"
+
+#include "access/heapam.h"
+#include "access/htup.h"
+#include "access/htup_details.h"
+#include "access/xact.h"
+
+#include "catalog/pg_subscription.h"
+
+#include "libpq/pqsignal.h"
+
+#include "postmaster/bgworker.h"
+#include "postmaster/fork_process.h"
+#include "postmaster/postmaster.h"
+
+#include "replication/logicalworker.h"
+#include "replication/subscription.h"
+
+#include "storage/ipc.h"
+#include "storage/proc.h"
+#include "storage/procarray.h"
+#include "storage/procsignal.h"
+
+#include "tcop/tcopprot.h"
+
+#include "utils/memutils.h"
+#include "utils/ps_status.h"
+#include "utils/timeout.h"
+#include "utils/snapmgr.h"
+
+int	max_logical_replication_workers = 4;
+LogicalRepWorker *MyLogicalRepWorker = NULL;
+
+typedef struct LogicalRepCtxStruct
+{
+	/* Supervisor process. */
+	pid_t		launcher_pid;
+
+	/* Background workers. */
+	LogicalRepWorker	workers[FLEXIBLE_ARRAY_MEMBER];
+} LogicalRepCtxStruct;
+
+LogicalRepCtxStruct *LogicalRepCtx;
+
+static LogicalRepWorker *logicalrep_worker_find(Oid subid);
+static void logicalrep_worker_launch(Oid dbid, Oid subid);
+static void logicalrep_worker_stop(LogicalRepWorker *worker);
+static void logicalrep_worker_onexit(int code, Datum arg);
+static void logicalrep_worker_detach(void);
+
+static bool xacthook_do_signal_launcher = false;
+
+/*
+ * Load the list of subscriptions.
+ *
+ * Only the fields interesting for worker start/stop functions are filled for
+ * each subscription.
+ */
+static List *
+get_subscription_list(void)
+{
+	List	   *res = NIL;
+	Relation	rel;
+	HeapScanDesc scan;
+	HeapTuple	tup;
+	MemoryContext resultcxt;
+
+	/* This is the context that we will allocate our output data in */
+	resultcxt = CurrentMemoryContext;
+
+	/*
+	 * Start a transaction so we can access pg_database, and get a snapshot.
+	 * We don't have a use for the snapshot itself, but we're interested in
+	 * the secondary effect that it sets RecentGlobalXmin.  (This is critical
+	 * for anything that reads heap pages, because HOT may decide to prune
+	 * them even if the process doesn't attempt to modify any tuples.)
+	 */
+	StartTransactionCommand();
+	(void) GetTransactionSnapshot();
+
+	rel = heap_open(SubscriptionRelationId, AccessShareLock);
+	scan = heap_beginscan_catalog(rel, 0, NULL);
+
+	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
+	{
+		Form_pg_subscription subform = (Form_pg_subscription) GETSTRUCT(tup);
+		Subscription   *sub;
+		MemoryContext	oldcxt;
+
+		/*
+		 * Allocate our results in the caller's context, not the
+		 * transaction's. We do this inside the loop, and restore the original
+		 * context at the end, so that leaky things like heap_getnext() are
+		 * not called in a potentially long-lived context.
+		 */
+		oldcxt = MemoryContextSwitchTo(resultcxt);
+
+		sub = (Subscription *) palloc(sizeof(Subscription));
+		sub->oid = HeapTupleGetOid(tup);
+		sub->dbid = subform->dbid;
+		sub->enabled = subform->subenabled;
+
+		/* We don't fill fields we are not intereste in. */
+		sub->name = NULL;
+		sub->conninfo = NULL;
+		sub->slotname = NULL;
+		sub->publications = NIL;
+
+		res = lappend(res, sub);
+		MemoryContextSwitchTo(oldcxt);
+	}
+
+	heap_endscan(scan);
+	heap_close(rel, AccessShareLock);
+
+	CommitTransactionCommand();
+
+	return res;
+}
+
+/*
+ * Wait for a background worker to start up and attach to the shmem context.
+ *
+ * This is like WaitForBackgroundWorkerStartup(), except that we wait for
+ * attaching, not just start and we also just exit if postmaster died.
+ */
+static bool
+WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
+							   BackgroundWorkerHandle *handle)
+{
+	BgwHandleStatus status;
+	int			rc;
+
+	for (;;)
+	{
+		pid_t		pid;
+
+		CHECK_FOR_INTERRUPTS();
+
+		status = GetBackgroundWorkerPid(handle, &pid);
+
+		/*
+		 * Worker started and attached to our shmem. This check is safe
+		 * because only laucher ever starts the workers, so nobody can steal
+		 * the worker slot.
+		 */
+		if (status == BGWH_STARTED && worker->proc)
+			return true;
+		/* Worker didn't start or died before attaching to our shmem. */
+		if (status == BGWH_STOPPED)
+			return false;
+
+		/*
+		 * We need timeout because we generaly don't get notified via latch
+		 * about the worker attach.
+		 */
+		rc = WaitLatch(MyLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 1000L);
+
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+		ResetLatch(MyLatch);
+	}
+
+	return status;
+}
+
+/*
+ * Walks the workers array and searches for one that matches given
+ * subscription id.
+ */
+static LogicalRepWorker *
+logicalrep_worker_find(Oid subid)
+{
+	int	i;
+	LogicalRepWorker   *res = NULL;
+
+	/* Block concurrent modification. */
+	LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+
+	/* Search for attached worker for a given subscription id. */
+	for (i = 0; i < max_logical_replication_workers; i++)
+	{
+		LogicalRepWorker   *w = &LogicalRepCtx->workers[i];
+		if (w->subid == subid && w->proc && IsBackendPid(w->proc->pid))
+		{
+			res = w;
+			break;
+		}
+	}
+
+	LWLockRelease(LogicalRepLauncherLock);
+
+	return res;
+}
+
+/*
+ * Start new apply background worker.
+ */
+static void
+logicalrep_worker_launch(Oid dbid, Oid subid)
+{
+	BackgroundWorker	bgw;
+	BackgroundWorkerHandle *bgw_handle;
+	int					slot;
+	LogicalRepWorker   *worker = NULL;
+
+	ereport(LOG,
+			(errmsg("starting logical replication worker for subscription %u",
+					subid)));
+
+	/*
+	 * We need to do the modification of the shared memory under lock so that
+	 * we have consistent view.
+	 */
+	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+
+	/* Find unused worker slot. */
+	for (slot = 0; slot < max_logical_replication_workers; slot++)
+	{
+		if (!LogicalRepCtx->workers[slot].proc)
+		{
+			worker = &LogicalRepCtx->workers[slot];
+			break;
+		}
+	}
+
+	/* Bail if not found */
+	if (worker == NULL)
+	{
+		ereport(WARNING,
+				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+				 errmsg("logical replication worker registration failed, "
+						"you might want to increase "
+						"max_logical_replication_workers setting")));
+		return;
+	}
+
+	/* Prepare the worker info. */
+	memset(worker, 0, sizeof(LogicalRepWorker));
+	worker->dbid = dbid;
+	worker->subid = subid;
+
+	LWLockRelease(LogicalRepLauncherLock);
+
+	/* Register the new dynamic worker. */
+	bgw.bgw_flags =	BGWORKER_SHMEM_ACCESS |
+		BGWORKER_BACKEND_DATABASE_CONNECTION;
+	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+	bgw.bgw_main = ApplyWorkerMain;
+
+	bgw.bgw_restart_time = BGW_NEVER_RESTART;
+	bgw.bgw_notify_pid = MyProcPid;
+	bgw.bgw_main_arg = slot;
+
+	if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
+	{
+		ereport(WARNING,
+				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+				 errmsg("logical replication worker registration failed, "
+						"you might want to increase "
+						"max_logical_replication_workers setting")));
+		return;
+	}
+
+	/* Now wait until it attaches. */
+	if (!WaitForReplicationWorkerAttach(worker, bgw_handle))
+	{
+		ereport(WARNING,
+				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+				 errmsg("could not launch logical replication worker")));
+		return;
+	}
+}
+
+/*
+ * Stop the logical replication worker and wait until it detaches from the
+ * slot.
+ */
+static void
+logicalrep_worker_stop(LogicalRepWorker *worker)
+{
+	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+
+	/* Check that the worker is up and what we expect. */
+	if (!worker->proc)
+		return;
+	if (!IsBackendPid(worker->proc->pid))
+		return;
+
+	/* Terminate the worker. */
+	kill(worker->proc->pid, SIGTERM);
+
+	LWLockRelease(LogicalRepLauncherLock);
+
+	/* Wait for it to detach. */
+	for (;;)
+	{
+		int	rc = WaitLatch(&MyProc->procLatch,
+						   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+						   1000L);
+
+        /* emergency bailout if postmaster has died */
+        if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+        ResetLatch(&MyProc->procLatch);
+
+		CHECK_FOR_INTERRUPTS();
+
+		if (!worker->proc)
+			return;
+	}
+}
+
+/*
+ * Attach to a slot.
+ */
+void
+logicalrep_worker_attach(int slot)
+{
+	/* Block concurrent access. */
+	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+
+	Assert(slot >= 0 && slot < max_logical_replication_workers);
+	MyLogicalRepWorker = &LogicalRepCtx->workers[slot];
+
+	if (MyLogicalRepWorker->proc)
+		ereport(ERROR,
+			   (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				errmsg("logical replication worker slot %d already used by "
+					   "another worker", slot)));
+
+	MyLogicalRepWorker->proc = MyProc;
+	before_shmem_exit(logicalrep_worker_onexit, (Datum) 0);
+
+	LWLockRelease(LogicalRepLauncherLock);
+}
+
+/*
+ * Detach the worker (cleans up the worker info).
+ */
+static void
+logicalrep_worker_detach(void)
+{
+	/* Block concurrent access. */
+	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+
+	MyLogicalRepWorker->dbid = InvalidOid;
+	MyLogicalRepWorker->subid = InvalidOid;
+	MyLogicalRepWorker->proc = NULL;
+
+	LWLockRelease(LogicalRepLauncherLock);
+}
+
+/*
+ * Cleanup function.
+ *
+ * Called on logical replication worker exit.
+ */
+static void
+logicalrep_worker_onexit(int code, Datum arg)
+{
+	logicalrep_worker_detach();
+}
+
+/*
+ * ApplyLauncherShmemSize
+ *		Compute space needed for replication launcher shared memory
+ */
+Size
+ApplyLauncherShmemSize(void)
+{
+	Size		size;
+
+	/*
+	 * Need the fixed struct and the array of LogicalRepWorker.
+	 */
+	size = sizeof(LogicalRepCtxStruct);
+	size = MAXALIGN(size);
+	size = add_size(size, mul_size(max_logical_replication_workers,
+								   sizeof(LogicalRepWorker)));
+	return size;
+}
+
+/*
+ * ApplyLauncherShmemInit
+ *		Allocate and initialize replication launcher shared memory
+ */
+void
+ApplyLauncherShmemInit(void)
+{
+	bool		found;
+
+	LogicalRepCtx = (LogicalRepCtxStruct *)
+		ShmemInitStruct("Logical Replication Launcher Data",
+						ApplyLauncherShmemSize(),
+						&found);
+
+	if (IsUnderPostmaster)
+	{
+		Assert(found);
+		return;
+	}
+
+	memset(LogicalRepCtx, 0, ApplyLauncherShmemSize());
+}
+
+static void
+xacthook_signal_launcher(XactEvent event, void *arg)
+{
+	switch (event)
+	{
+		case XACT_EVENT_COMMIT:
+			if (xacthook_do_signal_launcher)
+				ApplyLauncherWakeup();
+			break;
+		default:
+			/* We're not interested in other tx events */
+			break;
+	}
+}
+
+void
+ApplyLauncherWakeupOnCommit(void)
+{
+	if (!xacthook_do_signal_launcher)
+	{
+		RegisterXactCallback(xacthook_signal_launcher, NULL);
+		xacthook_do_signal_launcher = true;
+	}
+}
+
+void
+ApplyLauncherWakeup(void)
+{
+	if (IsBackendPid(LogicalRepCtx->launcher_pid))
+		kill(LogicalRepCtx->launcher_pid, SIGUSR1);
+}
+
+/*
+ * Main loop for the apply launcher process.
+ */
+void
+ApplyLauncherMain(Datum main_arg)
+{
+	ereport(LOG,
+			(errmsg("logical replication launcher started")));
+
+	/* Establish signal handlers. */
+	pqsignal(SIGTERM, die);
+	BackgroundWorkerUnblockSignals();
+
+	/* Make it easy to identify our processes. */
+	SetConfigOption("application_name", MyBgworkerEntry->bgw_name,
+					PGC_USERSET, PGC_S_SESSION);
+
+	LogicalRepCtx->launcher_pid = MyProcPid;
+
+	/*
+	 * Establish connection to nailed catalogs (we only ever access
+	 * pg_subscription).
+	 */
+	BackgroundWorkerInitializeConnection(NULL, NULL);
+
+	/* Enter main loop */
+	for (;;)
+    {
+		int			rc;
+		List	   *sublist;
+		ListCell   *lc;
+		Subscription   *startsub = NULL;
+		MemoryContext	subctx;
+		MemoryContext	oldctx;
+
+		CHECK_FOR_INTERRUPTS();
+
+		/* Use temporary context for the database list and worker info. */
+		subctx = AllocSetContextCreate(TopMemoryContext,
+									   "Logical Replication Launcher sublist",
+									   ALLOCSET_DEFAULT_MINSIZE,
+									   ALLOCSET_DEFAULT_INITSIZE,
+									   ALLOCSET_DEFAULT_MAXSIZE);
+		oldctx = MemoryContextSwitchTo(subctx);
+
+		/* Search for subscriptions to start or stop. */
+		sublist = get_subscription_list();
+		foreach(lc, sublist)
+		{
+			Subscription	   *sub = (Subscription *) lfirst(lc);
+			LogicalRepWorker   *w = logicalrep_worker_find(sub->oid);
+
+			if (sub->enabled && w == NULL && startsub == NULL)
+				startsub = sub;
+			else if (!sub->enabled && w != NULL)
+				logicalrep_worker_stop(w);
+		}
+
+		if (startsub)
+			logicalrep_worker_launch(startsub->dbid, startsub->oid);
+
+		/* Switch back to original memory context. */
+		MemoryContextSwitchTo(oldctx);
+		/* Clean the temporary memory. */
+		MemoryContextDelete(subctx);
+
+		/* Wait for more work. */
+		rc = WaitLatch(&MyProc->procLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   startsub ? 5000L : 180000L);
+
+        /* emergency bailout if postmaster has died */
+        if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+		ResetLatch(&MyProc->procLatch);
+	}
+
+	LogicalRepCtx->launcher_pid = 0;
+
+	/* ... and if it returns, we're done */
+	ereport(LOG,
+			(errmsg("logical replication launcher shutting down")));
+
+	proc_exit(0);
+}
diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c
index c04b17f..423cb0f 100644
--- a/src/backend/storage/ipc/ipci.c
+++ b/src/backend/storage/ipc/ipci.c
@@ -28,6 +28,7 @@
 #include "postmaster/bgworker_internals.h"
 #include "postmaster/bgwriter.h"
 #include "postmaster/postmaster.h"
+#include "replication/logicalworker.h"
 #include "replication/slot.h"
 #include "replication/walreceiver.h"
 #include "replication/walsender.h"
@@ -137,6 +138,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 		size = add_size(size, ReplicationOriginShmemSize());
 		size = add_size(size, WalSndShmemSize());
 		size = add_size(size, WalRcvShmemSize());
+		size = add_size(size, ApplyLauncherShmemSize());
 		size = add_size(size, SnapMgrShmemSize());
 		size = add_size(size, BTreeShmemSize());
 		size = add_size(size, SyncScanShmemSize());
@@ -245,6 +247,7 @@ CreateSharedMemoryAndSemaphores(bool makePrivate, int port)
 	ReplicationOriginShmemInit();
 	WalSndShmemInit();
 	WalRcvShmemInit();
+	ApplyLauncherShmemInit();
 
 	/*
 	 * Set up other modules that need some shared memory space
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index f8996cd..4488ff7 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -47,3 +47,4 @@ CommitTsLock						39
 ReplicationOriginLock				40
 MultiXactTruncationLock				41
 OldSnapshotTimeMapLock				42
+LogicalRepLauncherLock				43
diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c
index 9c93df0..32856db 100644
--- a/src/backend/utils/misc/guc.c
+++ b/src/backend/utils/misc/guc.c
@@ -58,6 +58,7 @@
 #include "postmaster/postmaster.h"
 #include "postmaster/syslogger.h"
 #include "postmaster/walwriter.h"
+#include "replication/logicalworker.h"
 #include "replication/slot.h"
 #include "replication/syncrep.h"
 #include "replication/walreceiver.h"
@@ -171,6 +172,7 @@ static const char *show_tcp_keepalives_interval(void);
 static const char *show_tcp_keepalives_count(void);
 static bool check_maxconnections(int *newval, void **extra, GucSource source);
 static bool check_max_worker_processes(int *newval, void **extra, GucSource source);
+static bool check_max_logical_replication_workers(int *newval, void **extra, GucSource source);
 static bool check_autovacuum_max_workers(int *newval, void **extra, GucSource source);
 static bool check_autovacuum_work_mem(int *newval, void **extra, GucSource source);
 static bool check_effective_io_concurrency(int *newval, void **extra, GucSource source);
@@ -2474,6 +2476,18 @@ static struct config_int ConfigureNamesInt[] =
 	},
 
 	{
+		{"max_logical_replication_processes",
+			PGC_POSTMASTER,
+			RESOURCES_ASYNCHRONOUS,
+			gettext_noop("Maximum number of logical replication worker processes."),
+			NULL,
+		},
+		&max_logical_replication_workers,
+		4, 1, MAX_BACKENDS,
+		check_max_logical_replication_workers, NULL, NULL
+	},
+
+	{
 		{"log_rotation_age", PGC_SIGHUP, LOGGING_WHERE,
 			gettext_noop("Automatic log file rotation will occur after N minutes."),
 			NULL,
@@ -10184,6 +10198,14 @@ check_max_worker_processes(int *newval, void **extra, GucSource source)
 }
 
 static bool
+check_max_logical_replication_workers(int *newval, void **extra, GucSource source)
+{
+	if (*newval > max_worker_processes)
+		return false;
+	return true;
+}
+
+static bool
 check_effective_io_concurrency(int *newval, void **extra, GucSource source)
 {
 #ifdef USE_PREFETCH
diff --git a/src/include/executor/nodeModifyTable.h b/src/include/executor/nodeModifyTable.h
index 6b66353..dfb7e7c 100644
--- a/src/include/executor/nodeModifyTable.h
+++ b/src/include/executor/nodeModifyTable.h
@@ -19,5 +19,25 @@ extern ModifyTableState *ExecInitModifyTable(ModifyTable *node, EState *estate,
 extern TupleTableSlot *ExecModifyTable(ModifyTableState *node);
 extern void ExecEndModifyTable(ModifyTableState *node);
 extern void ExecReScanModifyTable(ModifyTableState *node);
+extern TupleTableSlot *ExecInsert(ModifyTableState *mtstate,
+		   TupleTableSlot *slot,
+		   TupleTableSlot *planSlot,
+		   List *arbiterIndexes,
+		   OnConflictAction onconflict,
+		   EState *estate,
+		   bool canSetTag);
+extern TupleTableSlot *ExecDelete(ItemPointer tupleid,
+		   HeapTuple oldtuple,
+		   TupleTableSlot *planSlot,
+		   EPQState *epqstate,
+		   EState *estate,
+		   bool canSetTag);
+extern TupleTableSlot *ExecUpdate(ItemPointer tupleid,
+		   HeapTuple oldtuple,
+		   TupleTableSlot *slot,
+		   TupleTableSlot *planSlot,
+		   EPQState *epqstate,
+		   EState *estate,
+		   bool canSetTag);
 
 #endif   /* NODEMODIFYTABLE_H */
diff --git a/src/include/postmaster/bgworker_internals.h b/src/include/postmaster/bgworker_internals.h
index cd6cd44..3f8e764 100644
--- a/src/include/postmaster/bgworker_internals.h
+++ b/src/include/postmaster/bgworker_internals.h
@@ -52,4 +52,6 @@ extern void StartBackgroundWorker(void) pg_attribute_noreturn();
 extern BackgroundWorker *BackgroundWorkerEntry(int slotno);
 #endif
 
+extern bool internal_bgworker_registration_in_progress;
+
 #endif   /* BGWORKER_INTERNALS_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
new file mode 100644
index 0000000..64f36d3
--- /dev/null
+++ b/src/include/replication/logicalworker.h
@@ -0,0 +1,41 @@
+/*-------------------------------------------------------------------------
+ *
+ * logicalworker.h
+ *	  Exports for logical replication workers.
+ *
+ * Portions Copyright (c) 2010-2016, PostgreSQL Global Development Group
+ *
+ * src/include/replication/logicalworker.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef LOGICALWORKER_H
+#define LOGICALWORKER_H
+
+typedef struct LogicalRepWorker
+{
+	/* Pointer to proc array. NULL if not running. */
+	PGPROC *proc;
+
+	/* Database id to connect to. */
+	Oid		dbid;
+
+	/* Subscription id for the worker. */
+	Oid		subid;
+} LogicalRepWorker;
+
+extern int max_logical_replication_workers;
+extern LogicalRepWorker *MyLogicalRepWorker;
+
+extern void ApplyLauncherMain(Datum main_arg);
+extern void ApplyWorkerMain(Datum main_arg);
+
+extern Size ApplyLauncherShmemSize(void);
+extern void ApplyLauncherShmemInit(void);
+
+extern void ApplyLauncherWakeupOnCommit(void);
+extern void ApplyLauncherWakeup(void);
+
+extern void logicalrep_worker_attach(int slot);
+
+#endif   /* LOGICALWORKER_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 0f42008..3801949 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -152,6 +152,9 @@ typedef char *(*walrcvconn_create_slot_fn) (
 									WalReceiverConnHandle *handle,
 									char *slotname, bool logical,
 									XLogRecPtr *lsn);
+typedef void (*walrcvconn_drop_slot_fn) (
+									WalReceiverConnHandle *handle,
+									char *slotname);
 typedef bool (*walrcvconn_startstreaming_physical_fn) (
 									WalReceiverConnHandle *handle,
 									TimeLineID tli, XLogRecPtr startpoint,
@@ -174,6 +177,7 @@ typedef struct WalReceiverConnAPI {
 	walrcvconn_identify_system_fn			identify_system;
 	walrcvconn_readtimelinehistoryfile_fn	readtimelinehistoryfile;
 	walrcvconn_create_slot_fn				create_slot;
+	walrcvconn_drop_slot_fn					drop_slot;
 	walrcvconn_startstreaming_physical_fn	startstreaming_physical;
 	walrcvconn_startstreaming_logical_fn	startstreaming_logical;
 	walrcvconn_endstreaming_fn				endstreaming;
diff --git a/src/test/perl/PostgresNode.pm b/src/test/perl/PostgresNode.pm
index e629373..408a2c1 100644
--- a/src/test/perl/PostgresNode.pm
+++ b/src/test/perl/PostgresNode.pm
@@ -407,8 +407,16 @@ sub init
 
 	if ($params{allows_streaming})
 	{
-		print $conf "wal_level = replica\n";
+		if ($params{allows_streaming} eq "logical")
+		{
+			print $conf "wal_level = logical\n";
+		}
+		else
+		{
+			print $conf "wal_level = replica\n";
+		}
 		print $conf "max_wal_senders = 5\n";
+		print $conf "max_replication_slots = 5\n";
 		print $conf "wal_keep_segments = 20\n";
 		print $conf "max_wal_size = 128MB\n";
 		print $conf "shared_buffers = 1MB\n";
diff --git a/src/test/subscription/.gitignore b/src/test/subscription/.gitignore
new file mode 100644
index 0000000..871e943
--- /dev/null
+++ b/src/test/subscription/.gitignore
@@ -0,0 +1,2 @@
+# Generated by test suite
+/tmp_check/
diff --git a/src/test/subscription/Makefile b/src/test/subscription/Makefile
new file mode 100644
index 0000000..54c4d19
--- /dev/null
+++ b/src/test/subscription/Makefile
@@ -0,0 +1,20 @@
+#-------------------------------------------------------------------------
+#
+# Makefile for src/test/subscription
+#
+# Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+# Portions Copyright (c) 1994, Regents of the University of California
+#
+# src/test/subscription/Makefile
+#
+#-------------------------------------------------------------------------
+
+subdir = src/test/subscription
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+
+check:
+	$(prove_check)
+
+clean distclean maintainer-clean:
+	rm -rf tmp_check
diff --git a/src/test/subscription/README b/src/test/subscription/README
new file mode 100644
index 0000000..e9e9375
--- /dev/null
+++ b/src/test/subscription/README
@@ -0,0 +1,16 @@
+src/test/subscription/README
+
+Regression tests for subscription/logical replication
+=====================================================
+
+This directory contains a test suite for subscription/logical replication.
+
+Running the tests
+=================
+
+    make check
+
+NOTE: This creates a temporary installation, and some tests may
+create one or multiple nodes, for the purpose of the tests.
+
+NOTE: This requires the --enable-tap-tests argument to configure.
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
new file mode 100644
index 0000000..dca19c4
--- /dev/null
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -0,0 +1,89 @@
+# Basic logical replication test
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+# Initialize provider node
+my $node_provider = get_new_node('provider');
+$node_provider->init(allows_streaming => 'logical');
+$node_provider->start;
+
+# Create subscriber node
+my $node_subscriber = get_new_node('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Create some preexisting content on provider
+$node_provider->safe_psql('postgres',
+	"CREATE TABLE tab_notrep AS SELECT generate_series(1,10) AS a");
+$node_provider->safe_psql('postgres',
+	"CREATE TABLE tab_ins (a int)");
+$node_provider->safe_psql('postgres',
+	"CREATE TABLE tab_rep (a int primary key)");
+
+# Setup structure on subscriber
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE tab_notrep (a int)");
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE tab_ins (a int)");
+$node_subscriber->safe_psql('postgres',
+	"CREATE TABLE tab_rep (a int primary key)");
+
+# Setup logical replication
+my $provider_connstr = $node_provider->connstr . ' dbname=postgres';
+$node_provider->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub");
+$node_provider->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub_ins_only WITH noreplicate_delete noreplicate_update");
+$node_provider->safe_psql('postgres',
+	"ALTER PUBLICATION tap_pub ADD TABLE tab_rep");
+$node_provider->safe_psql('postgres',
+	"ALTER PUBLICATION tap_pub_ins_only ADD TABLE tab_ins");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION tap_sub WITH CONNECTION '$provider_connstr' PUBLICATION tap_pub, tap_pub_ins_only");
+
+# Wait for subscriber to finish table sync
+my $appname = 'tap_sub';
+my $caughtup_query =
+"SELECT pg_current_xlog_location() <= write_location FROM pg_stat_replication WHERE application_name = '$appname';";
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+my $result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_notrep");
+print "node_subscriber: $result\n";
+is($result, qq(0), 'check non-replicated table is empty on subscriber');
+
+
+$node_provider->safe_psql('postgres',
+	"INSERT INTO tab_ins SELECT generate_series(1,50)");
+$node_provider->safe_psql('postgres',
+	"DELETE FROM tab_ins WHERE a > 20");
+$node_provider->safe_psql('postgres',
+	"UPDATE tab_ins SET a = -a");
+
+$node_provider->safe_psql('postgres',
+	"INSERT INTO tab_rep SELECT generate_series(1,50)");
+$node_provider->safe_psql('postgres',
+	"DELETE FROM tab_rep WHERE a > 20");
+$node_provider->safe_psql('postgres',
+	"UPDATE tab_rep SET a = -a");
+
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_ins");
+print "node_subscriber: $result\n";
+is($result, qq(50|1|50), 'check replicated inserts on subscriber');
+
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_rep");
+print "node_subscriber: $result\n";
+is($result, qq(20|-20|-1), 'check replicated changes on subscriber');
+
+$node_subscriber->stop('fast');
+$node_provider->stop('fast');
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
new file mode 100644
index 0000000..a126201
--- /dev/null
+++ b/src/test/subscription/t/002_types.pl
@@ -0,0 +1,509 @@
+# This tests that more complex datatypes are replicated correctly
+# by logical replication
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More tests => 3;
+
+# Initialize provider node
+my $node_provider = get_new_node('provider');
+$node_provider->init(allows_streaming => 'logical');
+$node_provider->start;
+
+# Create subscriber node
+my $node_subscriber = get_new_node('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Create some preexisting content on provider
+my $ddl =  qq(
+	CREATE TABLE public.tst_one_array (
+		a INTEGER PRIMARY KEY,
+		b INTEGER[]
+		);
+	CREATE TABLE public.tst_arrays (
+		a INTEGER[] PRIMARY KEY,
+		b TEXT[],
+		c FLOAT[],
+		d INTERVAL[]
+		);
+
+	CREATE TYPE public.tst_enum_t AS ENUM ('a', 'b', 'c', 'd', 'e');
+	CREATE TABLE public.tst_one_enum (
+		a INTEGER PRIMARY KEY,
+		b public.tst_enum_t
+		);
+	CREATE TABLE public.tst_enums (
+		a public.tst_enum_t PRIMARY KEY,
+		b public.tst_enum_t[]
+		);
+
+	CREATE TYPE public.tst_comp_basic_t AS (a FLOAT, b TEXT, c INTEGER);
+	CREATE TYPE public.tst_comp_enum_t AS (a FLOAT, b public.tst_enum_t, c INTEGER);
+	CREATE TYPE public.tst_comp_enum_array_t AS (a FLOAT, b public.tst_enum_t[], c INTEGER);
+	CREATE TABLE public.tst_one_comp (
+		a INTEGER PRIMARY KEY,
+		b public.tst_comp_basic_t
+		);
+	CREATE TABLE public.tst_comps (
+		a public.tst_comp_basic_t PRIMARY KEY,
+		b public.tst_comp_basic_t[]
+		);
+	CREATE TABLE public.tst_comp_enum (
+		a INTEGER PRIMARY KEY,
+		b public.tst_comp_enum_t
+		);
+	CREATE TABLE public.tst_comp_enum_array (
+		a public.tst_comp_enum_t PRIMARY KEY,
+		b public.tst_comp_enum_t[]
+		);
+	CREATE TABLE public.tst_comp_one_enum_array (
+		a INTEGER PRIMARY KEY,
+		b public.tst_comp_enum_array_t
+		);
+	CREATE TABLE public.tst_comp_enum_what (
+		a public.tst_comp_enum_array_t PRIMARY KEY,
+		b public.tst_comp_enum_array_t[]
+		);
+
+	CREATE TYPE public.tst_comp_mix_t AS (
+		a public.tst_comp_basic_t,
+		b public.tst_comp_basic_t[],
+		c public.tst_enum_t,
+		d public.tst_enum_t[]
+		);
+	CREATE TABLE public.tst_comp_mix_array (
+		a public.tst_comp_mix_t PRIMARY KEY,
+		b public.tst_comp_mix_t[]
+		);
+	CREATE TABLE public.tst_range (
+		a INTEGER PRIMARY KEY,
+		b int4range
+	);
+	CREATE TABLE public.tst_range_array (
+		a INTEGER PRIMARY KEY,
+		b TSTZRANGE,
+		c int8range[]
+	););
+
+# Setup structure on both nodes
+$node_provider->safe_psql('postgres', $ddl);
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Setup logical replication
+my $provider_connstr = $node_provider->connstr . ' dbname=postgres';
+$node_provider->safe_psql('postgres',
+	"CREATE PUBLICATION tap_pub");
+$node_provider->safe_psql('postgres',
+	"ALTER PUBLICATION tap_pub ADD TABLE ALL IN SCHEMA public");
+
+$node_subscriber->safe_psql('postgres',
+	"CREATE SUBSCRIPTION tap_sub WITH CONNECTION '$provider_connstr' PUBLICATION tap_pub");
+
+# Wait for subscriber to finish table sync
+my $appname = 'tap_sub';
+my $caughtup_query =
+"SELECT pg_current_xlog_location() <= write_location FROM pg_stat_replication WHERE application_name = '$appname';";
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+# Insert initial test data
+$node_provider->safe_psql('postgres', qq(
+	-- test_tbl_one_array_col
+	INSERT INTO tst_one_array (a, b) VALUES
+		(1, '{1, 2, 3}'),
+		(2, '{2, 3, 1}'),
+		(3, '{3, 2, 1}'),
+		(4, '{4, 3, 2}'),
+		(5, '{5, NULL, 3}');
+
+	-- test_tbl_arrays
+	INSERT INTO tst_arrays (a, b, c, d) VALUES
+		('{1, 2, 3}', '{"a", "b", "c"}', '{1.1, 2.2, 3.3}', '{"1 day", "2 days", "3 days"}'),
+		('{2, 3, 1}', '{"b", "c", "a"}', '{2.2, 3.3, 1.1}', '{"2 minutes", "3 minutes", "1 minute"}'),
+		('{3, 1, 2}', '{"c", "a", "b"}', '{3.3, 1.1, 2.2}', '{"3 years", "1 year", "2 years"}'),
+		('{4, 1, 2}', '{"d", "a", "b"}', '{4.4, 1.1, 2.2}', '{"4 years", "1 year", "2 years"}'),
+		('{5, NULL, NULL}', '{"e", NULL, "b"}', '{5.5, 1.1, NULL}', '{"5 years", NULL, NULL}');
+
+	-- test_tbl_single_enum
+	INSERT INTO tst_one_enum (a, b) VALUES
+		(1, 'a'),
+		(2, 'b'),
+		(3, 'c'),
+		(4, 'd'),
+		(5, NULL);
+
+	-- test_tbl_enums
+	INSERT INTO tst_enums (a, b) VALUES
+		('a', '{b, c}'),
+		('b', '{c, a}'),
+		('c', '{b, a}'),
+		('d', '{c, b}'),
+		('e', '{d, NULL}');
+
+	-- test_tbl_single_composites
+	INSERT INTO tst_one_comp (a, b) VALUES
+		(1, ROW(1.0, 'a', 1)),
+		(2, ROW(2.0, 'b', 2)),
+		(3, ROW(3.0, 'c', 3)),
+		(4, ROW(4.0, 'd', 4)),
+		(5, ROW(NULL, NULL, 5));
+
+	-- test_tbl_composites
+	INSERT INTO tst_comps (a, b) VALUES
+		(ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_basic_t]),
+		(ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_basic_t]),
+		(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_basic_t]),
+		(ROW(4.0, 'd', 4), ARRAY[ROW(4, 'd', 3)::tst_comp_basic_t]),
+		(ROW(5.0, 'e', NULL), ARRAY[NULL, ROW(5, NULL, 5)::tst_comp_basic_t]);
+
+	-- test_tbl_composite_with_enums
+	INSERT INTO tst_comp_enum (a, b) VALUES
+		(1, ROW(1.0, 'a', 1)),
+		(2, ROW(2.0, 'b', 2)),
+		(3, ROW(3.0, 'c', 3)),
+		(4, ROW(4.0, 'd', 4)),
+		(5, ROW(NULL, 'e', NULL));
+
+	-- test_tbl_composite_with_enums_array
+	INSERT INTO tst_comp_enum_array (a, b) VALUES
+		(ROW(1.0, 'a', 1), ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t]),
+		(ROW(2.0, 'b', 2), ARRAY[ROW(2, 'b', 2)::tst_comp_enum_t]),
+		(ROW(3.0, 'c', 3), ARRAY[ROW(3, 'c', 3)::tst_comp_enum_t]),
+		(ROW(4.0, 'd', 3), ARRAY[ROW(3, 'd', 3)::tst_comp_enum_t]),
+		(ROW(5.0, 'e', 3), ARRAY[ROW(3, 'e', 3)::tst_comp_enum_t, NULL]);
+
+	-- test_tbl_composite_with_single_enums_array_in_composite
+	INSERT INTO tst_comp_one_enum_array (a, b) VALUES
+		(1, ROW(1.0, '{a, b, c}', 1)),
+		(2, ROW(2.0, '{a, b, c}', 2)),
+		(3, ROW(3.0, '{a, b, c}', 3)),
+		(4, ROW(4.0, '{c, b, d}', 4)),
+		(5, ROW(5.0, '{NULL, e, NULL}', 5));
+
+	-- test_tbl_composite_with_enums_array_in_composite
+	INSERT INTO tst_comp_enum_what (a, b) VALUES
+		(ROW(1.0, '{a, b, c}', 1), ARRAY[ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t]),
+		(ROW(2.0, '{b, c, a}', 2), ARRAY[ROW(2, '{b, c, a}', 1)::tst_comp_enum_array_t]),
+		(ROW(3.0, '{c, a, b}', 1), ARRAY[ROW(3, '{c, a, b}', 1)::tst_comp_enum_array_t]),
+		(ROW(4.0, '{c, b, d}', 4), ARRAY[ROW(4, '{c, b, d}', 4)::tst_comp_enum_array_t]),
+		(ROW(5.0, '{c, NULL, b}', NULL), ARRAY[ROW(5, '{c, e, b}', 1)::tst_comp_enum_array_t]);
+
+	-- test_tbl_mixed_composites
+	INSERT INTO tst_comp_mix_array (a, b) VALUES
+		(ROW(
+			ROW(1,'a',1),
+			ARRAY[ROW(1,'a',1)::tst_comp_basic_t, ROW(2,'b',2)::tst_comp_basic_t],
+			'a',
+			'{a,b,NULL,c}'),
+		ARRAY[
+			ROW(
+				ROW(1,'a',1),
+				ARRAY[
+					ROW(1,'a',1)::tst_comp_basic_t,
+					ROW(2,'b',2)::tst_comp_basic_t,
+					NULL
+					],
+				'a',
+				'{a,b,c}'
+				)::tst_comp_mix_t
+			]
+		);
+
+	-- test_tbl_range
+	INSERT INTO tst_range (a, b) VALUES
+		(1, '[1, 10]'),
+		(2, '[2, 20]'),
+		(3, '[3, 30]'),
+		(4, '[4, 40]'),
+		(5, '[5, 50]');
+
+	-- test_tbl_range_array
+	INSERT INTO tst_range_array (a, b, c) VALUES
+		(1, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), '{"[1,2]", "[10,20]"}'),
+		(2, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz - interval '2 days', 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[2,3]", "[20,30]"}'),
+		(3, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz - interval '3 days', 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[3,4]"}'),
+		(4, tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz - interval '4 days', 'Mon Aug 04 00:00:00 2014 CEST'::timestamptz), '{"[4,5]", NULL, "[40,50]"}'),
+		(5, NULL, NULL);
+));
+
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+# Check the data on subscriber
+my $result = $node_subscriber->safe_psql('postgres', qq(
+	SELECT a, b FROM tst_one_array ORDER BY a;
+	SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+	SELECT a, b FROM tst_one_enum ORDER BY a;
+	SELECT a, b FROM tst_enums ORDER BY a;
+	SELECT a, b FROM tst_one_comp ORDER BY a;
+	SELECT a, b FROM tst_comps ORDER BY a;
+	SELECT a, b FROM tst_comp_enum ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+	SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+	SELECT a, b FROM tst_range ORDER BY a;
+	SELECT a, b, c FROM tst_range_array ORDER BY a;
+));
+
+is($result, '1|{1,2,3}
+2|{2,3,1}
+3|{3,2,1}
+4|{4,3,2}
+5|{5,NULL,3}
+{1,2,3}|{a,b,c}|{1.1,2.2,3.3}|{"1 day","2 days","3 days"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
+{4,1,2}|{d,a,b}|{4.4,1.1,2.2}|{"4 years","1 year","2 years"}
+{5,NULL,NULL}|{e,NULL,b}|{5.5,1.1,NULL}|{"5 years",NULL,NULL}
+1|a
+2|b
+3|c
+4|d
+5|
+a|{b,c}
+b|{c,a}
+c|{b,a}
+d|{c,b}
+e|{d,NULL}
+1|(1,a,1)
+2|(2,b,2)
+3|(3,c,3)
+4|(4,d,4)
+5|(,,5)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+(3,c,3)|{"(3,c,3)"}
+(4,d,4)|{"(4,d,3)"}
+(5,e,)|{NULL,"(5,,5)"}
+1|(1,a,1)
+2|(2,b,2)
+3|(3,c,3)
+4|(4,d,4)
+5|(,e,)
+(1,a,1)|{"(1,a,1)"}
+(2,b,2)|{"(2,b,2)"}
+(3,c,3)|{"(3,c,3)"}
+(4,d,3)|{"(3,d,3)"}
+(5,e,3)|{"(3,e,3)",NULL}
+1|(1,"{a,b,c}",1)
+2|(2,"{a,b,c}",2)
+3|(3,"{a,b,c}",3)
+4|(4,"{c,b,d}",4)
+5|(5,"{NULL,e,NULL}",5)
+(1,"{a,b,c}",1)|{"(1,\"{a,b,c}\",1)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+(3,"{c,a,b}",1)|{"(3,\"{c,a,b}\",1)"}
+(4,"{c,b,d}",4)|{"(4,\"{c,b,d}\",4)"}
+(5,"{c,NULL,b}",)|{"(5,\"{c,e,b}\",1)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")"}
+1|[1,11)
+2|[2,21)
+3|[3,31)
+4|[4,41)
+5|[5,51)
+1|["2014-08-04 00:00:00+02",infinity)|{"[1,3)","[10,21)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
+4|["2014-07-31 00:00:00+02","2014-08-04 00:00:00+02")|{"[4,6)",NULL,"[40,51)"}
+5||',
+'check replicated inserts on subscriber');
+
+# Run batch of updates
+$node_provider->safe_psql('postgres', qq(
+	UPDATE tst_one_array SET b = '{4, 5, 6}' WHERE a = 1;
+	UPDATE tst_one_array SET b = '{4, 5, 6, 1}' WHERE a > 3;
+	UPDATE tst_arrays SET b = '{"1a", "2b", "3c"}', c = '{1.0, 2.0, 3.0}', d = '{"1 day 1 second", "2 days 2 seconds", "3 days 3 second"}' WHERE a = '{1, 2, 3}';
+	UPDATE tst_arrays SET b = '{"c", "d", "e"}', c = '{3.0, 4.0, 5.0}', d = '{"3 day 1 second", "4 days 2 seconds", "5 days 3 second"}' WHERE a[1] > 3;
+	UPDATE tst_one_enum SET b = 'c' WHERE a = 1;
+	UPDATE tst_one_enum SET b = NULL WHERE a > 3;
+	UPDATE tst_enums SET b = '{e, NULL}' WHERE a = 'a';
+	UPDATE tst_enums SET b = '{e, d}' WHERE a > 'c';
+	UPDATE tst_one_comp SET b = ROW(1.0, 'A', 1) WHERE a = 1;
+	UPDATE tst_one_comp SET b = ROW(NULL, 'x', -1) WHERE a > 3;
+	UPDATE tst_comps SET b = ARRAY[ROW(9, 'x', -1)::tst_comp_basic_t] WHERE (a).a = 1.0;
+	UPDATE tst_comps SET b = ARRAY[NULL, ROW(9, 'x', NULL)::tst_comp_basic_t] WHERE (a).a > 3.9;
+	UPDATE tst_comp_enum SET b = ROW(1.0, NULL, NULL) WHERE a = 1;
+	UPDATE tst_comp_enum SET b = ROW(4.0, 'd', 44) WHERE a > 3;
+	UPDATE tst_comp_enum_array SET b = ARRAY[NULL, ROW(3, 'd', 3)::tst_comp_enum_t] WHERE a = ROW(1.0, 'a', 1)::tst_comp_enum_t;
+	UPDATE tst_comp_enum_array SET b = ARRAY[ROW(1, 'a', 1)::tst_comp_enum_t, ROW(2, 'b', 2)::tst_comp_enum_t] WHERE (a).a > 3;
+	UPDATE tst_comp_one_enum_array SET b = ROW(1.0, '{a, e, c}', NULL) WHERE a = 1;
+	UPDATE tst_comp_one_enum_array SET b = ROW(4.0, '{c, b, d}', 4) WHERE a > 3;
+	UPDATE tst_comp_enum_what SET b = ARRAY[NULL, ROW(1, '{a, b, c}', 1)::tst_comp_enum_array_t, ROW(NULL, '{a, e, c}', 2)::tst_comp_enum_array_t] WHERE (a).a = 1;
+	UPDATE tst_comp_enum_what SET b = ARRAY[ROW(5, '{a, b, c}', 5)::tst_comp_enum_array_t] WHERE (a).a > 3;
+	UPDATE tst_comp_mix_array SET b[2] = NULL WHERE ((a).a).a = 1;
+	UPDATE tst_range SET b = '[100, 1000]' WHERE a = 1;
+	UPDATE tst_range SET b = '(1, 90)' WHERE a > 3;
+	UPDATE tst_range_array SET c = '{"[100, 1000]"}' WHERE a = 1;
+	UPDATE tst_range_array SET b = tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'infinity'), c = '{NULL, "[11,9999999]"}' WHERE a > 3;
+));
+
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql('postgres', qq(
+	SELECT a, b FROM tst_one_array ORDER BY a;
+	SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+	SELECT a, b FROM tst_one_enum ORDER BY a;
+	SELECT a, b FROM tst_enums ORDER BY a;
+	SELECT a, b FROM tst_one_comp ORDER BY a;
+	SELECT a, b FROM tst_comps ORDER BY a;
+	SELECT a, b FROM tst_comp_enum ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+	SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+	SELECT a, b FROM tst_range ORDER BY a;
+	SELECT a, b, c FROM tst_range_array ORDER BY a;
+));
+
+is($result, '1|{4,5,6}
+2|{2,3,1}
+3|{3,2,1}
+4|{4,5,6,1}
+5|{4,5,6,1}
+{1,2,3}|{1a,2b,3c}|{1,2,3}|{"1 day 00:00:01","2 days 00:00:02","3 days 00:00:03"}
+{2,3,1}|{b,c,a}|{2.2,3.3,1.1}|{00:02:00,00:03:00,00:01:00}
+{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
+{4,1,2}|{c,d,e}|{3,4,5}|{"3 days 00:00:01","4 days 00:00:02","5 days 00:00:03"}
+{5,NULL,NULL}|{c,d,e}|{3,4,5}|{"3 days 00:00:01","4 days 00:00:02","5 days 00:00:03"}
+1|c
+2|b
+3|c
+4|
+5|
+a|{e,NULL}
+b|{c,a}
+c|{b,a}
+d|{e,d}
+e|{e,d}
+1|(1,A,1)
+2|(2,b,2)
+3|(3,c,3)
+4|(,x,-1)
+5|(,x,-1)
+(1,a,1)|{"(9,x,-1)"}
+(2,b,2)|{"(2,b,2)"}
+(3,c,3)|{"(3,c,3)"}
+(4,d,4)|{NULL,"(9,x,)"}
+(5,e,)|{NULL,"(9,x,)"}
+1|(1,,)
+2|(2,b,2)
+3|(3,c,3)
+4|(4,d,44)
+5|(4,d,44)
+(1,a,1)|{NULL,"(3,d,3)"}
+(2,b,2)|{"(2,b,2)"}
+(3,c,3)|{"(3,c,3)"}
+(4,d,3)|{"(1,a,1)","(2,b,2)"}
+(5,e,3)|{"(1,a,1)","(2,b,2)"}
+1|(1,"{a,e,c}",)
+2|(2,"{a,b,c}",2)
+3|(3,"{a,b,c}",3)
+4|(4,"{c,b,d}",4)
+5|(4,"{c,b,d}",4)
+(1,"{a,b,c}",1)|{NULL,"(1,\"{a,b,c}\",1)","(,\"{a,e,c}\",2)"}
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+(3,"{c,a,b}",1)|{"(3,\"{c,a,b}\",1)"}
+(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
+(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+("(1,a,1)","{""(1,a,1)"",""(2,b,2)""}",a,"{a,b,NULL,c}")|{"(\"(1,a,1)\",\"{\"\"(1,a,1)\"\",\"\"(2,b,2)\"\",NULL}\",a,\"{a,b,c}\")",NULL}
+1|[100,1001)
+2|[2,21)
+3|[3,31)
+4|[2,90)
+5|[2,90)
+1|["2014-08-04 00:00:00+02",infinity)|{"[100,1001)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}
+4|["2014-08-04 00:00:00+02",infinity)|{NULL,"[11,10000000)"}
+5|["2014-08-04 00:00:00+02",infinity)|{NULL,"[11,10000000)"}',
+'check replicated updates on subscriber');
+
+# Run batch of deletes
+$node_provider->safe_psql('postgres', qq(
+	DELETE FROM tst_one_array WHERE a = 1;
+	DELETE FROM tst_one_array WHERE b = '{2, 3, 1}';
+	DELETE FROM tst_arrays WHERE a = '{1, 2, 3}';
+	DELETE FROM tst_arrays WHERE a[1] = 2;
+	DELETE FROM tst_one_enum WHERE a = 1;
+	DELETE FROM tst_one_enum WHERE b = 'b';
+	DELETE FROM tst_enums WHERE a = 'a';
+	DELETE FROM tst_enums WHERE b[1] = 'b';
+	DELETE FROM tst_one_comp WHERE a = 1;
+	DELETE FROM tst_one_comp WHERE (b).a = 2.0;
+	DELETE FROM tst_comps WHERE (a).b = 'a';
+	DELETE FROM tst_comps WHERE ROW(3, 'c', 3)::tst_comp_basic_t = ANY(b);
+	DELETE FROM tst_comp_enum WHERE a = 1;
+	DELETE FROM tst_comp_enum WHERE (b).a = 2.0;
+	DELETE FROM tst_comp_enum_array WHERE a = ROW(1.0, 'a', 1)::tst_comp_enum_t;
+	DELETE FROM tst_comp_enum_array WHERE ROW(3, 'c', 3)::tst_comp_enum_t = ANY(b);
+	DELETE FROM tst_comp_one_enum_array WHERE a = 1;
+	DELETE FROM tst_comp_one_enum_array WHERE 'a' = ANY((b).b);
+	DELETE FROM tst_comp_enum_what WHERE (a).a = 1;
+	DELETE FROM tst_comp_enum_what WHERE (b[1]).b = '{c, a, b}';
+	DELETE FROM tst_comp_mix_array WHERE ((a).a).a = 1;
+	DELETE FROM tst_range WHERE a = 1;
+	DELETE FROM tst_range WHERE '[10,20]' && b;
+	DELETE FROM tst_range_array WHERE a = 1;
+	DELETE FROM tst_range_array WHERE tstzrange('Mon Aug 04 00:00:00 2014 CEST'::timestamptz, 'Mon Aug 05 00:00:00 2014 CEST'::timestamptz) && b;
+));
+
+$node_provider->poll_query_until('postgres', $caughtup_query)
+  or die "Timed out while waiting for subscriber to catch up";
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql('postgres', qq(
+	SELECT a, b FROM tst_one_array ORDER BY a;
+	SELECT a, b, c, d FROM tst_arrays ORDER BY a;
+	SELECT a, b FROM tst_one_enum ORDER BY a;
+	SELECT a, b FROM tst_enums ORDER BY a;
+	SELECT a, b FROM tst_one_comp ORDER BY a;
+	SELECT a, b FROM tst_comps ORDER BY a;
+	SELECT a, b FROM tst_comp_enum ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_one_enum_array ORDER BY a;
+	SELECT a, b FROM tst_comp_enum_what ORDER BY a;
+	SELECT a, b FROM tst_comp_mix_array ORDER BY a;
+	SELECT a, b FROM tst_range ORDER BY a;
+	SELECT a, b, c FROM tst_range_array ORDER BY a;
+));
+
+is($result, '3|{3,2,1}
+4|{4,5,6,1}
+5|{4,5,6,1}
+{3,1,2}|{c,a,b}|{3.3,1.1,2.2}|{"3 years","1 year","2 years"}
+{4,1,2}|{c,d,e}|{3,4,5}|{"3 days 00:00:01","4 days 00:00:02","5 days 00:00:03"}
+{5,NULL,NULL}|{c,d,e}|{3,4,5}|{"3 days 00:00:01","4 days 00:00:02","5 days 00:00:03"}
+3|c
+4|
+5|
+b|{c,a}
+d|{e,d}
+e|{e,d}
+3|(3,c,3)
+4|(,x,-1)
+5|(,x,-1)
+(2,b,2)|{"(2,b,2)"}
+(4,d,4)|{NULL,"(9,x,)"}
+(5,e,)|{NULL,"(9,x,)"}
+3|(3,c,3)
+4|(4,d,44)
+5|(4,d,44)
+(2,b,2)|{"(2,b,2)"}
+(4,d,3)|{"(1,a,1)","(2,b,2)"}
+(5,e,3)|{"(1,a,1)","(2,b,2)"}
+4|(4,"{c,b,d}",4)
+5|(4,"{c,b,d}",4)
+(2,"{b,c,a}",2)|{"(2,\"{b,c,a}\",1)"}
+(4,"{c,b,d}",4)|{"(5,\"{a,b,c}\",5)"}
+(5,"{c,NULL,b}",)|{"(5,\"{a,b,c}\",5)"}
+2|["2014-08-02 00:00:00+02","2014-08-04 00:00:00+02")|{"[2,4)","[20,31)"}
+3|["2014-08-01 00:00:00+02","2014-08-04 00:00:00+02")|{"[3,5)"}',
+'check replicated deletes on subscriber');
+
+$node_subscriber->stop('fast');
+$node_provider->stop('fast');
-- 
2.7.4

