From 21bfbbe60ec81eb028ede812216c3ca2f035499c Mon Sep 17 00:00:00 2001
From: Petr Jelinek <pjmodos@pjmodos.net>
Date: Wed, 13 Jul 2016 18:11:54 +0200
Subject: [PATCH 1/6] Add PUBLICATION catalogs and DDL

---
 doc/src/sgml/catalogs.sgml                    | 128 ++++++
 doc/src/sgml/ref/allfiles.sgml                |   3 +
 doc/src/sgml/ref/alter_publication.sgml       | 159 ++++++++
 doc/src/sgml/ref/create_publication.sgml      | 164 ++++++++
 doc/src/sgml/ref/drop_publication.sgml        |  86 ++++
 src/backend/catalog/Makefile                  |   1 +
 src/backend/catalog/dependency.c              |  19 +
 src/backend/catalog/objectaddress.c           | 162 ++++++++
 src/backend/commands/Makefile                 |   5 +-
 src/backend/commands/event_trigger.c          |   5 +
 src/backend/commands/publicationcmds.c        | 550 ++++++++++++++++++++++++++
 src/backend/nodes/copyfuncs.c                 |  31 ++
 src/backend/nodes/equalfuncs.c                |  29 ++
 src/backend/parser/gram.y                     | 117 +++++-
 src/backend/replication/logical/Makefile      |   4 +-
 src/backend/replication/logical/publication.c | 343 ++++++++++++++++
 src/backend/tcop/utility.c                    |  38 ++
 src/backend/utils/cache/relcache.c            |   9 +
 src/backend/utils/cache/syscache.c            |  46 +++
 src/bin/psql/command.c                        |  27 +-
 src/bin/psql/describe.c                       |  96 +++++
 src/bin/psql/describe.h                       |   3 +
 src/bin/psql/help.c                           |   1 +
 src/include/catalog/dependency.h              |   2 +
 src/include/catalog/indexing.h                |  12 +
 src/include/catalog/pg_publication.h          |  64 +++
 src/include/catalog/pg_publication_rel.h      |  53 +++
 src/include/commands/replicationcmds.h        |  26 ++
 src/include/nodes/nodes.h                     |   2 +
 src/include/nodes/parsenodes.h                |  24 ++
 src/include/parser/kwlist.h                   |   1 +
 src/include/replication/publication.h         |  47 +++
 src/include/utils/rel.h                       |   4 +
 src/include/utils/syscache.h                  |   4 +
 src/test/regress/expected/publication.out     |  76 ++++
 src/test/regress/expected/sanity_check.out    |   2 +
 src/test/regress/parallel_schedule            |   2 +-
 src/test/regress/sql/publication.sql          |  40 ++
 38 files changed, 2370 insertions(+), 15 deletions(-)
 create mode 100644 doc/src/sgml/ref/alter_publication.sgml
 create mode 100644 doc/src/sgml/ref/create_publication.sgml
 create mode 100644 doc/src/sgml/ref/drop_publication.sgml
 create mode 100644 src/backend/commands/publicationcmds.c
 create mode 100644 src/backend/replication/logical/publication.c
 create mode 100644 src/include/catalog/pg_publication.h
 create mode 100644 src/include/catalog/pg_publication_rel.h
 create mode 100644 src/include/commands/replicationcmds.h
 create mode 100644 src/include/replication/publication.h
 create mode 100644 src/test/regress/expected/publication.out
 create mode 100644 src/test/regress/sql/publication.sql

diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 8f5332a..6d505ae 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -236,6 +236,16 @@
      </row>
 
      <row>
+      <entry><link linkend="catalog-pg-publication"><structname>pg_publication</structname></link></entry>
+      <entry>publications for logical replication</entry>
+     </row>
+
+     <row>
+      <entry><link linkend="catalog-pg-publication-rel"><structname>pg_publication_rel</structname></link></entry>
+      <entry>relation to publication mapping</entry>
+     </row>
+
+     <row>
       <entry><link linkend="catalog-pg-range"><structname>pg_range</structname></link></entry>
       <entry>information about range types</entry>
      </row>
@@ -5110,6 +5120,124 @@
 
  </sect1>
 
+ <sect1 id="catalog-pg-publication">
+  <title><structname>pg_publication</structname></title>
+
+  <indexterm zone="catalog-pg-publication">
+   <primary>pg_publication</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_publication</structname> catalog contains
+   all publications created in the database.
+  </para>
+
+  <table>
+
+   <title><structname>pg_publication</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>pubname</structfield></entry>
+      <entry><type>Name</type></entry>
+      <entry></entry>
+      <entry>A unique, database-wide identifier for the publication.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>pubreplins</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>If true, INSERT operations are replicated for tables in the
+       publication.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>pubreplupd</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>If true, UPDATE operations are replicated for tables in the
+       publication.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>pubrepldel</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry></entry>
+      <entry>If true, DELETE operations are replicated for tables in the
+       publication.</entry>
+     </row>
+
+    </tbody>
+   </tgroup>
+  </table>
+ </sect1>
+
+ <sect1 id="catalog-pg-publication-rel">
+  <title><structname>pg_publication_rel</structname></title>
+
+  <indexterm zone="catalog-pg-publication-rel">
+   <primary>pg_publication_rel</primary>
+  </indexterm>
+
+  <para>
+   The <structname>pg_publication_rel</structname> catalog contains
+   mapping between tables and publications in the database. This is many to
+   many mapping.
+  </para>
+
+  <table>
+
+   <title><structname>pg_publication_rel</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>pubid</structfield></entry>
+      <entry><type>oid</type></entry>
+      <entry><literal><link linkend="catalog-pg-publication"><structname>pg_publication</structname></link>.oid</literal></entry>
+      <entry>Publication reference.</entry>
+     </row>
+
+     <row>
+      <entry><structfield>relid</structfield></entry>
+      <entry><type>bool</type></entry>
+      <entry><literal><link linkend="catalog-pg-class"><structname>pg_class</structname></link>.oid</literal></entry>
+      <entry>Relation reference.</entry>
+     </row>
+
+    </tbody>
+   </tgroup>
+  </table>
+ </sect1>
+
  <sect1 id="catalog-pg-range">
   <title><structname>pg_range</structname></title>
 
diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml
index 77667bd..371a7b7 100644
--- a/doc/src/sgml/ref/allfiles.sgml
+++ b/doc/src/sgml/ref/allfiles.sgml
@@ -26,6 +26,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY alterOperatorClass SYSTEM "alter_opclass.sgml">
 <!ENTITY alterOperatorFamily SYSTEM "alter_opfamily.sgml">
 <!ENTITY alterPolicy        SYSTEM "alter_policy.sgml">
+<!ENTITY alterPublication   SYSTEM "alter_publication.sgml">
 <!ENTITY alterRole          SYSTEM "alter_role.sgml">
 <!ENTITY alterRule          SYSTEM "alter_rule.sgml">
 <!ENTITY alterSchema        SYSTEM "alter_schema.sgml">
@@ -72,6 +73,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY createOperatorClass SYSTEM "create_opclass.sgml">
 <!ENTITY createOperatorFamily SYSTEM "create_opfamily.sgml">
 <!ENTITY createPolicy       SYSTEM "create_policy.sgml">
+<!ENTITY createPublication  SYSTEM "create_publication.sgml">
 <!ENTITY createRole         SYSTEM "create_role.sgml">
 <!ENTITY createRule         SYSTEM "create_rule.sgml">
 <!ENTITY createSchema       SYSTEM "create_schema.sgml">
@@ -116,6 +118,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY dropOperatorFamily  SYSTEM "drop_opfamily.sgml">
 <!ENTITY dropOwned          SYSTEM "drop_owned.sgml">
 <!ENTITY dropPolicy         SYSTEM "drop_policy.sgml">
+<!ENTITY dropPublication    SYSTEM "drop_publication.sgml">
 <!ENTITY dropRole           SYSTEM "drop_role.sgml">
 <!ENTITY dropRule           SYSTEM "drop_rule.sgml">
 <!ENTITY dropSchema         SYSTEM "drop_schema.sgml">
diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml
new file mode 100644
index 0000000..246b39b
--- /dev/null
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -0,0 +1,159 @@
+<!--
+doc/src/sgml/ref/alter_publication.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-ALTERPUBLICATION">
+ <indexterm zone="sql-alterpublication">
+  <primary>ALTER PUBLICATION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>ALTER PUBLICATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>ALTER PUBLICATION</refname>
+  <refpurpose>change the definition of a publication</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+ALTER PUBLICATION <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replaceable class="PARAMETER">option</replaceable> [ ... ] ]
+
+<phrase>where <replaceable class="PARAMETER">option</replaceable> can be:</phrase>
+
+      REPLICATE_INSERT | NOREPLICATE_INSERT
+    | REPLICATE_UPDATE | NOREPLICATE_UPDATE
+    | REPLICATE_DELETE | NOREPLICATE_DELETE
+
+ALTER PUBLICATION <replaceable class="PARAMETER">name</replaceable> ADD TABLE <replaceable class="PARAMETER">table_name</replaceable> [, ...]
+ALTER PUBLICATION <replaceable class="PARAMETER">name</replaceable> ADD TABLE ALL IN SCHEMA <replaceable class="PARAMETER">schema_name</replaceable>
+ALTER PUBLICATION <replaceable class="PARAMETER">name</replaceable> DROP TABLE <replaceable class="PARAMETER">table_name</replaceable> [, ...]
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   The first variant of this command listed in the synopsis can change
+   all of the publication attributes that can be specified in
+   <xref linkend="sql-createpublication">.
+   Attributes not mentioned in the command retain their previous settings.
+   Database superusers can change any of these settings for any role.
+  </para>
+
+  <para>
+   The other variants this command deal with table membership in the
+   publication. The <literal>ADD TABLE</literal> subcommand will add one
+   or more tables to the publication. If the optional
+   <literal>ALL IN SCHEMA</literal> is specified all tables in that schema
+   will be added. The <literal>ALL IN SCHEMA</literal> variant of this
+   command will not complain about tables that are already present in the
+   publication, it only adds the missing ones.
+   The <literal>DROP TABLE</literal> will remove one or more table from
+   publication.
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name of an existing publication whose attributes are to be
+      altered.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><literal>REPLICATE_INSERT</literal></term>
+    <term><literal>NOREPLICATE_INSERT</literal></term>
+    <term><literal>REPLICATE_UPDATE</literal></term>
+    <term><literal>NOREPLICATE_UPDATE</literal></term>
+    <term><literal>REPLICATE_DELETE</literal></term>
+    <term><literal>NOREPLICATE_DELETE</literal></term>
+    <listitem>
+     <para>
+      These clauses alter attributes originally set by
+      <xref linkend="SQL-CREATEPUBLICATION">. For more information, see the
+      <command>CREATE PUBLICATION</command> reference page.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">table_name</replaceable></term>
+    <listitem>
+     <para>
+      Name of an existing table.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><replaceable class="parameter">schema_name</replaceable></term>
+    <listitem>
+     <para>
+      Name of a schema.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Change the publication to not replicate inserts:
+<programlisting>
+ALTER PUBLICATION noinsert NOREPLICATE_INSERT;
+</programlisting>
+  </para>
+
+  <para>
+   Add some tables to the publication:
+<programlisting>
+ALTER PUBLICATION mypublication ADD TABLE users, departments;
+</programlisting>
+  </para>
+
+  <para>
+   Add all tables from public schema to the publication:
+<programlisting>
+ALTER PUBLICATION mypublication ADD TABLE ALL IN SCHEMA public;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>ALTER PUBLICATION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-createpublication"></member>
+   <member><xref linkend="sql-droppublication"></member>
+   <member><xref linkend="sql-createsubscription"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
new file mode 100644
index 0000000..ac8b71e
--- /dev/null
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -0,0 +1,164 @@
+<!--
+doc/src/sgml/ref/create_publication.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-CREATEPUBLICATION">
+ <indexterm zone="sql-createpublication">
+  <primary>CREATE PUBLICATION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>CREATE PUBLICATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>CREATE PUBLICATION</refname>
+  <refpurpose>define new publication</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+CREATE PUBLICATION <replaceable class="PARAMETER">name</replaceable> [ [ WITH ] <replaceable class="PARAMETER">option</replaceable> [ ... ] ]
+
+<phrase>where <replaceable class="PARAMETER">option</replaceable> can be:</phrase>
+
+      REPLICATE_INSERT | NOREPLICATE_INSERT
+    | REPLICATE_UPDATE | NOREPLICATE_UPDATE
+    | REPLICATE_DELETE | NOREPLICATE_DELETE
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>CREATE PUBLICATION</command> adds a new publication
+   into the current database. The publication name must be distinct from
+   the name of any existing publication in the current database.
+  </para>
+
+  <para>
+   A publication is essentially a group of tables intended for managing
+   logical replication. See
+   <xref linkend="logical-replication-publication"> for details about how
+   publications fit into logical replication setup.
+   </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name of the new publication.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><literal>REPLICATE_INSERT</literal></term>
+    <term><literal>NOREPLICATE_INSERT</literal></term>
+    <listitem>
+     <para>
+      These clauses determine whether the new publication will send
+      the <command>INSERT</command> operations to the subscribers.
+      <literal>REPLICATE_INSERT</literal> is the default.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><literal>REPLICATE_UPDATE</literal></term>
+    <term><literal>NOREPLICATE_UPDATE</literal></term>
+    <listitem>
+     <para>
+      These clauses determine whether the new publication will send
+      the <command>UPDATE</command> operations to the subscribers.
+      <literal>REPLICATE_UPDATE</literal> is the default.
+     </para>
+    </listitem>
+   </varlistentry>
+
+   <varlistentry>
+    <term><literal>REPLICATE_DELETE</literal></term>
+    <term><literal>NOREPLICATE_DELETE</literal></term>
+    <listitem>
+     <para>
+      These clauses determine whether the new publication will send
+      the <command>DELETE</command> operations to the subscribers.
+      <literal>REPLICATE_DELETE</literal> is the default.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Notes</title>
+
+  <para>
+   This operation does not reserve any resources on the server. It only
+   defines grouping and filtering logic for future subscribers.
+  </para>
+
+  <para>
+   To create a publication, the invoking user must have the
+   <literal>CREATE</> privilege for the current database and the
+   <literal>SUBSCRIPTION</> role.
+   (Of course, superusers bypass this check.)
+  </para>
+
+  <para>
+   Replication of <command>UPDATE</command> and <command>DELETE</command>
+   operations requires the tables added to the publication to have a
+   <literal>REPLICA IDENTITY</> index specified.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Create a simple publication that just replicates all DML for tables in it:
+<programlisting>
+CREATE PUBLICATION mypublication;
+</programlisting>
+  </para>
+
+  <para>
+   Create an insert only publication (for example for tables without a
+   primary key):
+<programlisting>
+CREATE PUBLICATION insert_only NOREPLICATE_UPDATE NOREPLICATE_DELETE;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>CREATE PUBLICATION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-alterpublication"></member>
+   <member><xref linkend="sql-droppublication"></member>
+   <member><xref linkend="sql-createsubscription"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/ref/drop_publication.sgml b/doc/src/sgml/ref/drop_publication.sgml
new file mode 100644
index 0000000..c5b0c78
--- /dev/null
+++ b/doc/src/sgml/ref/drop_publication.sgml
@@ -0,0 +1,86 @@
+<!--
+doc/src/sgml/ref/drop_publication.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="SQL-DROPPUBLICATION">
+ <indexterm zone="sql-droppublication">
+  <primary>DROP PUBLICATION</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle>DROP PUBLICATION</refentrytitle>
+  <manvolnum>7</manvolnum>
+  <refmiscinfo>SQL - Language Statements</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>DROP PUBLICATION</refname>
+  <refpurpose>remove an existing publication</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+<synopsis>
+DROP PUBLCATION <replaceable class="PARAMETER">name</replaceable> [, ...]
+</synopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+
+  <para>
+   <command>DROP PUBLCATION</command> removes publications from the database.
+  </para>
+
+  <para>
+   A publication can only be dropped by its owner or a superuser.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>Parameters</title>
+
+  <variablelist>
+   <varlistentry>
+    <term><replaceable class="parameter">name</replaceable></term>
+    <listitem>
+     <para>
+      The name of an existing publication.
+     </para>
+    </listitem>
+   </varlistentry>
+
+  </variablelist>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   Drop a publication:
+<programlisting>
+DROP PUBLICATION mypublication;
+</programlisting>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Compatibility</title>
+
+  <para>
+   <command>DROP PUBLICATION</command> is a <productname>PostgreSQL</>
+   extension.
+  </para>
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="sql-createpublication"></member>
+   <member><xref linkend="sql-alterpublication"></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/src/backend/catalog/Makefile b/src/backend/catalog/Makefile
index 1ce7610..6bb3683 100644
--- a/src/backend/catalog/Makefile
+++ b/src/backend/catalog/Makefile
@@ -42,6 +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 \
 	toasting.h indexing.h \
     )
 
diff --git a/src/backend/catalog/dependency.c b/src/backend/catalog/dependency.c
index 04d7840..359e7bb 100644
--- a/src/backend/catalog/dependency.c
+++ b/src/backend/catalog/dependency.c
@@ -48,6 +48,8 @@
 #include "catalog/pg_opfamily.h"
 #include "catalog/pg_policy.h"
 #include "catalog/pg_proc.h"
+#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
@@ -64,6 +66,7 @@
 #include "commands/extension.h"
 #include "commands/policy.h"
 #include "commands/proclang.h"
+#include "commands/replicationcmds.h"
 #include "commands/schemacmds.h"
 #include "commands/seclabel.h"
 #include "commands/trigger.h"
@@ -163,6 +166,8 @@ static const Oid object_classes[] = {
 	ExtensionRelationId,		/* OCLASS_EXTENSION */
 	EventTriggerRelationId,		/* OCLASS_EVENT_TRIGGER */
 	PolicyRelationId,			/* OCLASS_POLICY */
+	PublicationRelationId,		/* OCLASS_PUBCLICATION */
+	PublicationRelRelationId,	/* OCLASS_PUBCLICATION_REL */
 	TransformRelationId			/* OCLASS_TRANSFORM */
 };
 
@@ -1279,6 +1284,14 @@ doDeletion(const ObjectAddress *object, int flags)
 			RemovePolicyById(object->objectId);
 			break;
 
+		case OCLASS_PUBLICATION:
+			DropPublicationById(object->objectId);
+			break;
+
+		case OCLASS_PUBLICATION_REL:
+			RemovePublicationRelById(object->objectId);
+			break;
+
 		case OCLASS_TRANSFORM:
 			DropTransformById(object->objectId);
 			break;
@@ -2436,6 +2449,12 @@ getObjectClass(const ObjectAddress *object)
 		case PolicyRelationId:
 			return OCLASS_POLICY;
 
+		case PublicationRelationId:
+			return OCLASS_PUBLICATION;
+
+		case PublicationRelRelationId:
+			return OCLASS_PUBLICATION_REL;
+
 		case TransformRelationId:
 			return OCLASS_TRANSFORM;
 	}
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index 8068b82..375f4b0 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -45,6 +45,8 @@
 #include "catalog/pg_operator.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_policy.h"
+#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
 #include "catalog/pg_tablespace.h"
 #include "catalog/pg_transform.h"
@@ -71,6 +73,7 @@
 #include "parser/parse_func.h"
 #include "parser/parse_oper.h"
 #include "parser/parse_type.h"
+#include "replication/publication.h"
 #include "rewrite/rewriteSupport.h"
 #include "storage/lmgr.h"
 #include "storage/sinval.h"
@@ -450,6 +453,18 @@ static const ObjectPropertyType ObjectProperty[] =
 		Anum_pg_type_typacl,
 		ACL_KIND_TYPE,
 		true
+	},
+	{
+		PublicationRelationId,
+		PublicationObjectIndexId,
+		PUBLICATIONOID,
+		PUBLICATIONNAME,
+		Anum_pg_publication_pubname,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		InvalidAttrNumber,
+		-1,
+		true
 	}
 };
 
@@ -653,6 +668,14 @@ static const struct object_type_map
 	{
 		"policy", OBJECT_POLICY
 	},
