From 2b2bc8c5d670c765ed3eb771dee266f6b9b3f3c2 Mon Sep 17 00:00:00 2001
From: Petr Jelinek <pjmodos@pjmodos.net>
Date: Tue, 19 Jul 2016 01:55:25 +0200
Subject: [PATCH 6/6] Logical replication support for initial data copy

---
 src/backend/catalog/Makefile                       |   2 +-
 src/backend/commands/subscriptioncmds.c            |  34 +-
 .../libpqwalreceiver/libpqwalreceiver.c            | 126 +++-
 src/backend/replication/logical/Makefile           |   2 +-
 src/backend/replication/logical/apply.c            | 173 +++++-
 src/backend/replication/logical/launcher.c         |  92 +--
 src/backend/replication/logical/logical.c          | 190 +++++-
 src/backend/replication/logical/proto.c            |  63 +-
 src/backend/replication/logical/publication.c      |  40 ++
 src/backend/replication/logical/snapbuild.c        |   5 +-
 src/backend/replication/logical/subscription.c     | 180 ++++++
 src/backend/replication/logical/tablesync.c        | 672 +++++++++++++++++++++
 src/backend/replication/pgoutput/pgoutput.c        | 108 ++++
 src/backend/replication/repl_gram.y                |  31 +-
 src/backend/replication/repl_scanner.l             |   3 +
 src/backend/replication/walsender.c                | 199 +++++-
 src/backend/utils/cache/syscache.c                 |  23 +
 src/include/catalog/indexing.h                     |   6 +
 src/include/catalog/pg_subscription_rel.h          |  61 ++
 src/include/commands/replicationcmds.h             |   1 +
 src/include/nodes/nodes.h                          |   4 +-
 src/include/nodes/replnodes.h                      |  23 +
 src/include/replication/logical.h                  |  22 +-
 src/include/replication/logicalproto.h             |   3 +-
 src/include/replication/logicalworker.h            |   9 +-
 src/include/replication/output_plugin.h            |  18 +
 src/include/replication/publication.h              |   1 +
 src/include/replication/subscription.h             |   6 +
 src/include/replication/walreceiver.h              |   9 +
 src/include/replication/worker_internal.h          |  32 +
 src/include/utils/syscache.h                       |   2 +
 src/test/Makefile                                  |   2 +-
 src/test/README                                    |   3 +
 src/test/regress/expected/sanity_check.out         |   1 +
 src/test/subscription/t/001_rep_changes.pl         |  18 +-
 src/test/subscription/t/002_types.pl               |   8 +-
 36 files changed, 2043 insertions(+), 129 deletions(-)
 create mode 100644 src/backend/replication/logical/tablesync.c
 create mode 100644 src/include/catalog/pg_subscription_rel.h
 create mode 100644 src/include/replication/worker_internal.h

diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 60737d4..22f78f8 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -43,7 +43,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_default_acl.h pg_init_privs.h pg_seclabel.h pg_shseclabel.h \
 	pg_collation.h pg_range.h pg_transform.h \
 	pg_publication.h pg_publication_rel.h pg_subscription.h \
-	toasting.h indexing.h \
+	pg_subscription_rel.h toasting.h indexing.h \
     )
 
 # location of Catalog.pm
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 43e2853..bfef492 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -22,11 +22,13 @@
 #include "access/htup_details.h"
 #include "access/xact.h"
 
+#include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/objectaddress.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_rel.h"
 
 #include "commands/defrem.h"
 #include "commands/replicationcmds.h"
@@ -40,9 +42,12 @@
 #include "replication/logicalworker.h"
 #include "replication/origin.h"
 #include "replication/reorderbuffer.h"
-#include "replication/logicalworker.h"
+#include "replication/subscription.h"
 #include "replication/walreceiver.h"
 
+#include "storage/ipc.h"
+#include "storage/proc.h"
+
 #include "utils/array.h"
 #include "utils/builtins.h"
 #include "utils/catcache.h"
@@ -172,10 +177,13 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 	char	   *conninfo;
 	char	   *slotname;
 	List	   *publications;
+	char	   *options;
 	WalReceiverConnHandle  *wrchandle = NULL;
 	WalReceiverConnAPI	   *wrcapi = NULL;
 	walrcvconn_init_fn		walrcvconn_init;
 	XLogRecPtr	lsn;
+	List	   *tables;
+	ListCell   *lc;
 
 	check_replication_permissions();
 
@@ -248,6 +256,12 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 		wrcapi->create_slot == NULL)
 		elog(ERROR, "libpqwalreceiver didn't initialize correctly");
 
+	/*
+	 * Create the replication slot on remote side for our newly created
+	 * subscription.
+	 *
+	 * TODO: ensure drop of the slot on subsequent failure/rollback?
+	 */
 	wrcapi->connect(wrchandle, conninfo, true, stmt->subname);
 	wrcapi->create_slot(wrchandle, slotname, true, &lsn);
 	ereport(NOTICE,
@@ -259,6 +273,24 @@ CreateSubscription(CreateSubscriptionStmt *stmt)
 	 */
 	replorigin_create(slotname);
 
+	/* Build option string for the plugin. */
+	options = logicalrep_build_options(publications);
+
+	/* Get the table list from provider and build local table status info. */
+	tables = wrcapi->list_tables(wrchandle, slotname, options);
+	foreach (lc, tables)
+	{
+		LogicalRepTableListEntry *entry = lfirst(lc);
+		Oid		nspid = LookupExplicitNamespace(entry->nspname, false);
+		Oid		relid = get_relname_relid(entry->relname, nspid);
+
+		SetSubscriptionRelState(subid, relid, SUBREL_STATE_INIT,
+								InvalidXLogRecPtr);
+	}
+
+	ereport(NOTICE,
+			(errmsg("synchronized table states")));
+
 	/* And we are done with the remote side. */
 	wrcapi->disconnect(wrchandle);
 
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 4c4d441..94648c7 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -23,6 +23,7 @@
 #include "pqexpbuffer.h"
 #include "access/xlog.h"
 #include "miscadmin.h"
+#include "replication/logical.h"
 #include "replication/walreceiver.h"
 #include "utils/builtins.h"
 #include "utils/pg_lsn.h"
@@ -73,6 +74,11 @@ static int	libpqrcv_receive(WalReceiverConnHandle *handle, char **buffer,
 							 pgsocket *wait_fd);
 static void libpqrcv_send(WalReceiverConnHandle *handle, const char *buffer,
 						  int nbytes);
+static List *libpqrcv_list_tables(WalReceiverConnHandle *handle,
+								 char *slotname, char *options);
+static bool libpqrcv_copy_table(WalReceiverConnHandle *handle,
+								char *slotname, char *nspname,
+								char *relname, char *options);
 static void libpqrcv_disconnect(WalReceiverConnHandle *handle);
 
 /* Prototypes for private functions */
@@ -104,6 +110,8 @@ _PG_walreceirver_conn_init(WalReceiverConnAPI *wrcapi)
 	wrcapi->receive = libpqrcv_receive;
 	wrcapi->send = libpqrcv_send;
 	wrcapi->disconnect = libpqrcv_disconnect;
+	wrcapi->copy_table = libpqrcv_copy_table;
+	wrcapi->list_tables = libpqrcv_list_tables;
 
 	return handle;
 }
@@ -416,15 +424,15 @@ libpqrcv_endstreaming(WalReceiverConnHandle *handle, TimeLineID *next_tli)
 			(errmsg("could not send end-of-streaming message to primary: %s",
 					PQerrorMessage(handle->streamConn))));
 
+	*next_tli = 0;
+
 	/*
 	 * After COPY is finished, we should receive a result set indicating the
 	 * next timeline's ID, or just CommandComplete if the server was shut
 	 * down.
 	 *
-	 * If we had not yet received CopyDone from the backend, PGRES_COPY_IN
-	 * would also be possible. However, at the moment this function is only
-	 * called after receiving CopyDone from the backend - the walreceiver
-	 * never terminates replication on its own initiative.
+	 * If we had not yet received CopyDone from the backend, PGRES_COPY_OUT
+	 * is also possible in case we aborted the copy in mid-stream.
 	 */
 	res = PQgetResult(handle->streamConn);
 	if (PQresultStatus(res) == PGRES_TUPLES_OK)
@@ -442,8 +450,16 @@ libpqrcv_endstreaming(WalReceiverConnHandle *handle, TimeLineID *next_tli)
 		/* the result set should be followed by CommandComplete */
 		res = PQgetResult(handle->streamConn);
 	}
-	else
-		*next_tli = 0;
+	else if (PQresultStatus(res) == PGRES_COPY_OUT)
+	{
+		PQclear(res);
+
+		/* End the copy */
+		PQendcopy(handle->streamConn);
+
+		/* CommandComplete should follow */
+		res = PQgetResult(handle->streamConn);
+	}
 
 	if (PQresultStatus(res) != PGRES_COMMAND_OK)
 		ereport(ERROR,
@@ -642,6 +658,104 @@ libpqrcv_PQexec(WalReceiverConnHandle *handle, const char *query)
 }
 
 /*
+ * Run the LIST_TABLES command which will send list of the tables to copy
+ * in whatever format the plugin choses.
+ */
+static List *
+libpqrcv_list_tables(WalReceiverConnHandle *handle, char *slotname,
+					 char *options)
+{
+	StringInfoData	cmd;
+	PGresult	   *res;
+	int				i;
+	List		   *tablelist = NIL;
+
+	initStringInfo(&cmd);
+	appendStringInfo(&cmd, "LIST_TABLES SLOT \"%s\"",
+					 slotname);
+
+	/* Add options */
+	if (options)
+		appendStringInfo(&cmd, "( %s )", options);
+
+	res = libpqrcv_PQexec(handle, cmd.data);
+	pfree(cmd.data);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("could not receive list of replicated tables from the provider: %s",
+						PQerrorMessage(handle->streamConn))));
+	}
+	if (PQnfields(res) != 3)
+	{
+		int nfields = PQnfields(res);
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("invalid response from provider"),
+				 errdetail("Expected 3 fields, got %d fields.", nfields)));
+	}
+
+	for (i = 0; i < PQntuples(res); i++)
+	{
+		LogicalRepTableListEntry *entry;
+
+		entry = palloc(sizeof(LogicalRepTableListEntry));
+		entry->nspname = pstrdup(PQgetvalue(res, i, 0));
+		entry->relname = pstrdup(PQgetvalue(res, i, 1));
+		if (!PQgetisnull(res, i, 2))
+			entry->info = pstrdup(PQgetvalue(res, i, 2));
+		else
+			entry->info = NULL;
+
+		tablelist = lappend(tablelist, entry);
+	}
+
+	PQclear(res);
+
+	return tablelist;
+}
+
+/*
+ * Run the COPY_TABLE command which will start streaming the existing data
+ * in the table.
+ */
+static bool
+libpqrcv_copy_table(WalReceiverConnHandle *handle, char *slotname,
+					char *nspname, char *relname, char *options)
+{
+	StringInfoData	cmd;
+	PGresult	   *res;
+
+	initStringInfo(&cmd);
+	appendStringInfo(&cmd, "COPY_TABLE SLOT \"%s\" TABLE \"%s\" \"%s\"",
+					 slotname, nspname, relname);
+
+	/* Add options */
+	if (options)
+		appendStringInfo(&cmd, "( %s )", options);
+
+	res = libpqrcv_PQexec(handle, cmd.data);
+
+	if (PQresultStatus(res) == PGRES_COMMAND_OK)
+	{
+		PQclear(res);
+		return false;
+	}
+	else if (PQresultStatus(res) != PGRES_COPY_BOTH)
+	{
+		PQclear(res);
+		ereport(ERROR,
+				(errmsg("could not start initial table contents streaming: %s",
+						PQerrorMessage(handle->streamConn))));
+	}
+	PQclear(res);
+	pfree(cmd.data);
+	return true;
+}
+
+/*
  * Disconnect connection to primary, if any.
  */
 static void
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index ab6e11e..a1e1d81 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -16,6 +16,6 @@ override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
 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
+	   subscription.o tablesync.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/apply.c b/src/backend/replication/logical/apply.c
index eb7af19..809b77b 100644
--- a/src/backend/replication/logical/apply.c
+++ b/src/backend/replication/logical/apply.c
@@ -24,6 +24,7 @@
 #include "access/xlog_internal.h"
 
 #include "catalog/namespace.h"
+#include "catalog/pg_subscription_rel.h"
 
 #include "commands/trigger.h"
 
@@ -51,6 +52,7 @@
 #include "replication/snapbuild.h"
 #include "replication/subscription.h"
 #include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
 
 #include "rewrite/rewriteHandler.h"
 
