From ef8ed0bed60f2cf5123a2cdc1335d0d02891d82c Mon Sep 17 00:00:00 2001
From: Petr Jelinek <pjmodos@pjmodos.net>
Date: Wed, 13 Jul 2016 18:12:05 +0200
Subject: [PATCH 2/6] Add SUBSCRIPTION catalog and DDL

---
 doc/src/sgml/catalogs.sgml                     |  90 +++++++
 doc/src/sgml/ref/allfiles.sgml                 |   3 +
 doc/src/sgml/ref/alter_subscription.sgml       | 135 ++++++++++
 doc/src/sgml/ref/create_subscription.sgml      | 159 ++++++++++++
 doc/src/sgml/ref/drop_subscription.sgml        | 101 ++++++++
 src/backend/catalog/Makefile                   |   2 +-
 src/backend/catalog/catalog.c                  |   8 +-
 src/backend/catalog/dependency.c               |   9 +
 src/backend/catalog/objectaddress.c            |  59 +++++
 src/backend/commands/Makefile                  |   5 +-
 src/backend/commands/event_trigger.c           |   3 +
 src/backend/commands/subscriptioncmds.c        | 331 +++++++++++++++++++++++++
 src/backend/nodes/copyfuncs.c                  |  28 +++
 src/backend/nodes/equalfuncs.c                 |  26 ++
 src/backend/parser/gram.y                      | 127 +++++++++-
 src/backend/replication/logical/Makefile       |   2 +-
 src/backend/replication/logical/subscription.c | 146 +++++++++++
 src/backend/tcop/utility.c                     |  32 +++
 src/backend/utils/cache/relcache.c             |   6 +-
 src/backend/utils/cache/syscache.c             |  23 ++
 src/bin/psql/command.c                         |   3 +
 src/bin/psql/describe.c                        |  66 +++++
 src/bin/psql/describe.h                        |   3 +
 src/bin/psql/help.c                            |   1 +
 src/include/catalog/dependency.h               |   1 +
 src/include/catalog/indexing.h                 |   6 +
 src/include/catalog/pg_subscription.h          |  52 ++++
 src/include/commands/replicationcmds.h         |   6 +-
 src/include/nodes/nodes.h                      |   2 +
 src/include/nodes/parsenodes.h                 |  15 ++
 src/include/parser/kwlist.h                    |   1 +
 src/include/replication/subscription.h         |  33 +++
 src/include/utils/rel.h                        |   1 +
 src/include/utils/syscache.h                   |   2 +
 src/test/regress/expected/sanity_check.out     |   1 +
 35 files changed, 1478 insertions(+), 10 deletions(-)
 create mode 100644 doc/src/sgml/ref/alter_subscription.sgml
 create mode 100644 doc/src/sgml/ref/create_subscription.sgml
 create mode 100644 doc/src/sgml/ref/drop_subscription.sgml
 create mode 100644 src/backend/commands/subscriptioncmds.c
 create mode 100644 src/backend/replication/logical/subscription.c
 create mode 100644 src/include/catalog/pg_subscription.h
 create mode 100644 src/include/replication/subscription.h

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 6d505ae..84211c1 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -286,6 +286,11 @@
      </row>
 
      <row>
+      <entry><link linkend="catalog-pg-subscription"><structname>pg_subscription</structname></link></entry>
+      <entry>logical replication subscriptions</entry>
+     </row>
+
+     <row>
       <entry><link linkend="catalog-pg-tablespace"><structname>pg_tablespace</structname></link></entry>
       <entry>tablespaces within this database cluster</entry>
      </row>
@@ -6037,6 +6042,91 @@
 
  </sect1>
 
+ <sect1 id="catalog-pg-subscription">
+  <title><structname>pg_subscription</structname></title>
+
+  <indexterm zone="catalog-pg-subscription">
+   <primary>pg_subscription</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_subscription</structname> catalog contains
+   all existing logical replication subscriptions.
+  </para>
+
+  <para>
+   Unlike most system catalogs, <structname>pg_subscription</structname>
+   is shared across all databases of a cluster: there is only one
+   copy of <structname>pg_subscription</structname> per cluster, not
+   one per database.
+  </para>
+
+  <table>
+
+   <title><structname>pg_subscription</structname> Columns</title>
+
+   <tgroup cols="4">
+    <thead>
+     <row>
+      <entry>Name</entry>
+      <entry>Type</entry>
+      <entry>References</entry>
+      <entry>Description</entry>
+     </row>
+    </thead>
+
+    <tbody>
+     <row>
+      <entry><structfield>oid</structfield></entry>
+      <entry><type>oid</type></entry>
+      <entry></entry>
+      <entry>Row identifier (hidden attribute; must be explicitly selected)</entry>
+     </row>
+
+     <row>
+      <entry><structfield>subname</structfield></entry>
+      <entry><type>Name</type></entry>
+      <entry></entry>
+      <entry>A unique, database-wide identifier for the replication
+      subscription.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>subenabled</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>If true, the subscription is enabled and should be replicating.
+      </entry>
+     </row>
+
+     <row>
+      <entry><structfield>subconninfo</structfield></entry>
+      <entry><type>text</type></entry>
+      <entry></entry>
+      <entry>Connection string to the upstream database.
+      </entry>
+     </row>
+
+     <row>
+      <entry><structfield>subslotname</structfield></entry>
+      <entry><type>text</type></entry>
+      <entry></entry>
+      <entry>Name of the replication slot in the upstream database. Also used
+       for local replication origin name.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>subpublications</structfield></entry>
+      <entry><type>text[]</type></entry>
+      <entry></entry>
+      <entry>Array of subscribed publication names. For more on publications
+       see <xref linkend="publications">.
+      </entry>
+     </row>
+    </tbody>
+   </tgroup>
+  </table>
+ </sect1>
 
  <sect1 id="catalog-pg-tablespace">
   <title><structname>pg_tablespace</structname></title>
diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml
index 371a7b7..0d09f81 100644
--- a/doc/src/sgml/ref/allfiles.sgml
+++ b/doc/src/sgml/ref/allfiles.sgml
@@ -32,6 +32,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY alterSchema        SYSTEM "alter_schema.sgml">
 <!ENTITY alterServer        SYSTEM "alter_server.sgml">
 <!ENTITY alterSequence      SYSTEM "alter_sequence.sgml">
+<!ENTITY alterSubscription  SYSTEM "alter_subscription.sgml">
 <!ENTITY alterSystem        SYSTEM "alter_system.sgml">
 <!ENTITY alterTable         SYSTEM "alter_table.sgml">
 <!ENTITY alterTableSpace    SYSTEM "alter_tablespace.sgml">
@@ -79,6 +80,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY createSchema       SYSTEM "create_schema.sgml">
 <!ENTITY createSequence     SYSTEM "create_sequence.sgml">
 <!ENTITY createServer       SYSTEM "create_server.sgml">
+<!ENTITY createSubscription SYSTEM "create_subscription.sgml">
 <!ENTITY createTable        SYSTEM "create_table.sgml">
 <!ENTITY createTableAs      SYSTEM "create_table_as.sgml">
 <!ENTITY createTableSpace   SYSTEM "create_tablespace.sgml">
@@ -124,6 +126,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY dropSchema         SYSTEM "drop_schema.sgml">
 <!ENTITY dropSequence       SYSTEM "drop_sequence.sgml">
 <!ENTITY dropServer         SYSTEM "drop_server.sgml">