+	/* OCLASS_PUBLICATION */
+	{
+		"publication", OBJECT_PUBLICATION
+	},
+	/* OCLASS_PUBLICATION_REL */
+	{
+		"publication relation", OBJECT_PUBLICATION_REL
+	},
 	/* OCLASS_TRANSFORM */
 	{
 		"transform", OBJECT_TRANSFORM
@@ -688,6 +711,9 @@ static ObjectAddress get_object_address_opf_member(ObjectType objtype,
 
 static ObjectAddress get_object_address_usermapping(List *objname,
 							   List *objargs, bool missing_ok);
+static ObjectAddress get_object_address_publication_rel(List *objname,
+								   List *objargs, Relation *relation,
+								   bool missing_ok);
 static ObjectAddress get_object_address_defacl(List *objname, List *objargs,
 						  bool missing_ok);
 static const ObjectPropertyType *get_object_property_data(Oid class_id);
@@ -812,6 +838,7 @@ get_object_address(ObjectType objtype, List *objname, List *objargs,
 			case OBJECT_FOREIGN_SERVER:
 			case OBJECT_EVENT_TRIGGER:
 			case OBJECT_ACCESS_METHOD:
+			case OBJECT_PUBLICATION:
 				address = get_object_address_unqualified(objtype,
 														 objname, missing_ok);
 				break;
@@ -926,6 +953,10 @@ get_object_address(ObjectType objtype, List *objname, List *objargs,
 				address = get_object_address_usermapping(objname, objargs,
 														 missing_ok);
 				break;
+			case OBJECT_PUBLICATION_REL:
+				address = get_object_address_publication_rel(objname, objargs,
+															 &relation,
+															 missing_ok);
 			case OBJECT_DEFACL:
 				address = get_object_address_defacl(objname, objargs,
 													missing_ok);
@@ -1091,6 +1122,9 @@ get_object_address_unqualified(ObjectType objtype,
 			case OBJECT_EVENT_TRIGGER:
 				msg = gettext_noop("event trigger name cannot be qualified");
 				break;
+			case OBJECT_PUBLICATION:
+				msg = gettext_noop("publication name cannot be qualified");
+				break;
 			default:
 				elog(ERROR, "unrecognized objtype: %d", (int) objtype);
 				msg = NULL;		/* placate compiler */
@@ -1156,6 +1190,11 @@ get_object_address_unqualified(ObjectType objtype,
 			address.objectId = get_event_trigger_oid(name, missing_ok);
 			address.objectSubId = 0;
 			break;
+		case OBJECT_PUBLICATION:
+			address.classId = PublicationRelationId;
+			address.objectId = get_publication_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 */
@@ -1743,6 +1782,50 @@ get_object_address_usermapping(List *objname, List *objargs, bool missing_ok)
 }
 
 /*
+ * Find the ObjectAddress for a publication relation.
+ */
+static ObjectAddress
+get_object_address_publication_rel(List *objname, List *objargs,
+								   Relation *relation, bool missing_ok)
+{
+	ObjectAddress address;
+	char	   *pubname;
+	Publication *pub;
+
+	ObjectAddressSet(address, UserMappingRelationId, InvalidOid);
+
+	*relation = relation_openrv_extended(makeRangeVarFromNameList(objname),
+										 AccessShareLock, missing_ok);
+	if (!relation)
+		return address;
+
+	/* fetch publication name from input list */
+	pubname = strVal(linitial(objargs));
+
+	/* Now look up the pg_publication tuple */
+	pub = GetPublicationByName(pubname, missing_ok);
+	if (!pub)
+		return address;
+
+	/* Find the publication relation mapping in syscache. */
+	address.objectId =
+		GetSysCacheOid2(PUBLICATIONRELMAP,
+						ObjectIdGetDatum(RelationGetRelid(*relation)),
+						ObjectIdGetDatum(pub->oid));
+	if (!OidIsValid(address.objectId))
+	{
+		if (!missing_ok)
+			ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("publication relation \"%s\" in publication \"%s\" does not exist",
+							RelationGetRelationName(*relation), pubname)));
+		return address;
+	}
+
+	return address;
+}
+
+/*
  * Find the ObjectAddress for a default ACL.
  */
 static ObjectAddress
@@ -2001,6 +2084,7 @@ pg_get_object_address(PG_FUNCTION_ARGS)
 		case OBJECT_DOMCONSTRAINT:
 		case OBJECT_CAST:
 		case OBJECT_USER_MAPPING:
+		case OBJECT_PUBLICATION_REL:
 		case OBJECT_DEFACL:
 		case OBJECT_TRANSFORM:
 			if (list_length(args) != 1)
@@ -2230,6 +2314,7 @@ check_object_ownership(Oid roleid, ObjectType objtype, ObjectAddress address,
 		case OBJECT_TSPARSER:
 		case OBJECT_TSTEMPLATE:
 		case OBJECT_ACCESS_METHOD:
+		case OBJECT_PUBLICATION:
 			/* We treat these object types as being owned by superusers */
 			if (!superuser_arg(roleid))
 				ereport(ERROR,
@@ -3195,6 +3280,42 @@ getObjectDescription(const ObjectAddress *object)
 				break;
 			}
 
+		case OCLASS_PUBLICATION:
+			{
+				HeapTuple	tup;
+
+				tup = SearchSysCache1(PUBLICATIONOID,
+									  ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(tup))
+					elog(ERROR, "cache lookup failed for publication %u",
+						 object->objectId);
+				appendStringInfo(&buffer, _("publicaton %s"),
+				   NameStr(((Form_pg_publication) GETSTRUCT(tup))->pubname));
+				ReleaseSysCache(tup);
+				break;
+			}
+
+		case OCLASS_PUBLICATION_REL:
+			{
+				HeapTuple	tup;
+				Publication *pub;
+				Form_pg_publication_rel	prform;
+
+				tup = SearchSysCache1(PUBLICATIONREL,
+									  ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(tup))
+					elog(ERROR, "cache lookup failed for publication table %u",
+						 object->objectId);
+
+				prform = (Form_pg_publication_rel) GETSTRUCT(tup);
+				pub = GetPublication(prform->pubid);
+
+				appendStringInfo(&buffer, _("publication table %s in publication %s"),
+								 get_rel_name(prform->relid), pub->name);
+				ReleaseSysCache(tup);
+				break;
+			}
+
 		default:
 			appendStringInfo(&buffer, "unrecognized object %u %u %d",
 							 object->classId,
@@ -3680,6 +3801,14 @@ getObjectTypeDescription(const ObjectAddress *object)
 			appendStringInfoString(&buffer, "access method");
 			break;
 
+		case OCLASS_PUBLICATION:
+			appendStringInfoString(&buffer, "publication");
+			break;
+
+		case OCLASS_PUBLICATION_REL:
+			appendStringInfoString(&buffer, "publication table");
+			break;
+
 		default:
 			appendStringInfo(&buffer, "unrecognized %u", object->classId);
 			break;
@@ -4650,6 +4779,39 @@ getObjectIdentityParts(const ObjectAddress *object,
 			}
 			break;
 
+		case OCLASS_PUBLICATION:
+			{
+				Publication *pub;
+
+				pub = GetPublication(object->objectId);
+				appendStringInfoString(&buffer,
+									   quote_identifier(pub->name));
+				if (objname)
+					*objname = list_make1(pstrdup(pub->name));
+				break;
+			}
+
+		case OCLASS_PUBLICATION_REL:
+			{
+				HeapTuple	tup;
+				Publication *pub;
+				Form_pg_publication_rel	prform;
+
+				tup = SearchSysCache1(PUBLICATIONREL,
+									  ObjectIdGetDatum(object->objectId));
+				if (!HeapTupleIsValid(tup))
+					elog(ERROR, "cache lookup failed for publication table %u",
+						 object->objectId);
+
+				prform = (Form_pg_publication_rel) GETSTRUCT(tup);
+				pub = GetPublication(prform->pubid);
+
+				appendStringInfo(&buffer, _("%s in publication %s"),
+								 get_rel_name(prform->relid), pub->name);
+				ReleaseSysCache(tup);
+				break;
+			}
+
 		default:
 			appendStringInfo(&buffer, "unrecognized object %u %u %d",
 							 object->classId,
diff --git a/src/backend/commands/Makefile b/src/backend/commands/Makefile
index 6b3742c..cb580377 100644
--- a/src/backend/commands/Makefile
+++ b/src/backend/commands/Makefile
@@ -17,9 +17,8 @@ OBJS = amcmds.o aggregatecmds.o alter.o analyze.o async.o cluster.o comment.o \
 	dbcommands.o define.o discard.o dropcmds.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 \
+	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
+	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 50c89b8..8aaf1a7 100644
--- a/src/backend/commands/event_trigger.c
+++ b/src/backend/commands/event_trigger.c
@@ -122,6 +122,7 @@ static event_trigger_support_data event_trigger_support[] = {
 	{"TYPE", true},
 	{"USER MAPPING", true},
 	{"VIEW", true},
+	{"PUBLICATION", true},
 	{NULL, false}
 };
 
@@ -1119,6 +1120,8 @@ EventTriggerSupportsObjectType(ObjectType obtype)
 		case OBJECT_TYPE:
 		case OBJECT_USER_MAPPING:
 		case OBJECT_VIEW:
+		case OBJECT_PUBLICATION:
+		case OBJECT_PUBLICATION_REL:
 			return true;
 	}
 	return true;
@@ -1170,6 +1173,8 @@ EventTriggerSupportsObjectClass(ObjectClass objclass)
 		case OCLASS_EXTENSION:
 		case OCLASS_POLICY:
 		case OCLASS_AM:
+		case OCLASS_PUBLICATION:
+		case OCLASS_PUBLICATION_REL:
 			return true;
 	}
 
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
new file mode 100644
index 0000000..19c6a97
--- /dev/null
+++ b/src/backend/commands/publicationcmds.c
@@ -0,0 +1,550 @@
+/*-------------------------------------------------------------------------
+ *
+ * publicationcmds.c
+ *		publication manipulation
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		publicationcmds.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_inherits_fn.h"
+#include "catalog/pg_type.h"
+#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
+
+#include "commands/defrem.h"
+#include "commands/event_trigger.h"
+#include "commands/replicationcmds.h"
+
+#include "executor/spi.h"
+
+#include "nodes/makefuncs.h"
+
+#include "parser/parse_clause.h"
+
+#include "replication/publication.h"
+#include "replication/reorderbuffer.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 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 publications"))));
+}
+
+static void
+parse_publication_options(List *options,
+						  bool *replicate_insert_given,
+						  bool *replicate_insert,
+						  bool *replicate_update_given,
+						  bool *replicate_update,
+						  bool *replicate_delete_given,
+						  bool *replicate_delete)
+{
+	ListCell   *lc;
+
+	*replicate_insert_given = false;
+	*replicate_update_given = false;
+	*replicate_delete_given = false;
+
+	/* Defaults are true */
+	*replicate_insert = true;
+	*replicate_update = true;
+	*replicate_delete = true;
+
+	/* Parse options */
+	foreach (lc, options)
+	{
+		DefElem    *defel = (DefElem *) lfirst(lc);
+
+		if (strcmp(defel->defname, "replicate_insert") == 0)
+		{
+			if (*replicate_insert_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			*replicate_insert_given = true;
+			*replicate_insert = defGetBoolean(defel);
+		}
+		else if (strcmp(defel->defname, "replicate_update") == 0)
+		{
+			if (*replicate_update_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			*replicate_update_given = true;
+			*replicate_update = defGetBoolean(defel);
+		}
+		else if (strcmp(defel->defname, "replicate_delete") == 0)
+		{
+			if (*replicate_delete_given)
+				ereport(ERROR,
+						(errcode(ERRCODE_SYNTAX_ERROR),
+						 errmsg("conflicting or redundant options")));
+
+			*replicate_delete_given = true;
+			*replicate_delete = defGetBoolean(defel);
+		}
+		else
+			elog(ERROR, "unrecognized option: %s", defel->defname);
+	}
+}
+
+/*
+ * Create new publication.
+ * TODO ACL check
+ */
+ObjectAddress
+CreatePublication(CreatePublicationStmt *stmt)
+{
+	Relation	rel;
+	ObjectAddress myself;
+	Oid			puboid;
+	bool		nulls[Natts_pg_publication];
+	Datum		values[Natts_pg_publication];
+	HeapTuple	tup;
+	bool		replicate_insert_given;
+	bool		replicate_update_given;
+	bool		replicate_delete_given;
+	bool		replicate_insert;
+	bool		replicate_update;
+	bool		replicate_delete;
+
+	check_replication_permissions();
+
+	rel = heap_open(PublicationRelationId, RowExclusiveLock);
+
+	/* Check if name is used */
+	puboid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(stmt->pubname));
+	if (OidIsValid(puboid))
+	{
+		ereport(ERROR,
+				(errcode(ERRCODE_DUPLICATE_OBJECT),
+				 errmsg("publication \"%s\" already exists",
+						stmt->pubname)));
+	}
+
+	/* Form a tuple. */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+
+	values[Anum_pg_publication_pubname - 1] =
+		DirectFunctionCall1(namein, CStringGetDatum(stmt->pubname));
+
+	parse_publication_options(stmt->options,
+							  &replicate_insert_given, &replicate_insert,
+							  &replicate_update_given, &replicate_update,
+							  &replicate_delete_given, &replicate_delete);
+
+	values[Anum_pg_publication_pubreplins - 1] =
+		BoolGetDatum(replicate_insert);
+	values[Anum_pg_publication_pubreplupd - 1] =
+		BoolGetDatum(replicate_update);
+	values[Anum_pg_publication_pubrepldel - 1] =
+		BoolGetDatum(replicate_delete);
+
+	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+
+	/* Insert tuple into catalog. */
+	puboid = simple_heap_insert(rel, tup);
+	CatalogUpdateIndexes(rel, tup);
+	heap_freetuple(tup);
+
+	ObjectAddressSet(myself, PublicationRelationId, puboid);
+
+	heap_close(rel, RowExclusiveLock);
+
+	/* Make the changes visible. */
+	CommandCounterIncrement();
+
+	return myself;
+}
+
+/*
+ * Change options of a publication.
+ */
+static void
+AlterPublicationOptions(AlterPublicationStmt *stmt, Relation rel,
+					   HeapTuple tup)
+{
+	bool		nulls[Natts_pg_publication];
+	bool		replaces[Natts_pg_publication];
+	Datum		values[Natts_pg_publication];
+	bool		replicate_insert_given;
+	bool		replicate_update_given;
+	bool		replicate_delete_given;
+	bool		replicate_insert;
+	bool		replicate_update;
+	bool		replicate_delete;
+	ObjectAddress		obj;
+	Form_pg_publication pub = (Form_pg_publication) GETSTRUCT(tup);
+
+	parse_publication_options(stmt->options,
+							  &replicate_insert_given, &replicate_insert,
+							  &replicate_update_given, &replicate_update,
+							  &replicate_delete_given, &replicate_delete);
+
+	/*
+	 * Validate that replication is not being changed to replicate UPDATEs
+	 * and DELETEs if it contains any tables without replication identity.
+	 */
+	if ((replicate_update_given && replicate_update) ||
+		(replicate_delete_given && replicate_delete))
+	{
+		Relation		pubrelsrel;
+		ScanKeyData		scankey;
+		SysScanDesc		scan;
+		HeapTuple		reltup;
+
+		pubrelsrel = heap_open(PublicationRelRelationId, AccessShareLock);
+
+		/* Loop over all relations in the publication. */
+		ScanKeyInit(&scankey,
+					Anum_pg_publication_rel_pubid,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(HeapTupleGetOid(tup)));
+
+		scan = systable_beginscan(pubrelsrel, 0, true, NULL, 1, &scankey);
+
+		/* Process every individual table in the publication. */
+		while (HeapTupleIsValid(reltup = systable_getnext(scan)))
+		{
+			Form_pg_publication_rel	t;
+			Relation				pubrel;
+
+			t = (Form_pg_publication_rel) GETSTRUCT(reltup);
+
+			pubrel = heap_open(t->relid, AccessShareLock);
+
+			/* Check if relation has replication index. */
+			if (RelationGetForm(pubrel)->relkind == RELKIND_RELATION)
+			{
+
+				if (pubrel->rd_indexvalid == 0)
+					RelationGetIndexList(pubrel);
+				if (!OidIsValid(pubrel->rd_replidindex))
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							 errmsg("publication %s cannot be altered to "
+									"replicate UPDATEs or DELETEs because it "
+									"contains tables without PRIMARY KEY",
+									NameStr(pub->pubname))));
+			}
+
+			heap_close(pubrel, NoLock);
+		}
+
+		systable_endscan(scan);
+		heap_close(pubrelsrel, NoLock);
+	}
+
+	/* Everything ok, form a new tuple. */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+	memset(replaces, false, sizeof(replaces));
+
+	if (replicate_insert_given)
+	{
+		values[Anum_pg_publication_pubreplins - 1] =
+			BoolGetDatum(replicate_insert);
+		replaces[Anum_pg_publication_pubreplins - 1] = true;
+	}
+	if (replicate_update_given)
+	{
+		values[Anum_pg_publication_pubreplupd - 1] =
+			BoolGetDatum(replicate_update);
+		replaces[Anum_pg_publication_pubreplupd - 1] = true;
+	}
+	if (replicate_delete_given)
+	{
+		values[Anum_pg_publication_pubrepldel - 1] =
+			BoolGetDatum(replicate_delete);
+		replaces[Anum_pg_publication_pubrepldel - 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(obj, PublicationRelationId, HeapTupleGetOid(tup));
+	EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
+									 (Node *) stmt);
+}
+
+/*
+ * Add or removes table to/from publication.
+ */
+static void
+AlterPublicationTables(AlterPublicationStmt *stmt, Relation rel,
+					   HeapTuple tup)
+{
+	Oid			pubid = HeapTupleGetOid(tup);
+	Oid			prid;
+	bool		missing_ok = false;
+	bool		if_not_exists = false;
+	List	   *rels = NIL;
+	ListCell   *lc;
+	ObjectAddress	obj;
+
+	if (stmt->schema)
+	{
+		Oid			nspoid;
+		Relation	rel;
+		SysScanDesc scan;
+		ScanKeyData key[2];
+		HeapTuple	tup;
+
+		Assert(list_length(stmt->tables) == 0);
+
+		/*
+		 * Open, share-lock, and check all relation in the specified schema
+		 */
+
+		nspoid = LookupExplicitNamespace(stmt->schema, false);
+		rel = heap_open(RelationRelationId, AccessShareLock);
+
+		ScanKeyInit(&key[0],
+					Anum_pg_class_relnamespace,
+					BTEqualStrategyNumber, F_OIDEQ,
+					ObjectIdGetDatum(nspoid));
+		ScanKeyInit(&key[1],
+					Anum_pg_class_relkind,
+					BTEqualStrategyNumber, F_CHAREQ,
+					CharGetDatum(RELKIND_RELATION));
+
+		scan = systable_beginscan(rel, InvalidOid, false,
+								  NULL, 2, key);
+
+		while ((tup = systable_getnext(scan)) != NULL)
+		{
+			Oid			schemarelid = HeapTupleGetOid(tup);
+			Relation	schemarel;
+
+			schemarel = heap_open(schemarelid, AccessShareLock);
+			rels = lappend(rels, schemarel);
+		}
+
+		systable_endscan(scan);
+		heap_close(rel, AccessShareLock);
+
+		/* Don't error on missing tables on DROP */
+		missing_ok = true;
+
+		/* Don't error on already existing tables on ADD. */
+		if_not_exists = true;
+	}
+	else
+	{
+		List	   *relids = NIL;
+
+		Assert(list_length(stmt->tables) > 0);
+
+		/*
+		 * Open, share-lock, and check all the explicitly-specified relations
+		 */
+
+		foreach(lc, stmt->tables)
+		{
+			RangeVar   *rv = lfirst(lc);
+			Relation	rel;
+			bool		recurse = interpretInhOption(rv->inhOpt);
+			Oid			myrelid;
+
+			rel = heap_openrv(rv, AccessShareLock);
+			myrelid = RelationGetRelid(rel);
+			/* don't throw error for "foo, foo" */
+			if (list_member_oid(relids, myrelid))
+			{
+				heap_close(rel, AccessShareLock);
+				continue;
+			}
+			rels = lappend(rels, rel);
+			relids = lappend_oid(relids, myrelid);
+
+			if (recurse)
+			{
+				ListCell   *child;
+				List	   *children;
+
+				children = find_all_inheritors(myrelid, AccessShareLock,
+											   NULL);
+
+				foreach(child, children)
+				{
+					Oid			childrelid = lfirst_oid(child);
+
+					if (list_member_oid(relids, childrelid))
+						continue;
+
+					/* find_all_inheritors already got lock */
+					rel = heap_open(childrelid, NoLock);
+					rels = lappend(rels, rel);
+					relids = lappend_oid(relids, childrelid);
+				}
+			}
+		}
+	}
+
+	/* Do the operation which was requested on all found relations. */
+	if (stmt->isDrop)
+	{
+		foreach(lc, rels)
+		{
+			Relation	rel = (Relation) lfirst(lc);
+			Oid			relid = RelationGetRelid(rel);
+
+			prid = GetSysCacheOid2(PUBLICATIONRELMAP, relid, pubid);
+			if (!OidIsValid(prid))
+			{
+				if (missing_ok)
+					continue;
+
+				ereport(ERROR,
+					(errcode(ERRCODE_UNDEFINED_OBJECT),
+					 errmsg("relation \"%s\" is not part of the publication",
+							RelationGetRelationName(rel))));
+			}
+
+			ObjectAddressSet(obj, PublicationRelRelationId, prid);
+			performDeletion(&obj, DROP_CASCADE, 0);
+		}
+	}
+	else
+	{
+		foreach(lc, rels)
+		{
+			Relation	rel = (Relation) lfirst(lc);
+
+			prid = publication_add_relation(pubid, rel, if_not_exists);
+			ObjectAddressSet(obj, PublicationRelRelationId, prid);
+			EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
+											 (Node *) stmt);
+		}
+	}
+
+	/* And close the rels */
+	foreach(lc, rels)
+	{
+		Relation	rel = (Relation) lfirst(lc);
+
+		heap_close(rel, NoLock);
+	}
+}
+
+/*
+ * Alter the existing publication.
+ *
+ * This is dispatcher function for AlterPublicationOptions and
+ * AlterPublicationTables.
+ */
+void
+AlterPublication(AlterPublicationStmt *stmt)
+{
+	Relation		rel;
+	HeapTuple		tup;
+
+	check_replication_permissions();
+
+	rel = heap_open(PublicationRelationId, RowExclusiveLock);
+
+	tup = SearchSysCacheCopy1(PUBLICATIONNAME,
+							  CStringGetDatum(stmt->pubname));
+
+	if (!HeapTupleIsValid(tup))
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("publication \"%s\" does not exist",
+						stmt->pubname)));
+
+	if (stmt->options)
+		AlterPublicationOptions(stmt, rel, tup);
+	else
+		AlterPublicationTables(stmt, rel, tup);
+
+	/* Cleanup. */
+	heap_freetuple(tup);
+	heap_close(rel, RowExclusiveLock);
+}
+
+/*
+ * Drop publication by OID
+ */
+void
+DropPublicationById(Oid pubid)
+{
+	Relation	rel;
+	HeapTuple	tup;
+
+	check_replication_permissions();
+
+	rel = heap_open(PublicationRelationId, RowExclusiveLock);
+
+	tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for publication %u", pubid);
+
+	simple_heap_delete(rel, &tup->t_self);
+
+	ReleaseSysCache(tup);
+
+	heap_close(rel, RowExclusiveLock);
+}
+
+/*
+ * Remove relation from publication by mapping OID.
+ */
+void
+RemovePublicationRelById(Oid prid)
+{
+	Relation        rel;
+	HeapTuple       tup;
+
+	rel = heap_open(PublicationRelRelationId, RowExclusiveLock);
+
+	tup = SearchSysCache1(PUBLICATIONREL, ObjectIdGetDatum(prid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for publication table %u",
+			 prid);
+
+	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 3244c76..bf76742 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4154,6 +4154,31 @@ _copyAlterPolicyStmt(const AlterPolicyStmt *from)
 	return newnode;
 }
 
+static CreatePublicationStmt *
+_copyCreatePublicationStmt(const CreatePublicationStmt *from)
+{
+	CreatePublicationStmt *newnode = makeNode(CreatePublicationStmt);
+
+	COPY_STRING_FIELD(pubname);
+	COPY_NODE_FIELD(options);
+
+	return newnode;
+}
+
+static AlterPublicationStmt *
+_copyAlterPublicationStmt(const AlterPublicationStmt *from)
+{
+	AlterPublicationStmt *newnode = makeNode(AlterPublicationStmt);
+
+	COPY_STRING_FIELD(pubname);
+	COPY_NODE_FIELD(options);
+	COPY_SCALAR_FIELD(isDrop);
+	COPY_STRING_FIELD(schema);
+	COPY_NODE_FIELD(tables);
+
+	return newnode;
+}
+
 /* ****************************************************************
  *					pg_list.h copy functions
  * ****************************************************************
@@ -4945,6 +4970,12 @@ copyObject(const void *from)
 		case T_AlterPolicyStmt:
 			retval = _copyAlterPolicyStmt(from);
 			break;
+		case T_CreatePublicationStmt:
+			retval = _copyCreatePublicationStmt(from);
+			break;
+		case T_AlterPublicationStmt:
+			retval = _copyAlterPublicationStmt(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 1eb6799..0c5f1d0 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2106,6 +2106,29 @@ _equalAlterTSConfigurationStmt(const AlterTSConfigurationStmt *a,
 }
 
 static bool
+_equalCreatePublicationStmt(const CreatePublicationStmt *a,
+							const CreatePublicationStmt *b)
+{
+	COMPARE_STRING_FIELD(pubname);
+	COMPARE_NODE_FIELD(options);
+
+	return true;
+}
+
+static bool
+_equalAlterPublicationStmt(const AlterPublicationStmt *a,
+						   const AlterPublicationStmt *b)
+{
+	COMPARE_STRING_FIELD(pubname);
+	COMPARE_NODE_FIELD(options);
+	COMPARE_SCALAR_FIELD(isDrop);
+	COMPARE_STRING_FIELD(schema);
+	COMPARE_NODE_FIELD(tables);
+
+	return true;
+}
+
+static bool
 _equalCreatePolicyStmt(const CreatePolicyStmt *a, const CreatePolicyStmt *b)
 {
 	COMPARE_STRING_FIELD(policy_name);
@@ -3249,6 +3272,12 @@ equal(const void *a, const void *b)
 		case T_AlterPolicyStmt:
 			retval = _equalAlterPolicyStmt(a, b);
 			break;
+		case T_CreatePublicationStmt:
+			retval = _equalCreatePublicationStmt(a, b);
+			break;
+		case T_AlterPublicationStmt:
+			retval = _equalAlterPublicationStmt(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 0cae446..b91e75a 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -266,6 +266,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 		DropOwnedStmt ReassignOwnedStmt
 		AlterTSConfigurationStmt AlterTSDictionaryStmt
 		CreateMatViewStmt RefreshMatViewStmt CreateAmStmt
+		CreatePublicationStmt AlterPublicationStmt
 
 %type <node>	select_no_parens select_with_parens select_clause
 				simple_select values_clause
@@ -372,13 +373,14 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 				create_generic_options alter_generic_options
 				relation_expr_list dostmt_opt_list
 				transform_element_list transform_type_list
+				publication_opt_list publication_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
-%type <defelt>	fdw_option
+%type <defelt>	fdw_option publication_opt_item
 
 %type <range>	OptTempTableName
 %type <into>	into_clause create_as_target create_mv_target
@@ -617,7 +619,7 @@ static Node *makeRecursiveViewSelect(char *relname, List *aliases, Node *query);
 
 	PARALLEL PARSER PARTIAL PARTITION PASSING PASSWORD PLACING PLANS POLICY
 	POSITION PRECEDING PRECISION PRESERVE PREPARE PREPARED PRIMARY
-	PRIOR PRIVILEGES PROCEDURAL PROCEDURE PROGRAM
+	PRIOR PRIVILEGES PROCEDURAL PROCEDURE PROGRAM PUBLICATION
 
 	QUOTE
 
@@ -778,6 +780,7 @@ stmt :
 			| AlterTableStmt
 			| AlterTblSpcStmt
 			| AlterCompositeTypeStmt
+			| AlterPublicationStmt
 			| AlterRoleSetStmt
 			| AlterRoleStmt
 			| AlterTSConfigurationStmt
@@ -807,6 +810,7 @@ stmt :
 			| CreateMatViewStmt
 			| CreateOpClassStmt
 			| CreateOpFamilyStmt
+			| CreatePublicationStmt
 			| AlterOpFamilyStmt
 			| CreatePolicyStmt
 			| CreatePLangStmt
@@ -5643,6 +5647,7 @@ drop_type:	TABLE									{ $$ = OBJECT_TABLE; }
 			| TEXT_P SEARCH DICTIONARY				{ $$ = OBJECT_TSDICTIONARY; }
 			| TEXT_P SEARCH TEMPLATE				{ $$ = OBJECT_TSTEMPLATE; }
 			| TEXT_P SEARCH CONFIGURATION			{ $$ = OBJECT_TSCONFIGURATION; }
+			| PUBLICATION							{ $$ = OBJECT_PUBLICATION; }
 		;
 
 any_name_list:
@@ -8498,6 +8503,113 @@ AlterOwnerStmt: ALTER AGGREGATE func_name aggr_args OWNER TO RoleSpec
 
 /*****************************************************************************
  *
+ * CREATE PUBLICATION name [ WITH options ]
+ *
+ *****************************************************************************/
+
+CreatePublicationStmt:
+			CREATE PUBLICATION name opt_with publication_opt_list
+				{
+					CreatePublicationStmt *n = makeNode(CreatePublicationStmt);
+					n->pubname = $3;
+					n->options = $5;
+					$$ = (Node *)n;
+				}
+		;
+
+publication_opt_list:
+			publication_opt_items					{ $$ = $1; }
+			| /* EMPTY */							{ $$ = NIL; }
+		;
+
+publication_opt_items:
+			publication_opt_item							{ $$ = list_make1($1); }
+			| publication_opt_items publication_opt_item	{ $$ = lappend($1, $2); }
+		;
+
+publication_opt_item:
+			IDENT
+				{
+					/*
+					 * We handle identifiers that aren't parser keywords with
+					 * the following special-case codes, to avoid bloating the
+					 * size of the main parser.
+					 */
+					if (strcmp($1, "replicate_insert") == 0)
+						$$ = makeDefElem("replicate_insert", (Node *)makeInteger(TRUE));
+					else if (strcmp($1, "noreplicate_insert") == 0)
+						$$ = makeDefElem("replicate_insert", (Node *)makeInteger(FALSE));
+					else if (strcmp($1, "replicate_update") == 0)
+						$$ = makeDefElem("replicate_update", (Node *)makeInteger(TRUE));
+					else if (strcmp($1, "noreplicate_update") == 0)
+						$$ = makeDefElem("replicate_update", (Node *)makeInteger(FALSE));
+					else if (strcmp($1, "replicate_delete") == 0)
+						$$ = makeDefElem("replicate_delete", (Node *)makeInteger(TRUE));
+					else if (strcmp($1, "noreplicate_delete") == 0)
+						$$ = makeDefElem("replicate_delete", (Node *)makeInteger(FALSE));
+					else
+						ereport(ERROR,
+								(errcode(ERRCODE_SYNTAX_ERROR),
+								 errmsg("unrecognized publication option \"%s\"", $1),
+									 parser_errposition(@1)));
+				}
+		;
+
+/*****************************************************************************
+ *
+ * ALTER PUBLICATION name [ WITH ] options
+ *
+ * ALTER PUBLICATION name ADD TABLE table [, table2]
+ *
+ * ALTER PUBLICATION name DROP TABLE table [, table2]
+ *
+ *****************************************************************************/
+
+AlterPublicationStmt:
+			ALTER PUBLICATION name opt_with publication_opt_items
+				{
+					AlterPublicationStmt *n = makeNode(AlterPublicationStmt);
+					n->pubname = $3;
+					n->options = $5;
+					n->isDrop = FALSE;
+					n->schema = NULL;
+					n->tables = NIL;
+					$$ = (Node *)n;
+				}
+			| ALTER PUBLICATION name ADD_P TABLE relation_expr_list
+				{
+					AlterPublicationStmt *n = makeNode(AlterPublicationStmt);
+					n->pubname = $3;
+					n->options = NIL;
+					n->isDrop = FALSE;
+					n->schema = NULL;
+					n->tables = $6;
+					$$ = (Node *)n;
+				}
+			| ALTER PUBLICATION name ADD_P TABLE ALL IN_P SCHEMA name
+				{
+					AlterPublicationStmt *n = makeNode(AlterPublicationStmt);
+					n->pubname = $3;
+					n->options = NIL;
+					n->isDrop = FALSE;
+					n->schema = $9;
+					n->tables = NIL;
+					$$ = (Node *)n;
+				}
+			| ALTER PUBLICATION name DROP TABLE relation_expr_list
+				{
+					AlterPublicationStmt *n = makeNode(AlterPublicationStmt);
+					n->pubname = $3;
+					n->options = NIL;
+					n->isDrop = TRUE;
+					n->schema = NULL;
+					n->tables = $6;
+					$$ = (Node *)n;
+				}
+		;
+
+/*****************************************************************************
+ *
  *		QUERY:	Define Rewrite Rule
  *
  *****************************************************************************/
@@ -13899,6 +14011,7 @@ unreserved_keyword:
 			| PROCEDURAL
 			| PROCEDURE
 			| PROGRAM
+			| PUBLICATION
 			| QUOTE
 			| RANGE
 			| READ
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 1d7ca06..3b3e90c 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -14,7 +14,7 @@ include $(top_builddir)/src/Makefile.global
 
 override CPPFLAGS := -I$(srcdir) $(CPPFLAGS)
 
-OBJS = decode.o logical.o logicalfuncs.o message.o origin.o reorderbuffer.o \
-	snapbuild.o
+OBJS = decode.o logical.o logicalfuncs.o message.o origin.o publication.o \
+	   reorderbuffer.o snapbuild.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/replication/logical/publication.c b/src/backend/replication/logical/publication.c
new file mode 100644
index 0000000..b86611e
--- /dev/null
+++ b/src/backend/replication/logical/publication.c
@@ -0,0 +1,343 @@
+/*-------------------------------------------------------------------------
+ *
+ * publication.c
+ *		publication C api manipulation
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		publication.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/dependency.h"
+#include "catalog/indexing.h"
+#include "catalog/namespace.h"
+#include "catalog/objectaddress.h"
+#include "catalog/pg_type.h"
+#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
+
+#include "executor/spi.h"
+
+#include "nodes/makefuncs.h"
+
+#include "replication/publication.h"
+#include "replication/reorderbuffer.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"
+
+
+/*
+ * Check if an action on a specified relation should be replicated for
+ * the publication list
+ */
+bool
+publication_change_is_replicated(Relation rel,
+								 PublicationChangeType change_type,
+								 List *pubnames)
+{
+	List	   *relpublications = GetRelationPublications(rel);
+	ListCell   *rellc;
+	bool		result = false;
+
+	/* TODO: optimize */
+	foreach (rellc, relpublications)
+	{
+		Oid	relpuboid = lfirst_oid(rellc);
+		ListCell   *namelc;
+
+		foreach (namelc, pubnames)
+		{
+			char	   *pubname = (char *) lfirst(namelc);
+			HeapTuple	tup;
+			Form_pg_publication	pub;
+
+			tup = SearchSysCache1(PUBLICATIONNAME, CStringGetDatum(pubname));
+			pub = (Form_pg_publication) GETSTRUCT(tup);
+
+			if (HeapTupleGetOid(tup) == relpuboid)
+			{
+				switch (change_type)
+				{
+					case PublicationChangeInsert:
+						result = pub->pubreplins;
+						break;
+					case PublicationChangeUpdate:
+						result = pub->pubreplupd;
+						break;
+					case PublicationChangeDelete:
+						result = pub->pubrepldel;
+						break;
+					default:
+						elog(ERROR, "unknown change_type %d", change_type);
+				}
+
+				/*
+				 * We don't need to search more once we found a publication
+				 * that replicates the change type.
+				 */
+				if (result)
+				{
+					list_free(relpublications);
+					return result;
+				}
+			}
+		}
+	}
+
+	list_free(relpublications);
+	return result;
+}
+
+
+/*
+ * Insert new publication / relation mapping.
+ */
+Oid
+publication_add_relation(Oid pubid, Relation targetrel,
+						 bool if_not_exists)
+{
+	Relation	rel;
+	HeapTuple	tup;
+	Datum		values[Natts_pg_publication_rel];
+	bool		nulls[Natts_pg_publication_rel];
+	Oid			relid = RelationGetRelid(targetrel);
+	Oid			prid;
+	Publication *pub = GetPublication(pubid);
+	ObjectAddress	myself,
+					referenced;
+
+	rel = heap_open(PublicationRelRelationId, RowExclusiveLock);
+
+	/* Check for duplicates */
+	if (SearchSysCacheExists2(PUBLICATIONRELMAP, relid, pubid))
+	{
+		heap_close(rel, RowExclusiveLock);
+
+		if (if_not_exists)
+			return InvalidOid;
+
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("relation %s is already member of publication %s",
+						RelationGetRelationName(rel), pub->name)));
+	}
+
+	/* Must be table */
+	if (RelationGetForm(rel)->relkind != RELKIND_RELATION)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("only tables can be added to publication"),
+				 errdetail("%s is not a table",
+						   RelationGetRelationName(rel))));
+
+	/* UNLOGGED and TEMP relations cannot be part of publication. */
+	if (!RelationNeedsWAL(targetrel))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("UNLOGGED and TEMP relations cannot be replicated")));
+
+	/* Must have replica identity index. */
+	if (targetrel->rd_indexvalid == 0)
+		RelationGetIndexList(targetrel);
+	if (!OidIsValid(targetrel->rd_replidindex) &&
+		(pub->replicate_update || pub->replicate_delete))
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("table %s cannot be added to publication %s",
+						RelationGetRelationName(targetrel), pub->name),
+				 errdetail("table does not have REPLICA IDENTITY index "
+						   "and given publication is configured to "
+						   "replicate UPDATEs and/or DELETEs"),
+				 errhint("Add a PRIMARY KEY to the table")));
+
+	/* Form a tuple. */
+	memset(values, 0, sizeof(values));
+	memset(nulls, false, sizeof(nulls));
+
+	values[Anum_pg_publication_rel_pubid - 1] =
+		ObjectIdGetDatum(pubid);
+	values[Anum_pg_publication_rel_relid - 1] =
+		ObjectIdGetDatum(RelationGetRelid(targetrel));
+
+	tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
+
+	/* Insert tuple into catalog. */
+	prid = simple_heap_insert(rel, tup);
+	CatalogUpdateIndexes(rel, tup);
+	heap_freetuple(tup);
+
+	/* Add dependency on the publication */
+	ObjectAddressSet(myself, PublicationRelRelationId, prid);
+	ObjectAddressSet(referenced, PublicationRelationId, pubid);
+	recordDependencyOn(&myself, &referenced, DEPENDENCY_AUTO);
+
+	/* Add dependency on the relation */
+	ObjectAddressSet(referenced, RelationRelationId, relid);
+	recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+
+	/*
+	 * In case the publication replicates updates and deletes we also
+	 * need to record dependency on the replica index.
+	 *
+	 * XXX: this has unpleasant sideeffect that replacing replica index
+	 * (PKey in most cases) means removal of table from publication. We
+	 * could habdle is better if we checked specifically for this during
+	 * execution of the related commands.
+	 */
+	if (pub->replicate_update || pub->replicate_delete)
+	{
+		ObjectAddressSet(referenced, IndexRelationId,
+						 targetrel->rd_replidindex);
+		recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
+	}
+
+	/* Close the table. */
+	heap_close(rel, RowExclusiveLock);
+
+	return prid;
+}
+
+
+/*
+ * Gets list of publication oids for a relation.
+ */
+List *
+GetRelationPublications(Relation rel)
+{
+	List		   *result;
+	List		   *oldlist;
+	Relation		pubrelsrel;
+	ScanKeyData		scankey;
+	SysScanDesc		scan;
+	HeapTuple		tup;
+	MemoryContext	oldcxt;
+
+	/* Quick exit if we already computed the list. */
+	if (rel->rd_publicationsvalid)
+		return list_copy(rel->rd_publications);
+
+	/* Find all publications associated with the relation. */
+	pubrelsrel = heap_open(PublicationRelRelationId, AccessShareLock);
+
+	ScanKeyInit(&scankey,
+				Anum_pg_publication_rel_relid,
+				BTEqualStrategyNumber, F_OIDEQ,
+				ObjectIdGetDatum(RelationGetRelid(rel)));
+
+	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->pubid);
+	}
+
+	systable_endscan(scan);
+	heap_close(pubrelsrel, NoLock);
+
+	/* Now save a copy of the completed list in the relcache entry. */
+	oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+	oldlist = rel->rd_publications;
+	rel->rd_publications = list_copy(result);
+	rel->rd_publicationsvalid = true;
+	MemoryContextSwitchTo(oldcxt);
+
+	/* Don't leak the old list, if there is one */
+	list_free(oldlist);
+
+	return result;
+}
+
+Publication *
+GetPublication(Oid pubid)
+{
+	HeapTuple		tup;
+	Publication	   *pub;
+	Form_pg_publication	pubform;
+
+	tup = SearchSysCache1(PUBLICATIONOID, ObjectIdGetDatum(pubid));
+
+	if (!HeapTupleIsValid(tup))
+		elog(ERROR, "cache lookup failed for publication %u", pubid);
+
+	pubform = (Form_pg_publication) GETSTRUCT(tup);
+
+	pub = (Publication *) palloc(sizeof(Publication));
+	pub->oid = pubid;
+	pub->name = NameStr(pubform->pubname);
+	pub->replicate_insert = pubform->pubreplins;
+	pub->replicate_update = pubform->pubreplupd;
+	pub->replicate_delete = pubform->pubrepldel;
+
+	ReleaseSysCache(tup);
+
+	return pub;
+}
+
+
+/*
+ * Get Publication using name.
+ */
+Publication *
+GetPublicationByName(const char *pubname, bool missing_ok)
+{
+	Oid			oid;
+
+	oid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(pubname));
+	if (!OidIsValid(oid))
+	{
+		if (missing_ok)
+			return NULL;
+
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("publication \"%s\" does not exist", pubname)));
+	}
+
+	return GetPublication(oid);
+}
+
+/*
+ * get_publication_oid - given a publication name, look up the OID
+ *
+ * If missing_ok is false, throw an error if name not found.  If true, just
+ * return InvalidOid.
+ */
+Oid
+get_publication_oid(const char *pubname, bool missing_ok)
+{
+	Oid			oid;
+
+	oid = GetSysCacheOid1(PUBLICATIONNAME, CStringGetDatum(pubname));
+	if (!OidIsValid(oid) && !missing_ok)
+		ereport(ERROR,
+				(errcode(ERRCODE_UNDEFINED_OBJECT),
+				 errmsg("publication \"%s\" does not exist", pubname)));
+	return oid;
+}
diff --git a/src/backend/tcop/utility.c b/src/backend/tcop/utility.c
index ac50c2a..886d2ff 100644
--- a/src/backend/tcop/utility.c
+++ b/src/backend/tcop/utility.c
@@ -44,6 +44,7 @@
 #include "commands/portalcmds.h"
 #include "commands/prepare.h"
 #include "commands/proclang.h"