@@ -62,6 +64,7 @@
 
 #include "utils/builtins.h"
 #include "utils/catcache.h"
+#include "utils/fmgroids.h"
 #include "utils/guc.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
@@ -72,17 +75,27 @@
 
 typedef struct FlushPosition
 {
-	dlist_node node;
-	XLogRecPtr local_end;
-	XLogRecPtr remote_end;
+	dlist_node	node;
+	XLogRecPtr	local_end;
+	XLogRecPtr	remote_end;
 } FlushPosition;
 
 static dlist_head lsn_mapping = DLIST_STATIC_INIT(lsn_mapping);
 
+
+typedef struct TableState
+{
+	dlist_node	node;
+	Oid			relid;
+	XLogRecPtr	lsn;
+	char		state;
+} TableState;
+
+static dlist_head table_states = DLIST_STATIC_INIT(table_states);
+static XLogRecPtr		last_commit_lsn;
+
 static MemoryContext	ApplyContext;
-static bool				in_remote_transaction = false;
 
-static Subscription	   *MySubscription = NULL;
 static bool				got_SIGTERM = false;
 
 typedef struct LogicalRepRelMapEntry {
@@ -95,13 +108,19 @@ typedef struct LogicalRepRelMapEntry {
 										 * local ones */
 	AttInMetadata	   *attin;			/* cached info used in type
 										 * conversion */
+	char				state;
 } LogicalRepRelMapEntry;
 
 static HTAB *LogicalRepRelMap = NULL;
 
-/* filled by libpqreceiver when loaded */
-static WalReceiverConnAPI *wrcapi = NULL;
-static WalReceiverConnHandle *wrchandle = NULL;
+WalReceiverConnAPI	   *wrcapi = NULL;
+WalReceiverConnHandle  *wrchandle = NULL;
+
+LogicalRepWorker   *MyLogicalRepWorker = NULL;
+Subscription	   *MySubscription = NULL;
+
+static char	   *myslotname = NULL;
+bool			in_remote_transaction = false;
 
 static void send_feedback(XLogRecPtr recvpos, int64 now, bool force);
 void pglogical_apply_main(Datum main_arg);
@@ -109,6 +128,19 @@ void pglogical_apply_main(Datum main_arg);
 static bool tuple_find_by_replidx(Relation rel, LockTupleMode lockmode,
 					  TupleTableSlot *searchslot, TupleTableSlot *slot);
 
+/*
+ * Should this worker apply changes for given relation.
+ *
+ * This is mainly needed for initial relation data sync as that runs in
+ * parallel worker and we need some way to skip changes coming to the main
+ * apply worker during the sync of a table.
+ */
+static bool
+interesting_relation(LogicalRepRelMapEntry *rel)
+{
+	return rel->state == SUBREL_STATE_READY ||
+		rel->reloid == MyLogicalRepWorker->relid;
+}
 
 
 /*
@@ -266,12 +298,16 @@ tupdesc_get_att_by_name(TupleDesc desc, const char *attname)
 
 /*
  * Open the local relation associated with the remote one.
+ *
+ * Optionally rebuilds the Relcache mapping if it was invalidated
+ * by local DDL.
  */
 static LogicalRepRelMapEntry *
 logicalreprel_open(uint32 remoteid, LOCKMODE lockmode)
 {
 	LogicalRepRelMapEntry  *entry;
 	bool		found;
+	XLogRecPtr	lsn;
 
 	if (LogicalRepRelMap == NULL)
 		remoterelmap_init();
@@ -309,6 +345,10 @@ logicalreprel_open(uint32 remoteid, LOCKMODE lockmode)
 	else
 		entry->rel = heap_open(entry->reloid, lockmode);
 
+	/* TODO cache this */
+	entry->state = GetSubscriptionRelState(MySubscription->oid,
+										   entry->reloid, &lsn, true);
+
 	return entry;
 }
 
@@ -322,7 +362,6 @@ logicalreprel_close(LogicalRepRelMapEntry *rel, LOCKMODE lockmode)
 	rel->rel = NULL;
 }
 
-
 /*
  * Make sure that we started local transaction.
  *
@@ -599,6 +638,9 @@ handle_commit(StringInfo s)
 
 	in_remote_transaction = false;
 
+	last_commit_lsn = end_lsn;
+	process_syncing_tables(myslotname, end_lsn);
+
 	pgstat_report_activity(STATE_IDLE, NULL);
 }
 
@@ -651,6 +693,15 @@ handle_insert(StringInfo s)
 
 	relid = logicalrep_read_insert(s, &newtup);
 	rel = logicalreprel_open(relid, RowExclusiveLock);
+	if (!interesting_relation(rel))
+	{
+		/*
+		 * The relation can't become interestin in the middle of the
+		 * transaction so it's safe to unlock it.
+		 */
+		logicalreprel_close(rel, RowExclusiveLock);
+		return;
+	}
 
 	/* Initialize the executor state. */
 	estate = create_estate_for_relation(rel);
@@ -710,6 +761,15 @@ handle_update(StringInfo s)
 	relid = logicalrep_read_update(s, &hasoldtup, &oldtup,
 								   &newtup);
 	rel = logicalreprel_open(relid, RowExclusiveLock);
+	if (!interesting_relation(rel))
+	{
+		/*
+		 * The relation can't become interestin in the middle of the
+		 * transaction so it's safe to unlock it.
+		 */
+		logicalreprel_close(rel, RowExclusiveLock);
+		return;
+	}
 
 	/* Initialize the executor state. */
 	estate = create_estate_for_relation(rel);
@@ -796,6 +856,15 @@ handle_delete(StringInfo s)
 
 	relid = logicalrep_read_delete(s, &oldtup);
 	rel = logicalreprel_open(relid, RowExclusiveLock);
+	if (!interesting_relation(rel))
+	{
+		/*
+		 * The relation can't become interestin in the middle of the
+		 * transaction so it's safe to unlock it.
+		 */
+		logicalreprel_close(rel, RowExclusiveLock);
+		return;
+	}
 
 	/* Initialize the executor state. */
 	estate = create_estate_for_relation(rel);
@@ -942,11 +1011,9 @@ get_flush_position(XLogRecPtr *write, XLogRecPtr *flush)
 /*
  * Apply main loop.
  */
-static void
-ApplyLoop(void)
+void
+LogicalRepApplyLoop(XLogRecPtr last_received)
 {
-	XLogRecPtr	last_received = InvalidXLogRecPtr;
-
 	/* Init the ApplyContext which we use for easier cleanup. */
 	ApplyContext = AllocSetContextCreate(TopMemoryContext,
 										 "ApplyContext",
@@ -1029,6 +1096,9 @@ ApplyLoop(void)
 						/* timestamp = */ pq_getmsgint64(&s);
 						reply_requested = pq_getmsgbyte(&s);
 
+						if (last_received < endpos)
+							last_received = endpos;
+
 						send_feedback(endpos,
 									  GetCurrentTimestamp(),
 									  reply_requested);
@@ -1040,6 +1110,18 @@ ApplyLoop(void)
 			}
 		}
 
+		if (!in_remote_transaction)
+		{
+			/*
+			 * If we didn't get any transactions for a while there might be
+			 * unconsumer invalidation messages in the queue, consume them now.
+			 */
+			AcceptInvalidationMessages();
+
+			/* Process any table synchronization changes. */
+			process_syncing_tables(myslotname, last_received);
+		}
+
 		/* confirm all writes at once */
 		send_feedback(last_received, GetCurrentTimestamp(), false);
 
@@ -1049,7 +1131,11 @@ ApplyLoop(void)
 
 		/* Check if we need to exit the streaming loop. */
 		if (endofstream)
+		{
+			TimeLineID	tli;
+			wrcapi->endstreaming(wrchandle, &tli);
 			break;
+		}
 
 		/*
 		 * Wait for more data or latch.
@@ -1156,7 +1242,6 @@ ApplyWorkerMain(Datum main_arg)
 {
 	int				worker_slot = DatumGetObjectId(main_arg);
 	MemoryContext	oldctx;
-	RepOriginId		originid;
 	XLogRecPtr		origin_startpos;
 	char		   *options;
 	walrcvconn_init_fn walrcvconn_init;
@@ -1207,41 +1292,65 @@ ApplyWorkerMain(Datum main_arg)
 	BackgroundWorkerInitializeConnectionByOid(MyLogicalRepWorker->dbid,
 											  InvalidOid);
 
-	StartTransactionCommand();
-
 	/* Load the subscription into persistent memory context. */
+	StartTransactionCommand();
 	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);
-
+	if (OidIsValid(MyLogicalRepWorker->relid))
+		elog(LOG, "logical replication sync for subscription %s, table %s started",
+			 MySubscription->name, get_rel_name(MyLogicalRepWorker->relid));
+	else
+		elog(LOG, "logical replication apply for subscription %s started",
+			 MySubscription->name);
 	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. */
+	if (OidIsValid(MyLogicalRepWorker->relid))
+	{
+		/* This is table synchroniation worker, call initial sync. */
+		myslotname = LogicalRepSyncTableStart(&origin_startpos);
+	}
+	else
+	{
+		/* This is main apply worker */
+		RepOriginId		originid;
+
+		myslotname = MySubscription->slotname;
+
+		StartTransactionCommand();
+		originid = replorigin_by_name(myslotname, false);
+		replorigin_session_setup(originid);
+		replorigin_session_origin = originid;
+		CommitTransactionCommand();
+
+		wrcapi->connect(wrchandle, MySubscription->conninfo, true,
+						myslotname);
+	}
+
+	/*
+	 * Setup callback for syscache so that we know when something
+	 * changes in the subscription relation state.
+	 */
+	CacheRegisterSyscacheCallback(SUBSCRIPTIONRELOID,
+								  invalidate_syncing_table_states,
+								  (Datum) 0);
+
+	/* Start normal logical streaming replication. */
 	wrcapi->startstreaming_logical(wrchandle, origin_startpos,
-								   MySubscription->slotname, options);
+								   myslotname, options);
+
+	pfree(options);
 
 	/* Run the main loop. */
-	ApplyLoop();
+	LogicalRepApplyLoop(origin_startpos);
 
 	wrcapi->disconnect(wrchandle);
 
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 385260e..927af69 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -30,6 +30,7 @@
 
 #include "replication/logicalworker.h"
 #include "replication/subscription.h"
+#include "replication/worker_internal.h"
 
 #include "storage/ipc.h"
 #include "storage/proc.h"
@@ -44,7 +45,6 @@
 #include "utils/snapmgr.h"
 
 int	max_logical_replication_workers = 4;
-LogicalRepWorker *MyLogicalRepWorker = NULL;
 
 typedef struct LogicalRepCtxStruct
 {
@@ -57,8 +57,6 @@ typedef struct 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);
@@ -183,29 +181,50 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
 
 /*
  * Walks the workers array and searches for one that matches given
- * subscription id.
+ * subscription id and relid.
  */