+<!ENTITY dropSubscription   SYSTEM "drop_subscription.sgml">
 <!ENTITY dropTable          SYSTEM "drop_table.sgml">
 <!ENTITY dropTableSpace     SYSTEM "drop_tablespace.sgml">
 <!ENTITY dropTransform      SYSTEM "drop_transform.sgml">
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
new file mode 100644
index 0000000..467e71f
--- /dev/null
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -0,0 +1,135 @@
+<!--
+doc/src/sgml/ref/alter_subscription.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-ALTERSUBSCRIPTION">
+ <indexterm zone="sql-altersubscription">
+  <primary>ALTER SUBSCRIPTION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>ALTER SUBSCRIPTION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>ALTER SUBSCRIPTION</refname>
+  <refpurpose>change the definition of a subscription</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+ALTER SUBSCRIPTION <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replaceable class="PARAMETER">option</replaceable> [ ... ] ]
+
+<phrase>where <replaceable class="PARAMETER">option</replaceable> can be:</phrase>
+
+    CONNECTION 'conninfo'
+    | PUBLICATION publication_name [, ...]
+
+ALTER SUBSCRIPTION <replaceable class="PARAMETER">name</replaceable> ENABLE
+ALTER SUBSCRIPTION <replaceable class="PARAMETER">name</replaceable> DISABLE
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>ALTER SUBSCRIPTION</command> can change most of the subscription
+   attributes that can be specified in
+   <xref linkend="sql-createsubscription">.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name of a subscription whose attributes are to be altered.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>CONNECTION '<replaceable class="parameter">conninfo</replaceable>'</term>
+    <term>PUBLICATION <replaceable class="parameter">publication_name</replaceable></term>
+    <listitem>
+     <para>
+      These clauses alter attributes originally set by
+      <xref linkend="SQL-CREATESUBSCRIPTION">. For more information, see the
+      <command>CREATE SUBSCRIPTION</command> reference page.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>ENABLE</term>
+    <listitem>
+     <para>
+      Enables the previously disabled subscription, starting the logical
+      replication worker at the end of transaction.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>DISABLE</term>
+    <listitem>
+     <para>
+      Disables the running subscription, stopping the logical replication
+      worker at the end of transaction.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Change the publication subscribed by a subscription to
+   <literal>insert_only</literal>:
+<programlisting>
+ALTER SUBSCRIPTION mysub
+        PUBLICATION insert_only;
+</programlisting>
+  </para>
+
+  <para>
+   Disable (stop) the subscription:
+<programlisting>
+ALTER SUBSCRIPTION mysub DISABLE;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>ALTER SUBSCRIPTION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-createsubscription"></member>
+   <member><xref linkend="sql-dropsubscription"></member>
+   <member><xref linkend="sql-createpublication"></member>
+   <member><xref linkend="sql-alterpublication"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
new file mode 100644
index 0000000..a2cb459
--- /dev/null
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -0,0 +1,159 @@
+<!--
+doc/src/sgml/ref/create_subscription.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-CREATESUBSCRIPTION">
+ <indexterm zone="sql-altersubscription">
+  <primary>CREATE SUBSCRIPTION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>CREATE SUBSCRIPTION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>CREATE SUBSCRIPTION</refname>
+  <refpurpose>define new subscription</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+CREATE SUBSCRIPTION <replaceable class="PARAMETER">subscription_name</replaceable> [ [ WITH ] <replaceable class="PARAMETER">option</replaceable> [ ... ] ]
+
+<phrase>where <replaceable class="PARAMETER">option</replaceable> can be:</phrase>
+
+    CONNECTION 'conninfo'
+    | PUBLICATION publication_name [, ...]
+    | INITIALLY ( ENABLED | DISABLED )
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>CREATE SUBSCRIPTION</command> adds a new subscription for
+   a current database. The subscription name must be distinct from
+   the name of any existing subscription in the database cluster.
+  </para>
+
+  <para>
+   The subscription represents a replication connection to the provider.
+   As such this command does not only add definition in the local catalogs
+   but also creates a replication slot on the provider.
+  </para>
+
+  <para>
+   A logical replication worker will be started to replicate data for the
+   new subscription at the commit of the transaction where this command
+   was run.
+  </para>
+
+  <para>
+   Additional info about subscriptions and logical replication as a whole
+   can is available at <xref linkend="logical-replication-subscription"> and
+   <xref linkend="logical-replication">.
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">subscription_name</replaceable></term>
+    <listitem>
+     <para>
+      The name of the new subscription.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>CONNECTION '<replaceable class="parameter">conninfo</replaceable>'</term>
+    <listitem>
+     <para>
+      The connection string to the provider.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>PUBLICATION <replaceable class="parameter">publication_name</replaceable></term>
+    <listitem>
+     <para>
+      Name(s) of the publications on the provider to subscribe to.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term>INITIALLY ENABLED</term>
+    <term>INITIALLY DISABLED</term>
+    <listitem>
+     <para>
+      Specifies if the subscription should be actively replicating or
+      if it should be just setup but not started yet.  Note that the
+      replication slot as described above is created in either case.
+      <literal>INITIALLY ENABLED</literal> is the default.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Create a subscription to a different server which replicates tables in
+   the publications <literal>mypubclication</literal> and
+   <literal>insert_only</literal> and starts replicating immediately on
+   commit:
+<programlisting>
+CREATE SUBSCRIPTION mysub
+         CONNECTION 'host=192.168.1.50 port=5432 user=foo dbname=foodb password=foopass'
+        PUBLICATION mypublication, insert_only;
+</programlisting>
+  </para>
+
+  <para>
+   Create a subscription to a different server which replicates tables in
+   the <literal>insert_only</literal> publication and does not replicate
+   until enabled at a later time.
+<programlisting>
+CREATE SUBSCRIPTION mysub
+         CONNECTION 'host=192.168.1.50 port=5432 user=foo dbname=foodb password=foopass'
+        PUBLICATION insert_only
+ INITIALLY DISABLED;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>CREATE SUBSCRIPTION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-altersubscription"></member>
+   <member><xref linkend="sql-dropsubscription"></member>
+   <member><xref linkend="sql-createpublication"></member>
+   <member><xref linkend="sql-alterpublication"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/ref/drop_subscription.sgml b/doc/src/sgml/ref/drop_subscription.sgml
new file mode 100644
index 0000000..38fb4b0
--- /dev/null
+++ b/doc/src/sgml/ref/drop_subscription.sgml
@@ -0,0 +1,101 @@
+<!--
+doc/src/sgml/ref/drop_subscription.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-DROPSUBSCRIPTION">
+ <indexterm zone="sql-dropsubscription">
+  <primary>DROP SUBSCRIPTION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>DROP SUBSCRIPTION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>DROP SUBSCRIPTION</refname>
+  <refpurpose>remove an existing subscription</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+DROP SUBSCRIPTION <replaceable class="PARAMETER">name</replaceable> [, ...]
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>DROP SUBSCRIPTION</command> removes subscriptions from the
+   cluster.
+  </para>
+
+  <para>
+   This command cannot be performed inside a transaction block.
+  </para>
+
+  <para>
+   A subscription can only be dropped by its owner or a superuser.
+  </para>
+
+  <warning>
+   <para>
+    While the <command>DROP SUBSCRIPTION</command> will try to remove the
+    replication slot on the provider, it will not fail when said action
+    is not successfull. This means the replication slot may need to be
+    dropped manually to not hold back removal of WAL on provider.
+   </para>
+  </warning>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name of a subscription to be dropped.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Drop a subscription:
+<programlisting>
+DROP SUBSCRIPTION mysub;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>DROP SUBSCRIPTION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-createsubscription"></member>
+   <member><xref linkend="sql-altersubscription"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 6bb3683..60737d4 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -42,7 +42,7 @@ POSTGRES_BKI_SRCS = $(addprefix $(top_srcdir)/src/include/catalog/,\
 	pg_foreign_table.h pg_policy.h pg_replication_origin.h \
 	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_publication.h pg_publication_rel.h pg_subscription.h \
 	toasting.h indexing.h \
     )
 
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index 1baaa0b..ba758bb 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -36,6 +36,7 @@
 #include "catalog/pg_shdepend.h"
 #include "catalog/pg_shdescription.h"
 #include "catalog/pg_shseclabel.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/toasting.h"
 #include "miscadmin.h"
@@ -227,7 +228,8 @@ IsSharedRelation(Oid relationId)
 		relationId == SharedSecLabelRelationId ||
 		relationId == TableSpaceRelationId ||
 		relationId == DbRoleSettingRelationId ||