+#include "commands/replicationcmds.h"
 #include "commands/schemacmds.h"
 #include "commands/seclabel.h"
 #include "commands/sequence.h"
@@ -210,6 +211,8 @@ check_xact_readonly(Node *parsetree)
 		case T_CreateForeignTableStmt:
 		case T_ImportForeignSchemaStmt:
 		case T_SecLabelStmt:
+		case T_CreatePublicationStmt:
+		case T_AlterPublicationStmt:
 			PreventCommandIfReadOnly(CreateCommandTag(parsetree));
 			PreventCommandIfParallelMode(CreateCommandTag(parsetree));
 			break;
@@ -1544,6 +1547,19 @@ ProcessUtilitySlow(Node *parsetree,
 				address = CreateAccessMethod((CreateAmStmt *) parsetree);
 				break;
 
+			case T_CreatePublicationStmt:
+				address = CreatePublication((CreatePublicationStmt *) parsetree);
+				break;
+
+			case T_AlterPublicationStmt:
+				AlterPublication((AlterPublicationStmt *) parsetree);
+				/*
+				 * AlterPublication calls EventTriggerCollectSimpleCommand
+				 * directly
+				 */
+				commandCollected = true;
+				break;
+
 			default:
 				elog(ERROR, "unrecognized node type: %d",
 					 (int) nodeTag(parsetree));
@@ -1902,6 +1918,9 @@ AlterObjectTypeCommandTag(ObjectType objtype)
 		case OBJECT_MATVIEW:
 			tag = "ALTER MATERIALIZED VIEW";
 			break;
+		case OBJECT_PUBLICATION:
+			tag = "PUBLICATION";
+			break;
 		default:
 			tag = "???";
 			break;
@@ -2187,6 +2206,9 @@ CreateCommandTag(Node *parsetree)
 				case OBJECT_ACCESS_METHOD:
 					tag = "DROP ACCESS METHOD";
 					break;
+				case OBJECT_PUBLICATION:
+					tag = "DROP PUBLICATION";
+					break;
 				default:
 					tag = "???";
 			}