-static LogicalRepWorker *
-logicalrep_worker_find(Oid subid)
+LogicalRepWorker *
+logicalrep_worker_find(Oid subid, Oid relid)
 {
 	int	i;
 	LogicalRepWorker   *res = NULL;
 
-	/* Block concurrent modification. */
-	LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+	Assert(LWLockHeldByMe(LogicalRepLauncherLock));
 
 	/* 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))
+		if (w->subid == subid && w->relid == relid &&
+			w->proc && IsBackendPid(w->proc->pid))
 		{
 			res = w;
 			break;
 		}
 	}
 
-	LWLockRelease(LogicalRepLauncherLock);
+	return res;
+}
+
+/*
+ * Walks the workers array and searches for ones that matches given
+ * subscription id and counts them.
+ */
+int
+logicalrep_worker_count(Oid subid)
+{
+	int	i;
+	int	res = 0;
+
+	Assert(LWLockHeldByMe(LogicalRepLauncherLock));
+
+	/* 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++;
+	}
 
 	return res;
 }
@@ -213,17 +232,18 @@ logicalrep_worker_find(Oid subid)
 /*
  * Start new apply background worker.
  */
-static void
-logicalrep_worker_launch(Oid dbid, Oid subid)
+void
+logicalrep_worker_launch(Oid dbid, Oid subid, Oid relid)
 {
 	BackgroundWorker	bgw;
 	BackgroundWorkerHandle *bgw_handle;
 	int					slot;
 	LogicalRepWorker   *worker = NULL;
 
-	ereport(LOG,
-			(errmsg("starting logical replication worker for subscription %u",
-					subid)));
+	ereport(DEBUG1,
+			(errmsg("starting logical replication worker for "
+					"subscription %u, relation %u",
+					subid, relid)));
 
 	/*
 	 * We need to do the modification of the shared memory under lock so that
@@ -256,6 +276,7 @@ logicalrep_worker_launch(Oid dbid, Oid subid)
 	memset(worker, 0, sizeof(LogicalRepWorker));
 	worker->dbid = dbid;
 	worker->subid = subid;
+	worker->relid = relid;
 
 	LWLockRelease(LogicalRepLauncherLock);
 
@@ -265,6 +286,13 @@ logicalrep_worker_launch(Oid dbid, Oid subid)
 	bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
 	bgw.bgw_main = ApplyWorkerMain;
 
+	if (OidIsValid(relid))
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "logical replication worker %u sync %u", subid, relid);
+	else
+		snprintf(bgw.bgw_name, BGW_MAXLEN,
+				 "logical replication worker %u", subid);
+
 	bgw.bgw_restart_time = BGW_NEVER_RESTART;
 	bgw.bgw_notify_pid = MyProcPid;
 	bgw.bgw_main_arg = slot;
@@ -283,7 +311,7 @@ logicalrep_worker_launch(Oid dbid, Oid subid)
 	if (!WaitForReplicationWorkerAttach(worker, bgw_handle))
 	{
 		ereport(WARNING,
-				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 				 errmsg("could not launch logical replication worker")));
 		return;
 	}
@@ -296,7 +324,7 @@ logicalrep_worker_launch(Oid dbid, Oid subid)
 static void
 logicalrep_worker_stop(LogicalRepWorker *worker)
 {
-	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+	Assert(LWLockHeldByMe(LogicalRepLauncherLock));
 
 	/* Check that the worker is up and what we expect. */
 	if (!worker->proc)
@@ -306,27 +334,6 @@ logicalrep_worker_stop(LogicalRepWorker *worker)
 
 	/* 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;
-	}
 }
 
 /*
@@ -504,16 +511,21 @@ ApplyLauncherMain(Datum main_arg)
 		foreach(lc, sublist)
 		{
 			Subscription	   *sub = (Subscription *) lfirst(lc);
-			LogicalRepWorker   *w = logicalrep_worker_find(sub->oid);
+			LogicalRepWorker   *w;
+
+			LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+			w = logicalrep_worker_find(sub->oid, InvalidOid);
 
 			if (sub->enabled && w == NULL && startsub == NULL)
 				startsub = sub;
 			else if (!sub->enabled && w != NULL)
 				logicalrep_worker_stop(w);
+			LWLockRelease(LogicalRepLauncherLock);
 		}
 
 		if (startsub)
-			logicalrep_worker_launch(startsub->dbid, startsub->oid);
+			logicalrep_worker_launch(startsub->dbid, startsub->oid,
+									 InvalidOid);
 
 		/* Switch back to original memory context. */
 		MemoryContextSwitchTo(oldctx);
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index ecf9a03..4f456f6 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -30,6 +30,8 @@
 
 #include "miscadmin.h"
 
+#include "access/heapam.h"
+#include "access/htup.h"
 #include "access/xact.h"
 #include "access/xlog_internal.h"
 
@@ -43,6 +45,7 @@
 #include "storage/procarray.h"
 
 #include "utils/memutils.h"
+#include "utils/tuplestore.h"
 
 /* data for errcontext callback */
 typedef struct LogicalErrorCallbackState
@@ -65,6 +68,9 @@ static void change_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 static void message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 				   XLogRecPtr message_lsn, bool transactional,
 				 const char *prefix, Size message_size, const char *message);
+static List *list_tables_cb_wrapper(LogicalDecodingContext *ctx);
+static void tuple_cb_wrapper(LogicalDecodingContext *ctx, Relation relation,
+							 HeapTuple tup);
 
 static void LoadOutputPlugin(OutputPluginCallbacks *callbacks, char *plugin);
 
@@ -401,6 +407,127 @@ CreateDecodingContext(XLogRecPtr start_lsn,
 }
 
 /*
+ * Create a new limited decoding context for base copy.
+ *
+ * nspname:
+ *		name of a schema
+ *
+ * relname
+ *		name of a relation
+ *
+ * output_plugin_options
+ *		contains options passed to the output plugin.
+ *
+ * prepare_write, do_write
+ *		callbacks that have to be filled to perform the use-case dependent,
+ *		actual work.
+ *
+ * Needs to be called while in a memory context that's at least as long lived
+ * as the decoding context because further memory contexts will be created
+ * inside it.
+ *
+ * Needs to be called inside transaction.
+ *
+ * Returns an initialized decoding context after calling the output plugin's
+ * startup function.
+ */
+LogicalDecodingContext *
+CreateCopyDecodingContext(List *output_plugin_options,
+						  LogicalOutputPluginWriterPrepareWrite prepare_write,
+						  LogicalOutputPluginWriterWrite do_write)
+{
+	LogicalDecodingContext *ctx;
+	ReplicationSlot *slot;
+	MemoryContext	context,
+					old_context;
+
+	/* shorter lines... */
+	slot = MyReplicationSlot;
+
+	/* first some sanity checks that are unlikely to be violated */
+	if (slot == NULL)
+		elog(ERROR, "cannot perform logical base copy without an acquired slot");
+
+	/* make sure the passed slot is suitable, these are user facing errors */
+	if (SlotIsPhysical(slot))
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 (errmsg("cannot use physical replication slot for logical base copy"))));
+
+	if (slot->data.database != MyDatabaseId)
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+		  (errmsg("replication slot \"%s\" was not created in this database",
+				  NameStr(slot->data.name)))));
+
+	if (!IsTransactionOrTransactionBlock())
+		ereport(ERROR,
+				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+				 (errmsg("cannot perform table copy without snapshot"))));
+
+
+	context = AllocSetContextCreate(CurrentMemoryContext,
+									"Table Copy Context",
+									ALLOCSET_DEFAULT_MINSIZE,
+									ALLOCSET_DEFAULT_INITSIZE,
+									ALLOCSET_DEFAULT_MAXSIZE);
+	old_context = MemoryContextSwitchTo(context);
+	ctx = palloc0(sizeof(LogicalDecodingContext));
+
+	ctx->context = context;
+
+	/*
+	 * (re-)load output plugins, so we detect a bad (removed) output plugin
+	 * now.
+	 */
+	LoadOutputPlugin(&ctx->callbacks, NameStr(slot->data.plugin));
+
+	/* CHeck if the output plugin actually supports the copy operations. */
+	if (ctx->callbacks.list_tables_cb == NULL ||
+		ctx->callbacks.tuple_cb == NULL)
+		elog(ERROR, "output plugin \"%s\" does not support LIST_TABLES and COPY_TABLE comamnds",
+			 NameStr(slot->data.plugin));
+
+	/* Initialize non NULL fields. */
+	ctx->slot = slot;
+	ctx->out = makeStringInfo();
+	ctx->prepare_write = prepare_write;
+	ctx->write = do_write;
+
+	/* Make sure plugin sees the options. */
+	ctx->output_plugin_options = output_plugin_options;
+
+	/* call output plugin initialization callback */
+	if (ctx->callbacks.startup_cb != NULL)
+		startup_cb_wrapper(ctx, &ctx->options, false);
+
+	MemoryContextSwitchTo(old_context);
+
+	return ctx;
+}
+
+/*
+ * Process the tuple tup of a relation rel - send it to the tuple
+ * callback of the plugin.
+ */
+void
+DecodingContextProccessTuple(LogicalDecodingContext *ctx, Relation rel,
+							 HeapTuple tup)
+{
+	tuple_cb_wrapper(ctx, rel, tup);
+}
+
+/*
+ * Process the tuple tup of a relation rel - send it to the tuple
+ * callback of the plugin.
+ */
+List *
+DecodingContextGetTableList(LogicalDecodingContext *ctx)
+{
+	return list_tables_cb_wrapper(ctx);
+}
+
+/*
  * Returns true if a consistent initial decoding snapshot has been built.
  */
 bool
@@ -461,9 +588,12 @@ FreeDecodingContext(LogicalDecodingContext *ctx)
 	if (ctx->callbacks.shutdown_cb != NULL)
 		shutdown_cb_wrapper(ctx);
 
-	ReorderBufferFree(ctx->reorder);
-	FreeSnapshotBuilder(ctx->snapshot_builder);
-	XLogReaderFree(ctx->reader);
+	if (ctx->reorder)
+		ReorderBufferFree(ctx->reorder);
+	if (ctx->snapshot_builder)
+		FreeSnapshotBuilder(ctx->snapshot_builder);
+	if (ctx->reader)
+		XLogReaderFree(ctx->reader);
 	MemoryContextDelete(ctx->context);
 }
 
@@ -748,6 +878,60 @@ message_cb_wrapper(ReorderBuffer *cache, ReorderBufferTXN *txn,
 	error_context_stack = errcallback.previous;
 }
 
+static List *
+list_tables_cb_wrapper(LogicalDecodingContext *ctx)
+{
+	LogicalErrorCallbackState state;
+	ErrorContextCallback errcallback;
+	List *res;
+
+	/* Push callback + info on the error context stack */
+	state.ctx = ctx;
+	state.callback_name = "list_tables";
+	state.report_location = InvalidXLogRecPtr;
+	errcallback.callback = output_plugin_error_callback;
+	errcallback.arg = (void *) &state;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
+	/* set output state */
+	ctx->accept_writes = true;
+
+	/* do the actual work: call callback */
+	res = ctx->callbacks.list_tables_cb(ctx);
+
+	/* Pop the error context stack */
+	error_context_stack = errcallback.previous;
+
+	return res;
+}
+
+static void
+tuple_cb_wrapper(LogicalDecodingContext *ctx, Relation relation,
+				 HeapTuple tup)
+{
+	LogicalErrorCallbackState state;
+	ErrorContextCallback errcallback;
+
+	/* Push callback + info on the error context stack */
+	state.ctx = ctx;
+	state.callback_name = "tuple";
+	state.report_location = InvalidXLogRecPtr;
+	errcallback.callback = output_plugin_error_callback;
+	errcallback.arg = (void *) &state;
+	errcallback.previous = error_context_stack;
+	error_context_stack = &errcallback;
+
+	/* set output state */
+	ctx->accept_writes = true;
+
+	/* do the actual work: call callback */
+	ctx->callbacks.tuple_cb(ctx, relation, tup);
+
+	/* Pop the error context stack */
+	error_context_stack = errcallback.previous;
+}
+
 /*
  * Set the required catalog xmin horizon for historic snapshots in the current
  * replication slot.
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index 2b82495..e2f203b 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -356,60 +356,79 @@ logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup)
 }
 
 /*
+ * Write qualified relation name to the output stream.
+ */
+void
+logicalrep_write_rel_name(StringInfo out, char *nspname, char *relname)
+{
+	uint8		nspnamelen;
+	uint8		relnamelen;
+
+	nspnamelen = strlen(nspname) + 1;
+	relnamelen = strlen(relname) + 1;
+
+	pq_sendbyte(out, nspnamelen);		/* schema name length */
+	pq_sendbytes(out, nspname, nspnamelen);
+
+	pq_sendbyte(out, relnamelen);		/* table name length */
+	pq_sendbytes(out, relname, relnamelen);
+}
+
+/*
+ * Read qualified relation name from the stream.
+ */
+void
+logicalrep_read_rel_name(StringInfo in, char **nspname, char **relname)
+{
+	int			len;
+
+	len = pq_getmsgbyte(in);
+	*nspname = (char *) pq_getmsgbytes(in, len);
+
+	len = pq_getmsgbyte(in);
+	*relname = (char *) pq_getmsgbytes(in, len);
+}
+
+
+/*
  * Write relation description to the output stream.
  */
 void
 logicalrep_write_rel(StringInfo out, Relation rel)
 {
 	char	   *nspname;
-	uint8		nspnamelen;
-	const char *relname;
-	uint8		relnamelen;
+	char	   *relname;
 
 	pq_sendbyte(out, 'R');		/* sending RELATION */
 
 	/* use Oid as relation identifier */
 	pq_sendint(out, RelationGetRelid(rel), 4);
 
+	/* send the relation name */
 	nspname = get_namespace_name(RelationGetNamespace(rel));
 	if (nspname == NULL)
 		elog(ERROR, "cache lookup failed for namespace %u",
 			 rel->rd_rel->relnamespace);
-	nspnamelen = strlen(nspname) + 1;
-
 	relname = RelationGetRelationName(rel);
-	relnamelen = strlen(relname) + 1;
-
-	pq_sendbyte(out, nspnamelen);		/* schema name length */
-	pq_sendbytes(out, nspname, nspnamelen);
 
-	pq_sendbyte(out, relnamelen);		/* table name length */
-	pq_sendbytes(out, relname, relnamelen);
+	logicalrep_write_rel_name(out, nspname, relname);
 
 	/* send the attribute info */
 	logicalrep_write_attrs(out, rel);
-
-	pfree(nspname);
 }
 
 /*
- * Read schema.relation from stream and return as LogicalRepRelation opened in
- * lockmode.
+ * Read the relation info from stream and return as LogicalRepRelation.
  */
 LogicalRepRelation *
 logicalrep_read_rel(StringInfo in)
 {
 	LogicalRepRelation	*rel = palloc(sizeof(LogicalRepRelation));
-	int			len;
 
 	rel->remoteid = pq_getmsgint(in, 4);
 
-	/* Read relation from stream */
-	len = pq_getmsgbyte(in);
-	rel->nspname = (char *) pq_getmsgbytes(in, len);
-
-	len = pq_getmsgbyte(in);
-	rel->relname = (char *) pq_getmsgbytes(in, len);
+	/* Read relation name from stream */
+	logicalrep_read_rel_name(in, &rel->nspname, &rel->relname);
 
 	/* Get attribute description */
 	logicalrep_read_attrs(in, &rel->attnames, &rel->natts);
diff --git a/src/backend/replication/logical/publication.c b/src/backend/replication/logical/publication.c
index b86611e..62f40e9 100644
--- a/src/backend/replication/logical/publication.c
+++ b/src/backend/replication/logical/publication.c
@@ -274,6 +274,46 @@ GetRelationPublications(Relation rel)
 	return result;
 }
 