-		relationId == ReplicationOriginRelationId)
+		relationId == ReplicationOriginRelationId ||
+		relationId == SubscriptionRelationId)
 		return true;
 	/* These are their indexes (see indexing.h) */
 	if (relationId == AuthIdRolnameIndexId ||
@@ -245,7 +247,9 @@ IsSharedRelation(Oid relationId)
 		relationId == TablespaceNameIndexId ||
 		relationId == DbRoleSettingDatidRolidIndexId ||
 		relationId == ReplicationOriginIdentIndex ||
-		relationId == ReplicationOriginNameIndex)
+		relationId == ReplicationOriginNameIndex ||
+		relationId == SubscriptionObjectIndexId ||
+		relationId == SubscriptionNameIndexId)
 		return true;
 	/* These are their toast tables and toast indexes (see toasting.h) */
 	if (relationId == PgShdescriptionToastTable ||
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 359e7bb..91564b2 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -51,6 +51,7 @@
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_trigger.h"
@@ -168,6 +169,7 @@ static const Oid object_classes[] = {
 	PolicyRelationId,			/* OCLASS_POLICY */
 	PublicationRelationId,		/* OCLASS_PUBCLICATION */
 	PublicationRelRelationId,	/* OCLASS_PUBCLICATION_REL */
+	SubscriptionRelationId,		/* OCLASS_SUBSCRIPTION */
 	TransformRelationId			/* OCLASS_TRANSFORM */
 };
 
@@ -1292,6 +1294,10 @@ doDeletion(const ObjectAddress *object, int flags)
 			RemovePublicationRelById(object->objectId);
 			break;
 
+		case OCLASS_SUBSCRIPTION:
+			DropSubscriptionById(object->objectId);
+			break;
+
 		case OCLASS_TRANSFORM:
 			DropTransformById(object->objectId);
 			break;
@@ -2455,6 +2461,9 @@ getObjectClass(const ObjectAddress *object)
 		case PublicationRelRelationId:
 			return OCLASS_PUBLICATION_REL;
 
+		case SubscriptionRelationId:
+			return OCLASS_SUBSCRIPTION;
+
 		case TransformRelationId:
 			return OCLASS_TRANSFORM;
 	}
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 375f4b0..f25c69b 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -48,6 +48,7 @@
 #include "catalog/pg_publication.h"
 #include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_trigger.h"
@@ -74,6 +75,7 @@
 #include "parser/parse_oper.h"
 #include "parser/parse_type.h"
 #include "replication/publication.h"
+#include "replication/subscription.h"
 #include "rewrite/rewriteSupport.h"
 #include "storage/lmgr.h"
 #include "storage/sinval.h"
@@ -465,6 +467,18 @@ static const ObjectPropertyType ObjectProperty[] =
 		InvalidAttrNumber,
 		-1,
 		true
+	},
+	{
+		SubscriptionRelationId,
+		SubscriptionObjectIndexId,
+		SUBSCRIPTIONOID,
+		SUBSCRIPTIONNAME,
+		Anum_pg_subscription_subname,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		-1,
+		true
 	}
 };
 
@@ -676,6 +690,10 @@ static const struct object_type_map
 	{
 		"publication relation", OBJECT_PUBLICATION_REL
 	},