@@ -2557,6 +2579,14 @@ CreateCommandTag(Node *parsetree)
 			tag = "CREATE ACCESS METHOD";
 			break;
 
+		case T_CreatePublicationStmt:
+			tag = "CREATE PUBLICATION";
+			break;
+
+		case T_AlterPublicationStmt:
+			tag = "ALTER PUBLICATION";
+			break;
+
 		case T_PrepareStmt:
 			tag = "PREPARE";
 			break;
@@ -3122,6 +3152,14 @@ GetCommandLogLevel(Node *parsetree)
 			lev = LOGSTMT_DDL;
 			break;
 
+		case T_CreatePublicationStmt:
+			lev = LOGSTMT_DDL;
+			break;
+
+		case T_AlterPublicationStmt:
+			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 8d2ad01..3f7027f 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -2052,6 +2052,8 @@ RelationDestroyRelation(Relation relation, bool remember_tupdesc)
 		MemoryContextDelete(relation->rd_rsdesc->rscxt);
 	if (relation->rd_fdwroutine)
 		pfree(relation->rd_fdwroutine);
+	if (relation->rd_publications)
+		list_free(relation->rd_publications);
 	pfree(relation);
 }
 
@@ -5053,6 +5055,13 @@ load_relcache_init_file(bool shared)
 		rel->rd_fdwroutine = NULL;
 
 		/*
+		 * Publications are not needed by most backends so we load them on
+		 * demand.
+		 */
+		rel->rd_publicationsvalid = false;
+		rel->rd_publications = NIL;
+
+		/*
 		 * Reset transient-state fields in the relcache entry
 		 */
 		rel->rd_smgr = NULL;
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index 65ffe84..d575b51 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -50,6 +50,8 @@
 #include "catalog/pg_opfamily.h"
 #include "catalog/pg_proc.h"
 #include "catalog/pg_range.h"