+/*
+ * Gets list of relation oids for a publication.
+ */
+List *
+GetPublicationRelations(Oid pubid)
+{
+	List		   *result;
+	Relation		pubrelsrel;
+	ScanKeyData		scankey;
+	SysScanDesc		scan;
+	HeapTuple		tup;
+
+	/* Find all publications associated with the relation. */
+	pubrelsrel = heap_open(PublicationRelRelationId, AccessShareLock);
+
+	ScanKeyInit(&scankey,
+				Anum_pg_publication_rel_pubid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(pubid));
+
+	scan = systable_beginscan(pubrelsrel, PublicationRelMapIndexId, true,
+							  NULL, 1, &scankey);
+
+	result = NIL;
+	while (HeapTupleIsValid(tup = systable_getnext(scan)))
+	{
+		Form_pg_publication_rel		pubrel;
+
+		pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
+
+		result = lappend_oid(result, pubrel->relid);
+	}
+
+	systable_endscan(scan);
+	heap_close(pubrelsrel, NoLock);
+
+	return result;
+}
+
+
 Publication *
 GetPublication(Oid pubid)
 {
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index b5fa3db..33e15ab 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -591,9 +591,10 @@ SnapBuildExportSnapshot(SnapBuild *builder)
 	snap->xip = newxip;
 
 	/*
-	 * now that we've built a plain snapshot, use the normal mechanisms for
-	 * exporting it
+	 * now that we've built a plain snapshot, make it active and use the
+	 * normal mechanisms for exporting it
 	 */
+	PushActiveSnapshot(snap);
 	snapname = ExportSnapshot(snap);
 
 	ereport(LOG,
diff --git a/src/backend/replication/logical/subscription.c b/src/backend/replication/logical/subscription.c
index 7d1de2c..3ba2b45 100644
--- a/src/backend/replication/logical/subscription.c
+++ b/src/backend/replication/logical/subscription.c
@@ -22,11 +22,15 @@
 #include "access/htup_details.h"
 #include "access/xact.h"
 
+#include "catalog/dependency.h"
 #include "catalog/indexing.h"
 #include "catalog/namespace.h"
 #include "catalog/objectaddress.h"
 #include "catalog/pg_type.h"
 #include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_rel.h"
+
+#include "commands/replicationcmds.h"
 
 #include "executor/spi.h"
 
@@ -41,6 +45,7 @@
 #include "utils/fmgroids.h"
 #include "utils/inval.h"
 #include "utils/lsyscache.h"
+#include "utils/pg_lsn.h"
 #include "utils/rel.h"
 #include "utils/syscache.h"
 
@@ -144,3 +149,178 @@ textarray_to_stringlist(ArrayType *textarray)
 
 	return res;
 }
+
+/*
+ * Set the state of a subscription table.
+ */
+Oid
+SetSubscriptionRelState(Oid subid, Oid relid, char state,
+						   XLogRecPtr sublsn)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	Oid			subrelid;
+	bool		nulls[Natts_pg_subscription_rel];
+	Datum		values[Natts_pg_subscription_rel];
+
+	rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+	/* Try finding existing mapping. */
+	tup = SearchSysCacheCopy2(SUBSCRIPTIONRELMAP,
+							  ObjectIdGetDatum(relid),
+							  ObjectIdGetDatum(subid));
+
+	memset(values, 0, sizeof(values));
+
+	/*
+	 * If the record for given table does not exist yet create new
+	 * record, otherwise update the existing one.
+	 */
+	if (!HeapTupleIsValid(tup))
+	{
+		ObjectAddress	myself,
+						referenced;
+
+		/* Form the tuple. */
+		memset(nulls, false, sizeof(nulls));
+		values[Anum_pg_subscription_rel_subid - 1] = ObjectIdGetDatum(subid);
+		values[Anum_pg_subscription_rel_subrelid - 1] = ObjectIdGetDatum(relid);
+		values[Anum_pg_subscription_rel_substate - 1] = CharGetDatum(state);
+		if (sublsn != InvalidXLogRecPtr)
+			values[Anum_pg_subscription_rel_sublsn - 1] = LSNGetDatum(sublsn);
+		else
+			nulls[Anum_pg_subscription_rel_sublsn - 1] = true;
+
+		tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+
+		/* Insert tuple into catalog. */
+		subrelid = simple_heap_insert(rel, tup);
+		CatalogUpdateIndexes(rel, tup);
+
+		heap_freetuple(tup);
+
+		/* Add dependency on the publication */
+		ObjectAddressSet(myself, SubscriptionRelRelationId, subrelid);
+		ObjectAddressSet(referenced, SubscriptionRelationId, subid);
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
+
+		/* Add dependency on the relation */
+		ObjectAddressSet(referenced, RelationRelationId, relid);
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
+	}
+	else
+	{
+		bool		replaces[Natts_pg_subscription_rel];
+
+		/* Update the tuple. */
+		memset(nulls, true, sizeof(nulls));
+		memset(replaces, false, sizeof(replaces));
+
+		replaces[Anum_pg_subscription_rel_substate - 1] = true;
+		nulls[Anum_pg_subscription_rel_substate - 1] = false;
+		values[Anum_pg_subscription_rel_substate - 1] = CharGetDatum(state);
+
+		replaces[Anum_pg_subscription_rel_sublsn - 1] = true;
+		if (sublsn != InvalidXLogRecPtr)
+		{
+			nulls[Anum_pg_subscription_rel_sublsn - 1] = false;
+			values[Anum_pg_subscription_rel_sublsn - 1] = LSNGetDatum(sublsn);
+		}
+
+		tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+								replaces);
+
+		/* Update the catalog. */
+		simple_heap_update(rel, &tup->t_self, tup);
+		CatalogUpdateIndexes(rel, tup);
+
+		subrelid = HeapTupleGetOid(tup);
+	}
+
+	/* Cleanup. */
+	heap_close(rel, NoLock);
+
+	/* Make the changes visible. */
+	CommandCounterIncrement();
+
+	return subrelid;
+}
+
+/*
+ * Get state of subscription table.
+ *
+ * Returns SUBREL_STATE_UNKNOWN when not found and missing_ok is true.
+ */
+char
+GetSubscriptionRelState(Oid subid, Oid relid, XLogRecPtr *sublsn,
+						bool missing_ok)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	char		substate;
+	bool		isnull;
+	Datum		d;
+
+	rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+	/* Try finding the mapping. */
+	tup = SearchSysCache2(SUBSCRIPTIONRELMAP,
+						  ObjectIdGetDatum(relid),
+						  ObjectIdGetDatum(subid));
+
+	if (!HeapTupleIsValid(tup))
+	{
+		if (missing_ok)
+		{
+			*sublsn = InvalidXLogRecPtr;
+			return '\0';
+		}
+
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("subscription table %u in subscription %d does not exist",
+						relid, subid)));
+	}
+
+	/* Get the state. */
+	d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
+						Anum_pg_subscription_rel_substate, &isnull);
+	Assert(!isnull);
+	substate = DatumGetChar(d);
+	d = SysCacheGetAttr(SUBSCRIPTIONRELMAP, tup,
+						Anum_pg_subscription_rel_sublsn, &isnull);
+	if (isnull)
+		*sublsn = InvalidXLogRecPtr;
+	else
+		*sublsn = DatumGetLSN(d);
+
+	/* Cleanup */
+	ReleaseSysCache(tup);
+	heap_close(rel, RowExclusiveLock);
+
+	return substate;
+}
+
+/*
+ * Drop subscription table by OID
+ */
+void
+DropSubscriptionRelById(Oid subrelid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+
+	rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+	tup = SearchSysCache1(SUBSCRIPTIONRELOID, ObjectIdGetDatum(subrelid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for subscription table %u",
+			 subrelid);
+
+	simple_heap_delete(rel, &tup->t_self);
+
+	ReleaseSysCache(tup);
+
+	heap_close(rel, RowExclusiveLock);
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
new file mode 100644
index 0000000..7e844e8
--- /dev/null
+++ b/src/backend/replication/logical/tablesync.c
@@ -0,0 +1,672 @@
+/*-------------------------------------------------------------------------
+ * tablesync.c
+ *	   PostgreSQL logical replication
+ *
+ * Copyright (c) 2012-2016, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/backend/replication/logical/tablesync.c
+ *
+ * NOTES
+ *	  This file contains code for initial table data synchronization for
+ *	  logical replication.
+ *
+ *    The initial data synchronization is done separately for each table,
+ *    in separate apply worker that only fetches the initial snapshot data
+ *    from the provider and then synchronizes the position in stream with
+ *    the main apply worker.
+ *
+ *    The stream position synchronization works in multiple steps.
+ *     - sync finishes copy and sets table state as SYNCWAIT and waits
+ *       for state to change in a loop
+ *     - apply periodically checks unsynced tables for SYNCWAIT, when it
+ *       appears it will compare its position in the stream with the
+ *       SYNCWAIT position and decides to either set it to CATCHUP when
+ *       the apply was infront (and wait for the sync to do the catchup),
+ *       or set the state to SYNCDONE if the sync was infront or in case
+ *       both sync and apply are at the same position it will set it to
+ *       READY and stops tracking it
+ *     - if the state was set to CATCHUP sync will read the stream and
+ *       apply changes until it catches up to the specified stream
+ *       position and then sets state to READY and signals apply that it
+ *       can stop waiting and exits, if the state was set to something
+ *       else than CATCHUP the sync process will simply end
+ *     - if the state was set to SYNCDONE by apply, the apply will
+ *       continue tracking the table until it reaches the SYNCDONE stream
+ *       position at which point it sets state to READY and stops tracking
+ *
+ *    Example flows look like this:
+ *     - Apply is infront:
+ *	      sync:8   -> set SYNCWAIT
+ *        apply:10 -> set CATCHUP
+ *        sync:10  -> set ready
+ *          exit
+ *        apply:10
+ *          stop tracking
+ *          continue rep
+ *    - Sync infront:
+ *        sync:10
+ *          set SYNCWAIT
+ *        apply:8
+ *          set SYNCDONE
+ *        sync:10
+ *          exit
+ *        apply:10
+ *          set READY
+ *          stop tracking
+ *          continue rep
+ *-------------------------------------------------------------------------
+ */
+
+#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 "catalog/pg_subscription_rel.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 "replication/worker_internal.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/fmgroids.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 TableState
+{
+	dlist_node	node;
+	Oid			relid;
+	XLogRecPtr	lsn;
+	char		state;
+} TableState;
+
+static dlist_head table_states = DLIST_STATIC_INIT(table_states);
+static bool table_states_valid = false;
+
+
+/*
+ * Exit routine for synchronization worker.
+ */
+static void
+finish_sync_worker(char *slotname)
+{
+	LogicalRepWorker   *worker;
+	RepOriginId			originid;
+	MemoryContext		oldctx = CurrentMemoryContext;
+
+	/*
+	 * Drop the replication slot on remote server.
+	 * We want to continue even in the case that the slot on remote side
+	 * is already gone. This means that we can leave slot on the remote
+	 * side but that can happen for other reasons as well so we can't
+	 * really protect against that.
+	 */
+	PG_TRY();
+	{
+		wrcapi->drop_slot(wrchandle, slotname);
+	}
+	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();
+
+	/* Also remove the origin tracking for the slot if it exists. */
+	StartTransactionCommand();
+	originid = replorigin_by_name(slotname, true);
+	if (originid != InvalidRepOriginId)
+	{
+		if (originid == replorigin_session_origin)
+		{
+			replorigin_session_reset();
+			replorigin_session_origin = InvalidRepOriginId;
+		}
+		replorigin_drop(originid);
+	}
+	CommitTransactionCommand();
+
+	/* Flush all writes. */
+	XLogFlush(GetXLogWriteRecPtr());
+
+	/* Find the main apply worker and signal it. */
+	LWLockAcquire(LogicalRepLauncherLock, LW_EXCLUSIVE);
+	worker = logicalrep_worker_find(MyLogicalRepWorker->subid, InvalidOid);
+	if (worker && worker->proc)
+		SetLatch(&worker->proc->procLatch);
+	LWLockRelease(LogicalRepLauncherLock);
+
+	ereport(LOG,
+			(errmsg("logical replication synchronization worker finished processing")));
+
+	/* Stop gracefully */
+	wrcapi->disconnect(wrchandle);
+	proc_exit(0);
+}
+
+/*
+ * Wait until the table synchronization change matches the desired state.
+ */
+static bool
+wait_for_sync_status_change(TableState *tstate)
+{
+	int		rc;
+	char	state = tstate->state;
+
+	for (;;)
+	{
+		CHECK_FOR_INTERRUPTS();
+
+		StartTransactionCommand();
+		tstate->state = GetSubscriptionRelState(MyLogicalRepWorker->subid,
+												tstate->relid,
+												&tstate->lsn,
+												true);
+		CommitTransactionCommand();
+
+		if (tstate->state != state)
+			return true;
+
+		rc = WaitLatch(&MyProc->procLatch,
+					   WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+					   10000L);
+
+		/* emergency bailout if postmaster has died */
+		if (rc & WL_POSTMASTER_DEATH)
+			proc_exit(1);
+
+        ResetLatch(&MyProc->procLatch);
+	}
+
+	return false;
+}
+
+/*
+ * Read the state of the tables in the subscription and update our table
+ * state list.
+ */
+static void
+reread_sync_state(Oid relid)
+{
+	dlist_mutable_iter	iter;
+	Relation	rel;
+	HeapTuple	tup;
+	ScanKeyData	skey[2];
+	HeapScanDesc	scan;
+
+	/* Clean the old list. */
+	dlist_foreach_modify(iter, &table_states)
+	{
+		TableState *tstate = dlist_container(TableState, node, iter.cur);
+
+		dlist_delete(iter.cur);
+		pfree(tstate);
+	}
+
+	/*
+	 * Fetch all the subscription relation states that are not marked as
+	 * ready and push them into our table state tracking list.
+	 */
+	rel = heap_open(SubscriptionRelRelationId, RowExclusiveLock);
+
+	ScanKeyInit(&skey[0],
+				Anum_pg_subscription_rel_subid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(MyLogicalRepWorker->subid));
+
+	if (OidIsValid(relid))
+	{
+		ScanKeyInit(&skey[1],
+					Anum_pg_subscription_rel_subrelid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(relid));
+	}
+	else
+	{
+		ScanKeyInit(&skey[1],
+					Anum_pg_subscription_rel_substate,
+					BTEqualStrategyNumber, F_CHARNE,
+					CharGetDatum(SUBREL_STATE_READY));
+	}
+
+	scan = heap_beginscan_catalog(rel, 2, skey);
+
+	while (HeapTupleIsValid(tup = heap_getnext(scan, ForwardScanDirection)))
+	{
+		Form_pg_subscription_rel	subrel;
+		TableState	   *tstate;
+		MemoryContext	oldctx;
+
+		subrel = (Form_pg_subscription_rel) GETSTRUCT(tup);
+
+		/* Allocate the tracking info in a permament memory context. */
+		oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+
+		tstate = (TableState *) palloc(sizeof(TableState));
+		tstate->relid = subrel->relid;
+		tstate->state = subrel->substate;
+		tstate->lsn = subrel->sublsn;
+
+		dlist_push_tail(&table_states, &tstate->node);
+		MemoryContextSwitchTo(oldctx);
+	}
+
+	/* Cleanup */
+	heap_endscan(scan);
+	heap_close(rel, RowExclusiveLock);
+
+	table_states_valid = true;
+}
+
+/*
+ * Callback from syscache invalidation.
+ */
+void
+invalidate_syncing_table_states(Datum arg, int cacheid, uint32 hashvalue)
+{
+	table_states_valid = false;
+}
+
+/*
+ * Handle table synchronization cooperation from the synchroniation
+ * worker.
+ */
+static void
+process_syncing_tables_sync(char *slotname, XLogRecPtr end_lsn)
+{
+	TableState *tstate;
+	TimeLineID	tli;
+
+	Assert(!IsTransactionState());
+
+	/*
+	 * Synchronization workers don't keep track of all synchronization
+	 * tables, they only care about their table.
+	 */
+	if (!table_states_valid)
+	{
+		StartTransactionCommand();
+		reread_sync_state(MyLogicalRepWorker->relid);
+		CommitTransactionCommand();
+	}
+
+	/* Somebody removed table underneath this worker, nothing more to do. */
+	if (dlist_is_empty(&table_states))
+	{
+		wrcapi->endstreaming(wrchandle, &tli);
+		finish_sync_worker(slotname);
+	}
+
+	/* Check if we are done with catchup now. */
+	tstate = dlist_container(TableState, node, dlist_head_node(&table_states));
+	if (tstate->state == SUBREL_STATE_CATCHUP)
+	{
+		Assert(tstate->lsn != InvalidXLogRecPtr);
+
+		if (tstate->lsn == end_lsn)
+		{
+			tstate->state = SUBREL_STATE_READY;
+			tstate->lsn = InvalidXLogRecPtr;
+			/* Update state of the synchronization. */
+			StartTransactionCommand();
+			SetSubscriptionRelState(MyLogicalRepWorker->subid,
+									tstate->relid, tstate->state,
+									tstate->lsn);
+			CommitTransactionCommand();
+
+			wrcapi->endstreaming(wrchandle, &tli);
+			finish_sync_worker(slotname);
+		}
+		return;
+	}
+}
+
+/*
+ * Handle table synchronization cooperation from the apply worker.
+ */
+static void
+process_syncing_tables_apply(char *slotname, XLogRecPtr end_lsn)
+{
+	dlist_mutable_iter	iter;
+
+	Assert(!IsTransactionState());
+
+	if (!table_states_valid)
+	{
+		StartTransactionCommand();
+		reread_sync_state(InvalidOid);
+		CommitTransactionCommand();
+	}
+
+	dlist_foreach_modify(iter, &table_states)
+	{
+		TableState *tstate = dlist_container(TableState, node, iter.cur);
+		bool		start_worker;
+		LogicalRepWorker   *worker;
+
+		/*
+		 * When the synchronization process is at the cachup phase we need
+		 * to ensure that we are not behind it (it's going to wait at this
+		 * point for the change of state). Once we are infront or at the same
+		 * position as the synchronization proccess we can signal it to
+		 * finish the catchup.
+		 */
+		if (tstate->state == SUBREL_STATE_SYNCWAIT)
+		{
+			if (end_lsn > tstate->lsn)
+			{
+				/*
+				 * Apply is infront, tell sync to catchup. and wait until
+				 * it does.
+				 */
+				tstate->state = SUBREL_STATE_CATCHUP;
+				tstate->lsn = end_lsn;
+				StartTransactionCommand();
+				SetSubscriptionRelState(MyLogicalRepWorker->subid,
+										tstate->relid, tstate->state,
+										tstate->lsn);
+				CommitTransactionCommand();
+
+				/* Signal the worker as it may be waiting for us. */
+				LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+				worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+												tstate->relid);
+				if (worker && worker->proc)
+					SetLatch(&worker->proc->procLatch);
+				LWLockRelease(LogicalRepLauncherLock);
+
+				if (wait_for_sync_status_change(tstate));
+					Assert(SUBREL_STATE_READY);
+			}
+			else
+			{
+				/*
+				 * Apply is either behind in which case sync worker is done
+				 * but apply needs to keep tracking the table until it
+				 * catches up to where sync finished.
+				 * Or apply and sync are at the same position in which case
+				 * table can be switched to standard replication mode
+				 * immediately.
+				 */
+				if (end_lsn < tstate->lsn)
+					tstate->state = SUBREL_STATE_SYNCDONE;
+				else
+					tstate->state = SUBREL_STATE_READY;
+
+				StartTransactionCommand();
+				SetSubscriptionRelState(MyLogicalRepWorker->subid,
+										tstate->relid, tstate->state,
+										tstate->lsn);
+				CommitTransactionCommand();
+
+				/* Signal the worker as it may be waiting for us. */
+				LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+				worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+												tstate->relid);
+				if (worker && worker->proc)
+					SetLatch(&worker->proc->procLatch);
+				LWLockRelease(LogicalRepLauncherLock);
+			}
+		}
+		else if (tstate->state == SUBREL_STATE_SYNCDONE &&
+				 end_lsn >= tstate->lsn)
+		{
+			/*
+			 * Apply catched up to the position where table sync finished,
+			 * mark the table as ready for normal replication.
+			 */
+			tstate->state = SUBREL_STATE_READY;
+			tstate->lsn = InvalidXLogRecPtr;
+			StartTransactionCommand();
+			SetSubscriptionRelState(MyLogicalRepWorker->subid,
+									tstate->relid, tstate->state,
+									tstate->lsn);
+			CommitTransactionCommand();
+		}
+
+		/*
+		 * In case table is supposed to be synchronizing but the
+		 * synchronization worker is not running, start it.
+		 * Limit the number of launched workers here to one (for now).
+		 */
+		if (tstate->state != SUBREL_STATE_READY &&
+			tstate->state != SUBREL_STATE_SYNCDONE)
+		{
+			LWLockAcquire(LogicalRepLauncherLock, LW_SHARED);
+			worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
+											tstate->relid);
+			start_worker = !worker &&
+				logicalrep_worker_count(MyLogicalRepWorker->subid) < 2;
+			LWLockRelease(LogicalRepLauncherLock);
+			if (start_worker)
+				logicalrep_worker_launch(MyLogicalRepWorker->dbid,
+										 MyLogicalRepWorker->subid,
+										 tstate->relid);
+
+		}
+	}
+}
+
+/*
+ * Proccess state possible change(s) of tables that are being synchronized
+ * in parallel.
+ */
+void
+process_syncing_tables(char *slotname, XLogRecPtr end_lsn)
+{
+	if (OidIsValid(MyLogicalRepWorker->relid))
+		process_syncing_tables_sync(slotname, end_lsn);
+	else
+		process_syncing_tables_apply(slotname, end_lsn);
+}
+
+/*
+ * Setup replication origin tracking.
+ */
+static XLogRecPtr
+setup_origin_tracking(char *origin_name)
+{
+	RepOriginId		originid;
+
+	StartTransactionCommand();
+	originid = replorigin_by_name(origin_name, true);
+	if (!OidIsValid(originid))
+		originid = replorigin_create(origin_name);
+	replorigin_session_setup(originid);
+	replorigin_session_origin = originid;
+	CommitTransactionCommand();
+	return replorigin_session_get_progress(false);
+}
+
+
+/*
+ * Start syncing the table in the sync worker.
+ */
+char *
+LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
+{
+	StringInfoData	s;
+	TableState		tstate;
+	MemoryContext	oldctx;
+	char		   *slotname;
+
+	/* Check the state of the table synchronization. */
+	StartTransactionCommand();
+	tstate.relid = MyLogicalRepWorker->relid;
+	tstate.state = GetSubscriptionRelState(MySubscription->oid, tstate.relid,
+										   &tstate.lsn, false);
+
+	/*
+	 * Build unique slot name.
+	 * TODO: protect against too long slot name.
+	 */
+	oldctx = MemoryContextSwitchTo(CacheMemoryContext);
+	initStringInfo(&s);
+	appendStringInfo(&s, "%s_sync_%s", MySubscription->slotname,
+					 get_rel_name(tstate.relid));
+	slotname = s.data;
+	MemoryContextSwitchTo(oldctx);
+
+	CommitTransactionCommand();
+
+	wrcapi->connect(wrchandle, MySubscription->conninfo, true, slotname);
+
+	switch (tstate.state)
+	{
+		case SUBREL_STATE_INIT:
+		case SUBREL_STATE_DATA:
+			{
+				Relation	rel;
+				XLogRecPtr	lsn;
+				char	   *options;
+
+				/* Update the state and make it visible to others. */
+				StartTransactionCommand();
+				SetSubscriptionRelState(MySubscription->oid, tstate.relid,
+										SUBREL_STATE_DATA,
+										InvalidXLogRecPtr);
+				CommitTransactionCommand();
+
+				*origin_startpos = setup_origin_tracking(slotname);
+
+				/*
+				 * We want to do the table data sync in single
+				 * transaction so do not close the transaction opened
+				 * above.
+				 * There will be no BEGIN or COMMIT messages coming via
+				 * logical replication while the copy table command is
+				 * running so start the transaction here.
+				 * Note the memory context for data handling will still
+				 * be done using ensure_transaction called by the insert
+				 * handler.
+				 */
+				StartTransactionCommand();
+
+				/*
+				 * Don't allow parallel access other than SELECT while
+				 * the initial contents are being copied.
+				 */
+				rel = heap_open(tstate.relid, ExclusiveLock);
+
+				/* Create temporary slot for the sync proccess. */
+				wrcapi->create_slot(wrchandle, slotname, true,  &lsn);
+
+				/* Build option string for the plugin. */
+				options = logicalrep_build_options(MySubscription->publications);
+
+				wrcapi->copy_table(wrchandle, slotname,
+								   get_namespace_name(RelationGetNamespace(rel)),
+								   RelationGetRelationName(rel),
+								   options);
+
+				/*
+				 * Run the standard apply loop for the initial data
+				 * stream.
+				 */
+				in_remote_transaction = true;
+				LogicalRepApplyLoop(*origin_startpos);
+
+				/*
+				 * We are done with the initial data synchronization,
+				 * update the state.
+				 */
+				SetSubscriptionRelState(MySubscription->oid, tstate.relid,
+										SUBREL_STATE_SYNCWAIT, lsn);
+				heap_close(rel, NoLock);
+
+				/* End the transaction. */
+				CommitTransactionCommand();
+				in_remote_transaction = false;
+
+				/*
+				 * Wait for main apply worker to either tell us to
+				 * catchup or that we are done.
+				 */
+				wait_for_sync_status_change(&tstate);
+				if (tstate.state != SUBREL_STATE_CATCHUP)
+					finish_sync_worker(slotname);
+				break;
+			}
+
+		case SUBREL_STATE_SYNCWAIT:
+			*origin_startpos = setup_origin_tracking(slotname);
+			/*
+			 * Wait for main apply worker to either tell us to
+			 * catchup or that we are done.
+			 */
+			wait_for_sync_status_change(&tstate);
+			if (tstate.state != SUBREL_STATE_CATCHUP)
+				finish_sync_worker(slotname);
+			break;
+		case SUBREL_STATE_CATCHUP:
+			/* Catchup is handled by streaming loop. */
+			*origin_startpos = setup_origin_tracking(slotname);
+			break;
+		case SUBREL_STATE_SYNCDONE:
+		case SUBREL_STATE_READY:
+			/* Nothing to do here but finish. */
+			finish_sync_worker(slotname);
+		default:
+			elog(ERROR, "unknown relation state \"%c\"", tstate.state);
+	}
+
+	return slotname;
+}
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index d74c7e9..e18939e 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -14,6 +14,8 @@
 
 #include "access/xact.h"
 