+	/* OCLASS_SUBSCRIPTION */
+	{
+		"subscription", OBJECT_SUBSCRIPTION
+	},
 	/* OCLASS_TRANSFORM */
 	{
 		"transform", OBJECT_TRANSFORM
@@ -839,6 +857,7 @@ get_object_address(ObjectType objtype, List *objname, List *objargs,
 			case OBJECT_EVENT_TRIGGER:
 			case OBJECT_ACCESS_METHOD:
 			case OBJECT_PUBLICATION:
+			case OBJECT_SUBSCRIPTION:
 				address = get_object_address_unqualified(objtype,
 														 objname, missing_ok);
 				break;
@@ -1125,6 +1144,9 @@ get_object_address_unqualified(ObjectType objtype,
 			case OBJECT_PUBLICATION:
 				msg = gettext_noop("publication name cannot be qualified");
 				break;
+			case OBJECT_SUBSCRIPTION:
+				msg = gettext_noop("subscription name cannot be qualified");
+				break;
 			default:
 				elog(ERROR, "unrecognized objtype: %d", (int) objtype);
 				msg = NULL;		/* placate compiler */
@@ -1195,6 +1217,11 @@ get_object_address_unqualified(ObjectType objtype,
 			address.objectId = get_publication_oid(name, missing_ok);
 			address.objectSubId = 0;
 			break;
+		case OBJECT_SUBSCRIPTION:
+			address.classId = SubscriptionRelationId;
+			address.objectId = get_subscription_oid(name, missing_ok);
+			address.objectSubId = 0;
+			break;
 		default:
 			elog(ERROR, "unrecognized objtype: %d", (int) objtype);
 			/* placate compiler, which doesn't know elog won't return */
@@ -2315,6 +2342,7 @@ check_object_ownership(Oid roleid, ObjectType objtype, ObjectAddress address,
 		case OBJECT_TSTEMPLATE:
 		case OBJECT_ACCESS_METHOD:
 		case OBJECT_PUBLICATION:
+		case OBJECT_SUBSCRIPTION:
 			/* We treat these object types as being owned by superusers */
 			if (!superuser_arg(roleid))
 				ereport(ERROR,
@@ -3316,6 +3344,21 @@ getObjectDescription(const ObjectAddress *object)
 				break;
 			}
 
+		case OCLASS_SUBSCRIPTION:
+			{
+				HeapTuple	tup;
+
+				tup = SearchSysCache1(SUBSCRIPTIONOID,
+									  ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(tup))
+					elog(ERROR, "cache lookup failed for subscription %u",
+						 object->objectId);
+				appendStringInfo(&buffer, _("subscription %s"),
+				   NameStr(((Form_pg_subscription) GETSTRUCT(tup))->subname));
+				ReleaseSysCache(tup);
+				break;
+			}
+
 		default:
 			appendStringInfo(&buffer, "unrecognized object %u %u %d",
 							 object->classId,
@@ -3809,6 +3852,10 @@ getObjectTypeDescription(const ObjectAddress *object)
 			appendStringInfoString(&buffer, "publication table");
 			break;
 
+		case OCLASS_SUBSCRIPTION:
+			appendStringInfoString(&buffer, "subscription");
+			break;
+
 		default:
 			appendStringInfo(&buffer, "unrecognized %u", object->classId);
 			break;
@@ -4812,6 +4859,18 @@ getObjectIdentityParts(const ObjectAddress *object,
 				break;
 			}
 
+		case OCLASS_SUBSCRIPTION:
+			{
+				Subscription *sub;
+
+				sub = GetSubscription(object->objectId);
+				appendStringInfoString(&buffer,
+									   quote_identifier(sub->name));
+				if (objname)
+					*objname = list_make1(pstrdup(sub->name));
+				break;
+			}
+
 		default:
 			appendStringInfo(&buffer, "unrecognized object %u %u %d",
 							 object->classId,
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index cb580377..e0fab38 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -18,7 +18,8 @@ OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \
 	event_trigger.o explain.o extension.o foreigncmds.o functioncmds.o \
 	indexcmds.o lockcmds.o matview.o operatorcmds.o opclasscmds.o \
 	policy.o portalcmds.o prepare.o proclang.o publicationcmds.o \
-	schemacmds.o seclabel.o sequence.o tablecmds.o tablespace.o trigger.o \
-	tsearchcmds.o typecmds.o user.o vacuum.o vacuumlazy.o variable.o view.o
+	schemacmds.o seclabel.o sequence.o subscriptioncmds.o tablecmds.o \
+	tablespace.o trigger.o tsearchcmds.o typecmds.o user.o vacuum.o \
+	vacuumlazy.o variable.o view.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/commands/event_trigger.c b/src/backend/commands/event_trigger.c
index 8aaf1a7..2a7951d 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -123,6 +123,7 @@ static event_trigger_support_data event_trigger_support[] = {
 	{"USER MAPPING", true},
 	{"VIEW", true},
 	{"PUBLICATION", true},
+	{"SUBSCRIPTION", true},
 	{NULL, false}
 };
 
@@ -1122,6 +1123,7 @@ EventTriggerSupportsObjectType(ObjectType obtype)
 		case OBJECT_VIEW:
 		case OBJECT_PUBLICATION:
 		case OBJECT_PUBLICATION_REL:
+		case OBJECT_SUBSCRIPTION:
 			return true;
 	}
 	return true;
@@ -1175,6 +1177,7 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_AM:
 		case OCLASS_PUBLICATION:
 		case OCLASS_PUBLICATION_REL:
+		case OCLASS_SUBSCRIPTION:
 			return true;
 	}
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
new file mode 100644
index 0000000..54d66d5
--- /dev/null
+++ b/src/backend/commands/subscriptioncmds.c
@@ -0,0 +1,331 @@
+/*-------------------------------------------------------------------------
+ *
+ * subscriptioncmds.c
+ *		subscription catalog manipulation functions
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		subscriptioncmds.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "funcapi.h"
+#include "miscadmin.h"
+
+#include "access/genam.h"
+#include "access/hash.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/xact.h"
+
+#include "catalog/indexing.h"
+#include "catalog/namespace.h"
+#include "catalog/objectaddress.h"
+#include "catalog/pg_type.h"
+#include "catalog/pg_subscription.h"
+
+#include "commands/defrem.h"
+
+#include "executor/spi.h"
+
+#include "nodes/makefuncs.h"
+
+#include "replication/reorderbuffer.h"
+#include "commands/replicationcmds.h"
+
+#include "utils/array.h"
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/fmgroids.h"
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+
+
+static void
+check_replication_permissions(void)
+{
+	if (!superuser() && !has_rolreplication(GetUserId()))
+		ereport(ERROR,
+				(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+				 (errmsg("must be superuser or replication role to manipulate subscriptions"))));
+}
+
+static void
+parse_subscription_options(List *options,
+						   bool *enabled_given, bool *enabled,
+						   char **conninfo, List **publications)
+{
+	ListCell   *lc;
+
+	*enabled_given = false;
+	*enabled = true;
+	*conninfo = NULL;
+	*publications = NIL;
+
+	/* Parse options */
+	foreach (lc, options)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+
+		if (strcmp(defel->defname, "enabled") == 0)
+		{
+			if (*enabled_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			*enabled_given = true;
+			*enabled = defGetBoolean(defel);
+		}
+		else if (strcmp(defel->defname, "conninfo") == 0)
+		{
+			if (*conninfo)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			*conninfo = defGetString(defel);
+		}
+		else if (strcmp(defel->defname, "publication") == 0)
+		{
+			if (*publications)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			if (defel->arg == NULL)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("%s requires a parameter", defel->defname)));
+
+			*publications = (List *) defel->arg;
+		}
+		else
+			elog(ERROR, "unrecognized option: %s", defel->defname);
+	}
+}
+
+/*
+ * Auxiliary function to return a TEXT array out of a list of String nodes.
+ */
+static Datum
+publicationListToArray(List *list)
+{
+	ArrayType  *arr;
+	Datum	   *datums;
+	int			j = 0;
+	ListCell   *cell;
+	MemoryContext memcxt;
+	MemoryContext oldcxt;
+
+	memcxt = AllocSetContextCreate(CurrentMemoryContext,
+								   "publicationListToArray to array",
+								   ALLOCSET_DEFAULT_MINSIZE,
+								   ALLOCSET_DEFAULT_INITSIZE,
+								   ALLOCSET_DEFAULT_MAXSIZE);
+	oldcxt = MemoryContextSwitchTo(memcxt);
+
+	datums = palloc(sizeof(text *) * list_length(list));
+	foreach(cell, list)
+	{
+		char	   *name = strVal(lfirst(cell));
+
+		datums[j++] = CStringGetTextDatum(name);
+	}
+
+	MemoryContextSwitchTo(oldcxt);
+
+	arr = construct_array(datums, list_length(list),
+						  TEXTOID, -1, false, 'i');
+	MemoryContextDelete(memcxt);
+
+	return PointerGetDatum(arr);
+}
+
+/*
+ * Create new subscription.
+ */
+ObjectAddress
+CreateSubscription(CreateSubscriptionStmt *stmt)
+{
+	Relation	rel;
+	ObjectAddress myself;
+	Oid			subid;
+	bool		nulls[Natts_pg_subscription];
+	Datum		values[Natts_pg_subscription];
+	HeapTuple	tup;
+	bool		enabled_given;
+	bool		enabled;
+	char	   *conninfo;
+	List	   *publications;
+
+	check_replication_permissions();
+
+	rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
+
+	/* Check if name is used */
+	subid = GetSysCacheOid1(SUBSCRIPTIONNAME,
+							CStringGetDatum(stmt->subname));
+	if (OidIsValid(subid))
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("subscription \"%s\" already exists",
+						stmt->subname)));
+	}
+
+	/* Parse and check options. */
+	parse_subscription_options(stmt->options, &enabled_given, &enabled,
+							   &conninfo, &publications);
+
+	/* TODO: improve error messages here. */
+	if (conninfo == NULL)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("connection not specified")));
+
+	if (list_length(publications) == 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_SYNTAX_ERROR),
+				 errmsg("publication not specified")));
+
+	/* Everything ok, form a new tuple. */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+
+	values[Anum_pg_subscription_dbid - 1] = ObjectIdGetDatum(MyDatabaseId);
+	values[Anum_pg_subscription_subname - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
+	values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
+	values[Anum_pg_subscription_subconninfo - 1] =
+		CStringGetTextDatum(conninfo);
+	values[Anum_pg_subscription_subslotname - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(stmt->subname));
+	values[Anum_pg_subscription_subpublications - 1] =
+		 publicationListToArray(publications);
+
+	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+
+	/* Insert tuple into catalog. */
+	subid = simple_heap_insert(rel, tup);
+	CatalogUpdateIndexes(rel, tup);
+	heap_freetuple(tup);
+
+	ObjectAddressSet(myself, SubscriptionRelationId, suboid);
+
+	heap_close(rel, RowExclusiveLock);
+
+	/* Make the changes visible. */
+	CommandCounterIncrement();
+
+	return myself;
+}
+
+/*
+ * Alter the existing subscription.
+ */
+ObjectAddress
+AlterSubscription(AlterSubscriptionStmt *stmt)
+{
+	Relation	rel;
+	ObjectAddress myself;
+	bool		nulls[Natts_pg_subscription];
+	bool		replaces[Natts_pg_subscription];
+	Datum		values[Natts_pg_subscription];
+	HeapTuple	tup;
+	Oid			subid;
+	bool		enabled_given;
+	bool		enabled;
+	char	   *conninfo;
+	List	   *publications;
+
+	check_replication_permissions();
+
+	rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
+
+	/* Fetch the existing tuple. */
+	tup = SearchSysCacheCopy1(SUBSCRIPTIONNAME,
+							  CStringGetDatum(stmt->subname));
+
+	if (!HeapTupleIsValid(tup))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("subscription \"%s\" does not exist",
+						stmt->subname)));
+
+	subid = HeapTupleGetOid(tup);
+
+	/* Parse options. */
+	parse_subscription_options(stmt->options, &enabled_given, &enabled,
+							   &conninfo, &publications);
+
+	/* Form a new tuple. */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+
+	if (enabled_given)
+	{
+		values[Anum_pg_subscription_subenabled - 1] = BoolGetDatum(enabled);
+		replaces[Anum_pg_subscription_subenabled - 1] = true;
+	}
+	if (conninfo)
+	{
+		values[Anum_pg_subscription_subconninfo - 1] =
+			CStringGetTextDatum(conninfo);
+		replaces[Anum_pg_subscription_subenabled - 1] = true;
+	}
+	if (publications != NIL)
+	{
+		values[Anum_pg_subscription_subpublications - 1] =
+			 publicationListToArray(publications);
+		replaces[Anum_pg_subscription_subpublications - 1] = true;
+	}
+
+	tup = heap_modify_tuple(tup, RelationGetDescr(rel), values, nulls,
+							replaces);
+
+	/* Update the catalog. */
+	simple_heap_update(rel, &tup->t_self, tup);
+	CatalogUpdateIndexes(rel, tup);
+
+	ObjectAddressSet(myself, SubscriptionRelationId, subid);
+
+	/* Cleanup. */
+	heap_freetuple(tup);
+	heap_close(rel, RowExclusiveLock);
+
+	return myself;
+}
+
+/*
+ * Drop subscription by OID
+ */
+void
+DropSubscriptionById(Oid subid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+
+	check_replication_permissions();
+
+	rel = heap_open(SubscriptionRelationId, RowExclusiveLock);
+
+	tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for subscription %u", subid);
+
+	simple_heap_delete(rel, &tup->t_self);
+
+	ReleaseSysCache(tup);
+
+	heap_close(rel, RowExclusiveLock);
+}
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index bf76742..3c39b7e 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4179,6 +4179,28 @@ _copyAlterPublicationStmt(const AlterPublicationStmt *from)
 	return newnode;
 }
 