+#include "catalog/pg_publication.h"
+#include "catalog/pg_publication_rel.h"
 #include "catalog/pg_rewrite.h"
 #include "catalog/pg_seclabel.h"
 #include "catalog/pg_shdepend.h"
@@ -645,6 +647,50 @@ static const struct cachedesc cacheinfo[] = {
 		},
 		16
 	},
+	{PublicationRelationId,			/* PUBLICATIONOID */
+		PublicationObjectIndexId,
+		1,
+		{
+			ObjectIdAttributeNumber,
+			0,
+			0,
+			0
+		},
+		8
+	},
+	{PublicationRelationId,			/* PUBLICATIONNAME */
+		PublicationNameIndexId,
+		1,
+		{
+			Anum_pg_publication_pubname,
+			0,
+			0,
+			0
+		},
+		8
+	},
+	{PublicationRelRelationId,		/* PUBLICATIONREL */
+		PublicationRelObjectIndexId,
+		1,
+		{
+			ObjectIdAttributeNumber,
+			0,
+			0,
+			0
+		},
+		64
+	},
+	{PublicationRelRelationId,		/* PUBLICATIONRELMAP */
+		PublicationRelMapIndexId,
+		2,
+		{
+			Anum_pg_publication_rel_relid,
+			Anum_pg_publication_rel_pubid,
+			0,
+			0
+		},
+		64
+	},
 	{RewriteRelationId,			/* RULERELNAME */
 		RewriteRelRulenameIndexId,
 		2,
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index 3f2cebf..a379c19 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -477,17 +477,30 @@ exec_command(const char *cmd,
 				success = listTables(&cmd[1], pattern, show_verbose, show_system);
 				break;
 			case 'r':
-				if (cmd[2] == 'd' && cmd[3] == 's')
+				switch (cmd[2])
 				{
-					char	   *pattern2 = NULL;
+					case 'd':
+					{
+						if (cmd[3] == 's')
+						{
+							char	   *pattern2 = NULL;
 
-					if (pattern)
-						pattern2 = psql_scan_slash_option(scan_state,
+							if (pattern)
+								pattern2 = psql_scan_slash_option(scan_state,
 													  OT_NORMAL, NULL, true);
-					success = listDbRoleSettings(pattern, pattern2);
+							success = listDbRoleSettings(pattern, pattern2);
+						}
+						else
+							success = PSQL_CMD_UNKNOWN;
+						break;
+					}
+					case 'p':
+						success = describePublications(pattern, show_verbose);
+						break;
+					default:
+						status = PSQL_CMD_UNKNOWN;
+						break;
 				}
-				else
-					success = PSQL_CMD_UNKNOWN;
 				break;
 			case 'u':
 				success = describeRoles(pattern, show_verbose, show_system);
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 27be102..573d980 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -2290,6 +2290,38 @@ describeOneTableDetails(const char *schemaname,
 			}
 			PQclear(result);
 		}
+
+		/* print any publications */
+		/* TODO: bump version */
+		if (pset.sversion >= 90600)
+		{
+			printfPQExpBuffer(&buf,
+							  "SELECT pub.pubname\n"
+							  "FROM pg_catalog.pg_publication pub,\n"
+							  "     pg_publication_rel pr\n"
+							  "WHERE pr.relid = '%s' AND pr.pubid = pub.oid\n"
+							  "ORDER BY 1;",
+							  oid);
+
+			result = PSQLexec(buf.data);
+			if (!result)
+				goto error_return;
+			else
+				tuples = PQntuples(result);
+
+			if (tuples > 0)
+				printTableAddFooter(&cont, _("Publications:"));
+
+			/* Might be an empty set - that's ok */
+			for (i = 0; i < tuples; i++)
+			{
+				printfPQExpBuffer(&buf, "    \"%s\"",
+								  PQgetvalue(result, i, 0));
+
+				printTableAddFooter(&cont, buf.data);
+			}
+			PQclear(result);
+		}
 	}
 
 	if (view_def)
@@ -4688,6 +4720,70 @@ listOneExtensionContents(const char *extname, const char *oid)
 	return true;
 }
 