+#include "catalog/pg_publication.h"
+
 #include "mb/pg_wchar.h"
 
 #include "replication/logical.h"
@@ -24,7 +26,10 @@
 
 #include "utils/builtins.h"
 #include "utils/inval.h"
+#include "utils/lsyscache.h"
 #include "utils/memutils.h"
+#include "utils/tuplestore.h"
+#include "utils/syscache.h"
 
 PG_MODULE_MAGIC;
 
@@ -40,6 +45,9 @@ static void pgoutput_commit_txn(LogicalDecodingContext *ctx,
 static void pgoutput_change(LogicalDecodingContext *ctx,
 				 ReorderBufferTXN *txn, Relation rel,
 				 ReorderBufferChange *change);
+static void pgoutput_tuple(LogicalDecodingContext *ctx, Relation relation,
+			   HeapTuple tuple);
+static List *pgoutput_list_tables(LogicalDecodingContext *ctx);
 static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
 						RepOriginId origin_id);
 
@@ -74,6 +82,8 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
 	cb->commit_cb = pgoutput_commit_txn;
 	cb->filter_by_origin_cb = pgoutput_origin_filter;
 	cb->shutdown_cb = pgoutput_shutdown;
+	cb->tuple_cb = pgoutput_tuple;
+	cb->list_tables_cb = pgoutput_list_tables;
 }
 
 /*
@@ -295,6 +305,104 @@ pgoutput_change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
 }
 
 /*
+ * Sends the tuple from relation over wire.
+ * Currenly this behaves same as the INSERT replication.
+ */
+static void
+pgoutput_tuple(LogicalDecodingContext *ctx, Relation relation,
+			   HeapTuple tuple)
+{
+	PGOutputData	   *data = (PGOutputData *) ctx->output_plugin_private;
+	MemoryContext		old;
+	RelSchemaSyncEntry *relentry = NULL;
+
+	/*
+	 * First check the table filter
+	 * TODO: do we actually need this?
+	 */
+	if (!publication_change_is_replicated(relation,
+										  PublicationChangeInsert,
+										  data->publication_names))
+		return;
+
+	/* Avoid leaking memory by using and resetting our own context */
+	old = MemoryContextSwitchTo(data->context);
+
+	/*
+	 * Write the relation schema if the current schema haven't been sent yet.
+	 */
+	relentry = get_rel_schema_sync_entry(RelationGetRelid(relation));
+	if (!relentry->schema_sent)
+	{
+		OutputPluginPrepareWrite(ctx, false);
+		logicalrep_write_rel(ctx->out, relation);
+		OutputPluginWrite(ctx, false);
+		relentry->schema_sent = true;
+	}
+
+	/* Send the data */
+	OutputPluginPrepareWrite(ctx, true);
+	logicalrep_write_insert(ctx->out, relation, tuple);
+	OutputPluginWrite(ctx, true);
+
+	/* Cleanup */
+	MemoryContextSwitchTo(old);
+	MemoryContextReset(data->context);
+}
+
+/*
+ * Get the list of tables replicated by current connection.
+ */
+static List *
+pgoutput_list_tables(LogicalDecodingContext *ctx)
+{
+	PGOutputData   *data = (PGOutputData *) ctx->output_plugin_private;
+	MemoryContext	old;
+	List		   *rellist = NIL,
+				   *res = NIL;
+	ListCell	   *lc;
+
+	/* Avoid leaking memory by using and resetting our own context */
+	old = MemoryContextSwitchTo(data->context);
+
+	/* Build unique list of relations in all subscribed publications. */
+	foreach(lc, data->publication_names)
+	{
+		char	   *pubname = (char *) lfirst(lc);
+		Oid			pubid;
+		List	   *pubrellist;
+
+		pubid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(pubname));
+		if (!OidIsValid(pubid))
+			elog(ERROR, "cache lookup failed for publication %u", pubid);
+
+		pubrellist = GetPublicationRelations(pubid);
+		rellist = list_concat_unique_oid(rellist, pubrellist);
+	}
+
+	MemoryContextSwitchTo(old);
+
+	/* Put all the relations list of LogicalRepTableListEntry. */
+	foreach (lc, rellist)
+	{
+		Oid			relid = lfirst_oid(lc);
+		LogicalRepTableListEntry *entry;
+
+		entry = palloc(sizeof(LogicalRepTableListEntry));
+		entry->nspname = get_namespace_name(get_rel_namespace(relid));
+		entry->relname = get_rel_name(relid);
+		entry->info = NULL;
+
+		res = lappend(res, entry);
+	}
+
+	/* Cleanup our memory context. */
+	MemoryContextReset(data->context);
+
+	return res;
+}
+
+/*
  * Currently we always forward.
  */
 static bool
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index d93db88..65b8ea7 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -77,11 +77,14 @@ Node *replication_parse_result;
 %token K_LOGICAL
 %token K_SLOT
 %token K_RESERVE_WAL