+static CreateSubscriptionStmt *
+_copyCreateSubscriptionStmt(const CreateSubscriptionStmt *from)
+{
+	CreateSubscriptionStmt *newnode = makeNode(CreateSubscriptionStmt);
+
+	COPY_STRING_FIELD(subname);
+	COPY_NODE_FIELD(options);
+
+	return newnode;
+}
+
+static AlterSubscriptionStmt *
+_copyAlterSubscriptionStmt(const AlterSubscriptionStmt *from)
+{
+	AlterSubscriptionStmt *newnode = makeNode(AlterSubscriptionStmt);
+
+	COPY_STRING_FIELD(subname);
+	COPY_NODE_FIELD(options);
+
+	return newnode;
+}
+
 /* ****************************************************************
  *					pg_list.h copy functions
  * ****************************************************************
@@ -4976,6 +4998,12 @@ copyObject(const void *from)
 		case T_AlterPublicationStmt:
 			retval = _copyAlterPublicationStmt(from);
 			break;
+		case T_CreateSubscriptionStmt:
+			retval = _copyCreateSubscriptionStmt(from);
+			break;
+		case T_AlterSubscriptionStmt:
+			retval = _copyAlterSubscriptionStmt(from);
+			break;
 		case T_A_Expr:
 			retval = _copyAExpr(from);
 			break;
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index 0c5f1d0..2a5ed58 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2129,6 +2129,26 @@ _equalAlterPublicationStmt(const AlterPublicationStmt *a,
 }
 
 static bool
+_equalCreateSubscriptionStmt(const CreateSubscriptionStmt *a,
+							 const CreateSubscriptionStmt *b)
+{
+	COMPARE_STRING_FIELD(subname);
+	COMPARE_NODE_FIELD(options);
+
+	return true;
+}
+
+static bool
+_equalAlterSubscriptionStmt(const AlterSubscriptionStmt *a,
+							const AlterSubscriptionStmt *b)
+{
+	COMPARE_STRING_FIELD(subname);
+	COMPARE_NODE_FIELD(options);
+
+	return true;
+}
+
+static bool
 _equalCreatePolicyStmt(const CreatePolicyStmt *a, const CreatePolicyStmt *b)
 {
 	COMPARE_STRING_FIELD(policy_name);
@@ -3278,6 +3298,12 @@ equal(const void *a, const void *b)
 		case T_AlterPublicationStmt:
 			retval = _equalAlterPublicationStmt(a, b);
 			break;
+		case T_CreateSubscriptionStmt:
+			retval = _equalCreateSubscriptionStmt(a, b);
+			break;
+		case T_AlterSubscriptionStmt:
+			retval = _equalAlterSubscriptionStmt(a, b);
+			break;
 		case T_A_Expr:
 			retval = _equalAExpr(a, b);
 			break;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index b91e75a..03640eb 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -267,6 +267,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 		AlterTSConfigurationStmt AlterTSDictionaryStmt
 		CreateMatViewStmt RefreshMatViewStmt CreateAmStmt
 		CreatePublicationStmt AlterPublicationStmt
+		CreateSubscriptionStmt AlterSubscriptionStmt
 
 %type <node>	select_no_parens select_with_parens select_clause
 				simple_select values_clause
@@ -374,13 +375,17 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				relation_expr_list dostmt_opt_list
 				transform_element_list transform_type_list
 				publication_opt_list publication_opt_items
+				subscription_create_opt_items subscription_opt_items
 
 %type <list>	group_by_list
 %type <node>	group_by_item empty_grouping_set rollup_clause cube_clause
 %type <node>	grouping_sets_clause
 
 %type <list>	opt_fdw_options fdw_options
+				publication_list
 %type <defelt>	fdw_option publication_opt_item
+				subscription_opt_item subscription_create_opt_item
+%type <value>	publication_item
 
 %type <range>	OptTempTableName
 %type <into>	into_clause create_as_target create_mv_target
@@ -631,8 +636,8 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 	SAVEPOINT SCHEMA SCROLL SEARCH SECOND_P SECURITY SELECT SEQUENCE SEQUENCES
 	SERIALIZABLE SERVER SESSION SESSION_USER SET SETS SETOF SHARE SHOW
 	SIMILAR SIMPLE SKIP SMALLINT SNAPSHOT SOME SQL_P STABLE STANDALONE_P START
-	STATEMENT STATISTICS STDIN STDOUT STORAGE STRICT_P STRIP_P SUBSTRING
-	SYMMETRIC SYSID SYSTEM_P
+	STATEMENT STATISTICS STDIN STDOUT STORAGE STRICT_P STRIP_P SUBSCRIPTION
+	SUBSTRING SYMMETRIC SYSID SYSTEM_P
 
 	TABLE TABLES TABLESAMPLE TABLESPACE TEMP TEMPLATE TEMPORARY TEXT_P THEN
 	TIME TIMESTAMP TO TRAILING TRANSACTION TRANSFORM TREAT TRIGGER TRIM TRUE_P
@@ -783,6 +788,7 @@ stmt :
 			| AlterPublicationStmt
 			| AlterRoleSetStmt
 			| AlterRoleStmt
+			| AlterSubscriptionStmt
 			| AlterTSConfigurationStmt
 			| AlterTSDictionaryStmt
 			| AlterUserMappingStmt
@@ -817,6 +823,7 @@ stmt :
 			| CreateSchemaStmt
 			| CreateSeqStmt
 			| CreateStmt
+			| CreateSubscriptionStmt
 			| CreateTableSpaceStmt
 			| CreateTransformStmt
 			| CreateTrigStmt
@@ -5648,6 +5655,7 @@ drop_type:	TABLE									{ $$ = OBJECT_TABLE; }
 			| TEXT_P SEARCH TEMPLATE				{ $$ = OBJECT_TSTEMPLATE; }
 			| TEXT_P SEARCH CONFIGURATION			{ $$ = OBJECT_TSCONFIGURATION; }
 			| PUBLICATION							{ $$ = OBJECT_PUBLICATION; }
+			| SUBSCRIPTION							{ $$ = OBJECT_SUBSCRIPTION; }
 		;
 
 any_name_list:
@@ -8610,6 +8618,120 @@ AlterPublicationStmt:
 
 /*****************************************************************************
  *
+ * CREATE SUBSCRIPTION name [ WITH ] options
+ *
+ *****************************************************************************/
+
+CreateSubscriptionStmt:
+			CREATE SUBSCRIPTION name opt_with subscription_create_opt_items
+				{
+					CreateSubscriptionStmt *n =
+						makeNode(CreateSubscriptionStmt);
+					n->subname = $3;
+					n->options = $5;
+					$$ = (Node *)n;
+				}
+		;
+
+subscription_create_opt_items:
+			subscription_create_opt_item
+				{
+					$$ = list_make1($1);
+				}
+			| subscription_create_opt_items subscription_create_opt_item
+				{
+					$$ = lappend($1, $2);
+				}
+		;
+
+subscription_create_opt_item:
+			subscription_opt_item
+			| INITIALLY IDENT
+				{
+					if (strcmp($2, "enabled") == 0)
+						$$ = makeDefElem("enabled", (Node *)makeInteger(TRUE));
+					else if (strcmp($2, "disabled") == 0)
+						$$ = makeDefElem("enabled", (Node *)makeInteger(FALSE));
+					else
+						ereport(ERROR,
+								(errcode(ERRCODE_SYNTAX_ERROR),
+								 errmsg("unrecognized subscription option \"%s\"", $1),
+									 parser_errposition(@2)));
+				}
+		;
+
+subscription_opt_item:
+			CONNECTION Sconst
+				{
+					$$ = makeDefElem("conninfo", (Node *)makeString($2));
+				}
+			| PUBLICATION publication_list
+				{
+					$$ = makeDefElem("publication", (Node *)$2);
+				}
+		;
+
+publication_list:
+			publication_item
+				{
+					$$ = list_make1($1);
+				}
+			| publication_list ',' publication_item
+				{
+					$$ = lappend($1, $3);
+				}
+		;
+
+publication_item:
+			ColLabel			{ $$ = makeString($1); };
+
+/*****************************************************************************
+ *
+ * ALTER SUBSCRIPTION name [ WITH ] options
+ *
+ *****************************************************************************/
+
+AlterSubscriptionStmt:
+			ALTER SUBSCRIPTION name opt_with subscription_opt_items
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+					n->subname = $3;
+					n->options = $5;
+					$$ = (Node *)n;
+				}
+			| ALTER SUBSCRIPTION name ENABLE_P
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+					n->subname = $3;
+					n->options = list_make1(makeDefElem("enabled",
+												   (Node *)makeInteger(TRUE)));
+					$$ = (Node *)n;
+				}
+			| ALTER SUBSCRIPTION name DISABLE_P
+				{
+					AlterSubscriptionStmt *n =
+						makeNode(AlterSubscriptionStmt);
+					n->subname = $3;
+					n->options = list_make1(makeDefElem("enabled",
+												  (Node *)makeInteger(FALSE)));
+					$$ = (Node *)n;
+				}		;
+
+subscription_opt_items:
+			subscription_opt_item
+				{
+					$$ = list_make1($1);
+				}
+			| subscription_opt_items subscription_opt_item
+				{
+					$$ = lappend($1, $2);
+				}
+		;
+
+/*****************************************************************************
+ *
  *		QUERY:	Define Rewrite Rule
  *
  *****************************************************************************/