+/* \drp
+ * Describes publications.
+ *
+ * Takes an optional regexp to select particular publications
+ */
+bool
+describePublications(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 publications.\n",
+				   pset.sversion / 10000, (pset.sversion / 100) % 100);
+		return true;
+	}
+
+	initPQExpBuffer(&buf);
+
+	printfPQExpBuffer(&buf,
+					  "SELECT pubname AS \"%s\",\n"
+					  "  pubreplins AS \"%s\",\n"
+					  "  pubreplupd AS \"%s\",\n"
+					  "  pubrepldel AS \"%s\"\n",
+					  gettext_noop("Name"),
+					  gettext_noop("Inserts"),
+					  gettext_noop("Updates"),
+					  gettext_noop("Deletes"));
+
+	/* TODO Show owner and ACL */
+	if (verbose)
+	{
+	}
+
+	appendPQExpBufferStr(&buf,
+						 "\nFROM pg_catalog.pg_publication\n");
+
+	processSQLNamePattern(pset.db, &buf, pattern, false, false,
+						  NULL, "pubname", NULL,
+						  NULL);
+
+	appendPQExpBufferStr(&buf, "ORDER BY 1;");
+
+	res = PSQLexec(buf.data);
+	termPQExpBuffer(&buf);
+	if (!res)
+		return false;
+
+	myopt.nullPrint = NULL;
+	myopt.title = _("List of publications");
+	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 20a6508..c4457da 100644
--- a/src/bin/psql/describe.h
+++ b/src/bin/psql/describe.h
@@ -102,4 +102,7 @@ extern bool listExtensionContents(const char *pattern);
 /* \dy */
 extern bool listEventTriggers(const char *pattern, bool verbose);
 
+/* \drp */
+bool describePublications(const char *pattern, bool verbose);
+
 #endif   /* DESCRIBE_H */
diff --git a/src/bin/psql/help.c b/src/bin/psql/help.c
index efc8454..5cc5bce 100644
--- a/src/bin/psql/help.c
+++ b/src/bin/psql/help.c
@@ -241,6 +241,7 @@ slashUsage(unsigned short int pager)
 	fprintf(output, _("  \\dO[S+] [PATTERN]      list collations\n"));
 	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, _("  \\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 09b36c5..01fd16a 100644
--- a/src/include/catalog/dependency.h
+++ b/src/include/catalog/dependency.h
@@ -161,6 +161,8 @@ typedef enum ObjectClass
 	OCLASS_EXTENSION,			/* pg_extension */
 	OCLASS_EVENT_TRIGGER,		/* pg_event_trigger */
 	OCLASS_POLICY,				/* pg_policy */
+	OCLASS_PUBLICATION,			/* pg_publication */
+	OCLASS_PUBLICATION_REL,		/* pg_publication_rel */
 	OCLASS_TRANSFORM			/* pg_transform */
 } ObjectClass;
 
diff --git a/src/include/catalog/indexing.h b/src/include/catalog/indexing.h
index ca5eb3d..ae46ed6 100644
--- a/src/include/catalog/indexing.h
+++ b/src/include/catalog/indexing.h
@@ -319,6 +319,18 @@ DECLARE_UNIQUE_INDEX(pg_replication_origin_roiident_index, 6001, on pg_replicati
 DECLARE_UNIQUE_INDEX(pg_replication_origin_roname_index, 6002, on pg_replication_origin using btree(roname text_pattern_ops));
 #define ReplicationOriginNameIndex 6002
 
+DECLARE_UNIQUE_INDEX(pg_publication_oid_index, 6110, on pg_publication using btree(oid oid_ops));
+#define PublicationObjectIndexId 6110
+
+DECLARE_UNIQUE_INDEX(pg_publication_pubname_index, 6111, on pg_publication using btree(pubname name_ops));
+#define PublicationNameIndexId 6111
+
+DECLARE_UNIQUE_INDEX(pg_publication_rel_object_index, 6112, on pg_publication_rel using btree(oid oid_ops));
+#define PublicationRelObjectIndexId 6112
+
+DECLARE_UNIQUE_INDEX(pg_publication_rel_map_index, 6113, on pg_publication_rel using btree(relid oid_ops, pubid oid_ops));
+#define PublicationRelMapIndexId 6113
+
 /* last step of initialization script: build the indexes declared above */
 BUILD_INDICES
 
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
new file mode 100644
index 0000000..6b5263a
--- /dev/null
+++ b/src/include/catalog/pg_publication.h
@@ -0,0 +1,64 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_publication.h
+ *	  definition of the relation sets relation (pg_publication)
+  *
+ * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_publication.h
+ *
+ * NOTES
+ *	  the genbki.pl script reads this file and generates .bki
+ *	  information from the DATA() statements.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_PUBLICATION_H
+#define PG_PUBLICATION_H
+
+#include "catalog/genbki.h"
+
+/* ----------------
+ *		pg_publication definition.  cpp turns this into
+ *		typedef struct FormData_pg_publication
+ *
+ * ----------------
+ */
+#define PublicationRelationId			6104
+#define PublicationRelation_Rowtype_Id	6105
+
+CATALOG(pg_publication,6104) BKI_ROWTYPE_OID(6105)
+{
+	NameData	pubname;			/* name of the publication */
+
+	/* true if inserts are replicated */
+	bool		pubreplins;
+
+	/* true if inserts are replicated */
+	bool		pubreplupd;
+
+	/* true if inserts are replicated */
+	bool		pubrepldel;
+
+} FormData_pg_publication;
+
+/* ----------------
+ *		Form_pg_publication corresponds to a pointer to a tuple with
+ *		the format of pg_publication relation.
+ * ----------------
+ */
+typedef FormData_pg_publication *Form_pg_publication;
+
+/* ----------------
+ *		compiler constants for pg_publication
+ * ----------------
+ */
+
+#define Natts_pg_publication				4
+#define Anum_pg_publication_pubname			1
+#define Anum_pg_publication_pubreplins		2
+#define Anum_pg_publication_pubreplupd		3
+#define Anum_pg_publication_pubrepldel		4
+
+#endif   /* PG_PUBLICATION_H */
diff --git a/src/include/catalog/pg_publication_rel.h b/src/include/catalog/pg_publication_rel.h
new file mode 100644
index 0000000..976c1a9
--- /dev/null
+++ b/src/include/catalog/pg_publication_rel.h
@@ -0,0 +1,53 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_publication_rel.h
+ *	  definition of the publication to relation map (pg_publication_rel)
+  *
+ * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/catalog/pg_publication_rel.h
+ *
+ * NOTES
+ *	  the genbki.pl script reads this file and generates .bki
+ *	  information from the DATA() statements.
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PG_PUBLICATION_REL_H
+#define PG_PUBLICATION_REL_H
+
+#include "catalog/genbki.h"
+
+/* ----------------
+ *		pg_publication_rel definition.  cpp turns this into
+ *		typedef struct FormData_pg_publication_rel
+ *
+ * ----------------
+ */
+#define PublicationRelRelationId				6106
+#define PublicationRelRelation_Rowtype_Id	6107
+
+CATALOG(pg_publication_rel,6106) BKI_ROWTYPE_OID(6107)
+{
+	Oid		pubid;				/* Oid of the publication */
+	Oid		relid;				/* Oid of the relation */
+} FormData_pg_publication_rel;
+
+/* ----------------
+ *		Form_pg_publication_rel corresponds to a pointer to a tuple with
+ *		the format of pg_publication_rel relation.
+ * ----------------
+ */
+typedef FormData_pg_publication_rel *Form_pg_publication_rel;
+
+/* ----------------
+ *		compiler constants for pg_publication_rel
+ * ----------------
+ */
+
+#define Natts_pg_publication_rel				2
+#define Anum_pg_publication_rel_pubid			1
+#define Anum_pg_publication_rel_relid			2
+
+#endif   /* PG_PUBLICATION_REL_H */
diff --git a/src/include/commands/replicationcmds.h b/src/include/commands/replicationcmds.h
new file mode 100644
index 0000000..717485f
--- /dev/null
+++ b/src/include/commands/replicationcmds.h
@@ -0,0 +1,26 @@
+/*-------------------------------------------------------------------------
+ *
+ * replicationcmds.h
+ *	  prototypes for publicationcmds.c.
+ *
+ *
+ * Portions Copyright (c) 1996-2016, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * src/include/commands/replicationcmds.h
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#ifndef REPLICATIONCMDS_H
+#define REPLICATIONCMDS_H
+
+#include "catalog/objectaddress.h"
+#include "nodes/parsenodes.h"
+
+extern ObjectAddress CreatePublication(CreatePublicationStmt *stmt);
+extern void AlterPublication(AlterPublicationStmt *stmt);
+extern void DropPublicationById(Oid pubid);
+extern void RemovePublicationRelById(Oid prid);
+
+#endif   /* REPLICATIONCMDS_H */
diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h
index 6b850e4..3cce3d9 100644
--- a/src/include/nodes/nodes.h
+++ b/src/include/nodes/nodes.h
@@ -405,6 +405,8 @@ typedef enum NodeTag
 	T_AlterPolicyStmt,
 	T_CreateTransformStmt,
 	T_CreateAmStmt,
+	T_CreatePublicationStmt,
+	T_AlterPublicationStmt,
 
 	/*
 	 * TAGS FOR PARSE TREE NODES (parsenodes.h)
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 1481fff..c10e6e7 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -1408,6 +1408,8 @@ typedef enum ObjectType
 	OBJECT_OPERATOR,
 	OBJECT_OPFAMILY,
 	OBJECT_POLICY,
+	OBJECT_PUBLICATION,
+	OBJECT_PUBLICATION_REL,
 	OBJECT_ROLE,
 	OBJECT_RULE,
 	OBJECT_SCHEMA,
@@ -3101,4 +3103,26 @@ typedef struct AlterTSConfigurationStmt
 	bool		missing_ok;		/* for DROP - skip error if missing? */
 } AlterTSConfigurationStmt;
 