+%token K_TABLE
+%token K_LIST_TABLES
+%token K_COPY_TABLE
 
 %type <node>	command
 %type <node>	base_backup start_replication start_logical_replication
 				create_replication_slot drop_replication_slot identify_system
-				timeline_history
+				timeline_history list_tables copy_table
 %type <list>	base_backup_opt_list
 %type <defelt>	base_backup_opt
 %type <uintval>	opt_timeline
@@ -111,6 +114,8 @@ command:
 			| create_replication_slot
 			| drop_replication_slot
 			| timeline_history
+			| list_tables
+			| copy_table
 			;
 
 /*
@@ -323,6 +328,30 @@ plugin_opt_arg:
 			SCONST							{ $$ = (Node *) makeString($1); }
 			| /* EMPTY */					{ $$ = NULL; }
 		;
+
+copy_table:
+			K_COPY_TABLE K_SLOT IDENT K_TABLE IDENT IDENT plugin_options
+				{
+					CopyTableCmd *cmd;
+					cmd = makeNode(CopyTableCmd);
+					cmd->slotname = $3;
+					cmd->relation = makeRangeVar($5, $6, -1);
+					cmd->options = $7;
+					$$ = (Node *) cmd;
+				}
+		;
+
+list_tables:
+			K_LIST_TABLES K_SLOT IDENT plugin_options
+				{
+					ListTablesCmd *cmd;
+					cmd = makeNode(ListTablesCmd);
+					cmd->slotname = $3;
+					cmd->options = $4;
+					$$ = (Node *) cmd;
+				}
+		;
+
 %%
 
 #include "repl_scanner.c"
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index f83ec53..69d6e86 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -98,6 +98,9 @@ PHYSICAL			{ return K_PHYSICAL; }
 RESERVE_WAL			{ return K_RESERVE_WAL; }
 LOGICAL				{ return K_LOGICAL; }
 SLOT				{ return K_SLOT; }
+LIST_TABLES			{ return K_LIST_TABLES; }
+COPY_TABLE			{ return K_COPY_TABLE; }
+TABLE				{ return K_TABLE; }
 
 ","				{ return ','; }
 ";"				{ return ';'; }
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index a0dba19..def88d3 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -43,6 +43,7 @@
 #include <signal.h>
 #include <unistd.h>
 
+#include "access/relscan.h"
 #include "access/timeline.h"
 #include "access/transam.h"
 #include "access/xact.h"
@@ -932,7 +933,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
 	pq_endmessage(&buf);
 
 	/*
-	 * release active status again, START_REPLICATION will reacquire it
+	 * release active status again, subsequent commands will reacquire it
 	 */
 	ReplicationSlotRelease();
 }