@@ -14066,6 +14188,7 @@ unreserved_keyword:
 			| STORAGE
 			| STRICT_P
 			| STRIP_P
+			| SUBSCRIPTION
 			| SYSID
 			| SYSTEM_P
 			| TABLES
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 3b3e90c..e4b093b 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -15,6 +15,6 @@ include $(top_builddir)/src/Makefile.global
 override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
 OBJS = decode.o logical.o logicalfuncs.o message.o origin.o publication.o \
-	   reorderbuffer.o snapbuild.o
+	   reorderbuffer.o snapbuild.o subscription.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/subscription.c b/src/backend/replication/logical/subscription.c
new file mode 100644
index 0000000..7d1de2c
--- /dev/null
+++ b/src/backend/replication/logical/subscription.c
@@ -0,0 +1,146 @@
+/*-------------------------------------------------------------------------
+ *
+ * subscription.c
+ *		replication subscriptions
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		subscription.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "funcapi.h"
+#include "miscadmin.h"
+
+#include "access/genam.h"
+#include "access/hash.h"
+#include "access/heapam.h"
+#include "access/htup_details.h"
+#include "access/xact.h"
+
+#include "catalog/indexing.h"
+#include "catalog/namespace.h"
+#include "catalog/objectaddress.h"
+#include "catalog/pg_type.h"
+#include "catalog/pg_subscription.h"
+
+#include "executor/spi.h"
+
+#include "nodes/makefuncs.h"
+
+#include "replication/reorderbuffer.h"
+#include "replication/subscription.h"
+
+#include "utils/array.h"
+#include "utils/builtins.h"
+#include "utils/catcache.h"
+#include "utils/fmgroids.h"
+#include "utils/inval.h"
+#include "utils/lsyscache.h"
+#include "utils/rel.h"
+#include "utils/syscache.h"
+
+
+static List *textarray_to_stringlist(ArrayType *textarray);
+
+/*
+ * Fetch the subscription from the syscache.
+ */
+Subscription *
+GetSubscription(Oid subid)
+{
+	HeapTuple		tup;
+	Subscription   *sub;
+	Form_pg_subscription	subform;
+	Datum			datum;
+	bool			isnull;
+
+	tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for subscription %u", subid);
+
+	subform = (Form_pg_subscription) GETSTRUCT(tup);
+
+	sub = (Subscription *) palloc(sizeof(Subscription));
+	sub->oid = subid;
+	sub->dbid = subform->dbid;
+	sub->name = NameStr(subform->subname);
+	sub->enabled = subform->subenabled;
+
+	/* Get conninfo */
+	datum = SysCacheGetAttr(SUBSCRIPTIONOID,
+							tup,
+							Anum_pg_subscription_subconninfo,
+							&isnull);
+	Assert(!isnull);
+	sub->conninfo = pstrdup(TextDatumGetCString(datum));
+
+	/* Get slotname */
+	datum = SysCacheGetAttr(SUBSCRIPTIONOID,
+							tup,
+							Anum_pg_subscription_subslotname,
+							&isnull);
+	Assert(!isnull);
+	sub->slotname = pstrdup(NameStr(*DatumGetName(datum)));
+
+	/* Get publications */
+	datum = SysCacheGetAttr(SUBSCRIPTIONOID,
+							tup,
+							Anum_pg_subscription_subpublications,
+							&isnull);
+	Assert(!isnull);
+	sub->publications = textarray_to_stringlist(DatumGetArrayTypeP(datum));
+
+	ReleaseSysCache(tup);
+
+	return sub;
+}
+
+/*
+ * get_subscription_oid - given a subscription name, look up the OID
+ *
+ * If missing_ok is false, throw an error if name not found.  If true, just
+ * return InvalidOid.
+ */
+Oid
+get_subscription_oid(const char *subname, bool missing_ok)
+{
+	Oid			oid;
+
+	oid = GetSysCacheOid1(SUBSCRIPTIONNAME, CStringGetDatum(subname));
+	if (!OidIsValid(oid) && !missing_ok)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("subscription \"%s\" does not exist", subname)));
+	return oid;
+}
+
+/*
+ * Convert text array to list of strings.
+ *
+ * Note: the resulting list of strings is pallocated here.
+ */
+static List *
+textarray_to_stringlist(ArrayType *textarray)
+{
+	Datum		   *elems;
+	int				nelems, i;
+	List		   *res = NIL;
+
+	deconstruct_array(textarray,
+					  TEXTOID, -1, false, 'i',
+					  &elems, NULL, &nelems);
+
+	if (nelems == 0)
+		return NIL;
+
+	for (i = 0; i < nelems; i++)
+		res = lappend(res, makeString(pstrdup(TextDatumGetCString(elems[i]))));
+
+	return res;
+}
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index 886d2ff..4cb2366 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -213,6 +213,8 @@ check_xact_readonly(Node *parsetree)
 		case T_SecLabelStmt:
 		case T_CreatePublicationStmt:
 		case T_AlterPublicationStmt:
+		case T_CreateSubscriptionStmt:
+		case T_AlterSubscriptionStmt:
 			PreventCommandIfReadOnly(CreateCommandTag(parsetree));
 			PreventCommandIfParallelMode(CreateCommandTag(parsetree));
 			break;
@@ -1560,6 +1562,14 @@ ProcessUtilitySlow(Node *parsetree,
 				commandCollected = true;
 				break;
 
+			case T_CreateSubscriptionStmt:
+				address = CreateSubscription((CreateSubscriptionStmt *) parsetree);
+				break;
+
+			case T_AlterSubscriptionStmt:
+				address = AlterSubscription((AlterSubscriptionStmt *) parsetree);
+				break;
+
 			default:
 				elog(ERROR, "unrecognized node type: %d",
 					 (int) nodeTag(parsetree));
@@ -1921,6 +1931,9 @@ AlterObjectTypeCommandTag(ObjectType objtype)
 		case OBJECT_PUBLICATION:
 			tag = "PUBLICATION";
 			break;
+		case OBJECT_SUBSCRIPTION:
+			tag = "SUBSCRIPTION";
+			break;
 		default:
 			tag = "???";
 			break;
@@ -2209,6 +2222,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_PUBLICATION:
 					tag = "DROP PUBLICATION";
 					break;
+				case OBJECT_SUBSCRIPTION:
+					tag = "DROP SUBSCRIPTION";
+					break;
 				default:
 					tag = "???";
 			}
@@ -2587,6 +2603,14 @@ CreateCommandTag(Node *parsetree)
 			tag = "ALTER PUBLICATION";
 			break;
 