+
+typedef struct CreatePublicationStmt
+{
+	NodeTag		type;
+	char	   *pubname;		/* Name of of the publication */
+	List	   *options;		/* List of DefElem nodes */
+} CreatePublicationStmt;
+
+typedef struct AlterPublicationStmt
+{
+	NodeTag		type;
+	char	   *pubname;		/* Name of of the publication */
+
+	/* parameters used for ALTER PUBLICATION ... WITH */
+	List	   *options;		/* List of DefElem nodes */
+
+	/* parameters used for ALTER PUBLICATION ... ADD/DROP TABLE */
+	bool		isDrop;			/* Are tables to be added or dropped? */
+	char	   *schema;			/* ALL IN SCHEMA ... */
+	List	   *tables;			/* List of tables to add/drop */
+} AlterPublicationStmt;
+
 #endif   /* PARSENODES_H */
diff --git a/src/include/parser/kwlist.h b/src/include/parser/kwlist.h
index 17ffef5..9430ff0 100644
--- a/src/include/parser/kwlist.h
+++ b/src/include/parser/kwlist.h
@@ -304,6 +304,7 @@ PG_KEYWORD("privileges", PRIVILEGES, UNRESERVED_KEYWORD)
 PG_KEYWORD("procedural", PROCEDURAL, UNRESERVED_KEYWORD)
 PG_KEYWORD("procedure", PROCEDURE, UNRESERVED_KEYWORD)
 PG_KEYWORD("program", PROGRAM, UNRESERVED_KEYWORD)
+PG_KEYWORD("publication", PUBLICATION, UNRESERVED_KEYWORD)
 PG_KEYWORD("quote", QUOTE, UNRESERVED_KEYWORD)
 PG_KEYWORD("range", RANGE, UNRESERVED_KEYWORD)
 PG_KEYWORD("read", READ, UNRESERVED_KEYWORD)
diff --git a/src/include/replication/publication.h b/src/include/replication/publication.h
new file mode 100644
index 0000000..08245ee
--- /dev/null
+++ b/src/include/replication/publication.h
@@ -0,0 +1,47 @@
+/*-------------------------------------------------------------------------
+ *
+ * publication.h
+ *		publication support structures and interfaces
+ *
+ * Copyright (c) 2015, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		publication.h
+ *
+ *-------------------------------------------------------------------------
+ */
+#ifndef PUBLICATION_H
+#define PUBLICATION_H
+
+#include "postgres.h"
+
+
+typedef enum PublicationChangeType
+{
+	PublicationChangeInsert,
+	PublicationChangeUpdate,
+	PublicationChangeDelete
+} PublicationChangeType;
+
+typedef struct Publication
+{
+	Oid		oid;
+	char   *name;
+	bool	replicate_insert;
+	bool	replicate_update;
+	bool	replicate_delete;
+} Publication;
+
+extern Publication *GetPublication(Oid pubid);
+extern Publication *GetPublicationByName(const char *pubname, bool missing_ok);
+extern List *GetRelationPublications(Relation rel);
+
+extern bool publication_change_is_replicated(Relation rel,
+								 PublicationChangeType change_type,
+								 List *inpublications);
+extern Oid publication_add_relation(Oid pubid, Relation targetrel,
+							 bool if_not_exists);
+
+extern Oid get_publication_oid(const char *pubname, bool missing_ok);
+
+#endif		/* PUBLICATION_H */
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index ed14442..ca2283a 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -159,6 +159,10 @@ typedef struct RelationData
 	/* use "struct" here to avoid needing to include fdwapi.h: */
 	struct FdwRoutine *rd_fdwroutine;	/* cached function pointers, or NULL */
 
+	/* Publication support */
+	bool		rd_publicationsvalid;		/* state of the rd_publications */
+	List	   *rd_publications;
+
 	/*
 	 * Hack for CLUSTER, rewriting ALTER TABLE, etc: when writing a new
 	 * version of a table, we need to make any toast pointers inserted into it
diff --git a/src/include/utils/syscache.h b/src/include/utils/syscache.h
index 256615b..632fcbc 100644
--- a/src/include/utils/syscache.h
+++ b/src/include/utils/syscache.h
@@ -79,6 +79,10 @@ enum SysCacheIdentifier
 	RELOID,
 	REPLORIGIDENT,
 	REPLORIGNAME,
+	PUBLICATIONOID,
+	PUBLICATIONNAME,
+	PUBLICATIONREL,
+	PUBLICATIONRELMAP,
 	RULERELNAME,
 	STATRELATTINH,
 	TABLESPACEOID,
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
new file mode 100644
index 0000000..917408e
--- /dev/null
+++ b/src/test/regress/expected/publication.out
@@ -0,0 +1,76 @@
+--
+-- PUBLICATION
+--
+CREATE PUBLICATION testpub_default;
+CREATE PUBLICATION testpib_ins_trunct WITH noreplicate_delete noreplicate_update;
+ALTER PUBLICATION testpub_default WITH noreplicate_insert noreplicate_delete;
+\drp
+               List of publications
+        Name        | Inserts | Updates | Deletes 
+--------------------+---------+---------+---------
+ testpib_ins_trunct | t       | f       | f
+ testpub_default    | f       | t       | f
+(2 rows)
+
+ALTER PUBLICATION testpub_default WITH replicate_insert replicate_delete;
+\drp
+               List of publications
+        Name        | Inserts | Updates | Deletes 
+--------------------+---------+---------+---------
+ testpib_ins_trunct | t       | f       | f
+ testpub_default    | t       | t       | t
+(2 rows)
+
+--- adding tables
+CREATE TABLE testpub_tbl1 (id serial primary key, data text);
+CREATE TABLE testpub_nopk (foo int, bar int);
+ALTER PUBLICATION testpub_default ADD TABLE testpub_tbl1;
+ALTER PUBLICATION testpub_default ADD TABLE testpub_nopk;
+ERROR:  table testpub_nopk cannot be added to publication testpub_default
+DETAIL:  table does not have REPLICA IDENTITY index and given publication is configured to replicate UPDATEs and/or DELETEs
+HINT:  Add a PRIMARY KEY to the table
+ALTER PUBLICATION testpib_ins_trunct ADD TABLE testpub_nopk, testpub_tbl1;
+\d+ testpub_nopk
+                     Table "public.testpub_nopk"
+ Column |  Type   | Modifiers | Storage | Stats target | Description 
+--------+---------+-----------+---------+--------------+-------------
+ foo    | integer |           | plain   |              | 
+ bar    | integer |           | plain   |              | 
+Publications:
+    "testpib_ins_trunct"
+
+\d+ testpub_tbl1
+                                             Table "public.testpub_tbl1"
+ Column |  Type   |                         Modifiers                         | Storage  | Stats target | Description 
+--------+---------+-----------------------------------------------------------+----------+--------------+-------------
+ id     | integer | not null default nextval('testpub_tbl1_id_seq'::regclass) | plain    |              | 
+ data   | text    |                                                           | extended |              | 
+Indexes:
+    "testpub_tbl1_pkey" PRIMARY KEY, btree (id)
+Publications:
+    "testpib_ins_trunct"
+    "testpub_default"
+
+ALTER PUBLICATION testpub_default DROP TABLE testpub_tbl1, testpub_nopk;
+ERROR:  relation "testpub_nopk" is not part of the publication
+ALTER PUBLICATION testpub_default DROP TABLE testpub_tbl1;
+\d+ testpub_tbl1
+                                             Table "public.testpub_tbl1"
+ Column |  Type   |                         Modifiers                         | Storage  | Stats target | Description 
+--------+---------+-----------------------------------------------------------+----------+--------------+-------------
+ id     | integer | not null default nextval('testpub_tbl1_id_seq'::regclass) | plain    |              | 
+ data   | text    |                                                           | extended |              | 
+Indexes:
+    "testpub_tbl1_pkey" PRIMARY KEY, btree (id)
+Publications:
+    "testpib_ins_trunct"
+
+DROP TABLE testpub_tbl1;
+ERROR:  cannot drop table testpub_tbl1 because other objects depend on it
+DETAIL:  publication table testpub_tbl1 in publication testpib_ins_trunct depends on table testpub_tbl1
+HINT:  Use DROP ... CASCADE to drop the dependent objects too.
+DROP TABLE testpub_tbl1 CASCADE;
+NOTICE:  drop cascades to publication table testpub_tbl1 in publication testpib_ins_trunct
+DROP PUBLICATION testpub_default;
+DROP PUBLICATION testpib_ins_trunct;
+DROP TABLE testpub_nopk;
diff --git a/src/test/regress/expected/sanity_check.out b/src/test/regress/expected/sanity_check.out
index 1c087a3..5ab04ae 100644
--- a/src/test/regress/expected/sanity_check.out
+++ b/src/test/regress/expected/sanity_check.out
@@ -121,6 +121,8 @@ pg_opfamily|t
 pg_pltemplate|t
 pg_policy|t
 pg_proc|t
+pg_publication|t
+pg_publication_rel|t
 pg_range|t
 pg_replication_origin|t
 pg_rewrite|t
diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule
index 4ebad04..871a93d 100644
--- a/src/test/regress/parallel_schedule
+++ b/src/test/regress/parallel_schedule
@@ -84,7 +84,7 @@ test: select_into select_distinct select_distinct_on select_implicit select_havi
 # ----------
 # Another group of parallel tests
 # ----------
-test: brin gin gist spgist privileges init_privs security_label collate matview lock replica_identity rowsecurity object_address tablesample groupingsets drop_operator
+test: brin gin gist spgist privileges init_privs security_label collate matview lock replica_identity rowsecurity object_address tablesample groupingsets drop_operator publication
 
 # ----------
 # Another group of parallel tests
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
new file mode 100644
index 0000000..9f85081
--- /dev/null
+++ b/src/test/regress/sql/publication.sql
@@ -0,0 +1,40 @@
+--
+-- PUBLICATION
+--
+
+CREATE PUBLICATION testpub_default;
+
+CREATE PUBLICATION testpib_ins_trunct WITH noreplicate_delete noreplicate_update;
+
+ALTER PUBLICATION testpub_default WITH noreplicate_insert noreplicate_delete;
+
+\drp
+
+ALTER PUBLICATION testpub_default WITH replicate_insert replicate_delete;
+
+\drp
+
+--- adding tables
+CREATE TABLE testpub_tbl1 (id serial primary key, data text);
+CREATE TABLE testpub_nopk (foo int, bar int);
+
+ALTER PUBLICATION testpub_default ADD TABLE testpub_tbl1;
+ALTER PUBLICATION testpub_default ADD TABLE testpub_nopk;
+
+ALTER PUBLICATION testpib_ins_trunct ADD TABLE testpub_nopk, testpub_tbl1;
+
+\d+ testpub_nopk
+\d+ testpub_tbl1
+
+ALTER PUBLICATION testpub_default DROP TABLE testpub_tbl1, testpub_nopk;
+ALTER PUBLICATION testpub_default DROP TABLE testpub_tbl1;
+
+\d+ testpub_tbl1
+
+DROP TABLE testpub_tbl1;
+DROP TABLE testpub_tbl1 CASCADE;
+
+DROP PUBLICATION testpub_default;
+DROP PUBLICATION testpib_ins_trunct;
+
+DROP TABLE testpub_nopk;
-- 
2.7.4