@@ -1035,6 +1036,176 @@ StartLogicalReplication(StartReplicationCmd *cmd)
 }
 
 /*
+ * Handle LIST_TABLES command.
+ */
+static void
+SendTableList(ListTablesCmd *cmd)
+{
+	List		   *tables;
+	ListCell	   *lc;
+	StringInfoData	buf;
+
+	/* make sure that our requirements are still fulfilled */
+	CheckLogicalDecodingRequirements();
+
+	Assert(!MyReplicationSlot);
+
+	ReplicationSlotAcquire(cmd->slotname);
+
+	/* Initialize the decoding context for table copy. */
+	logical_decoding_ctx = CreateCopyDecodingContext(cmd->options,
+													 WalSndPrepareWrite,
+													 WalSndWriteData);
+
+	/* Send a RowDescription message */
+	pq_beginmessage(&buf, 'T');
+	pq_sendint(&buf, 3, 2);		/* 3 fields */
+
+	/* first field: namespace name */
+	pq_sendstring(&buf, "nspname");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	/* second field: relation name */
+	pq_sendstring(&buf, "relname");	/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	/* third field: freeform relation info (the only NULLable field) */
+	pq_sendstring(&buf, "info");		/* col name */
+	pq_sendint(&buf, 0, 4);		/* table oid */
+	pq_sendint(&buf, 0, 2);		/* attnum */
+	pq_sendint(&buf, TEXTOID, 4);		/* type oid */
+	pq_sendint(&buf, -1, 2);	/* typlen */
+	pq_sendint(&buf, 0, 4);		/* typmod */
+	pq_sendint(&buf, 0, 2);		/* format code */
+
+	pq_endmessage(&buf);
+
+	/* Let the decoding contex send the list. */
+	tables = DecodingContextGetTableList(logical_decoding_ctx);
+
+	/* Send the table list as tuples. */
+	foreach(lc, tables)
+	{
+		LogicalRepTableListEntry *entry = lfirst(lc);
+		Size		len;
+
+		Assert(entry->nspname != NULL);
+		Assert(entry->relname != NULL);
+
+		/* Send a DataRow message */
+		pq_beginmessage(&buf, 'D');
+		pq_sendint(&buf, 3, 2);		/* # of columns */
+
+		/* namespace name */
+		len = strlen(entry->nspname);
+		pq_sendint(&buf, len, 4);	/* col1 len */
+		pq_sendbytes(&buf, entry->nspname, len);
+
+		/* relation name name */
+		len = strlen(entry->relname);
+		pq_sendint(&buf, len, 4);	/* col1 len */
+		pq_sendbytes(&buf, entry->relname, len);
+
+		/* relation info, or NULL if none */
+		if (entry->info != NULL)
+		{
+			len = strlen(entry->info);
+			pq_sendint(&buf, len, 4);
+			pq_sendbytes(&buf, entry->info, len);
+		}
+		else
+			pq_sendint(&buf, -1, 4);
+
+		pq_endmessage(&buf);
+	}
+
+	/* Clean up the logical decoding context. */
+	FreeDecodingContext(logical_decoding_ctx);
+
+	ReplicationSlotRelease();
+}
+
+/*
+ * LogicalDecodingContext 'write' callback.
+ *
+ * Actually write out data previously prepared by WalSndPrepareWrite out
+ * to the network.
+ */
+static void
+CopyTableWriteData(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId xid,
+				bool last_write)
+{
+	/* output previously gathered data in a CopyData packet */
+	if (pq_putmessage('d', ctx->out->data, ctx->out->len))
+		ereport(ERROR,
+			(errmsg("copy table could not send data, aborting")));
+}
+
+/*
+ * Handle OPY_TABLE command.
+ */
+static void
+CopyTable(CopyTableCmd *cmd)
+{
+	StringInfoData	buf;
+	Relation		rel;
+	HeapScanDesc	scandesc;
+	HeapTuple	tup;
+
+	/* make sure that our requirements are still fulfilled */
+	CheckLogicalDecodingRequirements();
+
+	Assert(!MyReplicationSlot);
+
+	ReplicationSlotAcquire(cmd->slotname);
+
+	WalSndSetState(WALSNDSTATE_BACKUP);
+
+	/* Send a CopyBothResponse message, and start streaming */
+	pq_beginmessage(&buf, 'W');
+	pq_sendbyte(&buf, 0);
+	pq_sendint(&buf, 0, 2);
+	pq_endmessage(&buf);
+	pq_flush();
+
+	/* Initialize the decoding context for table copy. */
+	logical_decoding_ctx = CreateCopyDecodingContext(cmd->options,
+													 WalSndPrepareWrite,
+													 CopyTableWriteData);
+
+	/* Open the relation and start the scan. */
+	rel = heap_openrv(cmd->relation, AccessShareLock);
+	scandesc = heap_beginscan(rel, GetActiveSnapshot(), 0, NULL);
+
+	/* Scan the whole table and pass the rows to the decoding context. */
+	while (HeapTupleIsValid(tup = heap_getnext(scandesc,
+											   ForwardScanDirection)))
+		DecodingContextProccessTuple(logical_decoding_ctx, rel, tup);
+
+	/* Close the scan and relation. */
+	heap_endscan(scandesc);
+	heap_close(rel, AccessShareLock);
+
+	/* Send CopyDone */
+	pq_putemptymessage('c');
+
+	FreeDecodingContext(logical_decoding_ctx);
+
+	ReplicationSlotRelease();
+}
+
+
+/*
  * LogicalDecodingContext 'prepare_write' callback.
  *
  * Prepare a write into a StringInfo.
@@ -1299,14 +1470,6 @@ exec_replication_command(const char *cmd_string)
 	ereport(log_replication_commands ? LOG : DEBUG1,
 			(errmsg("received replication command: %s", cmd_string)));
 
-	/*
-	 * CREATE_REPLICATION_SLOT ... LOGICAL exports a snapshot until the next
-	 * command arrives. Clean up the old stuff if there's anything.
-	 */
-	SnapBuildClearExportedSnapshot();
-
-	CHECK_FOR_INTERRUPTS();
-
 	cmd_context = AllocSetContextCreate(CurrentMemoryContext,
 										"Replication command context",
 										ALLOCSET_DEFAULT_MINSIZE,
@@ -1324,6 +1487,16 @@ exec_replication_command(const char *cmd_string)
 
 	cmd_node = replication_parse_result;
 
+	/*
+	 * CREATE_REPLICATION_SLOT ... LOGICAL exports a snapshot until the next
+	 * command arrives. Clean up the old stuff if there's anything unless
+	 * the command currently being executed needs the snapshot.
+	 */
+	if (cmd_node->type != T_ListTablesCmd && cmd_node->type != T_CopyTableCmd)
+		SnapBuildClearExportedSnapshot();
+
+	CHECK_FOR_INTERRUPTS();
+
 	switch (cmd_node->type)
 	{
 		case T_IdentifySystemCmd:
@@ -1357,6 +1530,14 @@ exec_replication_command(const char *cmd_string)
 			SendTimeLineHistory((TimeLineHistoryCmd *) cmd_node);
 			break;
 
+		case T_ListTablesCmd:
+			SendTableList((ListTablesCmd *) cmd_node);
+			break;
+
+		case T_CopyTableCmd:
+			CopyTable((CopyTableCmd *) cmd_node);
+			break;
+
 		default:
 			elog(ERROR, "unrecognized replication command node tag: %u",
 				 cmd_node->type);
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 03c8916..c6e7207 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -60,6 +60,7 @@
 #include "catalog/pg_replication_origin.h"
 #include "catalog/pg_statistic.h"
 #include "catalog/pg_subscription.h"
+#include "catalog/pg_subscription_rel.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_ts_config.h"
@@ -736,6 +737,28 @@ static const struct cachedesc cacheinfo[] = {
 		},
 		4
 	},
+	{SubscriptionRelRelationId,		/* SUBSCRIPTIONRELOID */
+		SubscriptionRelOidIndexId,
+		1,
+		{
+			ObjectIdAttributeNumber,
+			0,
+			0,
+			0
+		},
+		64
+	},
+	{SubscriptionRelRelationId,		/* SUBSCRIPTIONRELMAP */
+		SubscriptionRelMapIndexId,
+		2,
+		{
+			Anum_pg_subscription_rel_subrelid,
+			Anum_pg_subscription_rel_subid,
+			0,
+			0
+		},
+		64
+	},
 	{TableSpaceRelationId,		/* TABLESPACEOID */
 		TablespaceOidIndexId,
 		1,
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index 86e2939..02ebd12 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -337,6 +337,12 @@ DECLARE_UNIQUE_INDEX(pg_subscription_oid_index, 6114, on pg_subscription using b
 DECLARE_UNIQUE_INDEX(pg_subscription_subname_index, 6115, on pg_subscription using btree(subname name_ops));
 #define SubscriptionNameIndexId 6115
 
+DECLARE_UNIQUE_INDEX(pg_subscription_rel_oid_index, 6116, on pg_subscription_rel using btree(oid oid_ops));
+#define SubscriptionRelOidIndexId 6116
+
+DECLARE_UNIQUE_INDEX(pg_subscription_rel_map_index, 6117, on pg_subscription_rel using btree(relid oid_ops, subid oid_ops));
+#define SubscriptionRelMapIndexId 6117
+
 /* last step of initialization script: build the indexes declared above */
 BUILD_INDICES
 
diff --git a/src/include/catalog/pg_subscription_rel.h b/src/include/catalog/pg_subscription_rel.h
new file mode 100644
index 0000000..300ba17
--- /dev/null
+++ b/src/include/catalog/pg_subscription_rel.h
@@ -0,0 +1,61 @@
+/* -------------------------------------------------------------------------
+ *
+ * pg_subscription_rel.h
+ *		Local info about tables that come from the provider of a
+ *		subscription (pg_subscription_rel).
+ *
+ * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * -------------------------------------------------------------------------
+ */
+#ifndef PG_SUBSCRIPTION_REL_H
+#define PG_SUBSCRIPTION_REL_H
+
+#include "catalog/genbki.h"
+
+/* ----------------
+ *		pg_subscription_rel definition. cpp turns this into
+ *		typedef struct FormData_pg_subscription_rel
+ * ----------------
+ */
+#define SubscriptionRelRelationId			6102
+#define SubscriptionRelRelation_Rowtype_Id	6103
+
+/* Workaround for genbki not knowing about XLogRecPtr */
+#define pg_lsn XLogRecPtr
+
+CATALOG(pg_subscription_rel,6102) BKI_ROWTYPE_OID(6103)
+{
+	Oid			subid;			/* Oid of subscription */
+	Oid			relid;			/* Oid of relation */
+	char		substate;		/* state of the relation in subscription */
+	pg_lsn		sublsn;			/* remote lsn of the state change
+								 * used for synchronization coordination */
+} FormData_pg_subscription_rel;
+
+typedef FormData_pg_subscription_rel *Form_pg_subscription_rel;
+
+/* ----------------
+ *		compiler constants for pg_subscription_rel
+ * ----------------
+ */
+#define Natts_pg_subscription_rel			4
+#define Anum_pg_subscription_rel_subid		1
+#define Anum_pg_subscription_rel_subrelid	2
+#define Anum_pg_subscription_rel_substate	3
+#define Anum_pg_subscription_rel_sublsn		4
+
+/* ----------------
+ *		substate constants
+ * ----------------
+ */
+#define		  SUBREL_STATE_UNKNOWN			'\0'	/* unknown state (sublsn NULL) */
+#define		  SUBREL_STATE_INIT				'i'		/* initializing (sublsn NULL) */
+#define		  SUBREL_STATE_DATA				'd'		/* data copy (sublsn NULL) */
+#define		  SUBREL_STATE_SYNCWAIT			'w'		/* waiting for sync (sublsn set) */
+#define		  SUBREL_STATE_CATCHUP			'c'		/* catchup (sublsn set) */
+#define		  SUBREL_STATE_SYNCDONE			's'		/* synced (sublsn set) */
+#define		  SUBREL_STATE_READY			'r'		/* ready (sublsn NULL) */
+
+#endif   /* PG_SUBSCRIPTION_REL_H */
diff --git a/src/include/commands/replicationcmds.h b/src/include/commands/replicationcmds.h
index 7c35d72..a9895f6 100644
--- a/src/include/commands/replicationcmds.h
+++ b/src/include/commands/replicationcmds.h
@@ -26,5 +26,6 @@ extern void RemovePublicationRelById(Oid prid);
 extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt);
 extern ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt);
 extern void DropSubscriptionById(Oid subid);
+extern void DropSubscriptionRelById(Oid subrelid);
 
 #endif   /* REPLICATIONCMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 322286b..21128a1 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -466,6 +466,8 @@ typedef enum NodeTag
 	T_DropReplicationSlotCmd,
 	T_StartReplicationCmd,
 	T_TimeLineHistoryCmd,
+	T_ListTablesCmd,
+	T_CopyTableCmd,
 
 	/*
 	 * TAGS FOR RANDOM OTHER STUFF
@@ -475,7 +477,7 @@ typedef enum NodeTag
 	 * purposes (usually because they are involved in APIs where we want to
 	 * pass multiple object types through the same pointer).
 	 */
-	T_TriggerData = 950,		/* in commands/trigger.h */
+	T_TriggerData = 970,		/* in commands/trigger.h */
 	T_EventTriggerData,			/* in commands/event_trigger.h */
 	T_ReturnSetInfo,			/* in nodes/execnodes.h */
 	T_WindowObjectData,			/* private in nodeWindowAgg.c */
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index d2f1edb..b8180c1 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -95,4 +95,27 @@ typedef struct TimeLineHistoryCmd
 	TimeLineID	timeline;
 } TimeLineHistoryCmd;
 
+/* ----------------------
+ *		LIST_TABLES command
+ * ----------------------
+ */
+typedef struct ListTablesCmd
+{
+	NodeTag		type;
+	char	   *slotname;
+	List	   *options;
+} ListTablesCmd;
+
+/* ----------------------
+ *		COPY_TABLE command
+ * ----------------------
+ */
+typedef struct CopyTableCmd
+{
+	NodeTag		type;
+	char	   *slotname;
+	struct RangeVar   *relation;
+	List	   *options;
+} CopyTableCmd;
+
 #endif   /* REPLNODES_H */
diff --git a/src/include/replication/logical.h b/src/include/replication/logical.h
index 947000e..a356b43 100644
--- a/src/include/replication/logical.h
+++ b/src/include/replication/logical.h
@@ -31,9 +31,11 @@ typedef struct LogicalDecodingContext
 	/* memory context this is all allocated in */
 	MemoryContext context;
 
-	/* infrastructure pieces */
-	XLogReaderState *reader;
+	/* The associated replication slot */
 	ReplicationSlot *slot;