+		case T_CreateSubscriptionStmt:
+			tag = "CREATE SUBSCRIPTION";
+			break;
+
+		case T_AlterSubscriptionStmt:
+			tag = "ALTER SUBSCRIPTION";
+			break;
+
 		case T_PrepareStmt:
 			tag = "PREPARE";
 			break;
@@ -3160,6 +3184,14 @@ GetCommandLogLevel(Node *parsetree)
 			lev = LOGSTMT_DDL;
 			break;
 
+		case T_CreateSubscriptionStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
+		case T_AlterSubscriptionStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
 			/* already-planned queries */
 		case T_PlannedStmt:
 			{
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 3f7027f..7c3121b 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -52,6 +52,7 @@
 #include "catalog/pg_proc.h"
 #include "catalog/pg_rewrite.h"
 #include "catalog/pg_shseclabel.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_trigger.h"
 #include "catalog/pg_type.h"
@@ -100,6 +101,7 @@ static const FormData_pg_attribute Desc_pg_authid[Natts_pg_authid] = {Schema_pg_
 static const FormData_pg_attribute Desc_pg_auth_members[Natts_pg_auth_members] = {Schema_pg_auth_members};
 static const FormData_pg_attribute Desc_pg_index[Natts_pg_index] = {Schema_pg_index};
 static const FormData_pg_attribute Desc_pg_shseclabel[Natts_pg_shseclabel] = {Schema_pg_shseclabel};
+static const FormData_pg_attribute Desc_pg_subscription[Natts_pg_subscription] = {Schema_pg_subscription};
 
 /*
  *		Hash tables that index the relation cache
@@ -3266,8 +3268,10 @@ RelationCacheInitializePhase2(void)
 				  false, Natts_pg_auth_members, Desc_pg_auth_members);
 		formrdesc("pg_shseclabel", SharedSecLabelRelation_Rowtype_Id, true,
 				  false, Natts_pg_shseclabel, Desc_pg_shseclabel);
+		formrdesc("pg_subscription", SubscriptionRelation_Rowtype_Id, true,
+				  true, Natts_pg_subscription, Desc_pg_subscription);
 
-#define NUM_CRITICAL_SHARED_RELS	4	/* fix if you change list above */
+#define NUM_CRITICAL_SHARED_RELS	5	/* fix if you change list above */
 	}
 
 	MemoryContextSwitchTo(oldcxt);
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index d575b51..03c8916 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -59,6 +59,7 @@
 #include "catalog/pg_shseclabel.h"
 #include "catalog/pg_replication_origin.h"
 #include "catalog/pg_statistic.h"
+#include "catalog/pg_subscription.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
 #include "catalog/pg_ts_config.h"
@@ -713,6 +714,28 @@ static const struct cachedesc cacheinfo[] = {
 		},
 		128
 	},
+	{SubscriptionRelationId,		/* SUBSCRIPTIONOID */
+		SubscriptionObjectIndexId,
+		1,
+		{
+			ObjectIdAttributeNumber,
+			0,
+			0,
+			0
+		},
+		4
+	},
+	{SubscriptionRelationId,		/* SUBSCRIPTIONNAME */
+		SubscriptionNameIndexId,
+		1,
+		{
+			Anum_pg_subscription_subname,
+			0,
+			0,
+			0
+		},
+		4
+	},
 	{TableSpaceRelationId,		/* TABLESPACEOID */
 		TablespaceOidIndexId,
 		1,
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index a379c19..225ea25 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -497,6 +497,9 @@ exec_command(const char *cmd,
 					case 'p':
 						success = describePublications(pattern, show_verbose);
 						break;
+					case 's':
+						success = describeSubscriptions(pattern, show_verbose);
+						break;
 					default:
 						status = PSQL_CMD_UNKNOWN;
 						break;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 573d980..823cda4 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -4784,6 +4784,72 @@ describePublications(const char *pattern, bool verbose)
 	return true;
 }
 
+/* \drs
+ * Describes subscriptions.
+ *
+ * Takes an optional regexp to select particular subscriptions
+ */
+bool
+describeSubscriptions(const char *pattern, bool verbose)
+{
+	PQExpBufferData buf;
+	PGresult   *res;
+	printQueryOpt myopt = pset.popt;
+	static const bool translate_columns[] = {false, true, false, false, false};
+
+	/* TODO bump */
+	if (pset.sversion < 90600)
+	{
+		psql_error("The server (version %d.%d) does not support subscriptions.\n",
+				   pset.sversion / 10000, (pset.sversion / 100) % 100);
+		return true;
+	}
+
+	initPQExpBuffer(&buf);
+
+	printfPQExpBuffer(&buf,
+					  "SELECT subname AS \"%s\",\n"
+					  "  (SELECT datname FROM pg_catalog.pg_database WHERE oid = dbid) AS \"%s\",\n"
+					  "  subenabled AS \"%s\",\n"
+					  "  subpublications AS \"%s\",\n"
+					  "  subconninfo AS \"%s\"\n",
+					  gettext_noop("Name"),
+					  gettext_noop("Database"),
+					  gettext_noop("Enabled"),
+					  gettext_noop("Publication"),
+					  gettext_noop("Conninfo"));
+
+	/* TODO Show owner and ACL */
+	if (verbose)
+	{
+	}
+
+	appendPQExpBufferStr(&buf,
+						 "\nFROM pg_catalog.pg_subscription\n");
+
+	processSQLNamePattern(pset.db, &buf, pattern, false, false,
+						  NULL, "subname", NULL,
+						  NULL);
+
+	appendPQExpBufferStr(&buf, "ORDER BY 1;");
+
+	res = PSQLexec(buf.data);
+	termPQExpBuffer(&buf);
+	if (!res)
+		return false;
+
+	myopt.nullPrint = NULL;
+	myopt.title = _("List of subscriptions");
+	myopt.translate_header = true;
+	myopt.translate_columns = translate_columns;
+	myopt.n_translate_columns = lengthof(translate_columns);
+
+	printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+
+	PQclear(res);
+	return true;
+}
+
 /*
  * printACLColumn
  *
diff --git a/src/bin/psql/describe.h b/src/bin/psql/describe.h
index c4457da..a754fb8 100644
--- a/src/bin/psql/describe.h
+++ b/src/bin/psql/describe.h
@@ -105,4 +105,7 @@ extern bool listEventTriggers(const char *pattern, bool verbose);
 /* \drp */
 bool describePublications(const char *pattern, bool verbose);
 
+/* \drs */
+bool describeSubscriptions(const char *pattern, bool verbose);
+
 #endif   /* DESCRIBE_H */
diff --git a/src/bin/psql/help.c b/src/bin/psql/help.c
index 5cc5bce..88c3932 100644
--- a/src/bin/psql/help.c
+++ b/src/bin/psql/help.c
@@ -242,6 +242,7 @@ slashUsage(unsigned short int pager)
 	fprintf(output, _("  \\dp     [PATTERN]      list table, view, and sequence access privileges\n"));
 	fprintf(output, _("  \\drds [PATRN1 [PATRN2]] list per-database role settings\n"));
 	fprintf(output, _("  \\drp[+] [PATTERN]      list replication publications\n"));
+	fprintf(output, _("  \\drs[+] [PATTERN]      list replication subscriptions\n"));
 	fprintf(output, _("  \\ds[S+] [PATTERN]      list sequences\n"));
 	fprintf(output, _("  \\dt[S+] [PATTERN]      list tables\n"));
 	fprintf(output, _("  \\dT[S+] [PATTERN]      list data types\n"));
diff --git a/src/include/catalog/dependency.h b/src/include/catalog/dependency.h
index 01fd16a..789ce00 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -163,6 +163,7 @@ typedef enum ObjectClass
 	OCLASS_POLICY,				/* pg_policy */
 	OCLASS_PUBLICATION,			/* pg_publication */
 	OCLASS_PUBLICATION_REL,		/* pg_publication_rel */
+	OCLASS_SUBSCRIPTION,		/* pg_subscription */
 	OCLASS_TRANSFORM			/* pg_transform */
 } ObjectClass;
 
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index ae46ed6..86e2939 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -331,6 +331,12 @@ DECLARE_UNIQUE_INDEX(pg_publication_rel_object_index, 6112, on pg_publication_re
 DECLARE_UNIQUE_INDEX(pg_publication_rel_map_index, 6113, on pg_publication_rel using btree(relid oid_ops, pubid oid_ops));
 #define PublicationRelMapIndexId 6113
 
+DECLARE_UNIQUE_INDEX(pg_subscription_oid_index, 6114, on pg_subscription using btree(oid oid_ops));
+#define SubscriptionObjectIndexId 6114
+
+DECLARE_UNIQUE_INDEX(pg_subscription_subname_index, 6115, on pg_subscription using btree(subname name_ops));
+#define SubscriptionNameIndexId 6115
+
 /* last step of initialization script: build the indexes declared above */
 BUILD_INDICES
 
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
new file mode 100644
index 0000000..254d509
--- /dev/null
+++ b/src/include/catalog/pg_subscription.h
@@ -0,0 +1,52 @@
+/* -------------------------------------------------------------------------
+ *
+ * pg_subscription.h
+ *		Definition of the subscription catalog (pg_subscription).
+ *
+ * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * -------------------------------------------------------------------------
+ */
+#ifndef PG_SUBSCRIPTION_H
+#define PG_SUBSCRIPTION_H
+
+#include "catalog/genbki.h"
+
+/* ----------------
+ *		pg_subscription definition. cpp turns this into
+ *		typedef struct FormData_pg_subscription
+ * ----------------
+ */
+#define SubscriptionRelationId			6100
+#define SubscriptionRelation_Rowtype_Id	6101
+
+CATALOG(pg_subscription,6100) BKI_SHARED_RELATION BKI_ROWTYPE_OID(6101) BKI_SCHEMA_MACRO
+{
+	Oid			dbid;			/* Database the subscription is in. */
+	NameData	subname;		/* Name of the subscription */
+	bool		subenabled;		/* True if the subsription is enabled (running) */
+
+#ifdef CATALOG_VARLEN			/* variable-length fields start here */
+	text		subconninfo;	/* Connection string to the provider */
+	NameData	subslotname;	/* Slot name on provider */
+
+	text		subpublications[1];	/* List of publications subscribed to */
+#endif
+} FormData_pg_subscription;
+
+typedef FormData_pg_subscription *Form_pg_subscription;
+
+/* ----------------
+ *		compiler constants for pg_subscription
+ * ----------------
+ */
+#define Natts_pg_subscription				6
+#define Anum_pg_subscription_dbid			1
+#define Anum_pg_subscription_subname		2
+#define Anum_pg_subscription_subenabled		3
+#define Anum_pg_subscription_subconninfo	4
+#define Anum_pg_subscription_subslotname	5
+#define Anum_pg_subscription_subpublications	6
+
+#endif   /* PG_SUBSCRIPTION_H */
diff --git a/src/include/commands/replicationcmds.h b/src/include/commands/replicationcmds.h
index 717485f..7c35d72 100644
--- a/src/include/commands/replicationcmds.h
+++ b/src/include/commands/replicationcmds.h
@@ -1,7 +1,7 @@
 /*-------------------------------------------------------------------------
  *
  * replicationcmds.h
- *	  prototypes for publicationcmds.c.
+ *	  prototypes for publicationcmds.c and subscriptioncmds.c.
  *
  *
  * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
@@ -23,4 +23,8 @@ extern void AlterPublication(AlterPublicationStmt *stmt);
 extern void DropPublicationById(Oid pubid);
 extern void RemovePublicationRelById(Oid prid);
 
+extern ObjectAddress CreateSubscription(CreateSubscriptionStmt *stmt);
+extern ObjectAddress AlterSubscription(AlterSubscriptionStmt *stmt);
+extern void DropSubscriptionById(Oid subid);
+
 #endif   /* REPLICATIONCMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 3cce3d9..322286b 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -407,6 +407,8 @@ typedef enum NodeTag
 	T_CreateAmStmt,
 	T_CreatePublicationStmt,
 	T_AlterPublicationStmt,
+	T_CreateSubscriptionStmt,
+	T_AlterSubscriptionStmt,
 
 	/*
 	 * TAGS FOR PARSE TREE NODES (parsenodes.h)
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index c10e6e7..247234c 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -1414,6 +1414,7 @@ typedef enum ObjectType
 	OBJECT_RULE,
 	OBJECT_SCHEMA,
 	OBJECT_SEQUENCE,
+	OBJECT_SUBSCRIPTION,
 	OBJECT_TABCONSTRAINT,
 	OBJECT_TABLE,
 	OBJECT_TABLESPACE,
@@ -3125,4 +3126,18 @@ typedef struct AlterPublicationStmt
 	List	   *tables;			/* List of tables to add/drop */
 } AlterPublicationStmt;
 