+
+	/* infrastructure pieces for decoding */
+	XLogReaderState *reader;
 	struct ReorderBuffer *reorder;
 	struct SnapBuild *snapshot_builder;
 
@@ -75,6 +77,14 @@ typedef struct LogicalDecodingContext
 	TransactionId write_xid;
 } LogicalDecodingContext;
 
+
+/* Entry used for listing tables by logical decoding plugin. */
+typedef struct LogicalRepTableListEntry {
+	char *nspname;
+	char *relname;
+	char *info;
+} LogicalRepTableListEntry;
+
 extern void CheckLogicalDecodingRequirements(void);
 
 extern LogicalDecodingContext *CreateInitDecodingContext(char *plugin,
@@ -92,6 +102,14 @@ extern void DecodingContextFindStartpoint(LogicalDecodingContext *ctx);
 extern bool DecodingContextReady(LogicalDecodingContext *ctx);
 extern void FreeDecodingContext(LogicalDecodingContext *ctx);
 
+extern LogicalDecodingContext *CreateCopyDecodingContext(
+					  List *output_plugin_options,
+					  LogicalOutputPluginWriterPrepareWrite prepare_write,
+					  LogicalOutputPluginWriterWrite do_write);
+extern void DecodingContextProccessTuple(LogicalDecodingContext *ctx,
+							 Relation rel, HeapTuple tup);
+extern List *DecodingContextGetTableList(LogicalDecodingContext *ctx);
+
 extern void LogicalIncreaseXminForSlot(XLogRecPtr lsn, TransactionId xmin);
 extern void LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn,
 									  XLogRecPtr restart_lsn);
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index b69d015..2491cc7 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -69,8 +69,9 @@ extern LogicalRepRelId logicalrep_read_update(StringInfo in, bool *hasoldtup,
 extern void logicalrep_write_delete(StringInfo out, Relation rel,
 							 HeapTuple oldtuple);
 extern LogicalRepRelId logicalrep_read_delete(StringInfo in, LogicalRepTupleData *oldtup);
+extern void logicalrep_write_rel_name(StringInfo out, char *nspname, char *relname);
 extern void logicalrep_write_rel(StringInfo out, Relation rel);
-
+extern void logicalrep_read_rel_name(StringInfo in, char **nspname, char **relname);
 extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
 
 #endif /* LOGICALREP_PROTO_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index 64f36d3..6327067 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -22,20 +22,27 @@ typedef struct LogicalRepWorker
 
 	/* Subscription id for the worker. */
 	Oid		subid;
+
+	/* Used for initial table synchronization. */
+	Oid		relid;
 } 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 ApplyLauncherWakeup(void);
+extern void ApplyLauncherWakeupOnCommit(void);
 
 extern void ApplyLauncherWakeupOnCommit(void);
 extern void ApplyLauncherWakeup(void);
 
 extern void logicalrep_worker_attach(int slot);
+extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid);
+extern int logicalrep_worker_count(Oid subid);
+extern void logicalrep_worker_launch(Oid dbid, Oid subid, Oid relid);
 
 #endif   /* LOGICALWORKER_H */
diff --git a/src/include/replication/output_plugin.h b/src/include/replication/output_plugin.h
index 7911cc0..04ab611 100644
--- a/src/include/replication/output_plugin.h
+++ b/src/include/replication/output_plugin.h
@@ -93,6 +93,22 @@ typedef bool (*LogicalDecodeFilterByOriginCB) (
 													  RepOriginId origin_id);
 
 /*
+ * Called from the LIST_TABLES replication command.
+ */
+typedef List *(*BaseCopyListTablesCB) (
+											  struct LogicalDecodingContext *
+);
+
+/*
+ * Called for every individual tuple in a table during COPY_TABLE.
+ */
+typedef void (*BaseCopyTupleCB) (
+											  struct LogicalDecodingContext *,
+											  Relation relation,
+											  HeapTuple tup
+);
+
+/*
  * Called to shutdown an output plugin.
  */
 typedef void (*LogicalDecodeShutdownCB) (
@@ -111,6 +127,8 @@ typedef struct OutputPluginCallbacks
 	LogicalDecodeMessageCB message_cb;
 	LogicalDecodeFilterByOriginCB filter_by_origin_cb;
 	LogicalDecodeShutdownCB shutdown_cb;
+	BaseCopyListTablesCB list_tables_cb;
+	BaseCopyTupleCB tuple_cb;
 } OutputPluginCallbacks;
 
 void		OutputPluginPrepareWrite(struct LogicalDecodingContext *ctx, bool last_write);
diff --git a/src/include/replication/publication.h b/src/include/replication/publication.h
index 08245ee..78b4eb4 100644
--- a/src/include/replication/publication.h
+++ b/src/include/replication/publication.h
@@ -35,6 +35,7 @@ typedef struct Publication
 extern Publication *GetPublication(Oid pubid);
 extern Publication *GetPublicationByName(const char *pubname, bool missing_ok);
 extern List *GetRelationPublications(Relation rel);
+extern List * GetPublicationRelations(Oid pubid);
 
 extern bool publication_change_is_replicated(Relation rel,
 								 PublicationChangeType change_type,
diff --git a/src/include/replication/subscription.h b/src/include/replication/subscription.h
index a937f4b..18504da 100644
--- a/src/include/replication/subscription.h
+++ b/src/include/replication/subscription.h
@@ -30,4 +30,10 @@ typedef struct Subscription
 extern Subscription *GetSubscription(Oid subid);
 extern Oid get_subscription_oid(const char *subname, bool missing_ok);
 
+extern Oid SetSubscriptionRelState(Oid subid, Oid relid, char state,
+								   XLogRecPtr sublsn);
+extern char GetSubscriptionRelState(Oid subid, Oid relid,
+									XLogRecPtr *sublsn, bool missing_ok);
+extern void DropSubscriptionRelById(Oid subrelid);
+
 #endif		/* SUBSCRIPTION_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 3801949..d99cd72 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -169,6 +169,13 @@ typedef int (*walrcvconn_receive_fn) (WalReceiverConnHandle *handle,
 									  char **buffer, pgsocket *wait_fd);
 typedef void (*walrcvconn_send_fn) (WalReceiverConnHandle *handle,
 									const char *buffer, int nbytes);
+typedef List *(*walrcvconn_list_tables_fn) (
+									WalReceiverConnHandle *handle,
+									char *slotname, char *options);
+typedef bool (*walrcvconn_copy_table_fn) (
+									WalReceiverConnHandle *handle,
+									char *slotname, char *nspname,
+									char *relname, char *options);
 typedef void (*walrcvconn_disconnect_fn) (WalReceiverConnHandle *handle);
 
 typedef struct WalReceiverConnAPI {
@@ -183,6 +190,8 @@ typedef struct WalReceiverConnAPI {
 	walrcvconn_endstreaming_fn				endstreaming;
 	walrcvconn_receive_fn					receive;
 	walrcvconn_send_fn						send;
+	walrcvconn_list_tables_fn				list_tables;
+	walrcvconn_copy_table_fn				copy_table;
 	walrcvconn_disconnect_fn				disconnect;
 } WalReceiverConnAPI;
 
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
new file mode 100644
index 0000000..bd4402b
--- /dev/null
+++ b/src/include/replication/worker_internal.h
@@ -0,0 +1,32 @@
+/*-------------------------------------------------------------------------
+ *
+ * worker_internal.h
+ *	  Internal headers shared by logical replication workers.
+ *
+ * Portions Copyright (c) 2010-2016, PostgreSQL Global Development Group
+ *
+ * src/include/replication/worker_internal.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef WORKER_INTERNAL_H
+#define WORKER_INTERNAL_H
+
+/* filled by libpqreceiver when loaded */
+extern struct WalReceiverConnAPI	   *wrcapi;
+extern struct WalReceiverConnHandle    *wrchandle;
+
+/* Worker and subscription objects. */
+extern Subscription		   *MySubscription;
+extern LogicalRepWorker	   *MyLogicalRepWorker;
+
+extern bool					in_remote_transaction;
+
+extern void LogicalRepApplyLoop(XLogRecPtr last_received);
+extern char *LogicalRepSyncTableStart(XLogRecPtr *origin_startpos);
+void process_syncing_tables(char *slotname, XLogRecPtr end_lsn);
+void invalidate_syncing_table_states(Datum arg, int cacheid,
+									 uint32 hashvalue);
+
+
+#endif   /* WORKER_INTERNAL_H */
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index b1d03a5..5349c81 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -87,6 +87,8 @@ enum SysCacheIdentifier
 	STATRELATTINH,
 	SUBSCRIPTIONOID,
 	SUBSCRIPTIONNAME,
+	SUBSCRIPTIONRELOID,
+	SUBSCRIPTIONRELMAP,
 	TABLESPACEOID,
 	TRFOID,
 	TRFTYPELANG,
diff --git a/src/test/Makefile b/src/test/Makefile
index 7f7754f..8e8527a 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -12,7 +12,7 @@ subdir = src/test
 top_builddir = ../..
 include $(top_builddir)/src/Makefile.global
 
-SUBDIRS = regress isolation modules recovery
+SUBDIRS = regress isolation modules recovery subscription
 
 # We don't build or execute examples/, locale/, or thread/ by default,
 # but we do want "make clean" etc to recurse into them.  Likewise for ssl/,
diff --git a/src/test/README b/src/test/README
index 62395e7..74bab09 100644
--- a/src/test/README
+++ b/src/test/README
@@ -37,5 +37,8 @@ regress/
 ssl/
   Tests to exercise and verify SSL certificate handling
 
+subscription/
+  Test suite for subscriptions and logical replication
+
 thread/
   A thread-safety-testing utility used by configure
diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out
index ceac2c8..9e402f6 100644
--- a/src/test/regress/expected/sanity_check.out
+++ b/src/test/regress/expected/sanity_check.out
@@ -132,6 +132,7 @@ pg_shdescription|t
 pg_shseclabel|t
 pg_statistic|t
 pg_subscription|t
+pg_subscription_rel|t
 pg_tablespace|t
 pg_transform|t
 pg_trigger|t
diff --git a/src/test/subscription/t/001_rep_changes.pl b/src/test/subscription/t/001_rep_changes.pl
index dca19c4..f8121d9 100644
--- a/src/test/subscription/t/001_rep_changes.pl
+++ b/src/test/subscription/t/001_rep_changes.pl
@@ -3,7 +3,7 @@ use strict;
 use warnings;
 use PostgresNode;
 use TestLib;
-use Test::More tests => 3;
+use Test::More tests => 4;
 
 # Initialize provider node
 my $node_provider = get_new_node('provider');
@@ -19,7 +19,7 @@ $node_subscriber->start;
 $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)");
+	"CREATE TABLE tab_ins AS SELECT generate_series(1,1002) AS a");
 $node_provider->safe_psql('postgres',
 	"CREATE TABLE tab_rep (a int primary key)");
 
@@ -45,18 +45,28 @@ $node_provider->safe_psql('postgres',
 $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
+# Wait for subscriber to finish initialization
 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";
 
+# Also wait for initial table sync to finish
+my $synced_query =
+"SELECT count(1) = 0 FROM pg_subscription_rel WHERE substate != 'r';";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
 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');
 
+$result =
+  $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_ins");
+print "node_subscriber: $result\n";
+is($result, qq(1002), 'check initial data was copied to subscriber');
 
 $node_provider->safe_psql('postgres',
 	"INSERT INTO tab_ins SELECT generate_series(1,50)");
@@ -78,7 +88,7 @@ $node_provider->poll_query_until('postgres', $caughtup_query)
 $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');
+is($result, qq(1052|1|1002), 'check replicated inserts on subscriber');
 
 $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(*), min(a), max(a) FROM tab_rep");
diff --git a/src/test/subscription/t/002_types.pl b/src/test/subscription/t/002_types.pl
index a126201..a9c8526 100644
--- a/src/test/subscription/t/002_types.pl
+++ b/src/test/subscription/t/002_types.pl
@@ -101,13 +101,19 @@ $node_provider->safe_psql('postgres',
 $node_subscriber->safe_psql('postgres',
 	"CREATE SUBSCRIPTION tap_sub WITH CONNECTION '$provider_connstr' PUBLICATION tap_pub");
 
-# Wait for subscriber to finish table sync
+# Wait for subscriber to finish initialization
 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";
 
+# Wait for initial sync to finish as well
+my $synced_query =
+"SELECT count(1) = 0 FROM pg_subscription_rel WHERE substate != 'r';";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+  or die "Timed out while waiting for subscriber to synchronize data";
+
 # Insert initial test data
 $node_provider->safe_psql('postgres', qq(
 	-- test_tbl_one_array_col
-- 
2.7.4