+typedef struct CreateSubscriptionStmt
+{
+	NodeTag		type;
+	char	   *subname;		/* Name of of the subscription */
+	List	   *options;		/* List of DefElem nodes */
+} CreateSubscriptionStmt;
+
+typedef struct AlterSubscriptionStmt
+{
+	NodeTag		type;
+	char	   *subname;		/* Name of of the subscription */
+	List	   *options;		/* List of DefElem nodes */
+} AlterSubscriptionStmt;
+
 #endif   /* PARSENODES_H */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 9430ff0..c11a3f4 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -370,6 +370,7 @@ PG_KEYWORD("stdout", STDOUT, UNRESERVED_KEYWORD)
 PG_KEYWORD("storage", STORAGE, UNRESERVED_KEYWORD)
 PG_KEYWORD("strict", STRICT_P, UNRESERVED_KEYWORD)
 PG_KEYWORD("strip", STRIP_P, UNRESERVED_KEYWORD)
+PG_KEYWORD("subscription", SUBSCRIPTION, UNRESERVED_KEYWORD)
 PG_KEYWORD("substring", SUBSTRING, COL_NAME_KEYWORD)
 PG_KEYWORD("symmetric", SYMMETRIC, RESERVED_KEYWORD)
 PG_KEYWORD("sysid", SYSID, UNRESERVED_KEYWORD)
diff --git a/src/include/replication/subscription.h b/src/include/replication/subscription.h
new file mode 100644
index 0000000..a937f4b
--- /dev/null
+++ b/src/include/replication/subscription.h
@@ -0,0 +1,33 @@
+/*-------------------------------------------------------------------------
+ *
+ * subscription.h
+ *		replication subscription support struncture/function definition
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		subscription.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef SUBSCRIPTION_H
+#define SUBSCRIPTION_H
+
+#include "postgres.h"
+
+
+typedef struct Subscription
+{
+	Oid		oid;			/* Oid of the subscription */
+	Oid		dbid;			/* Oid of the database which dubscription is in */
+	char   *name;			/* Name of the subscription */
+	bool	enabled;		/* Indicates if the subscription is enabled */
+	char   *conninfo;		/* Connection string to the provider */
+	char   *slotname;		/* Name of the replication slot */
+	List   *publications;	/* List of publication names to subscribe to */
+} Subscription;
+
+extern Subscription *GetSubscription(Oid subid);
+extern Oid get_subscription_oid(const char *subname, bool missing_ok);
+
+#endif		/* SUBSCRIPTION_H */
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index ca2283a..7fcd139 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -540,5 +540,6 @@ typedef struct ViewOptions
 extern void RelationIncrementReferenceCount(Relation rel);
 extern void RelationDecrementReferenceCount(Relation rel);
 extern bool RelationHasUnloggedIndex(Relation rel);
+extern List *RelationGetRepsetList(Relation rel);
 
 #endif   /* REL_H */
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index 632fcbc..b1d03a5 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -85,6 +85,8 @@ enum SysCacheIdentifier
 	PUBLICATIONRELMAP,
 	RULERELNAME,
 	STATRELATTINH,
+	SUBSCRIPTIONOID,
+	SUBSCRIPTIONNAME,
 	TABLESPACEOID,
 	TRFOID,
 	TRFTYPELANG,
diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out
index 5ab04ae..ceac2c8 100644
--- a/src/test/regress/expected/sanity_check.out
+++ b/src/test/regress/expected/sanity_check.out
@@ -131,6 +131,7 @@ pg_shdepend|t
 pg_shdescription|t
 pg_shseclabel|t
 pg_statistic|t
+pg_subscription|t
 pg_tablespace|t
 pg_transform|t
 pg_trigger|t
-- 
2.7.4

