public inbox for [email protected]
help / color / mirror / Atom feedFrom: Tomas Vondra <[email protected]>
To: Peter Eisentraut <[email protected]>
To: Petr Jelinek <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: logical decoding and replication of sequences
Date: Wed, 23 Feb 2022 00:24:30 +0100
Message-ID: <[email protected]> (raw)
In-Reply-To: <[email protected]>
References: <[email protected]>
<CAA4eK1J5R=3H1kdt4B0-K6A=7PoELdOW23HpYaTXQfE0y4p6KQ@mail.gmail.com>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
<[email protected]>
Hi,
Here's an updated version of the patch, fixing most of the issues from
reviews so far. There's still a couple FIXME comments, but I think those
are minor and/or straightforward to deal with.
The main improvements:
1) Elimination of a lot of redundant code - one function handling
tables, and an almost exact copy handling sequences. Now a single
function handles both, possibly with "sequences" flag to tweak the behavior.
2) A couple other functions gained "sequences" flag, determining which
objects are "interesting". For example PublicationAddSchemas needs to
know whether it's FOR ALL SEQUENCES or FOR ALL TABLES IN SCHEMA. I don't
think we can just use relkind here easily, because for tables we need to
handle various types of tables (regular, partitioned, ...).
3) I also renamed a couple functions with "tables" in the name, which
are now used for sequences too. So for example OpenTablesList() is
renamed to OpenRelationList() and so on.
4) Addition of a number of regression tests to "publication.sql", which
showed a lot of issues, mostly related to not distinguishing tables and
sequences when handling "FOR ALL TABLES [IN SCHEMA]" and "FOR ALL
SEQUENCES [IN SCHEMA]".
5) Proper tracking of FOR ALL [TABLES|SEQUENCES] IN SCHEMA in a catalog.
The pg_publication_namespace gained a pnsequences flag, which determines
which case it is. So for example if you do
ALTER PUBLICATION p ADD ALL TABLES IN SCHEMA s;
ALTER PUBLICATION p ADD ALL SEQUENCES IN SCHEMA s;
there will be two rows in the catalog, one with 't' and one with 'f' in
the new column. I'm not sure this is the best way to track this - maybe
it'd be better to have two flags, and keep a single row. Or maybe we
should have an array of relkinds (but that has the issue with tables
having multiple relkinds mentioned before). Multiple rows make it more
convenient to add/remove publication schemas - with a single row it'd be
necessary to either insert a new row or update an existing one when
adding the schema, and similarly for dropping it.
But maybe there are reasons / precedent to design this differently ...
6) I'm not entirely sure the object_address changes (handling of the
pnsequences flag) are correct.
7) This updates psql to do roughly the same thing as for tables, so \dRp
now list sequences added either directly or through schema, so you might
get footer like this:
\dRp+ testpub_mix
...
Tables:
"public.testpub_tbl1"
Tables from schemas:
"pub_test"
Sequences:
"public.testpub_seq1"
Sequences from schemas:
"pub_test"
Maybe it's a bit too verbose, though. It also addes "All sequences" and
"Sequences" columns into the publication description, but I don't think
that can be done much differently.
FWIW I had to switch the describeOneTableDetails() chunk dealing with
sequences from printQuery() to printTable() in order to handle dynamic
footers.
There's also a change in \dn+ because a schema may be included in one
publication as "FOR ALL SEQUENCES IN SCHEMA" and in another publication
with "FOR ALL TABLES IN SCHEMA". So I modified the output to
\dn+ pub_test1
...
Publications:
"testpub_schemas" (sequences)
"testpub_schemas" (tables)
But maybe it'd be better to aggregate this into a single line like
\dn+ pub_test1
...
Publications:
"testpub_schemas" (tables, sequences)
Opinions?
8) There's a shortcoming in the current grammar, because you can specify
either
CREATE PUBLICATION p FOR ALL TABLES;
or
CREATE PUBLICATION p FOR ALL SEQUENCES;
but it's not possible to do
CREATE PUBLICATION p FOR ALL TABLES AND FOR ALL SEQUENCES;
which seems like a fairly reasonable thing users might want to do.
The problem is that "FOR ALL TABLES" (and same for sequences) is
hard-coded in the grammar, not defined as PublicationObjSpec. This also
means you can't do
ALTER PUBLICATION p ADD ALL TABLES;
AFAICS there are two ways to fix this - adding the combinations into the
definition of CreatePublicationStmt, or adding FOR ALL TABLES (and
sequences) to PublicationObjSpec.
9) Another grammar limitation is that we don't cross-check the relkind,
so for example
ALTER PUBLICATION p ADD TABLE sequence;
might actually work. Should be easy to fix, though.
10) Added pg_dump support (including tests). I'll add more tests, to
check more grammar combinations.
11) I need to test more grammar combinations in the TAP test too, to
verify the output plugin interprets the stuff correctly.
regards
--
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company
Attachments:
[text/x-patch] 0001-Add-support-for-decoding-sequences-to-built-20220222.patch (220.8K, ../[email protected]/2-0001-Add-support-for-decoding-sequences-to-built-20220222.patch)
download | inline diff:
From 66e38c7ef81b3a25db4f93d8d87f0e78b4d1bab7 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <[email protected]>
Date: Thu, 10 Feb 2022 15:18:59 +0100
Subject: [PATCH] Add support for decoding sequences to built-in replication
---
doc/src/sgml/catalogs.sgml | 81 ++
doc/src/sgml/ref/alter_publication.sgml | 24 +-
doc/src/sgml/ref/alter_subscription.sgml | 3 +-
doc/src/sgml/ref/create_publication.sgml | 3 +
src/backend/catalog/objectaddress.c | 5 +-
src/backend/catalog/pg_publication.c | 236 ++++-
src/backend/catalog/system_views.sql | 10 +
src/backend/commands/publicationcmds.c | 380 +++++--
src/backend/commands/sequence.c | 79 ++
src/backend/commands/subscriptioncmds.c | 272 +++++
src/backend/executor/execReplication.c | 2 +-
src/backend/nodes/copyfuncs.c | 1 +
src/backend/nodes/equalfuncs.c | 1 +
src/backend/parser/gram.y | 40 +
src/backend/replication/logical/proto.c | 52 +
src/backend/replication/logical/tablesync.c | 118 ++-
src/backend/replication/logical/worker.c | 56 +
src/backend/replication/pgoutput/pgoutput.c | 82 +-
src/backend/utils/cache/relcache.c | 13 +-
src/backend/utils/cache/syscache.c | 6 +-
src/bin/pg_dump/pg_dump.c | 65 +-
src/bin/pg_dump/pg_dump.h | 3 +
src/bin/pg_dump/t/002_pg_dump.pl | 36 +-
src/bin/psql/describe.c | 292 +++--
src/bin/psql/tab-complete.c | 12 +-
src/include/catalog/pg_proc.dat | 5 +
src/include/catalog/pg_publication.h | 25 +-
.../catalog/pg_publication_namespace.h | 3 +-
src/include/commands/sequence.h | 1 +
src/include/nodes/parsenodes.h | 6 +
src/include/replication/logicalproto.h | 19 +
src/include/replication/pgoutput.h | 1 +
src/interfaces/libpq/fe-exec.c | 9 +
src/test/regress/expected/publication.out | 999 ++++++++++++++----
src/test/regress/expected/rules.out | 8 +
src/test/regress/sql/object_address.sql | 1 -
src/test/regress/sql/publication.sql | 220 +++-
src/test/subscription/t/028_sequences.pl | 203 ++++
38 files changed, 2984 insertions(+), 388 deletions(-)
create mode 100644 src/test/subscription/t/028_sequences.pl
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 83987a99045..324fc1050d3 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -6263,6 +6263,16 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
Reference to schema
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>pnsequences</structfield> <type>bool</type>
+ Determines whether to include tables or sequences from this schema.
+ </para>
+ <para>
+ Reference to schema
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
@@ -9560,6 +9570,11 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
<entry>prepared transactions</entry>
</row>
+ <row>
+ <entry><link linkend="view-pg-publication-sequences"><structname>pg_publication_sequences</structname></link></entry>
+ <entry>publications and their associated sequences</entry>
+ </row>
+
<row>
<entry><link linkend="view-pg-publication-tables"><structname>pg_publication_tables</structname></link></entry>
<entry>publications and their associated tables</entry>
@@ -11395,6 +11410,72 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
</sect1>
+ <sect1 id="view-pg-publication-sequences">
+ <title><structname>pg_publication_sequences</structname></title>
+
+ <indexterm zone="view-pg-publication-sequences">
+ <primary>pg_publication_sequences</primary>
+ </indexterm>
+
+ <para>
+ The view <structname>pg_publication_sequences</structname> provides
+ information about the mapping between publications and the sequences they
+ contain. Unlike the underlying catalog
+ <link linkend="catalog-pg-publication-rel"><structname>pg_publication_rel</structname></link>,
+ this view expands
+ publications defined as <literal>FOR ALL SEQUENCES</literal>, so for such
+ publications there will be a row for each eligible sequence.
+ </para>
+
+ <table>
+ <title><structname>pg_publication_sequences</structname> Columns</title>
+ <tgroup cols="1">
+ <thead>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ Column Type
+ </para>
+ <para>
+ Description
+ </para></entry>
+ </row>
+ </thead>
+
+ <tbody>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>pubname</structfield> <type>name</type>
+ (references <link linkend="catalog-pg-publication"><structname>pg_publication</structname></link>.<structfield>pubname</structfield>)
+ </para>
+ <para>
+ Name of publication
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>schemaname</structfield> <type>name</type>
+ (references <link linkend="catalog-pg-namespace"><structname>pg_namespace</structname></link>.<structfield>nspname</structfield>)
+ </para>
+ <para>
+ Name of schema containing sequence
+ </para></entry>
+ </row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>sequencename</structfield> <type>name</type>
+ (references <link linkend="catalog-pg-class"><structname>pg_class</structname></link>.<structfield>relname</structfield>)
+ </para>
+ <para>
+ Name of sequence
+ </para></entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+ </sect1>
+
<sect1 id="view-pg-publication-tables">
<title><structname>pg_publication_tables</structname></title>
diff --git a/doc/src/sgml/ref/alter_publication.sgml b/doc/src/sgml/ref/alter_publication.sgml
index 32b75f6c78e..36c9a5f438a 100644
--- a/doc/src/sgml/ref/alter_publication.sgml
+++ b/doc/src/sgml/ref/alter_publication.sgml
@@ -31,7 +31,9 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
+ SEQUENCE <replaceable class="parameter">sequence_name</replaceable> [ * ] [, ... ]
ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
+ ALL SEQUENCES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
@@ -58,7 +60,18 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
</para>
<para>
- The fourth variant of this command listed in the synopsis can change
+ The next three variants change which sequences are part of the publication.
+ The <literal>SET SEQUENCE</literal> clause will replace the list of sequences
+ in the publication with the specified one. The <literal>ADD SEQUENCE</literal>
+ and <literal>DROP SEQUENCE</literal> clauses will add and remove one or more
+ sequences from the publication. Note that adding sequences to a publication
+ that is already subscribed to will require a <literal>ALTER SUBSCRIPTION
+ ... REFRESH PUBLICATION</literal> action on the subscribing side in order
+ to become effective.
+ </para>
+
+ <para>
+ The seventh variant of this command listed in the synopsis can change
all of the publication properties specified in
<xref linkend="sql-createpublication"/>. Properties not mentioned in the
command retain their previous settings.
@@ -122,6 +135,15 @@ ALTER PUBLICATION <replaceable class="parameter">name</replaceable> RENAME TO <r
</listitem>
</varlistentry>
+ <varlistentry>
+ <term><replaceable class="parameter">sequence_name</replaceable></term>
+ <listitem>
+ <para>
+ Name of an existing sequence.
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry>
<term><replaceable class="parameter">schema_name</replaceable></term>
<listitem>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 0d6f064f58d..264c9c5a10f 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -147,7 +147,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<listitem>
<para>
Fetch missing table information from publisher. This will start
- replication of tables that were added to the subscribed-to publications
+ replication of tables and sequences that were added to the subscribed-to publications
since <command>CREATE SUBSCRIPTION</command> or
the last invocation of <command>REFRESH PUBLICATION</command>.
</para>
@@ -168,6 +168,7 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<para>
Previously subscribed tables are not copied, even if a table's row
filter <literal>WHERE</literal> clause has since been modified.
+ Previously-subscribed sequences are not copied either.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index 4979b9b646d..f72318e97dd 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -23,13 +23,16 @@ PostgreSQL documentation
<synopsis>
CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
[ FOR ALL TABLES
+ | FOR ALL SEQUENCES
| FOR <replaceable class="parameter">publication_object</replaceable> [, ... ] ]
[ WITH ( <replaceable class="parameter">publication_parameter</replaceable> [= <replaceable class="parameter">value</replaceable>] [, ... ] ) ]
<phrase>where <replaceable class="parameter">publication_object</replaceable> is one of:</phrase>
TABLE [ ONLY ] <replaceable class="parameter">table_name</replaceable> [ * ] [ WHERE ( <replaceable class="parameter">expression</replaceable> ) ] [, ... ]
+ SEQUENCE <replaceable class="parameter">sequence_name</replaceable> [ * ] [, ... ]
ALL TABLES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
+ ALL SEQUENCES IN SCHEMA { <replaceable class="parameter">schema_name</replaceable> | CURRENT_SCHEMA } [, ... ]
</synopsis>
</refsynopsisdiv>
diff --git a/src/backend/catalog/objectaddress.c b/src/backend/catalog/objectaddress.c
index f30c742d48f..301eeb7caca 100644
--- a/src/backend/catalog/objectaddress.c
+++ b/src/backend/catalog/objectaddress.c
@@ -1976,10 +1976,11 @@ get_object_address_publication_schema(List *object, bool missing_ok)
/* Find the publication schema mapping in syscache */
address.objectId =
- GetSysCacheOid2(PUBLICATIONNAMESPACEMAP,
+ GetSysCacheOid3(PUBLICATIONNAMESPACEMAP,
Anum_pg_publication_namespace_oid,
ObjectIdGetDatum(schemaid),
- ObjectIdGetDatum(pub->oid));
+ ObjectIdGetDatum(pub->oid),
+ BoolGetDatum(false)); /* FIXME */
if (!OidIsValid(address.objectId) && !missing_ok)
ereport(ERROR,
(errcode(ERRCODE_UNDEFINED_OBJECT),
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 25998fbb39b..d866e8a9b22 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -54,7 +54,8 @@ check_publication_add_relation(Relation targetrel)
{
/* Must be a regular or partitioned table */
if (RelationGetForm(targetrel)->relkind != RELKIND_RELATION &&
- RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE)
+ RelationGetForm(targetrel)->relkind != RELKIND_PARTITIONED_TABLE &&
+ RelationGetForm(targetrel)->relkind != RELKIND_SEQUENCE)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("cannot add relation \"%s\" to publication",
@@ -131,7 +132,8 @@ static bool
is_publishable_class(Oid relid, Form_pg_class reltuple)
{
return (reltuple->relkind == RELKIND_RELATION ||
- reltuple->relkind == RELKIND_PARTITIONED_TABLE) &&
+ reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
+ reltuple->relkind == RELKIND_SEQUENCE) &&
!IsCatalogRelationOid(relid) &&
reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
relid >= FirstNormalObjectId;
@@ -205,7 +207,7 @@ is_schema_publication(Oid pubid)
ObjectIdGetDatum(pubid));
scan = systable_beginscan(pubschsrel,
- PublicationNamespacePnnspidPnpubidIndexId,
+ PublicationNamespacePnnspidPnpubidSeqIndexId,
true, NULL, 1, &scankey);
tup = systable_getnext(scan);
result = HeapTupleIsValid(tup);
@@ -419,7 +421,7 @@ publication_add_relation(Oid pubid, PublicationRelInfo *pri,
* Insert new publication / schema mapping.
*/
ObjectAddress
-publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
+publication_add_schema(Oid pubid, Oid schemaid, bool sequences, bool if_not_exists)
{
Relation rel;
HeapTuple tup;
@@ -438,9 +440,10 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
* duplicates, it's here just to provide nicer error message in common
* case. The real protection is the unique key on the catalog.
*/
- if (SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
+ if (SearchSysCacheExists3(PUBLICATIONNAMESPACEMAP,
ObjectIdGetDatum(schemaid),
- ObjectIdGetDatum(pubid)))
+ ObjectIdGetDatum(pubid),
+ BoolGetDatum(sequences)))
{
table_close(rel, RowExclusiveLock);
@@ -466,6 +469,8 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
ObjectIdGetDatum(pubid);
values[Anum_pg_publication_namespace_pnnspid - 1] =
ObjectIdGetDatum(schemaid);
+ values[Anum_pg_publication_namespace_pnsequences - 1] =
+ BoolGetDatum(sequences);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -492,6 +497,7 @@ publication_add_schema(Oid pubid, Oid schemaid, bool if_not_exists)
* partitions.
*/
schemaRels = GetSchemaPublicationRelations(schemaid,
+ sequences,
PUBLICATION_PART_ALL);
InvalidatePublicationRels(schemaRels);
@@ -525,11 +531,14 @@ GetRelationPublications(Oid relid)
/*
* Gets list of relation oids for a publication.
*
- * This should only be used FOR TABLE publications, the FOR ALL TABLES
- * should use GetAllTablesPublicationRelations().
+ * This should only be used FOR TABLE / FOR SEQUENCE publications, the FOR
+ * ALL TABLES / SEQUENCES should use GetAllTablesPublicationRelations()
+ * and GetAllSequencesPublicationRelations().
+ *
+ * XXX It's a bit weird the pub_partopt does not really matter for sequences.
*/
List *
-GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
+GetPublicationRelations(Oid pubid, bool sequences, PublicationPartOpt pub_partopt)
{
List *result;
Relation pubrelsrel;
@@ -551,11 +560,21 @@ GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
result = NIL;
while (HeapTupleIsValid(tup = systable_getnext(scan)))
{
+ char relkind;
Form_pg_publication_rel pubrel;
pubrel = (Form_pg_publication_rel) GETSTRUCT(tup);
- result = GetPubPartitionOptionRelations(result, pub_partopt,
- pubrel->prrelid);
+ relkind = get_rel_relkind(pubrel->prrelid);
+
+ /*
+ * Handle tables and sequences, depending on the combination of relkind
+ * and sequences flag. If there's a mismatch, we ignore the relation.
+ */
+ if (sequences && (relkind == RELKIND_SEQUENCE))
+ result = lappend_oid(result, pubrel->prrelid);
+ else if (!sequences && (relkind != RELKIND_SEQUENCE))
+ result = GetPubPartitionOptionRelations(result, pub_partopt,
+ pubrel->prrelid);
}
systable_endscan(scan);
@@ -605,6 +624,43 @@ GetAllTablesPublications(void)
return result;
}
+/*
+ * Gets list of publication oids for publications marked as FOR ALL SEQUENCES.
+ */
+List *
+GetAllSequencesPublications(void)
+{
+ List *result;
+ Relation rel;
+ ScanKeyData scankey;
+ SysScanDesc scan;
+ HeapTuple tup;
+
+ /* Find all publications that are marked as for all tables. */
+ rel = table_open(PublicationRelationId, AccessShareLock);
+
+ ScanKeyInit(&scankey,
+ Anum_pg_publication_puballsequences,
+ BTEqualStrategyNumber, F_BOOLEQ,
+ BoolGetDatum(true));
+
+ scan = systable_beginscan(rel, InvalidOid, false,
+ NULL, 1, &scankey);
+
+ result = NIL;
+ while (HeapTupleIsValid(tup = systable_getnext(scan)))
+ {
+ Oid oid = ((Form_pg_publication) GETSTRUCT(tup))->oid;
+
+ result = lappend_oid(result, oid);
+ }
+
+ systable_endscan(scan);
+ table_close(rel, AccessShareLock);
+
+ return result;
+}
+
/*
* Gets list of all relation published by FOR ALL TABLES publication(s).
*
@@ -671,28 +727,36 @@ GetAllTablesPublicationRelations(bool pubviaroot)
/*
* Gets the list of schema oids for a publication.
*
- * This should only be used FOR ALL TABLES IN SCHEMA publications.
+ * This should only be used FOR ALL TABLES IN SCHEMA and FOR ALL SEQUENCES
+ * publications.
+ *
+ * 'sequences' determines whether to get FOR TABLE or FOR SEQUENCES schemas
*/
List *
-GetPublicationSchemas(Oid pubid)
+GetPublicationSchemas(Oid pubid, bool sequences)
{
List *result = NIL;
Relation pubschsrel;
- ScanKeyData scankey;
+ ScanKeyData scankey[2];
SysScanDesc scan;
HeapTuple tup;
/* Find all schemas associated with the publication */
pubschsrel = table_open(PublicationNamespaceRelationId, AccessShareLock);
- ScanKeyInit(&scankey,
+ ScanKeyInit(&scankey[0],
Anum_pg_publication_namespace_pnpubid,
BTEqualStrategyNumber, F_OIDEQ,
ObjectIdGetDatum(pubid));
+ ScanKeyInit(&scankey[1],
+ Anum_pg_publication_namespace_pnsequences,
+ BTEqualStrategyNumber, F_BOOLEQ,
+ BoolGetDatum(sequences));
+
scan = systable_beginscan(pubschsrel,
- PublicationNamespacePnnspidPnpubidIndexId,
- true, NULL, 1, &scankey);
+ PublicationNamespacePnnspidPnpubidSeqIndexId,
+ true, NULL, 2, scankey);
while (HeapTupleIsValid(tup = systable_getnext(scan)))
{
Form_pg_publication_namespace pubsch;
@@ -710,6 +774,9 @@ GetPublicationSchemas(Oid pubid)
/*
* Gets the list of publication oids associated with a specified schema.
+ *
+ * XXX probably should handle pnsequences somehow, either by taking a
+ * parameter or returning the flag, somehow
*/
List *
GetSchemaPublications(Oid schemaid)
@@ -738,7 +805,8 @@ GetSchemaPublications(Oid schemaid)
* Get the list of publishable relation oids for a specified schema.
*/
List *
-GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
+GetSchemaPublicationRelations(Oid schemaid, bool sequences,
+ PublicationPartOpt pub_partopt)
{
Relation classRel;
ScanKeyData key[1];
@@ -763,11 +831,19 @@ GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
Oid relid = relForm->oid;
char relkind;
+ /* FIXME need to differentiate tables and sequences */
if (!is_publishable_class(relid, relForm))
continue;
relkind = get_rel_relkind(relid);
- if (relkind == RELKIND_RELATION)
+
+ /* Filter by relkind depending on FOR ALL TABLES / SEQUENCES */
+ if ((sequences && relkind != RELKIND_SEQUENCE) ||
+ (!sequences && relkind == RELKIND_SEQUENCE))
+ continue;
+
+ if ((relkind == RELKIND_RELATION) ||
+ (relkind == RELKIND_SEQUENCE))
result = lappend_oid(result, relid);
else if (relkind == RELKIND_PARTITIONED_TABLE)
{
@@ -792,13 +868,14 @@ GetSchemaPublicationRelations(Oid schemaid, PublicationPartOpt pub_partopt)
/*
* Gets the list of all relations published by FOR ALL TABLES IN SCHEMA
- * publication.
+ * or FOR ALL SEQUENCES IN SCHEMA publication.
*/
List *
-GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
+GetAllSchemaPublicationRelations(Oid pubid, bool sequences,
+ PublicationPartOpt pub_partopt)
{
List *result = NIL;
- List *pubschemalist = GetPublicationSchemas(pubid);
+ List *pubschemalist = GetPublicationSchemas(pubid, sequences);
ListCell *cell;
foreach(cell, pubschemalist)
@@ -806,13 +883,54 @@ GetAllSchemaPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt)
Oid schemaid = lfirst_oid(cell);
List *schemaRels = NIL;
- schemaRels = GetSchemaPublicationRelations(schemaid, pub_partopt);
+ schemaRels = GetSchemaPublicationRelations(schemaid, sequences,
+ pub_partopt);
result = list_concat(result, schemaRels);
}
return result;
}
+/*
+ * Gets list of all relation published by FOR ALL TABLES publication(s).
+ *
+ * If the publication publishes partition changes via their respective root
+ * partitioned tables, we must exclude partitions in favor of including the
+ * root partitioned tables.
+ */
+List *
+GetAllSequencesPublicationRelations(void)
+{
+ Relation classRel;
+ ScanKeyData key[1];
+ TableScanDesc scan;
+ HeapTuple tuple;
+ List *result = NIL;
+
+ classRel = table_open(RelationRelationId, AccessShareLock);
+
+ ScanKeyInit(&key[0],
+ Anum_pg_class_relkind,
+ BTEqualStrategyNumber, F_CHAREQ,
+ CharGetDatum(RELKIND_SEQUENCE));
+
+ scan = table_beginscan_catalog(classRel, 1, key);
+
+ while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL)
+ {
+ Form_pg_class relForm = (Form_pg_class) GETSTRUCT(tuple);
+ Oid relid = relForm->oid;
+
+ if (is_publishable_class(relid, relForm))
+ result = lappend_oid(result, relid);
+ }
+
+ table_endscan(scan);
+
+ table_close(classRel, AccessShareLock);
+ return result;
+}
+
/*
* Get publication using oid
*
@@ -835,10 +953,12 @@ GetPublication(Oid pubid)
pub->oid = pubid;
pub->name = pstrdup(NameStr(pubform->pubname));
pub->alltables = pubform->puballtables;
+ pub->allsequences = pubform->puballsequences;
pub->pubactions.pubinsert = pubform->pubinsert;
pub->pubactions.pubupdate = pubform->pubupdate;
pub->pubactions.pubdelete = pubform->pubdelete;
pub->pubactions.pubtruncate = pubform->pubtruncate;
+ pub->pubactions.pubsequence = pubform->pubsequence;
pub->pubviaroot = pubform->pubviaroot;
ReleaseSysCache(tup);
@@ -949,10 +1069,12 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
*schemarelids;
relids = GetPublicationRelations(publication->oid,
+ false, /* tables only */
publication->pubviaroot ?
PUBLICATION_PART_ROOT :
PUBLICATION_PART_LEAF);
schemarelids = GetAllSchemaPublicationRelations(publication->oid,
+ false, /* tables only */
publication->pubviaroot ?
PUBLICATION_PART_ROOT :
PUBLICATION_PART_LEAF);
@@ -988,3 +1110,71 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
SRF_RETURN_DONE(funcctx);
}
+
+/*
+ * Returns Oids of sequences in a publication.
+ */
+Datum
+pg_get_publication_sequences(PG_FUNCTION_ARGS)
+{
+ FuncCallContext *funcctx;
+ char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
+ Publication *publication;
+ List *sequences;
+
+ /* stuff done only on the first call of the function */
+ if (SRF_IS_FIRSTCALL())
+ {
+ MemoryContext oldcontext;
+
+ /* create a function context for cross-call persistence */
+ funcctx = SRF_FIRSTCALL_INIT();
+
+ /* switch to memory context appropriate for multiple function calls */
+ oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
+
+ publication = GetPublicationByName(pubname, false);
+
+ /*
+ * Publications support partitioned tables, although all changes are
+ * replicated using leaf partition identity and schema, so we only
+ * need those.
+ */
+ if (publication->allsequences)
+ sequences = GetAllSequencesPublicationRelations();
+ else
+ {
+ List *relids,
+ *schemarelids;
+
+ relids = GetPublicationRelations(publication->oid,
+ true, /* sequences */
+ publication->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ schemarelids = GetAllSchemaPublicationRelations(publication->oid,
+ true, /* sequences only */
+ publication->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ sequences = list_concat_unique_oid(relids, schemarelids);
+ }
+
+ funcctx->user_fctx = (void *) sequences;
+
+ MemoryContextSwitchTo(oldcontext);
+ }
+
+ /* stuff done on every call of the function */
+ funcctx = SRF_PERCALL_SETUP();
+ sequences = (List *) funcctx->user_fctx;
+
+ if (funcctx->call_cntr < list_length(sequences))
+ {
+ Oid relid = list_nth_oid(sequences, funcctx->call_cntr);
+
+ SRF_RETURN_NEXT(funcctx, ObjectIdGetDatum(relid));
+ }
+
+ SRF_RETURN_DONE(funcctx);
+}
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 3cb69b1f87b..072cfb2655a 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -374,6 +374,16 @@ CREATE VIEW pg_publication_tables AS
pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)
WHERE C.oid = GPT.relid;
+CREATE VIEW pg_publication_sequences AS
+ SELECT
+ P.pubname AS pubname,
+ N.nspname AS schemaname,
+ C.relname AS sequencename
+ FROM pg_publication P,
+ LATERAL pg_get_publication_sequences(P.pubname) GPS,
+ pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)
+ WHERE C.oid = GPS.relid;
+
CREATE VIEW pg_locks AS
SELECT * FROM pg_lock_status() AS L;
diff --git a/src/backend/commands/publicationcmds.c b/src/backend/commands/publicationcmds.c
index 16b8661a1b7..db78dc86646 100644
--- a/src/backend/commands/publicationcmds.c
+++ b/src/backend/commands/publicationcmds.c
@@ -16,6 +16,7 @@
#include "access/genam.h"
#include "access/htup_details.h"
+#include "access/relation.h"
#include "access/table.h"
#include "access/xact.h"
#include "catalog/catalog.h"
@@ -67,15 +68,16 @@ typedef struct rf_context
} rf_context;
static List *OpenRelIdList(List *relids);
-static List *OpenTableList(List *tables);
-static void CloseTableList(List *rels);
+static List *OpenRelationList(List *rels);
+static void CloseRelationList(List *rels);
static void LockSchemaList(List *schemalist);
-static void PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
+static void PublicationAddRelations(Oid pubid, List *rels, bool if_not_exists,
AlterPublicationStmt *stmt);
-static void PublicationDropTables(Oid pubid, List *rels, bool missing_ok);
-static void PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
- AlterPublicationStmt *stmt);
-static void PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok);
+static void PublicationDropRelations(Oid pubid, List *rels, bool missing_ok);
+static void PublicationAddSchemas(Oid pubid, List *schemas, bool sequences,
+ bool if_not_exists, AlterPublicationStmt *stmt);
+static void PublicationDropSchemas(Oid pubid, List *schemas, bool sequences, bool missing_ok);
+
static void
parse_publication_options(ParseState *pstate,
@@ -95,6 +97,7 @@ parse_publication_options(ParseState *pstate,
pubactions->pubupdate = true;
pubactions->pubdelete = true;
pubactions->pubtruncate = true;
+ pubactions->pubsequence = true;
*publish_via_partition_root = false;
/* Parse options */
@@ -119,6 +122,7 @@ parse_publication_options(ParseState *pstate,
pubactions->pubupdate = false;
pubactions->pubdelete = false;
pubactions->pubtruncate = false;
+ pubactions->pubsequence = false;
*publish_given = true;
publish = defGetString(defel);
@@ -141,6 +145,8 @@ parse_publication_options(ParseState *pstate,
pubactions->pubdelete = true;
else if (strcmp(publish_opt, "truncate") == 0)
pubactions->pubtruncate = true;
+ else if (strcmp(publish_opt, "sequence") == 0)
+ pubactions->pubsequence = true;
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -167,7 +173,9 @@ parse_publication_options(ParseState *pstate,
*/
static void
ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
- List **rels, List **schemas)
+ List **tables, List **sequences,
+ List **tables_schemas, List **sequences_schemas,
+ List **schemas)
{
ListCell *cell;
PublicationObjSpec *pubobj;
@@ -185,12 +193,23 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
switch (pubobj->pubobjtype)
{
case PUBLICATIONOBJ_TABLE:
- *rels = lappend(*rels, pubobj->pubtable);
+ *tables = lappend(*tables, pubobj->pubtable);
+ break;
+ case PUBLICATIONOBJ_SEQUENCE:
+ *sequences = lappend(*sequences, pubobj->pubtable);
break;
case PUBLICATIONOBJ_TABLES_IN_SCHEMA:
schemaid = get_namespace_oid(pubobj->name, false);
/* Filter out duplicates if user specifies "sch1, sch1" */
+ *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid);
+ *schemas = list_append_unique_oid(*schemas, schemaid);
+ break;
+ case PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA:
+ schemaid = get_namespace_oid(pubobj->name, false);
+
+ /* Filter out duplicates if user specifies "sch1, sch1" */
+ *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid);
*schemas = list_append_unique_oid(*schemas, schemaid);
break;
case PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA:
@@ -204,6 +223,21 @@ ObjectsInPublicationToOids(List *pubobjspec_list, ParseState *pstate,
list_free(search_path);
/* Filter out duplicates if user specifies "sch1, sch1" */
+ *tables_schemas = list_append_unique_oid(*tables_schemas, schemaid);
+ *schemas = list_append_unique_oid(*schemas, schemaid);
+ break;
+ case PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA:
+ search_path = fetch_search_path(false);
+ if (search_path == NIL) /* nothing valid in search_path? */
+ ereport(ERROR,
+ errcode(ERRCODE_UNDEFINED_SCHEMA),
+ errmsg("no schema has been selected for CURRENT_SCHEMA"));
+
+ schemaid = linitial_oid(search_path);
+ list_free(search_path);
+
+ /* Filter out duplicates if user specifies "sch1, sch1" */
+ *sequences_schemas = list_append_unique_oid(*sequences_schemas, schemaid);
*schemas = list_append_unique_oid(*schemas, schemaid);
break;
default:
@@ -240,6 +274,14 @@ CheckObjSchemaNotAlreadyInPublication(List *rels, List *schemaidlist,
errdetail("Table \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
RelationGetRelationName(rel),
get_namespace_name(relSchemaId)));
+ else if (checkobjtype == PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add schema \"%s\" to publication",
+ get_namespace_name(relSchemaId)),
+ errdetail("SEquence \"%s\" in schema \"%s\" is already part of the publication, adding the same schema is not supported.",
+ RelationGetRelationName(rel),
+ get_namespace_name(relSchemaId)));
else if (checkobjtype == PUBLICATIONOBJ_TABLE)
ereport(ERROR,
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
@@ -248,6 +290,14 @@ CheckObjSchemaNotAlreadyInPublication(List *rels, List *schemaidlist,
RelationGetRelationName(rel)),
errdetail("Table's schema \"%s\" is already part of the publication or part of the specified schema list.",
get_namespace_name(relSchemaId)));
+ else if (checkobjtype == PUBLICATIONOBJ_SEQUENCE)
+ ereport(ERROR,
+ errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("cannot add relation \"%s.%s\" to publication",
+ get_namespace_name(relSchemaId),
+ RelationGetRelationName(rel)),
+ errdetail("Sequence's schema \"%s\" is already part of the publication or part of the specified schema list.",
+ get_namespace_name(relSchemaId)));
}
}
}
@@ -625,7 +675,10 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
bool publish_via_partition_root_given;
bool publish_via_partition_root;
AclResult aclresult;
- List *relations = NIL;
+ List *tables = NIL;
+ List *sequences = NIL;
+ List *tables_schemaidlist = NIL;
+ List *sequences_schemaidlist = NIL;
List *schemaidlist = NIL;
/* must have CREATE privilege on database */
@@ -672,6 +725,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
values[Anum_pg_publication_oid - 1] = ObjectIdGetDatum(puboid);
values[Anum_pg_publication_puballtables - 1] =
BoolGetDatum(stmt->for_all_tables);
+ values[Anum_pg_publication_puballsequences - 1] =
+ BoolGetDatum(stmt->for_all_sequences);
values[Anum_pg_publication_pubinsert - 1] =
BoolGetDatum(pubactions.pubinsert);
values[Anum_pg_publication_pubupdate - 1] =
@@ -680,6 +735,8 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
BoolGetDatum(pubactions.pubdelete);
values[Anum_pg_publication_pubtruncate - 1] =
BoolGetDatum(pubactions.pubtruncate);
+ values[Anum_pg_publication_pubsequence - 1] =
+ BoolGetDatum(pubactions.pubsequence);
values[Anum_pg_publication_pubviaroot - 1] =
BoolGetDatum(publish_via_partition_root);
@@ -704,38 +761,62 @@ CreatePublication(ParseState *pstate, CreatePublicationStmt *stmt)
}
else
{
- ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
+ ObjectsInPublicationToOids(stmt->pubobjects, pstate,
+ &tables, &sequences,
+ &tables_schemaidlist,
+ &sequences_schemaidlist,
&schemaidlist);
/* FOR ALL TABLES IN SCHEMA requires superuser */
- if (list_length(schemaidlist) > 0 && !superuser())
+ if (list_length(tables_schemaidlist) > 0 && !superuser())
ereport(ERROR,
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("must be superuser to create FOR ALL TABLES IN SCHEMA publication"));
- if (list_length(relations) > 0)
+ if (list_length(tables) > 0)
{
List *rels;
- rels = OpenTableList(relations);
- CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
+ rels = OpenRelationList(tables);
+ CheckObjSchemaNotAlreadyInPublication(rels, tables_schemaidlist,
PUBLICATIONOBJ_TABLE);
TransformPubWhereClauses(rels, pstate->p_sourcetext,
publish_via_partition_root);
- PublicationAddTables(puboid, rels, true, NULL);
- CloseTableList(rels);
+ PublicationAddRelations(puboid, rels, true, NULL);
+ CloseRelationList(rels);
}
- if (list_length(schemaidlist) > 0)
+ if (list_length(sequences) > 0)
+ {
+ List *rels;
+
+ rels = OpenRelationList(sequences);
+ CheckObjSchemaNotAlreadyInPublication(rels, sequences_schemaidlist,
+ PUBLICATIONOBJ_SEQUENCE);
+ PublicationAddRelations(puboid, rels, true, NULL);
+ CloseRelationList(rels);
+ }
+
+ if (list_length(tables_schemaidlist) > 0)
{
/*
* Schema lock is held until the publication is created to prevent
* concurrent schema deletion.
*/
- LockSchemaList(schemaidlist);
- PublicationAddSchemas(puboid, schemaidlist, true, NULL);
+ LockSchemaList(tables_schemaidlist);
+ PublicationAddSchemas(puboid, tables_schemaidlist, false, true, NULL);
+ }
+
+ if (list_length(sequences_schemaidlist) > 0)
+ {
+ /*
+ * Schema lock is held until the publication is created to prevent
+ * concurrent schema deletion.
+ */
+ LockSchemaList(sequences_schemaidlist);
+ PublicationAddSchemas(puboid, sequences_schemaidlist, true, true, NULL);
}
}
@@ -797,7 +878,7 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
LockDatabaseObject(PublicationRelationId, pubform->oid, 0,
AccessShareLock);
- root_relids = GetPublicationRelations(pubform->oid,
+ root_relids = GetPublicationRelations(pubform->oid, false,
PUBLICATION_PART_ROOT);
foreach(lc, root_relids)
@@ -856,6 +937,9 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
values[Anum_pg_publication_pubtruncate - 1] = BoolGetDatum(pubactions.pubtruncate);
replaces[Anum_pg_publication_pubtruncate - 1] = true;
+
+ values[Anum_pg_publication_pubsequence - 1] = BoolGetDatum(pubactions.pubsequence);
+ replaces[Anum_pg_publication_pubsequence - 1] = true;
}
if (publish_via_partition_root_given)
@@ -875,7 +959,7 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
pubform = (Form_pg_publication) GETSTRUCT(tup);
/* Invalidate the relcache. */
- if (pubform->puballtables)
+ if (pubform->puballtables) /* FIXME probably needs to handle puballsequences too? */
{
CacheInvalidateRelcacheAll();
}
@@ -890,7 +974,7 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
* trees, not just those explicitly mentioned in the publication.
*/
if (root_relids == NIL)
- relids = GetPublicationRelations(pubform->oid,
+ relids = GetPublicationRelations(pubform->oid, false,
PUBLICATION_PART_ALL);
else
{
@@ -904,7 +988,17 @@ AlterPublicationOptions(ParseState *pstate, AlterPublicationStmt *stmt,
lfirst_oid(lc));
}
- schemarelids = GetAllSchemaPublicationRelations(pubform->oid,
+ /* tables */
+ schemarelids = GetAllSchemaPublicationRelations(pubform->oid, false,
+ PUBLICATION_PART_ALL);
+ relids = list_concat_unique_oid(relids, schemarelids);
+
+ /* sequences */
+ relids = list_concat_unique_oid(relids,
+ GetPublicationRelations(pubform->oid,
+ true, PUBLICATION_PART_ALL));
+
+ schemarelids = GetAllSchemaPublicationRelations(pubform->oid, true,
PUBLICATION_PART_ALL);
relids = list_concat_unique_oid(relids, schemarelids);
@@ -951,6 +1045,11 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
Oid pubid = pubform->oid;
+ /*
+ * FIXME Do we need to test relkind of the relations? Otherwise we
+ * can do "ADD TABLE sequence" and it'll "work".
+ */
+
/*
* Nothing to do if no objects, except in SET: for that it is quite
* possible that user has not specified any tables in which case we need
@@ -959,7 +1058,7 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
if (!tables && stmt->action != AP_SetObjects)
return;
- rels = OpenTableList(tables);
+ rels = OpenRelationList(tables);
if (stmt->action == AP_AddObjects)
{
@@ -969,19 +1068,19 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
* Check if the relation is member of the existing schema in the
* publication or member of the schema list specified.
*/
- schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid));
+ schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid, false));
CheckObjSchemaNotAlreadyInPublication(rels, schemas,
PUBLICATIONOBJ_TABLE);
TransformPubWhereClauses(rels, queryString, pubform->pubviaroot);
- PublicationAddTables(pubid, rels, false, stmt);
+ PublicationAddRelations(pubid, rels, false, stmt);
}
else if (stmt->action == AP_DropObjects)
- PublicationDropTables(pubid, rels, false);
+ PublicationDropRelations(pubid, rels, false);
else /* AP_SetObjects */
{
- List *oldrelids = GetPublicationRelations(pubid,
+ List *oldrelids = GetPublicationRelations(pubid, false,
PUBLICATION_PART_ROOT);
List *delrels = NIL;
ListCell *oldlc;
@@ -1054,6 +1153,10 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
*/
if (!found)
{
+ /* don't drop sequences (not in list of tables) */
+ if (get_rel_relkind(oldrelid) == RELKIND_SEQUENCE)
+ continue;
+
oldrel = palloc(sizeof(PublicationRelInfo));
oldrel->whereClause = NULL;
oldrel->relation = table_open(oldrelid,
@@ -1063,18 +1166,18 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
}
/* And drop them. */
- PublicationDropTables(pubid, delrels, true);
+ PublicationDropRelations(pubid, delrels, true);
/*
* Don't bother calculating the difference for adding, we'll catch and
* skip existing ones when doing catalog update.
*/
- PublicationAddTables(pubid, rels, true, stmt);
+ PublicationAddRelations(pubid, rels, true, stmt);
- CloseTableList(delrels);
+ CloseRelationList(delrels);
}
- CloseTableList(rels);
+ CloseRelationList(rels);
}
/*
@@ -1084,7 +1187,8 @@ AlterPublicationTables(AlterPublicationStmt *stmt, HeapTuple tup,
*/
static void
AlterPublicationSchemas(AlterPublicationStmt *stmt,
- HeapTuple tup, List *schemaidlist)
+ HeapTuple tup, List *schemaidlist,
+ bool sequences)
{
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
@@ -1106,20 +1210,20 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt,
List *rels;
List *reloids;
- reloids = GetPublicationRelations(pubform->oid, PUBLICATION_PART_ROOT);
+ reloids = GetPublicationRelations(pubform->oid, sequences, PUBLICATION_PART_ROOT);
rels = OpenRelIdList(reloids);
CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
PUBLICATIONOBJ_TABLES_IN_SCHEMA);
- CloseTableList(rels);
- PublicationAddSchemas(pubform->oid, schemaidlist, false, stmt);
+ CloseRelationList(rels);
+ PublicationAddSchemas(pubform->oid, schemaidlist, sequences, false, stmt);
}
else if (stmt->action == AP_DropObjects)
- PublicationDropSchemas(pubform->oid, schemaidlist, false);
+ PublicationDropSchemas(pubform->oid, schemaidlist, sequences, false);
else /* AP_SetObjects */
{
- List *oldschemaids = GetPublicationSchemas(pubform->oid);
+ List *oldschemaids = GetPublicationSchemas(pubform->oid, sequences);
List *delschemas = NIL;
/* Identify which schemas should be dropped */
@@ -1132,13 +1236,13 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt,
LockSchemaList(delschemas);
/* And drop them */
- PublicationDropSchemas(pubform->oid, delschemas, true);
+ PublicationDropSchemas(pubform->oid, delschemas, sequences, true);
/*
* Don't bother calculating the difference for adding, we'll catch and
* skip existing ones when doing catalog update.
*/
- PublicationAddSchemas(pubform->oid, schemaidlist, true, stmt);
+ PublicationAddSchemas(pubform->oid, schemaidlist, sequences, true, stmt);
}
}
@@ -1148,12 +1252,13 @@ AlterPublicationSchemas(AlterPublicationStmt *stmt,
*/
static void
CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
- List *tables, List *schemaidlist)
+ List *tables, List *tables_schemaidlist,
+ List *sequences, List *sequences_schemaidlist)
{
Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
if ((stmt->action == AP_AddObjects || stmt->action == AP_SetObjects) &&
- schemaidlist && !superuser())
+ (tables_schemaidlist || sequences_schemaidlist) && !superuser())
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("must be superuser to add or set schemas")));
@@ -1162,13 +1267,24 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
* Check that user is allowed to manipulate the publication tables in
* schema
*/
- if (schemaidlist && pubform->puballtables)
+ if (tables_schemaidlist && pubform->puballtables)
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("publication \"%s\" is defined as FOR ALL TABLES",
NameStr(pubform->pubname)),
errdetail("Tables from schema cannot be added to, dropped from, or set on FOR ALL TABLES publications.")));
+ /*
+ * Check that user is allowed to manipulate the publication sequences in
+ * schema
+ */
+ if (sequences_schemaidlist && pubform->puballsequences)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES",
+ NameStr(pubform->pubname)),
+ errdetail("Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications.")));
+
/* Check that user is allowed to manipulate the publication tables. */
if (tables && pubform->puballtables)
ereport(ERROR,
@@ -1176,6 +1292,118 @@ CheckAlterPublication(AlterPublicationStmt *stmt, HeapTuple tup,
errmsg("publication \"%s\" is defined as FOR ALL TABLES",
NameStr(pubform->pubname)),
errdetail("Tables cannot be added to or dropped from FOR ALL TABLES publications.")));
+
+ /* Check that user is allowed to manipulate the publication tables. */
+ if (sequences && pubform->puballsequences)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("publication \"%s\" is defined as FOR ALL SEQUENCES",
+ NameStr(pubform->pubname)),
+ errdetail("Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications.")));
+}
+
+/*
+ * Add or remove sequence to/from publication.
+ */
+static void
+AlterPublicationSequences(AlterPublicationStmt *stmt, HeapTuple tup,
+ List *sequences, List *schemaidlist)
+{
+ List *rels = NIL;
+ Form_pg_publication pubform = (Form_pg_publication) GETSTRUCT(tup);
+ Oid pubid = pubform->oid;
+
+ /*
+ * FIXME Do we need to test relkind of the relations? Otherwise we
+ * can do "ADD SEQUENCE table" and it'll "work".
+ */
+
+ /*
+ * It is quite possible that for the SET case user has not specified any
+ * tables in which case we need to remove all the existing tables.
+ */
+ if (!sequences && stmt->action != AP_SetObjects)
+ return;
+
+ rels = OpenRelationList(sequences);
+
+ if (stmt->action == AP_AddObjects)
+ {
+ List *schemas = NIL;
+
+ /*
+ * Check if the relation is member of the existing schema in the
+ * publication or member of the schema list specified.
+ */
+ schemas = list_concat_copy(schemaidlist, GetPublicationSchemas(pubid, true));
+ CheckObjSchemaNotAlreadyInPublication(rels, schemas,
+ PUBLICATIONOBJ_SEQUENCE);
+ PublicationAddRelations(pubid, rels, false, stmt);
+ }
+ else if (stmt->action == AP_DropObjects)
+ PublicationDropRelations(pubid, rels, false);
+ else /* DEFELEM_SET */
+ {
+ List *oldrelids = GetPublicationRelations(pubid, true,
+ PUBLICATION_PART_ROOT);
+ List *delrels = NIL;
+ ListCell *oldlc;
+
+ CheckObjSchemaNotAlreadyInPublication(rels, schemaidlist,
+ PUBLICATIONOBJ_SEQUENCE);
+
+ /* Calculate which relations to drop. */
+ foreach(oldlc, oldrelids)
+ {
+ Oid oldrelid = lfirst_oid(oldlc);
+ ListCell *newlc;
+ bool found = false;
+
+ foreach(newlc, rels)
+ {
+ PublicationRelInfo *newpubrel;
+
+ newpubrel = (PublicationRelInfo *) lfirst(newlc);
+ if (RelationGetRelid(newpubrel->relation) == oldrelid)
+ {
+ found = true;
+ break;
+ }
+ }
+ /* Not yet in the list, open it and add to the list */
+ if (!found)
+ {
+ Relation oldrel;
+ PublicationRelInfo *pubrel;
+
+ /* don't drop non-sequences (not in list of sequences) */
+ if (get_rel_relkind(oldrelid) != RELKIND_SEQUENCE)
+ continue;
+
+ /* Wrap relation into PublicationRelInfo */
+ oldrel = table_open(oldrelid, ShareUpdateExclusiveLock);
+
+ pubrel = palloc(sizeof(PublicationRelInfo));
+ pubrel->relation = oldrel;
+ pubrel->whereClause = NULL;
+
+ delrels = lappend(delrels, pubrel);
+ }
+ }
+
+ /* And drop them. */
+ PublicationDropRelations(pubid, delrels, true);
+
+ /*
+ * Don't bother calculating the difference for adding, we'll catch and
+ * skip existing ones when doing catalog update.
+ */
+ PublicationAddRelations(pubid, rels, true, stmt);
+
+ CloseRelationList(delrels);
+ }
+
+ CloseRelationList(rels);
}
/*
@@ -1213,14 +1441,22 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
AlterPublicationOptions(pstate, stmt, rel, tup);
else
{
- List *relations = NIL;
+ List *tables = NIL;
+ List *sequences = NIL;
+ List *tables_schemaidlist = NIL;
+ List *sequences_schemaidlist = NIL;
List *schemaidlist = NIL;
Oid pubid = pubform->oid;
- ObjectsInPublicationToOids(stmt->pubobjects, pstate, &relations,
+ ObjectsInPublicationToOids(stmt->pubobjects, pstate,
+ &tables, &sequences,
+ &tables_schemaidlist,
+ &sequences_schemaidlist,
&schemaidlist);
- CheckAlterPublication(stmt, tup, relations, schemaidlist);
+ CheckAlterPublication(stmt, tup,
+ tables, tables_schemaidlist,
+ sequences, sequences_schemaidlist);
heap_freetuple(tup);
@@ -1248,9 +1484,14 @@ AlterPublication(ParseState *pstate, AlterPublicationStmt *stmt)
errmsg("publication \"%s\" does not exist",
stmt->pubname));
- AlterPublicationTables(stmt, tup, relations, schemaidlist,
+ AlterPublicationTables(stmt, tup, tables, tables_schemaidlist,
pstate->p_sourcetext);
- AlterPublicationSchemas(stmt, tup, schemaidlist);
+
+ AlterPublicationSequences(stmt, tup, sequences, sequences_schemaidlist);
+
+ AlterPublicationSchemas(stmt, tup, tables_schemaidlist, false);
+
+ AlterPublicationSchemas(stmt, tup, sequences_schemaidlist, true);
}
/* Cleanup. */
@@ -1318,7 +1559,7 @@ RemovePublicationById(Oid pubid)
pubform = (Form_pg_publication) GETSTRUCT(tup);
/* Invalidate relcache so that publication info is rebuilt. */
- if (pubform->puballtables)
+ if (pubform->puballtables) /* FIXME handle puballsequences too? */
CacheInvalidateRelcacheAll();
CatalogTupleDelete(rel, &tup->t_self);
@@ -1354,6 +1595,7 @@ RemovePublicationSchemaById(Oid psoid)
* partitions.
*/
schemaRels = GetSchemaPublicationRelations(pubsch->pnnspid,
+ pubsch->pnsequences,
PUBLICATION_PART_ALL);
InvalidatePublicationRels(schemaRels);
@@ -1396,17 +1638,17 @@ OpenRelIdList(List *relids)
* add them to a publication.
*/
static List *
-OpenTableList(List *tables)
+OpenRelationList(List *rels)
{
List *relids = NIL;
- List *rels = NIL;
+ List *result = NIL;
ListCell *lc;
List *relids_with_rf = NIL;
/*
* Open, share-lock, and check all the explicitly-specified relations
*/
- foreach(lc, tables)
+ foreach(lc, rels)
{
PublicationTable *t = lfirst_node(PublicationTable, lc);
bool recurse = t->relation->inh;
@@ -1443,7 +1685,7 @@ OpenTableList(List *tables)
pub_rel = palloc(sizeof(PublicationRelInfo));
pub_rel->relation = rel;
pub_rel->whereClause = t->whereClause;
- rels = lappend(rels, pub_rel);
+ result = lappend(result, pub_rel);
relids = lappend_oid(relids, myrelid);
if (t->whereClause)
@@ -1497,7 +1739,7 @@ OpenTableList(List *tables)
pub_rel->relation = rel;
/* child inherits WHERE clause from parent */
pub_rel->whereClause = t->whereClause;
- rels = lappend(rels, pub_rel);
+ result = lappend(result, pub_rel);
relids = lappend_oid(relids, childrelid);
if (t->whereClause)
@@ -1509,14 +1751,14 @@ OpenTableList(List *tables)
list_free(relids);
list_free(relids_with_rf);
- return rels;
+ return result;
}
/*
* Close all relations in the list.
*/
static void
-CloseTableList(List *rels)
+CloseRelationList(List *rels)
{
ListCell *lc;
@@ -1564,7 +1806,7 @@ LockSchemaList(List *schemalist)
* Add listed tables to the publication.
*/
static void
-PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
+PublicationAddRelations(Oid pubid, List *rels, bool if_not_exists,
AlterPublicationStmt *stmt)
{
ListCell *lc;
@@ -1598,7 +1840,7 @@ PublicationAddTables(Oid pubid, List *rels, bool if_not_exists,
* Remove listed tables from the publication.
*/
static void
-PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
+PublicationDropRelations(Oid pubid, List *rels, bool missing_ok)
{
ObjectAddress obj;
ListCell *lc;
@@ -1638,8 +1880,8 @@ PublicationDropTables(Oid pubid, List *rels, bool missing_ok)
* Add listed schemas to the publication.
*/
static void
-PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
- AlterPublicationStmt *stmt)
+PublicationAddSchemas(Oid pubid, List *schemas, bool sequences,
+ bool if_not_exists, AlterPublicationStmt *stmt)
{
ListCell *lc;
@@ -1650,7 +1892,7 @@ PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
Oid schemaid = lfirst_oid(lc);
ObjectAddress obj;
- obj = publication_add_schema(pubid, schemaid, if_not_exists);
+ obj = publication_add_schema(pubid, schemaid, sequences, if_not_exists);
if (stmt)
{
EventTriggerCollectSimpleCommand(obj, InvalidObjectAddress,
@@ -1666,7 +1908,7 @@ PublicationAddSchemas(Oid pubid, List *schemas, bool if_not_exists,
* Remove listed schemas from the publication.
*/
static void
-PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok)
+PublicationDropSchemas(Oid pubid, List *schemas, bool sequences, bool missing_ok)
{
ObjectAddress obj;
ListCell *lc;
@@ -1676,10 +1918,11 @@ PublicationDropSchemas(Oid pubid, List *schemas, bool missing_ok)
{
Oid schemaid = lfirst_oid(lc);
- psid = GetSysCacheOid2(PUBLICATIONNAMESPACEMAP,
+ psid = GetSysCacheOid3(PUBLICATIONNAMESPACEMAP,
Anum_pg_publication_namespace_oid,
ObjectIdGetDatum(schemaid),
- ObjectIdGetDatum(pubid));
+ ObjectIdGetDatum(pubid),
+ BoolGetDatum(sequences));
if (!OidIsValid(psid))
{
if (missing_ok)
@@ -1734,6 +1977,13 @@ AlterPublicationOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
NameStr(form->pubname)),
errhint("The owner of a FOR ALL TABLES publication must be a superuser.")));
+ if (form->puballsequences && !superuser_arg(newOwnerId))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied to change owner of publication \"%s\"",
+ NameStr(form->pubname)),
+ errhint("The owner of a FOR ALL SEQUENCES publication must be a superuser.")));
+
if (!superuser_arg(newOwnerId) && is_schema_publication(form->oid))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
diff --git a/src/backend/commands/sequence.c b/src/backend/commands/sequence.c
index ab592ce2f15..fe4f21ec438 100644
--- a/src/backend/commands/sequence.c
+++ b/src/backend/commands/sequence.c
@@ -336,6 +336,85 @@ ResetSequence(Oid seq_relid)
relation_close(seq_rel, NoLock);
}
+/*
+ * Reset a sequence to its initial value.
+ *
+ * The change is made transactionally, so that on failure of the current
+ * transaction, the sequence will be restored to its previous state.
+ * We do that by creating a whole new relfilenode for the sequence; so this
+ * works much like the rewriting forms of ALTER TABLE.
+ *
+ * Caller is assumed to have acquired AccessExclusiveLock on the sequence,
+ * which must not be released until end of transaction. Caller is also
+ * responsible for permissions checking.
+ */
+void
+ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called)
+{
+ Relation seq_rel;
+ SeqTable elm;
+ Form_pg_sequence_data seq;
+ Buffer buf;
+ HeapTupleData seqdatatuple;
+ HeapTuple tuple;
+
+ /*
+ * Read the old sequence. This does a bit more work than really
+ * necessary, but it's simple, and we do want to double-check that it's
+ * indeed a sequence.
+ */
+ init_sequence(seq_relid, &elm, &seq_rel);
+ (void) read_seq_tuple(seq_rel, &buf, &seqdatatuple);
+
+ /*
+ * Copy the existing sequence tuple.
+ */
+ tuple = heap_copytuple(&seqdatatuple);
+
+ /* Now we're done with the old page */
+ UnlockReleaseBuffer(buf);
+
+ /*
+ * Modify the copied tuple to execute the restart (compare the RESTART
+ * action in AlterSequence)
+ */
+ seq = (Form_pg_sequence_data) GETSTRUCT(tuple);
+ seq->last_value = last_value;
+ seq->is_called = is_called;
+ seq->log_cnt = log_cnt;
+
+ /*
+ * Create a new storage file for the sequence.
+ */
+ RelationSetNewRelfilenode(seq_rel, seq_rel->rd_rel->relpersistence);
+
+ /*
+ * Ensure sequence's relfrozenxid is at 0, since it won't contain any
+ * unfrozen XIDs. Same with relminmxid, since a sequence will never
+ * contain multixacts.
+ */
+ Assert(seq_rel->rd_rel->relfrozenxid == InvalidTransactionId);
+ Assert(seq_rel->rd_rel->relminmxid == InvalidMultiXactId);
+
+ /*
+ * Insert the modified tuple into the new storage file.
+ *
+ * XXX Maybe this should also use created=true, just like the other places
+ * calling fill_seq_with_data. That's probably needed for correct cascading
+ * replication.
+ *
+ * XXX That'd mean all fill_seq_with_data callers use created=true, making
+ * the parameter unnecessary.
+ */
+ fill_seq_with_data(seq_rel, tuple);
+
+ /* Clear local cache so that we don't think we have cached numbers */
+ /* Note that we do not change the currval() state */
+ elm->cached = elm->last;
+
+ relation_close(seq_rel, NoLock);
+}
+
/*
* Initialize a sequence's relation with the specified tuple as content
*/
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 3ef6607d246..5beb67e7652 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -85,6 +85,7 @@ typedef struct SubOpts
} SubOpts;
static List *fetch_table_list(WalReceiverConn *wrconn, List *publications);
+static List *fetch_sequence_list(WalReceiverConn *wrconn, List *publications);
static void check_duplicates_in_publist(List *publist, Datum *datums);
static List *merge_publications(List *oldpublist, List *newpublist, bool addpub, const char *subname);
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
@@ -496,6 +497,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
char *err;
WalReceiverConn *wrconn;
List *tables;
+ List *sequences;
ListCell *lc;
char table_state;
@@ -534,6 +536,26 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
InvalidXLogRecPtr);
}
+ /*
+ * Get the sequence list from publisher and build local sequence
+ * status info.
+ */
+ sequences = fetch_sequence_list(wrconn, publications);
+ foreach(lc, sequences)
+ {
+ RangeVar *rv = (RangeVar *) lfirst(lc);
+ Oid relid;
+
+ relid = RangeVarGetRelid(rv, AccessShareLock, false);
+
+ /* Check for supported relkind. */
+ CheckSubscriptionRelkind(get_rel_relkind(relid),
+ rv->schemaname, rv->relname);
+
+ AddSubscriptionRelState(subid, relid, table_state,
+ InvalidXLogRecPtr);
+ }
+
/*
* If requested, create permanent slot for the subscription. We
* won't use the initial snapshot for anything, so no need to
@@ -706,6 +728,10 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
{
Oid relid = subrel_local_oids[off];
+ /* XXX ignore sequences - maybe do this in GetSubscriptionRelations? */
+ if (get_rel_relkind(relid) == RELKIND_SEQUENCE)
+ continue;
+
if (!bsearch(&relid, pubrel_local_oids,
list_length(pubrel_names), sizeof(Oid), oid_cmp))
{
@@ -797,6 +823,183 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data)
ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
}
}
+
+ /*
+ * XXX now do the same thing for sequences, maybe before the preceding
+ * block, or earlier?
+ */
+
+ /* Get the table list from publisher. */
+ pubrel_names = fetch_sequence_list(wrconn, sub->publications);
+
+ /* Get local table list. */
+ subrel_states = GetSubscriptionRelations(sub->oid);
+
+ /*
+ * Build qsorted array of local table oids for faster lookup. This can
+ * potentially contain all tables in the database so speed of lookup
+ * is important.
+ */
+ subrel_local_oids = palloc(list_length(subrel_states) * sizeof(Oid));
+ off = 0;
+ foreach(lc, subrel_states)
+ {
+ SubscriptionRelState *relstate = (SubscriptionRelState *) lfirst(lc);
+
+ subrel_local_oids[off++] = relstate->relid;
+ }
+ qsort(subrel_local_oids, list_length(subrel_states),
+ sizeof(Oid), oid_cmp);
+
+ /*
+ * Rels that we want to remove from subscription and drop any slots
+ * and origins corresponding to them.
+ */
+ sub_remove_rels = palloc(list_length(subrel_states) * sizeof(SubRemoveRels));
+
+ /*
+ * Walk over the remote tables and try to match them to locally known
+ * tables. If the table is not known locally create a new state for
+ * it.
+ *
+ * Also builds array of local oids of remote tables for the next step.
+ */
+ off = 0;
+ pubrel_local_oids = palloc(list_length(pubrel_names) * sizeof(Oid));
+
+ foreach(lc, pubrel_names)
+ {
+ RangeVar *rv = (RangeVar *) lfirst(lc);
+ Oid relid;
+
+ relid = RangeVarGetRelid(rv, AccessShareLock, false);
+
+ /* Check for supported relkind. */
+ CheckSubscriptionRelkind(get_rel_relkind(relid),
+ rv->schemaname, rv->relname);
+
+ pubrel_local_oids[off++] = relid;
+
+ if (!bsearch(&relid, subrel_local_oids,
+ list_length(subrel_states), sizeof(Oid), oid_cmp))
+ {
+ AddSubscriptionRelState(sub->oid, relid,
+ copy_data ? SUBREL_STATE_INIT : SUBREL_STATE_READY,
+ InvalidXLogRecPtr);
+ ereport(DEBUG1,
+ (errmsg_internal("table \"%s.%s\" added to subscription \"%s\"",
+ rv->schemaname, rv->relname, sub->name)));
+ }
+ }
+
+ /*
+ * Next remove state for tables we should not care about anymore using
+ * the data we collected above
+ */
+ qsort(pubrel_local_oids, list_length(pubrel_names),
+ sizeof(Oid), oid_cmp);
+
+ remove_rel_len = 0;
+ for (off = 0; off < list_length(subrel_states); off++)
+ {
+ Oid relid = subrel_local_oids[off];
+
+ /* XXX ignore non-sequences - maybe do this in GetSubscriptionRelations? */
+ if (get_rel_relkind(relid) != RELKIND_SEQUENCE)
+ continue;
+
+ if (!bsearch(&relid, pubrel_local_oids,
+ list_length(pubrel_names), sizeof(Oid), oid_cmp))
+ {
+ char state;
+ XLogRecPtr statelsn;
+
+ /*
+ * Lock pg_subscription_rel with AccessExclusiveLock to
+ * prevent any race conditions with the apply worker
+ * re-launching workers at the same time this code is trying
+ * to remove those tables.
+ *
+ * Even if new worker for this particular rel is restarted it
+ * won't be able to make any progress as we hold exclusive
+ * lock on subscription_rel till the transaction end. It will
+ * simply exit as there is no corresponding rel entry.
+ *
+ * This locking also ensures that the state of rels won't
+ * change till we are done with this refresh operation.
+ */
+ if (!rel)
+ rel = table_open(SubscriptionRelRelationId, AccessExclusiveLock);
+
+ /* Last known rel state. */
+ state = GetSubscriptionRelState(sub->oid, relid, &statelsn);
+
+ sub_remove_rels[remove_rel_len].relid = relid;
+ sub_remove_rels[remove_rel_len++].state = state;
+
+ RemoveSubscriptionRel(sub->oid, relid);
+
+ logicalrep_worker_stop(sub->oid, relid);
+
+ /*
+ * For READY state, we would have already dropped the
+ * tablesync origin.
+ */
+ if (state != SUBREL_STATE_READY)
+ {
+ char originname[NAMEDATALEN];
+
+ /*
+ * Drop the tablesync's origin tracking if exists.
+ *
+ * It is possible that the origin is not yet created for
+ * tablesync worker, this can happen for the states before
+ * SUBREL_STATE_FINISHEDCOPY. The apply worker can also
+ * concurrently try to drop the origin and by this time
+ * the origin might be already removed. For these reasons,
+ * passing missing_ok = true.
+ */
+ ReplicationOriginNameForTablesync(sub->oid, relid, originname,
+ sizeof(originname));
+ replorigin_drop_by_name(originname, true, false);
+ }
+
+ ereport(DEBUG1,
+ (errmsg_internal("table \"%s.%s\" removed from subscription \"%s\"",
+ get_namespace_name(get_rel_namespace(relid)),
+ get_rel_name(relid),
+ sub->name)));
+ }
+ }
+
+ /*
+ * Drop the tablesync slots associated with removed tables. This has
+ * to be at the end because otherwise if there is an error while doing
+ * the database operations we won't be able to rollback dropped slots.
+ */
+ for (off = 0; off < remove_rel_len; off++)
+ {
+ if (sub_remove_rels[off].state != SUBREL_STATE_READY &&
+ sub_remove_rels[off].state != SUBREL_STATE_SYNCDONE)
+ {
+ char syncslotname[NAMEDATALEN] = {0};
+
+ /*
+ * For READY/SYNCDONE states we know the tablesync slot has
+ * already been dropped by the tablesync worker.
+ *
+ * For other states, there is no certainty, maybe the slot
+ * does not exist yet. Also, if we fail after removing some of
+ * the slots, next time, it will again try to drop already
+ * dropped slots and fail. For these reasons, we allow
+ * missing_ok = true for the drop.
+ */
+ ReplicationSlotNameForTablesync(sub->oid, sub_remove_rels[off].relid,
+ syncslotname, sizeof(syncslotname));
+ ReplicationSlotDropAtPubNode(wrconn, syncslotname, true);
+ }
+ }
+
}
PG_FINALLY();
{
@@ -1616,6 +1819,75 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
return tablelist;
}
+/*
+ * Get the list of sequences which belong to specified publications on the
+ * publisher connection.
+ */
+static List *
+fetch_sequence_list(WalReceiverConn *wrconn, List *publications)
+{
+ WalRcvExecResult *res;
+ StringInfoData cmd;
+ TupleTableSlot *slot;
+ Oid tableRow[2] = {TEXTOID, TEXTOID};
+ ListCell *lc;
+ bool first;
+ List *tablelist = NIL;
+
+ Assert(list_length(publications) > 0);
+
+ initStringInfo(&cmd);
+ appendStringInfoString(&cmd, "SELECT DISTINCT s.schemaname, s.sequencename\n"
+ " FROM pg_catalog.pg_publication_sequences s\n"
+ " WHERE s.pubname IN (");
+ first = true;
+ foreach(lc, publications)
+ {
+ char *pubname = strVal(lfirst(lc));
+
+ if (first)
+ first = false;
+ else
+ appendStringInfoString(&cmd, ", ");
+
+ appendStringInfoString(&cmd, quote_literal_cstr(pubname));
+ }
+ appendStringInfoChar(&cmd, ')');
+
+ res = walrcv_exec(wrconn, cmd.data, 2, tableRow);
+ pfree(cmd.data);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not receive list of replicated tables from the publisher: %s",
+ res->err)));
+
+ /* Process tables. */
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ char *nspname;
+ char *relname;
+ bool isnull;
+ RangeVar *rv;
+
+ nspname = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+ relname = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ rv = makeRangeVar(nspname, relname, -1);
+ tablelist = lappend(tablelist, rv);
+
+ ExecClearTuple(slot);
+ }
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+
+ return tablelist;
+}
+
/*
* This is to report the connection failure while dropping replication slots.
* Here, we report the WARNING for all tablesync slots so that user can drop
diff --git a/src/backend/executor/execReplication.c b/src/backend/executor/execReplication.c
index de106d767d1..f8f8ade180c 100644
--- a/src/backend/executor/execReplication.c
+++ b/src/backend/executor/execReplication.c
@@ -635,7 +635,7 @@ void
CheckSubscriptionRelkind(char relkind, const char *nspname,
const char *relname)
{
- if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE)
+ if (relkind != RELKIND_RELATION && relkind != RELKIND_PARTITIONED_TABLE && relkind != RELKIND_SEQUENCE)
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
errmsg("cannot use relation \"%s.%s\" as logical replication target",
diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c
index d4f8455a2bd..cc936328a32 100644
--- a/src/backend/nodes/copyfuncs.c
+++ b/src/backend/nodes/copyfuncs.c
@@ -4863,6 +4863,7 @@ _copyCreatePublicationStmt(const CreatePublicationStmt *from)
COPY_NODE_FIELD(options);
COPY_NODE_FIELD(pubobjects);
COPY_SCALAR_FIELD(for_all_tables);
+ COPY_SCALAR_FIELD(for_all_sequences);
return newnode;
}
diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c
index f1002afe7a0..692cf0e1352 100644
--- a/src/backend/nodes/equalfuncs.c
+++ b/src/backend/nodes/equalfuncs.c
@@ -2346,6 +2346,7 @@ _equalAlterPublicationStmt(const AlterPublicationStmt *a,
COMPARE_NODE_FIELD(options);
COMPARE_NODE_FIELD(pubobjects);
COMPARE_SCALAR_FIELD(for_all_tables);
+ COMPARE_SCALAR_FIELD(for_all_sequences);
COMPARE_SCALAR_FIELD(action);
return true;
diff --git a/src/backend/parser/gram.y b/src/backend/parser/gram.y
index a03b33b53bd..9097ac3fabd 100644
--- a/src/backend/parser/gram.y
+++ b/src/backend/parser/gram.y
@@ -9701,12 +9701,16 @@ AlterOwnerStmt: ALTER AGGREGATE aggregate_with_argtypes OWNER TO RoleSpec
*
* CREATE PUBLICATION FOR ALL TABLES [WITH options]
*
+ * CREATE PUBLICATION FOR ALL SEQUENCES [WITH options]
+ *
* CREATE PUBLICATION FOR pub_obj [, ...] [WITH options]
*
* pub_obj is one of:
*
* TABLE table [, ...]
+ * SEQUENCE table [, ...]
* ALL TABLES IN SCHEMA schema [, ...]
+ * ALL SEQUENCES IN SCHEMA schema [, ...]
*
*****************************************************************************/
@@ -9726,6 +9730,14 @@ CreatePublicationStmt:
n->for_all_tables = true;
$$ = (Node *)n;
}
+ | CREATE PUBLICATION name FOR ALL SEQUENCES opt_definition
+ {
+ CreatePublicationStmt *n = makeNode(CreatePublicationStmt);
+ n->pubname = $3;
+ n->options = $7;
+ n->for_all_sequences = true;
+ $$ = (Node *)n;
+ }
| CREATE PUBLICATION name FOR pub_obj_list opt_definition
{
CreatePublicationStmt *n = makeNode(CreatePublicationStmt);
@@ -9772,6 +9784,26 @@ PublicationObjSpec:
$$->pubobjtype = PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA;
$$->location = @5;
}
+ | SEQUENCE relation_expr
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_SEQUENCE;
+ $$->pubtable = makeNode(PublicationTable);
+ $$->pubtable->relation = $2;
+ }
+ | ALL SEQUENCES IN_P SCHEMA ColId
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA;
+ $$->name = $5;
+ $$->location = @5;
+ }
+ | ALL SEQUENCES IN_P SCHEMA CURRENT_SCHEMA
+ {
+ $$ = makeNode(PublicationObjSpec);
+ $$->pubobjtype = PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA;
+ $$->location = @5;
+ }
| ColId OptWhereClause
{
$$ = makeNode(PublicationObjSpec);
@@ -9839,7 +9871,9 @@ pub_obj_list: PublicationObjSpec
* pub_obj is one of:
*
* TABLE table_name [, ...]
+ * SEQUENCE table_name [, ...]
* ALL TABLES IN SCHEMA schema_name [, ...]
+ * ALL SEQUENCES IN SCHEMA schema_name [, ...]
*
*****************************************************************************/
@@ -10124,6 +10158,12 @@ UnlistenStmt:
}
;
+/*
+ * FIXME
+ *
+ * opt_publication_for_sequences and publication_for_sequences should be
+ * copies for sequences
+ */
/*****************************************************************************
*
diff --git a/src/backend/replication/logical/proto.c b/src/backend/replication/logical/proto.c
index c9b0eeefd7e..3dbe85d61a2 100644
--- a/src/backend/replication/logical/proto.c
+++ b/src/backend/replication/logical/proto.c
@@ -648,6 +648,56 @@ logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
pq_sendbytes(out, message, sz);
}
+/*
+ * Write SEQUENCE to stream
+ */
+void
+logicalrep_write_sequence(StringInfo out, Relation rel, TransactionId xid,
+ XLogRecPtr lsn, bool transactional,
+ int64 last_value, int64 log_cnt, bool is_called)
+{
+ uint8 flags = 0;
+ char *relname;
+
+ pq_sendbyte(out, LOGICAL_REP_MSG_SEQUENCE);
+
+ /* transaction ID (if not valid, we're not streaming) */
+ if (TransactionIdIsValid(xid))
+ pq_sendint32(out, xid);
+
+ pq_sendint8(out, flags);
+ pq_sendint64(out, lsn);
+
+ logicalrep_write_namespace(out, RelationGetNamespace(rel));
+ relname = RelationGetRelationName(rel);
+ pq_sendstring(out, relname);
+
+ pq_sendint8(out, transactional);
+ pq_sendint64(out, last_value);
+ pq_sendint64(out, log_cnt);
+ pq_sendint8(out, is_called);
+}
+
+/*
+ * Read SEQUENCE from the stream.
+ */
+void
+logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata)
+{
+ /* XXX skipping flags and lsn */
+ pq_getmsgint(in, 1);
+ pq_getmsgint64(in);
+
+ /* Read relation name from stream */
+ seqdata->nspname = pstrdup(logicalrep_read_namespace(in));
+ seqdata->seqname = pstrdup(pq_getmsgstring(in));
+
+ seqdata->transactional = pq_getmsgint(in, 1);
+ seqdata->last_value = pq_getmsgint64(in);
+ seqdata->log_cnt = pq_getmsgint64(in);
+ seqdata->is_called = pq_getmsgint(in, 1);
+}
+
/*
* Write relation description to the output stream.
*/
@@ -1203,6 +1253,8 @@ logicalrep_message_type(LogicalRepMsgType action)
return "STREAM ABORT";
case LOGICAL_REP_MSG_STREAM_PREPARE:
return "STREAM PREPARE";
+ case LOGICAL_REP_MSG_SEQUENCE:
+ return "SEQUENCE";
}
elog(ERROR, "invalid logical replication message type \"%c\"", action);
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 1659964571c..315697069dc 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -100,6 +100,7 @@
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/copy.h"
+#include "commands/sequence.h"
#include "miscadmin.h"
#include "parser/parse_relation.h"
#include "pgstat.h"
@@ -359,6 +360,12 @@ process_syncing_tables_for_sync(XLogRecPtr current_lsn)
*
* If the synchronization position is reached (SYNCDONE), then the table can
* be marked as READY and is no longer tracked.
+ *
+ * XXX This needs to handle sequences too - after AlterSubscription_refresh
+ * starts caring about sequences, GetSubscriptionNotReadyRelations won't
+ * return just tables, and we'll have to sync them here. Not sure it's worth
+ * creating a new "sync" worker per sequence, maybe we should just sync them
+ * in the current process (it's pretty light-weight).
*/
static void
process_syncing_tables_for_apply(XLogRecPtr current_lsn)
@@ -999,6 +1006,99 @@ copy_table(Relation rel)
logicalrep_rel_close(relmapentry, NoLock);
}
+
+
+/*
+ * FIXME add comment
+ */
+static void
+fetch_sequence_data(char *nspname, char *relname,
+ int64 *last_value, int64 *log_cnt, bool *is_called)
+{
+ WalRcvExecResult *res;
+ StringInfoData cmd;
+ TupleTableSlot *slot;
+ Oid tableRow[3] = {INT8OID, INT8OID, BOOLOID};
+
+ initStringInfo(&cmd);
+ appendStringInfo(&cmd, "SELECT last_value, log_cnt, is_called\n"
+ " FROM %s", quote_qualified_identifier(nspname, relname));
+
+ res = walrcv_exec(LogRepWorkerWalRcvConn, cmd.data, 3, tableRow);
+ pfree(cmd.data);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not receive list of replicated tables from the publisher: %s",
+ res->err)));
+
+ /* Process the sequence. */
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ bool isnull;
+
+ *last_value = DatumGetInt64(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+
+ *log_cnt = DatumGetInt64(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ *is_called = DatumGetBool(slot_getattr(slot, 3, &isnull));
+ Assert(!isnull);
+
+ ExecClearTuple(slot);
+ }
+ ExecDropSingleTupleTableSlot(slot);
+
+ walrcv_clear_result(res);
+}
+
+/*
+ * Copy existing data of a sequence from publisher.
+ *
+ * Caller is responsible for locking the local relation.
+ */
+static void
+copy_sequence(Relation rel)
+{
+ LogicalRepRelMapEntry *relmapentry;
+ LogicalRepRelation lrel;
+ List *qual = NIL;
+ StringInfoData cmd;
+ int64 last_value = 0,
+ log_cnt = 0;
+ bool is_called = 0;
+
+ /* Get the publisher relation info. */
+ fetch_remote_table_info(get_namespace_name(RelationGetNamespace(rel)),
+ RelationGetRelationName(rel), &lrel, &qual);
+
+ /* sequences don't have row filters */
+ Assert(!qual);
+
+ /* Put the relation into relmap. */
+ logicalrep_relmap_update(&lrel);
+
+ /* Map the publisher relation to local one. */
+ relmapentry = logicalrep_rel_open(lrel.remoteid, NoLock);
+ Assert(rel == relmapentry->localrel);
+
+ /* Start copy on the publisher. */
+ initStringInfo(&cmd);
+
+ Assert(lrel.relkind == RELKIND_SEQUENCE);
+
+ fetch_sequence_data(lrel.nspname, lrel.relname, &last_value, &log_cnt, &is_called);
+
+ ResetSequence2(RelationGetRelid(rel), last_value, log_cnt, is_called);
+
+ logicalrep_rel_close(relmapentry, NoLock);
+}
+
+
+
+
/*
* Determine the tablesync slot name.
*
@@ -1260,10 +1360,20 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
originname)));
}
- /* Now do the initial data copy */
- PushActiveSnapshot(GetTransactionSnapshot());
- copy_table(rel);
- PopActiveSnapshot();
+ if (get_rel_relkind(RelationGetRelid(rel)) == RELKIND_SEQUENCE)
+ {
+ /* Now do the initial sequence copy */
+ PushActiveSnapshot(GetTransactionSnapshot());
+ copy_sequence(rel);
+ PopActiveSnapshot();
+ }
+ else
+ {
+ /* Now do the initial data copy */
+ PushActiveSnapshot(GetTransactionSnapshot());
+ copy_table(rel);
+ PopActiveSnapshot();
+ }
res = walrcv_exec(LogRepWorkerWalRcvConn, "COMMIT", 0, NULL);
if (res->status != WALRCV_OK_COMMAND)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 5d9acc61733..f4d45d83152 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -144,6 +144,7 @@
#include "catalog/pg_tablespace.h"
#include "commands/tablecmds.h"
#include "commands/tablespace.h"
+#include "commands/sequence.h"
#include "commands/trigger.h"
#include "executor/executor.h"
#include "executor/execPartition.h"
@@ -1093,6 +1094,57 @@ apply_handle_origin(StringInfo s)
errmsg_internal("ORIGIN message sent out of order")));
}
+/*
+ * Handle SEQUENCE message.
+ */
+static void
+apply_handle_sequence(StringInfo s)
+{
+ LogicalRepSequence seq;
+ Oid relid;
+
+ if (handle_streamed_transaction(LOGICAL_REP_MSG_SEQUENCE, s))
+ return;
+
+ logicalrep_read_sequence(s, &seq);
+
+ /*
+ * Non-transactional sequence updates should not be part of a remote
+ * transaction. There should not be any running transaction.
+ */
+ Assert((!seq.transactional) || in_remote_transaction);
+ Assert(!(!seq.transactional && in_remote_transaction));
+ Assert(!(!seq.transactional && IsTransactionState()));
+
+ /*
+ * Make sure we're in a transaction (needed by ResetSequence2). For
+ * non-transactional updates we're guaranteed to start a new one,
+ * and we'll commit it at the end.
+ */
+ if (!IsTransactionState())
+ {
+ StartTransactionCommand();
+ maybe_reread_subscription();
+ }
+
+ relid = RangeVarGetRelid(makeRangeVar(seq.nspname,
+ seq.seqname, -1),
+ RowExclusiveLock, false);
+
+ /* lock the sequence in AccessExclusiveLock, as expected by ResetSequence2 */
+ LockRelationOid(relid, AccessExclusiveLock);
+
+ /* apply the sequence change */
+ ResetSequence2(relid, seq.last_value, seq.log_cnt, seq.is_called);
+
+ /*
+ * Commit the per-stream transaction (we only do this when not in
+ * remote transaction, i.e. for non-transactional sequence updates.
+ */
+ if (!in_remote_transaction)
+ CommitTransactionCommand();
+}
+
/*
* Handle STREAM START message.
*/
@@ -2421,6 +2473,10 @@ apply_dispatch(StringInfo s)
*/
break;
+ case LOGICAL_REP_MSG_SEQUENCE:
+ apply_handle_sequence(s);
+ return;
+
case LOGICAL_REP_MSG_STREAM_START:
apply_handle_stream_start(s);
break;
diff --git a/src/backend/replication/pgoutput/pgoutput.c b/src/backend/replication/pgoutput/pgoutput.c
index ea57a0477f0..60bbc524b19 100644
--- a/src/backend/replication/pgoutput/pgoutput.c
+++ b/src/backend/replication/pgoutput/pgoutput.c
@@ -53,6 +53,10 @@ static void pgoutput_message(LogicalDecodingContext *ctx,
ReorderBufferTXN *txn, XLogRecPtr message_lsn,
bool transactional, const char *prefix,
Size sz, const char *message);
+static void pgoutput_sequence(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
+ Relation relation, bool transactional,
+ int64 last_value, int64 log_cnt, bool is_called);
static bool pgoutput_origin_filter(LogicalDecodingContext *ctx,
RepOriginId origin_id);
static void pgoutput_begin_prepare_txn(LogicalDecodingContext *ctx,
@@ -208,6 +212,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->change_cb = pgoutput_change;
cb->truncate_cb = pgoutput_truncate;
cb->message_cb = pgoutput_message;
+ cb->sequence_cb = pgoutput_sequence;
cb->commit_cb = pgoutput_commit_txn;
cb->begin_prepare_cb = pgoutput_begin_prepare_txn;
@@ -224,6 +229,7 @@ _PG_output_plugin_init(OutputPluginCallbacks *cb)
cb->stream_commit_cb = pgoutput_stream_commit;
cb->stream_change_cb = pgoutput_change;
cb->stream_message_cb = pgoutput_message;
+ cb->stream_sequence_cb = pgoutput_sequence;
cb->stream_truncate_cb = pgoutput_truncate;
/* transaction streaming - two-phase commit */
cb->stream_prepare_cb = pgoutput_stream_prepare_txn;
@@ -237,6 +243,7 @@ parse_output_parameters(List *options, PGOutputData *data)
bool publication_names_given = false;
bool binary_option_given = false;
bool messages_option_given = false;
+ bool sequences_option_given = false;
bool streaming_given = false;
bool two_phase_option_given = false;
@@ -244,6 +251,7 @@ parse_output_parameters(List *options, PGOutputData *data)
data->streaming = false;
data->messages = false;
data->two_phase = false;
+ data->sequences = true;
foreach(lc, options)
{
@@ -312,6 +320,16 @@ parse_output_parameters(List *options, PGOutputData *data)
data->messages = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "sequences") == 0)
+ {
+ if (sequences_option_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ sequences_option_given = true;
+
+ data->sequences = defGetBoolean(defel);
+ }
else if (strcmp(defel->defname, "streaming") == 0)
{
if (streaming_given)
@@ -1440,6 +1458,51 @@ pgoutput_message(LogicalDecodingContext *ctx, ReorderBufferTXN *txn,
OutputPluginWrite(ctx, true);
}
+static void
+pgoutput_sequence(LogicalDecodingContext *ctx,
+ ReorderBufferTXN *txn, XLogRecPtr sequence_lsn,
+ Relation relation, bool transactional,
+ int64 last_value, int64 log_cnt, bool is_called)
+{
+ PGOutputData *data = (PGOutputData *) ctx->output_plugin_private;
+ TransactionId xid = InvalidTransactionId;
+ RelationSyncEntry *relentry;
+
+ if (!data->sequences)
+ return;
+
+ if (!is_publishable_relation(relation))
+ return;
+
+ /*
+ * Remember the xid for the message in streaming mode. See
+ * pgoutput_change.
+ */
+ if (in_streaming)
+ xid = txn->xid;
+
+ relentry = get_rel_sync_entry(data, relation);
+
+ /*
+ * First check the sequence filter.
+ *
+ * We handle just REORDER_BUFFER_CHANGE_SEQUENCE here.
+ */
+ if (!relentry->pubactions.pubsequence)
+ return;
+
+ OutputPluginPrepareWrite(ctx, true);
+ logicalrep_write_sequence(ctx->out,
+ relation,
+ xid,
+ sequence_lsn,
+ transactional,
+ last_value,
+ log_cnt,
+ is_called);
+ OutputPluginWrite(ctx, true);
+}
+
/*
* Currently we always forward.
*/
@@ -1725,7 +1788,8 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->schema_sent = false;
entry->streamed_txns = NIL;
entry->pubactions.pubinsert = entry->pubactions.pubupdate =
- entry->pubactions.pubdelete = entry->pubactions.pubtruncate = false;
+ entry->pubactions.pubdelete = entry->pubactions.pubtruncate =
+ entry->pubactions.pubsequence = false;
entry->new_slot = NULL;
entry->old_slot = NULL;
memset(entry->exprstate, 0, sizeof(entry->exprstate));
@@ -1751,6 +1815,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
bool am_partition = get_rel_relispartition(relid);
char relkind = get_rel_relkind(relid);
List *rel_publications = NIL;
+ bool is_sequence = (get_rel_relkind(relid) == RELKIND_SEQUENCE);
/* Reload publications if needed before use. */
if (!publications_valid)
@@ -1779,6 +1844,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->pubactions.pubupdate = false;
entry->pubactions.pubdelete = false;
entry->pubactions.pubtruncate = false;
+ entry->pubactions.pubsequence = false;
/*
* Tuple slots cleanups. (Will be rebuilt later if needed).
@@ -1815,12 +1881,23 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
Publication *pub = lfirst(lc);
bool publish = false;
- if (pub->alltables)
+ if (pub->alltables && (!is_sequence))
{
publish = true;
if (pub->pubviaroot && am_partition)
publish_as_relid = llast_oid(get_partition_ancestors(relid));
}
+ else if (pub->allsequences && is_sequence)
+ {
+ publish = true;
+ }
+
+ /* if a sequence, just cross-check the list of publications */
+ if (!publish && is_sequence)
+ {
+ if (list_member_oid(pubids, pub->oid))
+ publish = true;
+ }
if (!publish)
{
@@ -1866,6 +1943,7 @@ get_rel_sync_entry(PGOutputData *data, Relation relation)
entry->pubactions.pubupdate |= pub->pubactions.pubupdate;
entry->pubactions.pubdelete |= pub->pubactions.pubdelete;
entry->pubactions.pubtruncate |= pub->pubactions.pubtruncate;
+ entry->pubactions.pubsequence |= pub->pubactions.pubsequence;
rel_publications = lappend(rel_publications, pub);
}
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index fccffce5729..c29d822313d 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -5587,7 +5587,15 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
GetSchemaPublications(schemaid));
}
}
- puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
+
+ /*
+ * Consider also FOR ALL TABLES and FOR ALL SEQUENCES publications,
+ * depending on the relkind of the relation.
+ */
+ if (relation->rd_rel->relkind == RELKIND_SEQUENCE)
+ puboids = list_concat_unique_oid(puboids, GetAllSequencesPublications());
+ else
+ puboids = list_concat_unique_oid(puboids, GetAllTablesPublications());
foreach(lc, puboids)
{
@@ -5606,6 +5614,7 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
pubdesc->pubactions.pubupdate |= pubform->pubupdate;
pubdesc->pubactions.pubdelete |= pubform->pubdelete;
pubdesc->pubactions.pubtruncate |= pubform->pubtruncate;
+ pubdesc->pubactions.pubsequence |= pubform->pubsequence;
/*
* Check if all columns referenced in the filter expression are part of
@@ -5631,6 +5640,8 @@ RelationBuildPublicationDesc(Relation relation, PublicationDesc *pubdesc)
* If we know everything is replicated and the row filter is invalid
* for update and delete, there is no point to check for other
* publications.
+ *
+ * XXX We ignore sequences here, because those don't use row filters.
*/
if (pubdesc->pubactions.pubinsert && pubdesc->pubactions.pubupdate &&
pubdesc->pubactions.pubdelete && pubdesc->pubactions.pubtruncate &&
diff --git a/src/backend/utils/cache/syscache.c b/src/backend/utils/cache/syscache.c
index f4e7819f1e2..763aac81596 100644
--- a/src/backend/utils/cache/syscache.c
+++ b/src/backend/utils/cache/syscache.c
@@ -630,12 +630,12 @@ static const struct cachedesc cacheinfo[] = {
64
},
{PublicationNamespaceRelationId, /* PUBLICATIONNAMESPACEMAP */
- PublicationNamespacePnnspidPnpubidIndexId,
- 2,
+ PublicationNamespacePnnspidPnpubidSeqIndexId,
+ 3,
{
Anum_pg_publication_namespace_pnnspid,
Anum_pg_publication_namespace_pnpubid,
- 0,
+ Anum_pg_publication_namespace_pnsequences,
0
},
64
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index e69dcf8a484..ef8c6e43c61 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -3788,10 +3788,12 @@ getPublications(Archive *fout, int *numPublications)
int i_pubname;
int i_pubowner;
int i_puballtables;
+ int i_puballsequences;
int i_pubinsert;
int i_pubupdate;
int i_pubdelete;
int i_pubtruncate;
+ int i_pubsequence;
int i_pubviaroot;
int i,
ntups;
@@ -3807,23 +3809,29 @@ getPublications(Archive *fout, int *numPublications)
resetPQExpBuffer(query);
/* Get the publications. */
- if (fout->remoteVersion >= 130000)
+ if (fout->remoteVersion >= 150000)
+ appendPQExpBuffer(query,
+ "SELECT p.tableoid, p.oid, p.pubname, "
+ "p.pubowner, "
+ "p.puballtables, p.puballsequences, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubsequence, p.pubviaroot "
+ "FROM pg_publication p");
+ else if (fout->remoteVersion >= 130000)
appendPQExpBuffer(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, p.pubviaroot "
+ "p.puballtables, false AS p.puballsequences, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS p.pubsequence, p.pubviaroot "
"FROM pg_publication p");
else if (fout->remoteVersion >= 110000)
appendPQExpBuffer(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS pubviaroot "
+ "p.puballtables, false AS p.puballsequences, p.pubinsert, p.pubupdate, p.pubdelete, p.pubtruncate, false AS p.pubsequence, false AS pubviaroot "
"FROM pg_publication p");
else
appendPQExpBuffer(query,
"SELECT p.tableoid, p.oid, p.pubname, "
"p.pubowner, "
- "p.puballtables, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false AS pubviaroot "
+ "p.puballtables, false AS p.puballsequences, p.pubinsert, p.pubupdate, p.pubdelete, false AS pubtruncate, false AS p.pubsequence, false AS pubviaroot "
"FROM pg_publication p");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -3835,10 +3843,12 @@ getPublications(Archive *fout, int *numPublications)
i_pubname = PQfnumber(res, "pubname");
i_pubowner = PQfnumber(res, "pubowner");
i_puballtables = PQfnumber(res, "puballtables");
+ i_puballsequences = PQfnumber(res, "puballsequences");
i_pubinsert = PQfnumber(res, "pubinsert");
i_pubupdate = PQfnumber(res, "pubupdate");
i_pubdelete = PQfnumber(res, "pubdelete");
i_pubtruncate = PQfnumber(res, "pubtruncate");
+ i_pubsequence = PQfnumber(res, "pubsequence");
i_pubviaroot = PQfnumber(res, "pubviaroot");
pubinfo = pg_malloc(ntups * sizeof(PublicationInfo));
@@ -3854,6 +3864,8 @@ getPublications(Archive *fout, int *numPublications)
pubinfo[i].rolname = getRoleName(PQgetvalue(res, i, i_pubowner));
pubinfo[i].puballtables =
(strcmp(PQgetvalue(res, i, i_puballtables), "t") == 0);
+ pubinfo[i].puballsequences =
+ (strcmp(PQgetvalue(res, i, i_puballsequences), "t") == 0);
pubinfo[i].pubinsert =
(strcmp(PQgetvalue(res, i, i_pubinsert), "t") == 0);
pubinfo[i].pubupdate =
@@ -3862,6 +3874,8 @@ getPublications(Archive *fout, int *numPublications)
(strcmp(PQgetvalue(res, i, i_pubdelete), "t") == 0);
pubinfo[i].pubtruncate =
(strcmp(PQgetvalue(res, i, i_pubtruncate), "t") == 0);
+ pubinfo[i].pubsequence =
+ (strcmp(PQgetvalue(res, i, i_pubsequence), "t") == 0);
pubinfo[i].pubviaroot =
(strcmp(PQgetvalue(res, i, i_pubviaroot), "t") == 0);
@@ -3907,6 +3921,9 @@ dumpPublication(Archive *fout, const PublicationInfo *pubinfo)
if (pubinfo->puballtables)
appendPQExpBufferStr(query, " FOR ALL TABLES");
+ if (pubinfo->puballsequences)
+ appendPQExpBufferStr(query, " FOR ALL SEQUENCES");
+
appendPQExpBufferStr(query, " WITH (publish = '");
if (pubinfo->pubinsert)
{
@@ -3941,6 +3958,15 @@ dumpPublication(Archive *fout, const PublicationInfo *pubinfo)
first = false;
}
+ if (pubinfo->pubsequence)
+ {
+ if (!first)
+ appendPQExpBufferStr(query, ", ");
+
+ appendPQExpBufferStr(query, "sequence");
+ first = false;
+ }
+
appendPQExpBufferStr(query, "'");
if (pubinfo->pubviaroot)
@@ -3987,6 +4013,7 @@ getPublicationNamespaces(Archive *fout)
int i_oid;
int i_pnpubid;
int i_pnnspid;
+ int i_pnsequences;
int i,
j,
ntups;
@@ -3998,7 +4025,7 @@ getPublicationNamespaces(Archive *fout)
/* Collect all publication membership info. */
appendPQExpBufferStr(query,
- "SELECT tableoid, oid, pnpubid, pnnspid "
+ "SELECT tableoid, oid, pnpubid, pnnspid, pnsequences "
"FROM pg_catalog.pg_publication_namespace");
res = ExecuteSqlQuery(fout, query->data, PGRES_TUPLES_OK);
@@ -4008,6 +4035,7 @@ getPublicationNamespaces(Archive *fout)
i_oid = PQfnumber(res, "oid");
i_pnpubid = PQfnumber(res, "pnpubid");
i_pnnspid = PQfnumber(res, "pnnspid");
+ i_pnsequences = PQfnumber(res, "pnsequences");
/* this allocation may be more than we need */
pubsinfo = pg_malloc(ntups * sizeof(PublicationSchemaInfo));
@@ -4017,6 +4045,7 @@ getPublicationNamespaces(Archive *fout)
{
Oid pnpubid = atooid(PQgetvalue(res, i, i_pnpubid));
Oid pnnspid = atooid(PQgetvalue(res, i, i_pnnspid));
+ bool pnsequences = (strcmp(PQgetvalue(res, i, i_pnsequences), "t") == 0);
PublicationInfo *pubinfo;
NamespaceInfo *nspinfo;
@@ -4048,6 +4077,7 @@ getPublicationNamespaces(Archive *fout)
pubsinfo[j].dobj.name = nspinfo->dobj.name;
pubsinfo[j].publication = pubinfo;
pubsinfo[j].pubschema = nspinfo;
+ pubsinfo[j].pubsequences = pnsequences;
/* Decide whether we want to dump it */
selectDumpablePublicationObject(&(pubsinfo[j].dobj), fout);
@@ -4181,7 +4211,11 @@ dumpPublicationNamespace(Archive *fout, const PublicationSchemaInfo *pubsinfo)
query = createPQExpBuffer();
appendPQExpBuffer(query, "ALTER PUBLICATION %s ", fmtId(pubinfo->dobj.name));
- appendPQExpBuffer(query, "ADD ALL TABLES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
+
+ if (pubsinfo->pubsequences)
+ appendPQExpBuffer(query, "ADD ALL SEQUENCES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
+ else
+ appendPQExpBuffer(query, "ADD ALL TABLES IN SCHEMA %s;\n", fmtId(schemainfo->dobj.name));
/*
* There is no point in creating drop query as the drop is done by schema
@@ -4214,6 +4248,7 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
TableInfo *tbinfo = pubrinfo->pubtable;
PQExpBuffer query;
char *tag;
+ char *description;
/* Do nothing in data-only dump */
if (dopt->dataOnly)
@@ -4223,10 +4258,22 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
query = createPQExpBuffer();
- appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY",
- fmtId(pubinfo->dobj.name));
+ if (tbinfo->relkind == RELKIND_SEQUENCE)
+ {
+ appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD SEQUENCE",
+ fmtId(pubinfo->dobj.name));
+ description = "PUBLICATION SEQUENCE";
+ }
+ else
+ {
+ appendPQExpBuffer(query, "ALTER PUBLICATION %s ADD TABLE ONLY",
+ fmtId(pubinfo->dobj.name));
+ description = "PUBLICATION TABLE";
+ }
+
appendPQExpBuffer(query, " %s",
fmtQualifiedDumpable(tbinfo));
+
if (pubrinfo->pubrelqual)
{
/*
@@ -4249,7 +4296,7 @@ dumpPublicationTable(Archive *fout, const PublicationRelInfo *pubrinfo)
ARCHIVE_OPTS(.tag = tag,
.namespace = tbinfo->dobj.namespace->dobj.name,
.owner = pubinfo->rolname,
- .description = "PUBLICATION TABLE",
+ .description = description,
.section = SECTION_POST_DATA,
.createStmt = query->data));
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 997a3b60719..8739af69122 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -615,10 +615,12 @@ typedef struct _PublicationInfo
DumpableObject dobj;
const char *rolname;
bool puballtables;
+ bool puballsequences;
bool pubinsert;
bool pubupdate;
bool pubdelete;
bool pubtruncate;
+ bool pubsequence;
bool pubviaroot;
} PublicationInfo;
@@ -643,6 +645,7 @@ typedef struct _PublicationSchemaInfo
DumpableObject dobj;
PublicationInfo *publication;
NamespaceInfo *pubschema;
+ bool pubsequences;
} PublicationSchemaInfo;
/*
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index dd065c758fa..380c929f815 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -2357,7 +2357,7 @@ my %tests = (
create_order => 50,
create_sql => 'CREATE PUBLICATION pub1;',
regexp => qr/^
- \QCREATE PUBLICATION pub1 WITH (publish = 'insert, update, delete, truncate');\E
+ \QCREATE PUBLICATION pub1 WITH (publish = 'insert, update, delete, truncate, sequence');\E
/xm,
like => { %full_runs, section_post_data => 1, },
},
@@ -2377,7 +2377,18 @@ my %tests = (
create_order => 50,
create_sql => 'CREATE PUBLICATION pub3;',
regexp => qr/^
- \QCREATE PUBLICATION pub3 WITH (publish = 'insert, update, delete, truncate');\E
+ \QCREATE PUBLICATION pub3 WITH (publish = 'insert, update, delete, truncate, sequence');\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ },
+
+ 'CREATE PUBLICATION pub4' => {
+ create_order => 50,
+ create_sql => 'CREATE PUBLICATION pub4
+ FOR ALL SEQUENCES
+ WITH (publish = \'\');',
+ regexp => qr/^
+ \QCREATE PUBLICATION pub4 FOR ALL SEQUENCES WITH (publish = '');\E
/xm,
like => { %full_runs, section_post_data => 1, },
},
@@ -2439,6 +2450,27 @@ my %tests = (
like => { %full_runs, section_post_data => 1, },
},
+ 'ALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA dump_test' => {
+ create_order => 51,
+ create_sql =>
+ 'ALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA dump_test;',
+ regexp => qr/^
+ \QALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA dump_test;\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ unlike => { exclude_dump_test_schema => 1, },
+ },
+
+ 'ALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA public' => {
+ create_order => 52,
+ create_sql =>
+ 'ALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA public;',
+ regexp => qr/^
+ \QALTER PUBLICATION pub3 ADD ALL SEQUENCES IN SCHEMA public;\E
+ /xm,
+ like => { %full_runs, section_post_data => 1, },
+ },
+
'CREATE SCHEMA public' => {
regexp => qr/^CREATE SCHEMA public;/m,
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index e3382933d98..c0df70f0f9a 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -1621,28 +1621,19 @@ describeOneTableDetails(const char *schemaname,
if (tableinfo.relkind == RELKIND_SEQUENCE)
{
PGresult *result = NULL;
- printQueryOpt myopt = pset.popt;
- char *footers[2] = {NULL, NULL};
if (pset.sversion >= 100000)
{
printfPQExpBuffer(&buf,
- "SELECT pg_catalog.format_type(seqtypid, NULL) AS \"%s\",\n"
- " seqstart AS \"%s\",\n"
- " seqmin AS \"%s\",\n"
- " seqmax AS \"%s\",\n"
- " seqincrement AS \"%s\",\n"
- " CASE WHEN seqcycle THEN '%s' ELSE '%s' END AS \"%s\",\n"
- " seqcache AS \"%s\"\n",
- gettext_noop("Type"),
- gettext_noop("Start"),
- gettext_noop("Minimum"),
- gettext_noop("Maximum"),
- gettext_noop("Increment"),
+ "SELECT pg_catalog.format_type(seqtypid, NULL),\n"
+ " seqstart,\n"
+ " seqmin,\n"
+ " seqmax,\n"
+ " seqincrement,\n"
+ " CASE WHEN seqcycle THEN '%s' ELSE '%s' END,\n"
+ " seqcache\n",
gettext_noop("yes"),
- gettext_noop("no"),
- gettext_noop("Cycles?"),
- gettext_noop("Cache"));
+ gettext_noop("no"));
appendPQExpBuffer(&buf,
"FROM pg_catalog.pg_sequence\n"
"WHERE seqrelid = '%s';",
@@ -1651,22 +1642,15 @@ describeOneTableDetails(const char *schemaname,
else
{
printfPQExpBuffer(&buf,
- "SELECT 'bigint' AS \"%s\",\n"
- " start_value AS \"%s\",\n"
- " min_value AS \"%s\",\n"
- " max_value AS \"%s\",\n"
- " increment_by AS \"%s\",\n"
- " CASE WHEN is_cycled THEN '%s' ELSE '%s' END AS \"%s\",\n"
- " cache_value AS \"%s\"\n",
- gettext_noop("Type"),
- gettext_noop("Start"),
- gettext_noop("Minimum"),
- gettext_noop("Maximum"),
- gettext_noop("Increment"),
+ "SELECT 'bigint',\n"
+ " start_value,\n"
+ " min_value,\n"
+ " max_value,\n"
+ " increment_by,\n"
+ " CASE WHEN is_cycled THEN '%s' ELSE '%s' END,\n"
+ " cache_value\n",
gettext_noop("yes"),
- gettext_noop("no"),
- gettext_noop("Cycles?"),
- gettext_noop("Cache"));
+ gettext_noop("no"));
appendPQExpBuffer(&buf, "FROM %s", fmtId(schemaname));
/* must be separate because fmtId isn't reentrant */
appendPQExpBuffer(&buf, ".%s;", fmtId(relationname));
@@ -1676,6 +1660,51 @@ describeOneTableDetails(const char *schemaname,
if (!res)
goto error_return;
+ numrows = PQntuples(res);
+
+ /* XXX reset to use expanded output for sequences (maybe we should
+ * keep this disabled, just like for tables?) */
+ myopt.expanded = pset.popt.topt.expanded;
+
+ printTableInit(&cont, &myopt, title.data, 7, numrows);
+ printTableInitialized = true;
+
+ printfPQExpBuffer(&title, _("Sequence \"%s.%s\""),
+ schemaname, relationname);
+
+ printTableAddHeader(&cont, gettext_noop("Type"), true, 'l');
+ printTableAddHeader(&cont, gettext_noop("Start"), true, 'r');
+ printTableAddHeader(&cont, gettext_noop("Minimum"), true, 'r');
+ printTableAddHeader(&cont, gettext_noop("Maximum"), true, 'r');
+ printTableAddHeader(&cont, gettext_noop("Increment"), true, 'r');
+ printTableAddHeader(&cont, gettext_noop("Cycles?"), true, 'l');
+ printTableAddHeader(&cont, gettext_noop("Cache"), true, 'r');
+
+ /* Generate table cells to be printed */
+ for (i = 0; i < numrows; i++)
+ {
+ /* Type */
+ printTableAddCell(&cont, PQgetvalue(res, i, 0), false, false);
+
+ /* Start */
+ printTableAddCell(&cont, PQgetvalue(res, i, 1), false, false);
+
+ /* Minimum */
+ printTableAddCell(&cont, PQgetvalue(res, i, 2), false, false);
+
+ /* Maximum */
+ printTableAddCell(&cont, PQgetvalue(res, i, 3), false, false);
+
+ /* Increment */
+ printTableAddCell(&cont, PQgetvalue(res, i, 4), false, false);
+
+ /* Cycles? */
+ printTableAddCell(&cont, PQgetvalue(res, i, 5), false, false);
+
+ /* Cache */
+ printTableAddCell(&cont, PQgetvalue(res, i, 6), false, false);
+ }
+
/* Footer information about a sequence */
/* Get the column that owns this sequence */
@@ -1709,29 +1738,63 @@ describeOneTableDetails(const char *schemaname,
switch (PQgetvalue(result, 0, 1)[0])
{
case 'a':
- footers[0] = psprintf(_("Owned by: %s"),
- PQgetvalue(result, 0, 0));
+ printTableAddFooter(&cont,
+ psprintf(_("Owned by: %s"),
+ PQgetvalue(result, 0, 0)));
break;
case 'i':
- footers[0] = psprintf(_("Sequence for identity column: %s"),
- PQgetvalue(result, 0, 0));
+ printTableAddFooter(&cont,
+ psprintf(_("Sequence for identity column: %s"),
+ PQgetvalue(result, 0, 0)));
break;
}
}
PQclear(result);
- printfPQExpBuffer(&title, _("Sequence \"%s.%s\""),
- schemaname, relationname);
+ /* print any publications */
+ if (pset.sversion >= 150000)
+ {
+ int tuples = 0;
- myopt.footers = footers;
- myopt.topt.default_footer = false;
- myopt.title = title.data;
- myopt.translate_header = true;
+ printfPQExpBuffer(&buf,
+ "SELECT pubname\n"
+ "FROM pg_catalog.pg_publication p\n"
+ " JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n"
+ " JOIN pg_catalog.pg_class pc ON pc.relnamespace = pn.pnnspid\n"
+ "WHERE pc.oid ='%s' and pn.pnsequences and pg_catalog.pg_relation_is_publishable('%s')\n"
+ "UNION\n"
+ "SELECT pubname\n"
+ "FROM pg_catalog.pg_publication p\n"
+ " JOIN pg_catalog.pg_publication_rel pr ON p.oid = pr.prpubid\n"
+ "WHERE pr.prrelid = '%s'\n"
+ "UNION\n"
+ "SELECT pubname\n"
+ "FROM pg_catalog.pg_publication p\n"
+ "WHERE p.puballsequences AND pg_catalog.pg_relation_is_publishable('%s')\n"
+ "ORDER BY 1;",
+ oid, oid, oid, oid);
- printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+ 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 (footers[0])
- free(footers[0]);
+ printTable(&cont, pset.queryFout, false, pset.logfile);
retval = true;
goto error_return; /* not an error, just return early */
@@ -1958,6 +2021,11 @@ describeOneTableDetails(const char *schemaname,
for (i = 0; i < cols; i++)
printTableAddHeader(&cont, headers[i], true, 'l');
+ res = PSQLexec(buf.data);
+ if (!res)
+ goto error_return;
+ numrows = PQntuples(res);
+
/* Generate table cells to be printed */
for (i = 0; i < numrows; i++)
{
@@ -2883,7 +2951,7 @@ describeOneTableDetails(const char *schemaname,
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n"
" JOIN pg_catalog.pg_class pc ON pc.relnamespace = pn.pnnspid\n"
- "WHERE pc.oid ='%s' and pg_catalog.pg_relation_is_publishable('%s')\n"
+ "WHERE pc.oid ='%s' and (not pn.pnsequences) and pg_catalog.pg_relation_is_publishable('%s')\n"
"UNION\n"
"SELECT pubname\n"
" , pg_get_expr(pr.prqual, c.oid)\n"
@@ -4764,7 +4832,7 @@ listSchemas(const char *pattern, bool verbose, bool showSystem)
int i;
printfPQExpBuffer(&buf,
- "SELECT pubname \n"
+ "SELECT pubname, (CASE WHEN pnsequences THEN 'sequences' ELSE 'tables' END) AS pubtype\n"
"FROM pg_catalog.pg_publication p\n"
" JOIN pg_catalog.pg_publication_namespace pn ON p.oid = pn.pnpubid\n"
" JOIN pg_catalog.pg_namespace n ON n.oid = pn.pnnspid \n"
@@ -4793,8 +4861,9 @@ listSchemas(const char *pattern, bool verbose, bool showSystem)
/* Might be an empty set - that's ok */
for (i = 0; i < pub_schema_tuples; i++)
{
- printfPQExpBuffer(&buf, " \"%s\"",
- PQgetvalue(result, i, 0));
+ printfPQExpBuffer(&buf, " \"%s\" (%s)",
+ PQgetvalue(result, i, 0),
+ PQgetvalue(result, i, 1));
footers[i + 1] = pg_strdup(buf.data);
}
@@ -5799,7 +5868,7 @@ listPublications(const char *pattern)
PQExpBufferData buf;
PGresult *res;
printQueryOpt myopt = pset.popt;
- static const bool translate_columns[] = {false, false, false, false, false, false, false, false};
+ static const bool translate_columns[] = {false, false, false, false, false, false, false, false, false, false};
if (pset.sversion < 100000)
{
@@ -5813,23 +5882,45 @@ listPublications(const char *pattern)
initPQExpBuffer(&buf);
- printfPQExpBuffer(&buf,
- "SELECT pubname AS \"%s\",\n"
- " pg_catalog.pg_get_userbyid(pubowner) AS \"%s\",\n"
- " puballtables AS \"%s\",\n"
- " pubinsert AS \"%s\",\n"
- " pubupdate AS \"%s\",\n"
- " pubdelete AS \"%s\"",
- gettext_noop("Name"),
- gettext_noop("Owner"),
- gettext_noop("All tables"),
- gettext_noop("Inserts"),
- gettext_noop("Updates"),
- gettext_noop("Deletes"));
+ if (pset.sversion >= 150000)
+ printfPQExpBuffer(&buf,
+ "SELECT pubname AS \"%s\",\n"
+ " pg_catalog.pg_get_userbyid(pubowner) AS \"%s\",\n"
+ " puballtables AS \"%s\",\n"
+ " puballsequences AS \"%s\",\n"
+ " pubinsert AS \"%s\",\n"
+ " pubupdate AS \"%s\",\n"
+ " pubdelete AS \"%s\"",
+ gettext_noop("Name"),
+ gettext_noop("Owner"),
+ gettext_noop("All tables"),
+ gettext_noop("All sequences"),
+ gettext_noop("Inserts"),
+ gettext_noop("Updates"),
+ gettext_noop("Deletes"));
+ else
+ printfPQExpBuffer(&buf,
+ "SELECT pubname AS \"%s\",\n"
+ " pg_catalog.pg_get_userbyid(pubowner) AS \"%s\",\n"
+ " puballtables AS \"%s\",\n"
+ " pubinsert AS \"%s\",\n"
+ " pubupdate AS \"%s\",\n"
+ " pubdelete AS \"%s\"",
+ gettext_noop("Name"),
+ gettext_noop("Owner"),
+ gettext_noop("All tables"),
+ gettext_noop("Inserts"),
+ gettext_noop("Updates"),
+ gettext_noop("Deletes"));
+
if (pset.sversion >= 110000)
appendPQExpBuffer(&buf,
",\n pubtruncate AS \"%s\"",
gettext_noop("Truncates"));
+ if (pset.sversion >= 150000)
+ appendPQExpBuffer(&buf,
+ ",\n pubsequence AS \"%s\"",
+ gettext_noop("Sequences"));
if (pset.sversion >= 130000)
appendPQExpBuffer(&buf,
",\n pubviaroot AS \"%s\"",
@@ -5915,6 +6006,7 @@ describePublications(const char *pattern)
PGresult *res;
bool has_pubtruncate;
bool has_pubviaroot;
+ bool has_pubsequence;
PQExpBufferData title;
printTableContent cont;
@@ -5931,13 +6023,21 @@ describePublications(const char *pattern)
has_pubtruncate = (pset.sversion >= 110000);
has_pubviaroot = (pset.sversion >= 130000);
+ has_pubsequence = (pset.sversion >= 150000);
initPQExpBuffer(&buf);
- printfPQExpBuffer(&buf,
- "SELECT oid, pubname,\n"
- " pg_catalog.pg_get_userbyid(pubowner) AS owner,\n"
- " puballtables, pubinsert, pubupdate, pubdelete");
+ if (pset.sversion >= 150000)
+ printfPQExpBuffer(&buf,
+ "SELECT oid, pubname,\n"
+ " pg_catalog.pg_get_userbyid(pubowner) AS owner,\n"
+ " puballtables, puballsequences, pubinsert, pubupdate, pubdelete, pubsequence");
+ else
+ printfPQExpBuffer(&buf,
+ "SELECT oid, pubname,\n"
+ " pg_catalog.pg_get_userbyid(pubowner) AS owner,\n"
+ " puballtables, false as puballsequences, pubinsert, pubupdate, pubdelete, false as pubsequence");
+
if (has_pubtruncate)
appendPQExpBufferStr(&buf,
", pubtruncate");
@@ -5984,12 +6084,15 @@ describePublications(const char *pattern)
char *pubid = PQgetvalue(res, i, 0);
char *pubname = PQgetvalue(res, i, 1);
bool puballtables = strcmp(PQgetvalue(res, i, 3), "t") == 0;
+ bool puballsequences = strcmp(PQgetvalue(res, i, 4), "t") == 0;
printTableOpt myopt = pset.popt.topt;
if (has_pubtruncate)
ncols++;
if (has_pubviaroot)
ncols++;
+ if (has_pubsequence)
+ ncols += 2; /* adds "truncate" action and "all sequences" */
initPQExpBuffer(&title);
printfPQExpBuffer(&title, _("Publication %s"), pubname);
@@ -5997,23 +6100,30 @@ describePublications(const char *pattern)
printTableAddHeader(&cont, gettext_noop("Owner"), true, align);
printTableAddHeader(&cont, gettext_noop("All tables"), true, align);
+ if (has_pubsequence)
+ printTableAddHeader(&cont, gettext_noop("All sequences"), true, align);
printTableAddHeader(&cont, gettext_noop("Inserts"), true, align);
printTableAddHeader(&cont, gettext_noop("Updates"), true, align);
printTableAddHeader(&cont, gettext_noop("Deletes"), true, align);
+ if (has_pubsequence)
+ printTableAddHeader(&cont, gettext_noop("Sequences"), true, align);
if (has_pubtruncate)
printTableAddHeader(&cont, gettext_noop("Truncates"), true, align);
if (has_pubviaroot)
printTableAddHeader(&cont, gettext_noop("Via root"), true, align);
- printTableAddCell(&cont, PQgetvalue(res, i, 2), false, false);
- printTableAddCell(&cont, PQgetvalue(res, i, 3), false, false);
- printTableAddCell(&cont, PQgetvalue(res, i, 4), false, false);
- printTableAddCell(&cont, PQgetvalue(res, i, 5), false, false);
- printTableAddCell(&cont, PQgetvalue(res, i, 6), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, 2), false, false); /* owner */
+ printTableAddCell(&cont, PQgetvalue(res, i, 3), false, false); /* all tables */
+ printTableAddCell(&cont, PQgetvalue(res, i, 4), false, false); /* all sequences */
+ printTableAddCell(&cont, PQgetvalue(res, i, 5), false, false); /* insert */
+ printTableAddCell(&cont, PQgetvalue(res, i, 6), false, false); /* update */
+ printTableAddCell(&cont, PQgetvalue(res, i, 7), false, false); /* delete */
+ if (has_pubsequence)
+ printTableAddCell(&cont, PQgetvalue(res, i, 8), false, false); /* sequence */
if (has_pubtruncate)
- printTableAddCell(&cont, PQgetvalue(res, i, 7), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, 9), false, false); /* truncate */
if (has_pubviaroot)
- printTableAddCell(&cont, PQgetvalue(res, i, 8), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, 10), false, false); /* via root */
if (!puballtables)
{
@@ -6033,6 +6143,7 @@ describePublications(const char *pattern)
"WHERE c.relnamespace = n.oid\n"
" AND c.oid = pr.prrelid\n"
" AND pr.prpubid = '%s'\n"
+ " AND c.relkind != 'S'\n" /* exclude sequences */
"ORDER BY 1,2", pubid);
if (!addFooterToPublicationDesc(&buf, "Tables:", false, &cont))
goto error_return;
@@ -6044,7 +6155,7 @@ describePublications(const char *pattern)
"SELECT n.nspname\n"
"FROM pg_catalog.pg_namespace n\n"
" JOIN pg_catalog.pg_publication_namespace pn ON n.oid = pn.pnnspid\n"
- "WHERE pn.pnpubid = '%s'\n"
+ "WHERE pn.pnpubid = '%s' AND NOT pn.pnsequences\n"
"ORDER BY 1", pubid);
if (!addFooterToPublicationDesc(&buf, "Tables from schemas:",
true, &cont))
@@ -6052,6 +6163,37 @@ describePublications(const char *pattern)
}
}
+ if (!puballsequences)
+ {
+ /* Get the tables for the specified publication */
+ printfPQExpBuffer(&buf,
+ "SELECT n.nspname, c.relname, NULL\n"
+ "FROM pg_catalog.pg_class c,\n"
+ " pg_catalog.pg_namespace n,\n"
+ " pg_catalog.pg_publication_rel pr\n"
+ "WHERE c.relnamespace = n.oid\n"
+ " AND c.oid = pr.prrelid\n"
+ " AND pr.prpubid = '%s'\n"
+ " AND c.relkind = 'S'\n" /* only sequences */
+ "ORDER BY 1,2", pubid);
+ if (!addFooterToPublicationDesc(&buf, "Sequences:", false, &cont))
+ goto error_return;
+
+ if (pset.sversion >= 150000)
+ {
+ /* Get the schemas for the specified publication */
+ printfPQExpBuffer(&buf,
+ "SELECT n.nspname\n"
+ "FROM pg_catalog.pg_namespace n\n"
+ " JOIN pg_catalog.pg_publication_namespace pn ON n.oid = pn.pnnspid\n"
+ "WHERE pn.pnpubid = '%s' AND pn.pnsequences\n"
+ "ORDER BY 1", pubid);
+ if (!addFooterToPublicationDesc(&buf, "Sequences from schemas:",
+ true, &cont))
+ goto error_return;
+ }
+ }
+
printTable(&cont, pset.queryFout, false, pset.logfile);
printTableCleanup(&cont);
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 6957567264a..a8cdfc4cfba 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -1782,11 +1782,15 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH("ADD", "DROP", "OWNER TO", "RENAME TO", "SET");
/* ALTER PUBLICATION <name> ADD */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD"))
- COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") ||
(HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "TABLE") &&
ends_with(prev_wd, ',')))
COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_tables);
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "SEQUENCE") ||
+ (HeadMatches("ALTER", "PUBLICATION", MatchAny, "ADD|SET", "SEQUENCE") &&
+ ends_with(prev_wd, ',')))
+ COMPLETE_WITH_SCHEMA_QUERY(Query_for_list_of_sequences);
/*
* "ALTER PUBLICATION <name> SET TABLE <name> WHERE (" - complete with
* table attributes
@@ -1805,11 +1809,11 @@ psql_completion(const char *text, int start, int end)
COMPLETE_WITH(",");
/* ALTER PUBLICATION <name> DROP */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "DROP"))
- COMPLETE_WITH("ALL TABLES IN SCHEMA", "TABLE");
+ COMPLETE_WITH("ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
/* ALTER PUBLICATION <name> SET */
else if (Matches("ALTER", "PUBLICATION", MatchAny, "SET"))
- COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "TABLE");
- else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES", "IN", "SCHEMA"))
+ COMPLETE_WITH("(", "ALL TABLES IN SCHEMA", "ALL SEQUENCES IN SCHEMA", "TABLE", "SEQUENCE");
+ else if (Matches("ALTER", "PUBLICATION", MatchAny, "ADD|DROP|SET", "ALL", "TABLES|SEQUENCES", "IN", "SCHEMA"))
COMPLETE_WITH_QUERY_PLUS(Query_for_list_of_schemas
" AND nspname NOT LIKE E'pg\\\\_%'",
"CURRENT_SCHEMA");
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 7f1ee97f55c..29b84af04b6 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11546,6 +11546,11 @@
provolatile => 's', prorettype => 'oid', proargtypes => 'text',
proallargtypes => '{text,oid}', proargmodes => '{i,o}',
proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_tables' },
+{ oid => '8000', descr => 'get OIDs of sequences in a publication',
+ proname => 'pg_get_publication_sequences', prorows => '1000', proretset => 't',
+ provolatile => 's', prorettype => 'oid', proargtypes => 'text',
+ proallargtypes => '{text,oid}', proargmodes => '{i,o}',
+ proargnames => '{pubname,relid}', prosrc => 'pg_get_publication_sequences' },
{ oid => '6121',
descr => 'returns whether a relation can be part of a publication',
proname => 'pg_relation_is_publishable', provolatile => 's',
diff --git a/src/include/catalog/pg_publication.h b/src/include/catalog/pg_publication.h
index ba72e62e614..f2b4e838c77 100644
--- a/src/include/catalog/pg_publication.h
+++ b/src/include/catalog/pg_publication.h
@@ -40,6 +40,12 @@ CATALOG(pg_publication,6104,PublicationRelationId)
*/
bool puballtables;
+ /*
+ * indicates that this is special publication which should encompass all
+ * sequences in the database (except for the unlogged and temp ones)
+ */
+ bool puballsequences;
+
/* true if inserts are published */
bool pubinsert;
@@ -52,6 +58,9 @@ CATALOG(pg_publication,6104,PublicationRelationId)
/* true if truncates are published */
bool pubtruncate;
+ /* true if sequences are published */
+ bool pubsequence;
+
/* true if partition changes are published using root schema */
bool pubviaroot;
} FormData_pg_publication;
@@ -72,6 +81,7 @@ typedef struct PublicationActions
bool pubupdate;
bool pubdelete;
bool pubtruncate;
+ bool pubsequence;
} PublicationActions;
typedef struct PublicationDesc
@@ -92,6 +102,7 @@ typedef struct Publication
Oid oid;
char *name;
bool alltables;
+ bool allsequences;
bool pubviaroot;
PublicationActions pubactions;
} Publication;
@@ -122,26 +133,30 @@ typedef enum PublicationPartOpt
PUBLICATION_PART_ALL,
} PublicationPartOpt;
-extern List *GetPublicationRelations(Oid pubid, PublicationPartOpt pub_partopt);
+extern List *GetPublicationRelations(Oid pubid, bool sequences,
+ PublicationPartOpt pub_partopt);
extern List *GetAllTablesPublications(void);
extern List *GetAllTablesPublicationRelations(bool pubviaroot);
-extern List *GetPublicationSchemas(Oid pubid);
+extern List *GetPublicationSchemas(Oid pubid, bool sequence);
extern List *GetSchemaPublications(Oid schemaid);
-extern List *GetSchemaPublicationRelations(Oid schemaid,
+extern List *GetSchemaPublicationRelations(Oid schemaid, bool sequences,
PublicationPartOpt pub_partopt);
-extern List *GetAllSchemaPublicationRelations(Oid puboid,
+extern List *GetAllSchemaPublicationRelations(Oid puboid, bool sequences,
PublicationPartOpt pub_partopt);
extern List *GetPubPartitionOptionRelations(List *result,
PublicationPartOpt pub_partopt,
Oid relid);
extern Oid GetTopMostAncestorInPublication(Oid puboid, List *ancestors);
+extern List *GetAllSequencesPublications(void);
+extern List *GetAllSequencesPublicationRelations(void);
+
extern bool is_publishable_relation(Relation rel);
extern bool is_schema_publication(Oid pubid);
extern ObjectAddress publication_add_relation(Oid pubid, PublicationRelInfo *pri,
bool if_not_exists);
extern ObjectAddress publication_add_schema(Oid pubid, Oid schemaid,
- bool if_not_exists);
+ bool sequences, bool if_not_exists);
extern Oid get_publication_oid(const char *pubname, bool missing_ok);
extern char *get_publication_name(Oid pubid, bool missing_ok);
diff --git a/src/include/catalog/pg_publication_namespace.h b/src/include/catalog/pg_publication_namespace.h
index e4306da02e7..bbc22dec9f8 100644
--- a/src/include/catalog/pg_publication_namespace.h
+++ b/src/include/catalog/pg_publication_namespace.h
@@ -32,6 +32,7 @@ CATALOG(pg_publication_namespace,8901,PublicationNamespaceRelationId)
Oid oid; /* oid */
Oid pnpubid BKI_LOOKUP(pg_publication); /* Oid of the publication */
Oid pnnspid BKI_LOOKUP(pg_namespace); /* Oid of the schema */
+ bool pnsequences; /* tables or sequences from the schema? */
} FormData_pg_publication_namespace;
/* ----------------
@@ -42,6 +43,6 @@ CATALOG(pg_publication_namespace,8901,PublicationNamespaceRelationId)
typedef FormData_pg_publication_namespace *Form_pg_publication_namespace;
DECLARE_UNIQUE_INDEX_PKEY(pg_publication_namespace_oid_index, 8902, PublicationNamespaceObjectIndexId, on pg_publication_namespace using btree(oid oid_ops));
-DECLARE_UNIQUE_INDEX(pg_publication_namespace_pnnspid_pnpubid_index, 8903, PublicationNamespacePnnspidPnpubidIndexId, on pg_publication_namespace using btree(pnnspid oid_ops, pnpubid oid_ops));
+DECLARE_UNIQUE_INDEX(pg_publication_namespace_pnnspid_pnpubid_pnseq_index, 8903, PublicationNamespacePnnspidPnpubidSeqIndexId, on pg_publication_namespace using btree(pnnspid oid_ops, pnpubid oid_ops, pnsequences bool_ops));
#endif /* PG_PUBLICATION_NAMESPACE_H */
diff --git a/src/include/commands/sequence.h b/src/include/commands/sequence.h
index 9fecc41954e..d8c255a7af5 100644
--- a/src/include/commands/sequence.h
+++ b/src/include/commands/sequence.h
@@ -60,6 +60,7 @@ extern ObjectAddress DefineSequence(ParseState *pstate, CreateSeqStmt *stmt);
extern ObjectAddress AlterSequence(ParseState *pstate, AlterSeqStmt *stmt);
extern void DeleteSequenceTuple(Oid relid);
extern void ResetSequence(Oid seq_relid);
+extern void ResetSequence2(Oid seq_relid, int64 last_value, int64 log_cnt, bool is_called);
extern void ResetSequenceCaches(void);
extern void seq_redo(XLogReaderState *rptr);
diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h
index 1617702d9d6..7adc3710408 100644
--- a/src/include/nodes/parsenodes.h
+++ b/src/include/nodes/parsenodes.h
@@ -3663,6 +3663,10 @@ typedef enum PublicationObjSpecType
PUBLICATIONOBJ_TABLES_IN_SCHEMA, /* All tables in schema */
PUBLICATIONOBJ_TABLES_IN_CUR_SCHEMA, /* All tables in first element of
* search_path */
+ PUBLICATIONOBJ_SEQUENCE, /* Sequence type */
+ PUBLICATIONOBJ_SEQUENCES_IN_SCHEMA, /* Sequences in schema type */
+ PUBLICATIONOBJ_SEQUENCES_IN_CUR_SCHEMA, /* Get the first element of
+ * search_path */
PUBLICATIONOBJ_CONTINUATION /* Continuation of previous type */
} PublicationObjSpecType;
@@ -3682,6 +3686,7 @@ typedef struct CreatePublicationStmt
List *options; /* List of DefElem nodes */
List *pubobjects; /* Optional list of publication objects */
bool for_all_tables; /* Special publication for all tables in db */
+ bool for_all_sequences; /* Special publication for all sequences in db */
} CreatePublicationStmt;
typedef enum AlterPublicationAction
@@ -3705,6 +3710,7 @@ typedef struct AlterPublicationStmt
*/
List *pubobjects; /* Optional list of publication objects */
bool for_all_tables; /* Special publication for all tables in db */
+ bool for_all_sequences; /* Special publication for all sequences in db */
AlterPublicationAction action; /* What action to perform with the given
* objects */
} AlterPublicationStmt;
diff --git a/src/include/replication/logicalproto.h b/src/include/replication/logicalproto.h
index 4d2c881644a..415757d8a2d 100644
--- a/src/include/replication/logicalproto.h
+++ b/src/include/replication/logicalproto.h
@@ -61,6 +61,7 @@ typedef enum LogicalRepMsgType
LOGICAL_REP_MSG_RELATION = 'R',
LOGICAL_REP_MSG_TYPE = 'Y',
LOGICAL_REP_MSG_MESSAGE = 'M',
+ LOGICAL_REP_MSG_SEQUENCE = 'X', /* FIXME change */
LOGICAL_REP_MSG_BEGIN_PREPARE = 'b',
LOGICAL_REP_MSG_PREPARE = 'P',
LOGICAL_REP_MSG_COMMIT_PREPARED = 'K',
@@ -118,6 +119,18 @@ typedef struct LogicalRepTyp
char *typname; /* name of the remote type */
} LogicalRepTyp;
+/* Sequence info */
+typedef struct LogicalRepSequence
+{
+ Oid remoteid; /* unique id of the remote sequence */
+ char *nspname; /* schema name of remote sequence */
+ char *seqname; /* name of the remote sequence */
+ bool transactional;
+ int64 last_value;
+ int64 log_cnt;
+ bool is_called;
+} LogicalRepSequence;
+
/* Transaction info */
typedef struct LogicalRepBeginData
{
@@ -230,6 +243,12 @@ extern List *logicalrep_read_truncate(StringInfo in,
bool *cascade, bool *restart_seqs);
extern void logicalrep_write_message(StringInfo out, TransactionId xid, XLogRecPtr lsn,
bool transactional, const char *prefix, Size sz, const char *message);
+extern void logicalrep_write_sequence(StringInfo out, Relation rel,
+ TransactionId xid, XLogRecPtr lsn,
+ bool transactional,
+ int64 last_value, int64 log_cnt,
+ bool is_called);
+extern void logicalrep_read_sequence(StringInfo in, LogicalRepSequence *seqdata);
extern void logicalrep_write_rel(StringInfo out, TransactionId xid,
Relation rel);
extern LogicalRepRelation *logicalrep_read_rel(StringInfo in);
diff --git a/src/include/replication/pgoutput.h b/src/include/replication/pgoutput.h
index eafedd610a5..f4e9f35d09d 100644
--- a/src/include/replication/pgoutput.h
+++ b/src/include/replication/pgoutput.h
@@ -29,6 +29,7 @@ typedef struct PGOutputData
bool streaming;
bool messages;
bool two_phase;
+ bool sequences;
} PGOutputData;
#endif /* PGOUTPUT_H */
diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index 45dddaf5562..6e87d6f5c36 100644
--- a/src/interfaces/libpq/fe-exec.c
+++ b/src/interfaces/libpq/fe-exec.c
@@ -3725,7 +3725,10 @@ char *
PQgetvalue(const PGresult *res, int tup_num, int field_num)
{
if (!check_tuple_field_number(res, tup_num, field_num))
+ {
+ printf("getvalue\n");
return NULL;
+ }
return res->tuples[tup_num][field_num].value;
}
@@ -3736,7 +3739,10 @@ int
PQgetlength(const PGresult *res, int tup_num, int field_num)
{
if (!check_tuple_field_number(res, tup_num, field_num))
+ {
+ printf("getlength\n");
return 0;
+ }
if (res->tuples[tup_num][field_num].len != NULL_LEN)
return res->tuples[tup_num][field_num].len;
else
@@ -3750,7 +3756,10 @@ int
PQgetisnull(const PGresult *res, int tup_num, int field_num)
{
if (!check_tuple_field_number(res, tup_num, field_num))
+ {
+ printf("getisnull\n");
return 1; /* pretend it is null */
+ }
if (res->tuples[tup_num][field_num].len == NULL_LEN)
return 1;
else
diff --git a/src/test/regress/expected/publication.out b/src/test/regress/expected/publication.out
index 3c382e520e4..bc8ba71925e 100644
--- a/src/test/regress/expected/publication.out
+++ b/src/test/regress/expected/publication.out
@@ -30,20 +30,20 @@ ERROR: conflicting or redundant options
LINE 1: ...ub_xxx WITH (publish_via_partition_root = 'true', publish_vi...
^
\dRp
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f
- testpub_default | regress_publication_user | f | f | t | f | f | f
+ List of publications
+ Name | Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ testpib_ins_trunct | regress_publication_user | f | f | t | f | f | f | f | f
+ testpub_default | regress_publication_user | f | f | f | t | f | f | f | f
(2 rows)
-ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete');
+ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete, sequence');
\dRp
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpib_ins_trunct | regress_publication_user | f | t | f | f | f | f
- testpub_default | regress_publication_user | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+--------------------+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ testpib_ins_trunct | regress_publication_user | f | f | t | f | f | f | f | f
+ testpub_default | regress_publication_user | f | f | t | t | t | f | t | f
(2 rows)
--- adding tables
@@ -87,10 +87,10 @@ RESET client_min_messages;
-- should be able to add schema to 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable ADD ALL TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"public.testpub_tbl1"
Tables from schemas:
@@ -99,20 +99,20 @@ Tables from schemas:
-- should be able to drop schema from 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable DROP ALL TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"public.testpub_tbl1"
-- should be able to set schema to 'FOR TABLE' publication
ALTER PUBLICATION testpub_fortable SET ALL TABLES IN SCHEMA pub_test;
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test"
@@ -134,10 +134,10 @@ ERROR: relation "testpub_nopk" is not part of the publication
-- should be able to set table to schema publication
ALTER PUBLICATION testpub_forschema SET TABLE pub_test.testpub_nopk;
\dRp+ testpub_forschema
- Publication testpub_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
@@ -159,10 +159,10 @@ Publications:
"testpub_foralltables"
\dRp+ testpub_foralltables
- Publication testpub_foralltables
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | t | t | t | f | f | f
+ Publication testpub_foralltables
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | t | f | t | t | f | f | f | f
(1 row)
DROP TABLE testpub_tbl2;
@@ -174,24 +174,520 @@ CREATE PUBLICATION testpub3 FOR TABLE testpub_tbl3;
CREATE PUBLICATION testpub4 FOR TABLE ONLY testpub_tbl3;
RESET client_min_messages;
\dRp+ testpub3
- Publication testpub3
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"public.testpub_tbl3"
"public.testpub_tbl3a"
\dRp+ testpub4
- Publication testpub4
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub4
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"public.testpub_tbl3"
DROP TABLE testpub_tbl3, testpub_tbl3a;
DROP PUBLICATION testpub3, testpub4;
+--- adding sequences
+CREATE SEQUENCE testpub_seq0;
+CREATE SEQUENCE pub_test.testpub_seq1;
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forallsequences FOR ALL SEQUENCES WITH (publish = 'sequence');
+RESET client_min_messages;
+ALTER PUBLICATION testpub_forallsequences SET (publish = 'insert, sequence');
+CREATE SEQUENCE testpub_seq2;
+-- fail - can't add to for all sequences publication
+ALTER PUBLICATION testpub_forallsequences ADD SEQUENCE testpub_seq2;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications.
+-- fail - can't drop from all sequences publication
+ALTER PUBLICATION testpub_forallsequences DROP SEQUENCE testpub_seq2;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications.
+-- fail - can't add to for all sequences publication
+ALTER PUBLICATION testpub_forallsequences SET SEQUENCE pub_test.testpub_seq1;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences cannot be added to or dropped from FOR ALL SEQUENCES publications.
+-- fail - can't add schema to 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences ADD ALL SEQUENCES IN SCHEMA pub_test;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications.
+-- fail - can't drop schema from 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences DROP ALL SEQUENCES IN SCHEMA pub_test;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications.
+-- fail - can't set schema to 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences SET ALL SEQUENCES IN SCHEMA pub_test;
+ERROR: publication "testpub_forallsequences" is defined as FOR ALL SEQUENCES
+DETAIL: Sequences from schema cannot be added to, dropped from, or set on FOR ALL SEQUENCES publications.
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forsequence FOR SEQUENCE testpub_seq0;
+RESET client_min_messages;
+-- should be able to add schema to 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence ADD ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+ Publication testpub_forsequence
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Sequences:
+ "public.testpub_seq0"
+Sequences from schemas:
+ "pub_test"
+
+-- should be able to drop schema from 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence DROP ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+ Publication testpub_forsequence
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Sequences:
+ "public.testpub_seq0"
+
+-- should be able to set schema to 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence SET ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+ Publication testpub_forsequence
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Sequences from schemas:
+ "pub_test"
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forschema FOR ALL SEQUENCES IN SCHEMA pub_test;
+RESET client_min_messages;
+-- fail - can't create publication with schema and table of the same schema
+CREATE PUBLICATION testpub_for_seq_schema FOR ALL SEQUENCES IN SCHEMA pub_test, SEQUENCE pub_test.testpub_seq1;
+ERROR: cannot add relation "pub_test.testpub_seq1" to publication
+DETAIL: Sequence's schema "pub_test" is already part of the publication or part of the specified schema list.
+-- fail - can't add a sequence of the same schema to the schema publication
+ALTER PUBLICATION testpub_forschema ADD SEQUENCE pub_test.testpub_seq1;
+ERROR: cannot add relation "pub_test.testpub_seq1" to publication
+DETAIL: Sequence's schema "pub_test" is already part of the publication or part of the specified schema list.
+-- fail - can't drop a sequence from the schema publication which isn't in the
+-- publication
+ALTER PUBLICATION testpub_forschema DROP SEQUENCE pub_test.testpub_seq1;
+ERROR: relation "testpub_seq1" is not part of the publication
+-- should be able to set sequence to schema publication
+ALTER PUBLICATION testpub_forschema SET SEQUENCE pub_test.testpub_seq1;
+\dRp+ testpub_forschema
+ Publication testpub_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Sequences:
+ "pub_test.testpub_seq1"
+
+SELECT pubname, puballtables, puballsequences FROM pg_publication WHERE pubname = 'testpub_forallsequences';
+ pubname | puballtables | puballsequences
+-------------------------+--------------+-----------------
+ testpub_forallsequences | f | t
+(1 row)
+
+\d+ pub_test.testpub_seq1
+ Sequence "pub_test.testpub_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_forallsequences"
+ "testpub_forschema"
+ "testpub_forsequence"
+
+\dRp+ testpub_forallsequences
+ Publication testpub_forallsequences
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | t | t | f | f | t | f | f
+(1 row)
+
+DROP SEQUENCE testpub_seq0, pub_test.testpub_seq1, testpub_seq2;
+DROP PUBLICATION testpub_forallsequences, testpub_forsequence, testpub_forschema;
+-- Publication mixing tables and sequences
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_mix;
+RESET client_min_messages;
+CREATE SEQUENCE testpub_seq1;
+CREATE SEQUENCE pub_test.testpub_seq2;
+ALTER PUBLICATION testpub_mix ADD SEQUENCE testpub_seq1, TABLE testpub_tbl1;
+\dRp+ testpub_mix
+ Publication testpub_mix
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables:
+ "public.testpub_tbl1"
+Sequences:
+ "public.testpub_seq1"
+
+ALTER PUBLICATION testpub_mix ADD ALL SEQUENCES IN SCHEMA pub_test, ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+ Publication testpub_mix
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables:
+ "public.testpub_tbl1"
+Tables from schemas:
+ "pub_test"
+Sequences:
+ "public.testpub_seq1"
+Sequences from schemas:
+ "pub_test"
+
+ALTER PUBLICATION testpub_mix DROP ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+ Publication testpub_mix
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables:
+ "public.testpub_tbl1"
+Tables from schemas:
+ "pub_test"
+Sequences:
+ "public.testpub_seq1"
+
+ALTER PUBLICATION testpub_mix DROP ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+ Publication testpub_mix
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables:
+ "public.testpub_tbl1"
+Sequences:
+ "public.testpub_seq1"
+
+DROP PUBLICATION testpub_mix;
+DROP SEQUENCE testpub_seq1;
+DROP SEQUENCE pub_test.testpub_seq2;
+-- make sure we replicate only the correct relation type
+CREATE SCHEMA pub_test1;
+CREATE SEQUENCE pub_test1.test_seq1;
+CREATE TABLE pub_test1.test_tbl1 (a int primary key, b int);
+CREATE SCHEMA pub_test2;
+CREATE SEQUENCE pub_test2.test_seq2;
+CREATE TABLE pub_test2.test_tbl2 (a int primary key, b int);
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_schemas;
+RESET client_min_messages;
+-- add tables from one schema, sequences from the other
+ALTER PUBLICATION testpub_schemas ADD ALL TABLES IN SCHEMA pub_test2;
+ALTER PUBLICATION testpub_schemas ADD ALL SEQUENCES IN SCHEMA pub_test1;
+\dRp+ testpub_schemas
+ Publication testpub_schemas
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables from schemas:
+ "pub_test2"
+Sequences from schemas:
+ "pub_test1"
+
+\dn+ pub_test1
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test1 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (sequences)
+
+\dn+ pub_test2
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test2 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (tables)
+
+\d+ pub_test1.test_seq1;
+ Sequence "pub_test1.test_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test1.test_tbl1;
+ Table "pub_test1.test_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl1_pkey" PRIMARY KEY, btree (a)
+
+\d+ pub_test2.test_seq2;
+ Sequence "pub_test2.test_seq2"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+
+\d+ pub_test2.test_tbl2;
+ Table "pub_test2.test_tbl2"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl2_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+-- add the other object type from each schema
+ALTER PUBLICATION testpub_schemas ADD ALL TABLES IN SCHEMA pub_test1;
+ALTER PUBLICATION testpub_schemas ADD ALL SEQUENCES IN SCHEMA pub_test2;
+\dRp+ testpub_schemas
+ Publication testpub_schemas
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables from schemas:
+ "pub_test1"
+ "pub_test2"
+Sequences from schemas:
+ "pub_test1"
+ "pub_test2"
+
+\dn+ pub_test1
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test1 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (sequences)
+ "testpub_schemas" (tables)
+
+\dn+ pub_test2
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test2 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (tables)
+ "testpub_schemas" (sequences)
+
+\d+ pub_test1.test_seq1;
+ Sequence "pub_test1.test_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test1.test_tbl1;
+ Table "pub_test1.test_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl1_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_seq2;
+ Sequence "pub_test2.test_seq2"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_tbl2;
+ Table "pub_test2.test_tbl2"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl2_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+-- now drop the object type added first
+ALTER PUBLICATION testpub_schemas DROP ALL TABLES IN SCHEMA pub_test2;
+ALTER PUBLICATION testpub_schemas DROP ALL SEQUENCES IN SCHEMA pub_test1;
+\dRp+ testpub_schemas
+ Publication testpub_schemas
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables from schemas:
+ "pub_test1"
+Sequences from schemas:
+ "pub_test2"
+
+\dn+ pub_test1
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test1 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (tables)
+
+\dn+ pub_test2
+ List of schemas
+ Name | Owner | Access privileges | Description
+-----------+--------------------------+-------------------+-------------
+ pub_test2 | regress_publication_user | |
+Publications:
+ "testpub_schemas" (sequences)
+
+\d+ pub_test1.test_seq1;
+ Sequence "pub_test1.test_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+
+\d+ pub_test1.test_tbl1;
+ Table "pub_test1.test_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl1_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_seq2;
+ Sequence "pub_test2.test_seq2"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_tbl2;
+ Table "pub_test2.test_tbl2"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl2_pkey" PRIMARY KEY, btree (a)
+
+-- should fail (publication contains the whole schema)
+ALTER PUBLICATION testpub_schemas ADD TABLE pub_test1.test_tbl1;
+ERROR: cannot add relation "pub_test1.test_tbl1" to publication
+DETAIL: Table's schema "pub_test1" is already part of the publication or part of the specified schema list.
+ALTER PUBLICATION testpub_schemas ADD SEQUENCE pub_test2.test_seq2;
+ERROR: cannot add relation "pub_test2.test_seq2" to publication
+DETAIL: Sequence's schema "pub_test2" is already part of the publication or part of the specified schema list.
+-- should work (different schema)
+ALTER PUBLICATION testpub_schemas ADD TABLE pub_test2.test_tbl2;
+ALTER PUBLICATION testpub_schemas ADD SEQUENCE pub_test1.test_seq1;
+\dRp+ testpub_schemas
+ Publication testpub_schemas
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables:
+ "pub_test2.test_tbl2"
+Tables from schemas:
+ "pub_test1"
+Sequences:
+ "pub_test1.test_seq1"
+Sequences from schemas:
+ "pub_test2"
+
+\d+ pub_test1.test_seq1;
+ Sequence "pub_test1.test_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test1.test_tbl1;
+ Table "pub_test1.test_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl1_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_seq2;
+ Sequence "pub_test2.test_seq2"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_tbl2;
+ Table "pub_test2.test_tbl2"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl2_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+-- now drop the explicitly added objects again
+ALTER PUBLICATION testpub_schemas DROP TABLE pub_test2.test_tbl2;
+ALTER PUBLICATION testpub_schemas DROP SEQUENCE pub_test1.test_seq1;
+\dRp+ testpub_schemas
+ Publication testpub_schemas
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
+Tables from schemas:
+ "pub_test1"
+Sequences from schemas:
+ "pub_test2"
+
+\d+ pub_test1.test_seq1;
+ Sequence "pub_test1.test_seq1"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+
+\d+ pub_test1.test_tbl1;
+ Table "pub_test1.test_tbl1"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl1_pkey" PRIMARY KEY, btree (a)
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_seq2;
+ Sequence "pub_test2.test_seq2"
+ Type | Start | Minimum | Maximum | Increment | Cycles? | Cache
+--------+-------+---------+---------------------+-----------+---------+-------
+ bigint | 1 | 1 | 9223372036854775807 | 1 | no | 1
+Publications:
+ "testpub_schemas"
+
+\d+ pub_test2.test_tbl2;
+ Table "pub_test2.test_tbl2"
+ Column | Type | Collation | Nullable | Default | Storage | Stats target | Description
+--------+---------+-----------+----------+---------+---------+--------------+-------------
+ a | integer | | not null | | plain | |
+ b | integer | | | | plain | |
+Indexes:
+ "test_tbl2_pkey" PRIMARY KEY, btree (a)
+
+DROP PUBLICATION testpub_schemas;
+DROP TABLE pub_test1.test_tbl1, pub_test2.test_tbl2;
+DROP SEQUENCE pub_test1.test_seq1, pub_test2.test_seq2;
+DROP SCHEMA pub_test1, pub_test2;
-- Tests for partitioned tables
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forparted;
@@ -207,10 +703,10 @@ UPDATE testpub_parted1 SET a = 1;
-- only parent is listed as being in publication, not the partition
ALTER PUBLICATION testpub_forparted ADD TABLE testpub_parted;
\dRp+ testpub_forparted
- Publication testpub_forparted
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forparted
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"public.testpub_parted"
@@ -223,10 +719,10 @@ ALTER TABLE testpub_parted DETACH PARTITION testpub_parted1;
UPDATE testpub_parted1 SET a = 1;
ALTER PUBLICATION testpub_forparted SET (publish_via_partition_root = true);
\dRp+ testpub_forparted
- Publication testpub_forparted
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | t
+ Publication testpub_forparted
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | t
Tables:
"public.testpub_parted"
@@ -255,20 +751,20 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub5 FOR TABLE testpub_rf_tbl1, testpub_rf_tbl2 WHERE (c <> 'test' AND d < 5) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl2" WHERE ((c <> 'test'::text) AND (d < 5))
ALTER PUBLICATION testpub5 ADD TABLE testpub_rf_tbl3 WHERE (e > 1000 AND e < 2000);
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl2" WHERE ((c <> 'test'::text) AND (d < 5))
@@ -276,10 +772,10 @@ Tables:
ALTER PUBLICATION testpub5 DROP TABLE testpub_rf_tbl2;
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl3" WHERE ((e > 1000) AND (e < 2000))
@@ -287,10 +783,10 @@ Tables:
-- remove testpub_rf_tbl1 and add testpub_rf_tbl3 again (another WHERE expression)
ALTER PUBLICATION testpub5 SET TABLE testpub_rf_tbl3 WHERE (e > 300 AND e < 500);
\dRp+ testpub5
- Publication testpub5
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub5
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl3" WHERE ((e > 300) AND (e < 500))
@@ -315,10 +811,10 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax1 FOR TABLE testpub_rf_tbl1, ONLY testpub_rf_tbl3 WHERE (e < 999) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub_syntax1
- Publication testpub_syntax1
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub_syntax1
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"public.testpub_rf_tbl3" WHERE (e < 999)
@@ -328,10 +824,10 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_syntax2 FOR TABLE testpub_rf_tbl1, testpub_rf_schema1.testpub_rf_tbl5 WHERE (h < 999) WITH (publish = 'insert');
RESET client_min_messages;
\dRp+ testpub_syntax2
- Publication testpub_syntax2
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | f | f | f | f
+ Publication testpub_syntax2
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | f | f | f | f | f
Tables:
"public.testpub_rf_tbl1"
"testpub_rf_schema1.testpub_rf_tbl5" WHERE (h < 999)
@@ -636,10 +1132,10 @@ ERROR: relation "testpub_tbl1" is already member of publication "testpub_fortbl
CREATE PUBLICATION testpub_fortbl FOR TABLE testpub_tbl1;
ERROR: publication "testpub_fortbl" already exists
\dRp+ testpub_fortbl
- Publication testpub_fortbl
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortbl
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"pub_test.testpub_nopk"
"public.testpub_tbl1"
@@ -677,10 +1173,10 @@ Publications:
"testpub_fortbl"
\dRp+ testpub_default
- Publication testpub_default
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | f | f
+ Publication testpub_default
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f | f
Tables:
"pub_test.testpub_nopk"
"public.testpub_tbl1"
@@ -758,10 +1254,10 @@ REVOKE CREATE ON DATABASE regression FROM regress_publication_user2;
DROP TABLE testpub_parted;
DROP TABLE testpub_tbl1;
\dRp+ testpub_default
- Publication testpub_default
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | f | f
+ Publication testpub_default
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | f | f
(1 row)
-- fail - must be owner of publication
@@ -771,20 +1267,20 @@ ERROR: must be owner of publication testpub_default
RESET ROLE;
ALTER PUBLICATION testpub_default RENAME TO testpub_foo;
\dRp testpub_foo
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
--------------+--------------------------+------------+---------+---------+---------+-----------+----------
- testpub_foo | regress_publication_user | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+-------------+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ testpub_foo | regress_publication_user | f | f | t | t | t | f | t | f
(1 row)
-- rename back to keep the rest simple
ALTER PUBLICATION testpub_foo RENAME TO testpub_default;
ALTER PUBLICATION testpub_default OWNER TO regress_publication_user2;
\dRp testpub_default
- List of publications
- Name | Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
------------------+---------------------------+------------+---------+---------+---------+-----------+----------
- testpub_default | regress_publication_user2 | f | t | t | t | f | f
+ List of publications
+ Name | Owner | All tables | All sequences | Inserts | Updates | Deletes | Truncates | Sequences | Via root
+-----------------+---------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ testpub_default | regress_publication_user2 | f | f | t | t | t | f | t | f
(1 row)
-- adding schemas and tables
@@ -800,19 +1296,19 @@ CREATE TABLE "CURRENT_SCHEMA"."CURRENT_SCHEMA"(id int);
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub1_forschema FOR ALL TABLES IN SCHEMA pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
CREATE PUBLICATION testpub2_forschema FOR ALL TABLES IN SCHEMA pub_test1, pub_test2, pub_test3;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -826,44 +1322,44 @@ CREATE PUBLICATION testpub6_forschema FOR ALL TABLES IN SCHEMA "CURRENT_SCHEMA",
CREATE PUBLICATION testpub_fortable FOR TABLE "CURRENT_SCHEMA"."CURRENT_SCHEMA";
RESET client_min_messages;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"public"
\dRp+ testpub4_forschema
- Publication testpub4_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub4_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
\dRp+ testpub5_forschema
- Publication testpub5_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub5_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
"public"
\dRp+ testpub6_forschema
- Publication testpub6_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub6_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"CURRENT_SCHEMA"
"public"
\dRp+ testpub_fortable
- Publication testpub_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"CURRENT_SCHEMA.CURRENT_SCHEMA"
@@ -897,10 +1393,10 @@ ERROR: schema "testpub_view" does not exist
-- dropping the schema should reflect the change in publication
DROP SCHEMA pub_test3;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -908,20 +1404,20 @@ Tables from schemas:
-- renaming the schema should reflect the change in publication
ALTER SCHEMA pub_test1 RENAME to pub_test1_renamed;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1_renamed"
"pub_test2"
ALTER SCHEMA pub_test1_renamed RENAME to pub_test1;
\dRp+ testpub2_forschema
- Publication testpub2_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub2_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -929,10 +1425,10 @@ Tables from schemas:
-- alter publication add schema
ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -941,10 +1437,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -953,10 +1449,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema ADD ALL TABLES IN SCHEMA pub_test1;
ERROR: schema "pub_test1" is already member of publication "testpub1_forschema"
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -964,10 +1460,10 @@ Tables from schemas:
-- alter publication drop schema
ALTER PUBLICATION testpub1_forschema DROP ALL TABLES IN SCHEMA pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -975,10 +1471,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema DROP ALL TABLES IN SCHEMA pub_test2;
ERROR: tables from schema "pub_test2" are not part of the publication
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -986,29 +1482,29 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema DROP ALL TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
-- drop all schemas
ALTER PUBLICATION testpub1_forschema DROP ALL TABLES IN SCHEMA pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
(1 row)
-- alter publication set multiple schema
ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA pub_test1, pub_test2;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1017,10 +1513,10 @@ Tables from schemas:
ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA non_existent_schema;
ERROR: schema "non_existent_schema" does not exist
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
"pub_test2"
@@ -1029,10 +1525,10 @@ Tables from schemas:
-- removing the duplicate schemas
ALTER PUBLICATION testpub1_forschema SET ALL TABLES IN SCHEMA pub_test1, pub_test1;
\dRp+ testpub1_forschema
- Publication testpub1_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub1_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1102,18 +1598,18 @@ SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub3_forschema;
RESET client_min_messages;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
(1 row)
ALTER PUBLICATION testpub3_forschema SET ALL TABLES IN SCHEMA pub_test1;
\dRp+ testpub3_forschema
- Publication testpub3_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub3_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables from schemas:
"pub_test1"
@@ -1123,20 +1619,20 @@ CREATE PUBLICATION testpub_forschema_fortable FOR ALL TABLES IN SCHEMA pub_test1
CREATE PUBLICATION testpub_fortable_forschema FOR TABLE pub_test2.tbl1, ALL TABLES IN SCHEMA pub_test1;
RESET client_min_messages;
\dRp+ testpub_forschema_fortable
- Publication testpub_forschema_fortable
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_forschema_fortable
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"pub_test2.tbl1"
Tables from schemas:
"pub_test1"
\dRp+ testpub_fortable_forschema
- Publication testpub_fortable_forschema
- Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
---------------------------+------------+---------+---------+---------+-----------+----------
- regress_publication_user | f | t | t | t | t | f
+ Publication testpub_fortable_forschema
+ Owner | All tables | All sequences | Inserts | Updates | Deletes | Sequences | Truncates | Via root
+--------------------------+------------+---------------+---------+---------+---------+-----------+-----------+----------
+ regress_publication_user | f | f | t | t | t | t | t | f
Tables:
"pub_test2.tbl1"
Tables from schemas:
@@ -1180,40 +1676,85 @@ CREATE SCHEMA sch1;
CREATE SCHEMA sch2;
CREATE TABLE sch1.tbl1 (a int) PARTITION BY RANGE(a);
CREATE TABLE sch2.tbl1_part1 PARTITION OF sch1.tbl1 FOR VALUES FROM (1) to (10);
+CREATE SEQUENCE sch1.seq1;
+CREATE SEQUENCE sch2.seq2;
-- Schema publication that does not include the schema that has the parent table
CREATE PUBLICATION pub FOR ALL TABLES IN SCHEMA sch2 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch2;
SELECT * FROM pg_publication_tables;
pubname | schemaname | tablename
---------+------------+------------
pub | sch2 | tbl1_part1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
DROP PUBLICATION pub;
-- Table publication that does not include the parent table
CREATE PUBLICATION pub FOR TABLE sch2.tbl1_part1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
+ALTER PUBLICATION pub ADD SEQUENCE sch2.seq2;
SELECT * FROM pg_publication_tables;
pubname | schemaname | tablename
---------+------------+------------
pub | sch2 | tbl1_part1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
SELECT * FROM pg_publication_tables;
pubname | schemaname | tablename
---------+------------+-----------
pub | sch1 | tbl1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch1 | seq1
+ pub | sch2 | seq2
+(2 rows)
+
DROP PUBLICATION pub;
-- Schema publication that does not include the schema that has the parent table
CREATE PUBLICATION pub FOR ALL TABLES IN SCHEMA sch2 WITH (PUBLISH_VIA_PARTITION_ROOT=0);
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
SELECT * FROM pg_publication_tables;
pubname | schemaname | tablename
---------+------------+------------
pub | sch2 | tbl1_part1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch1 | seq1
+(1 row)
+
+DROP PUBLICATION pub;
+-- Sequence publication
+CREATE PUBLICATION pub FOR SEQUENCE sch2.seq2;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
DROP PUBLICATION pub;
-- Table publication that does not include the parent table
CREATE PUBLICATION pub FOR TABLE sch2.tbl1_part1 WITH (PUBLISH_VIA_PARTITION_ROOT=0);
@@ -1223,14 +1764,26 @@ SELECT * FROM pg_publication_tables;
pub | sch2 | tbl1_part1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+(0 rows)
+
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch2;
SELECT * FROM pg_publication_tables;
pubname | schemaname | tablename
---------+------------+------------
pub | sch2 | tbl1_part1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
DROP PUBLICATION pub;
DROP TABLE sch2.tbl1_part1;
DROP TABLE sch1.tbl1;
@@ -1246,9 +1799,81 @@ SELECT * FROM pg_publication_tables;
pub | sch1 | tbl1
(1 row)
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+(0 rows)
+
+DROP PUBLICATION pub;
+-- Schema publication
+CREATE PUBLICATION pub FOR SEQUENCE sch2.seq2;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
+DROP PUBLICATION pub;
+-- Sequence publication
+CREATE PUBLICATION pub FOR ALL SEQUENCES IN SCHEMA sch2;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch1 | seq1
+ pub | sch2 | seq2
+(2 rows)
+
+ALTER PUBLICATION pub DROP SEQUENCE sch1.seq1;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch2 | seq2
+(1 row)
+
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch1;
+SELECT * FROM pg_publication_tables;
+ pubname | schemaname | tablename
+---------+------------+-----------
+(0 rows)
+
+SELECT * FROM pg_publication_sequences;
+ pubname | schemaname | sequencename
+---------+------------+--------------
+ pub | sch1 | seq1
+ pub | sch2 | seq2
+(2 rows)
+
RESET client_min_messages;
DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
+DROP SEQUENCE sch1.seq1, sch2.seq2;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
RESET SESSION AUTHORIZATION;
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 1420288d67b..8b95e1ffaaa 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1429,6 +1429,14 @@ pg_prepared_xacts| SELECT p.transaction,
FROM ((pg_prepared_xact() p(transaction, gid, prepared, ownerid, dbid)
LEFT JOIN pg_authid u ON ((p.ownerid = u.oid)))
LEFT JOIN pg_database d ON ((p.dbid = d.oid)));
+pg_publication_sequences| SELECT p.pubname,
+ n.nspname AS schemaname,
+ c.relname AS sequencename
+ FROM pg_publication p,
+ LATERAL pg_get_publication_sequences((p.pubname)::text) gps(relid),
+ (pg_class c
+ JOIN pg_namespace n ON ((n.oid = c.relnamespace)))
+ WHERE (c.oid = gps.relid);
pg_publication_tables| SELECT p.pubname,
n.nspname AS schemaname,
c.relname AS tablename
diff --git a/src/test/regress/sql/object_address.sql b/src/test/regress/sql/object_address.sql
index 2f40156eb48..f90afad8045 100644
--- a/src/test/regress/sql/object_address.sql
+++ b/src/test/regress/sql/object_address.sql
@@ -143,7 +143,6 @@ SELECT pg_get_object_address('publication', '{one}', '{}');
SELECT pg_get_object_address('publication', '{one,two}', '{}');
SELECT pg_get_object_address('subscription', '{one}', '{}');
SELECT pg_get_object_address('subscription', '{one,two}', '{}');
-
-- test successful cases
WITH objects (type, name, args) AS (VALUES
('table', '{addr_nsp, gentable}'::text[], '{}'::text[]),
diff --git a/src/test/regress/sql/publication.sql b/src/test/regress/sql/publication.sql
index 3f04d34264a..5449bef8df9 100644
--- a/src/test/regress/sql/publication.sql
+++ b/src/test/regress/sql/publication.sql
@@ -27,7 +27,7 @@ CREATE PUBLICATION testpub_xxx WITH (publish_via_partition_root = 'true', publis
\dRp
-ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete');
+ALTER PUBLICATION testpub_default SET (publish = 'insert, update, delete, sequence');
\dRp
@@ -104,6 +104,179 @@ RESET client_min_messages;
DROP TABLE testpub_tbl3, testpub_tbl3a;
DROP PUBLICATION testpub3, testpub4;
+--- adding sequences
+CREATE SEQUENCE testpub_seq0;
+CREATE SEQUENCE pub_test.testpub_seq1;
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forallsequences FOR ALL SEQUENCES WITH (publish = 'sequence');
+RESET client_min_messages;
+ALTER PUBLICATION testpub_forallsequences SET (publish = 'insert, sequence');
+
+CREATE SEQUENCE testpub_seq2;
+-- fail - can't add to for all sequences publication
+ALTER PUBLICATION testpub_forallsequences ADD SEQUENCE testpub_seq2;
+-- fail - can't drop from all sequences publication
+ALTER PUBLICATION testpub_forallsequences DROP SEQUENCE testpub_seq2;
+-- fail - can't add to for all sequences publication
+ALTER PUBLICATION testpub_forallsequences SET SEQUENCE pub_test.testpub_seq1;
+
+-- fail - can't add schema to 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences ADD ALL SEQUENCES IN SCHEMA pub_test;
+-- fail - can't drop schema from 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences DROP ALL SEQUENCES IN SCHEMA pub_test;
+-- fail - can't set schema to 'FOR ALL SEQUENCES' publication
+ALTER PUBLICATION testpub_forallsequences SET ALL SEQUENCES IN SCHEMA pub_test;
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forsequence FOR SEQUENCE testpub_seq0;
+RESET client_min_messages;
+-- should be able to add schema to 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence ADD ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+-- should be able to drop schema from 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence DROP ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+-- should be able to set schema to 'FOR SEQUENCE' publication
+ALTER PUBLICATION testpub_forsequence SET ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_forsequence
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_forschema FOR ALL SEQUENCES IN SCHEMA pub_test;
+RESET client_min_messages;
+-- fail - can't create publication with schema and table of the same schema
+CREATE PUBLICATION testpub_for_seq_schema FOR ALL SEQUENCES IN SCHEMA pub_test, SEQUENCE pub_test.testpub_seq1;
+-- fail - can't add a sequence of the same schema to the schema publication
+ALTER PUBLICATION testpub_forschema ADD SEQUENCE pub_test.testpub_seq1;
+-- fail - can't drop a sequence from the schema publication which isn't in the
+-- publication
+ALTER PUBLICATION testpub_forschema DROP SEQUENCE pub_test.testpub_seq1;
+-- should be able to set sequence to schema publication
+ALTER PUBLICATION testpub_forschema SET SEQUENCE pub_test.testpub_seq1;
+\dRp+ testpub_forschema
+
+SELECT pubname, puballtables, puballsequences FROM pg_publication WHERE pubname = 'testpub_forallsequences';
+\d+ pub_test.testpub_seq1
+\dRp+ testpub_forallsequences
+DROP SEQUENCE testpub_seq0, pub_test.testpub_seq1, testpub_seq2;
+DROP PUBLICATION testpub_forallsequences, testpub_forsequence, testpub_forschema;
+
+-- Publication mixing tables and sequences
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_mix;
+RESET client_min_messages;
+
+CREATE SEQUENCE testpub_seq1;
+CREATE SEQUENCE pub_test.testpub_seq2;
+
+ALTER PUBLICATION testpub_mix ADD SEQUENCE testpub_seq1, TABLE testpub_tbl1;
+\dRp+ testpub_mix
+
+ALTER PUBLICATION testpub_mix ADD ALL SEQUENCES IN SCHEMA pub_test, ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+
+ALTER PUBLICATION testpub_mix DROP ALL SEQUENCES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+
+ALTER PUBLICATION testpub_mix DROP ALL TABLES IN SCHEMA pub_test;
+\dRp+ testpub_mix
+
+DROP PUBLICATION testpub_mix;
+DROP SEQUENCE testpub_seq1;
+DROP SEQUENCE pub_test.testpub_seq2;
+
+
+-- make sure we replicate only the correct relation type
+CREATE SCHEMA pub_test1;
+CREATE SEQUENCE pub_test1.test_seq1;
+CREATE TABLE pub_test1.test_tbl1 (a int primary key, b int);
+
+CREATE SCHEMA pub_test2;
+CREATE SEQUENCE pub_test2.test_seq2;
+CREATE TABLE pub_test2.test_tbl2 (a int primary key, b int);
+
+SET client_min_messages = 'ERROR';
+CREATE PUBLICATION testpub_schemas;
+RESET client_min_messages;
+
+-- add tables from one schema, sequences from the other
+ALTER PUBLICATION testpub_schemas ADD ALL TABLES IN SCHEMA pub_test2;
+ALTER PUBLICATION testpub_schemas ADD ALL SEQUENCES IN SCHEMA pub_test1;
+
+\dRp+ testpub_schemas
+
+\dn+ pub_test1
+\dn+ pub_test2
+
+\d+ pub_test1.test_seq1;
+\d+ pub_test1.test_tbl1;
+
+\d+ pub_test2.test_seq2;
+\d+ pub_test2.test_tbl2;
+
+-- add the other object type from each schema
+ALTER PUBLICATION testpub_schemas ADD ALL TABLES IN SCHEMA pub_test1;
+ALTER PUBLICATION testpub_schemas ADD ALL SEQUENCES IN SCHEMA pub_test2;
+
+\dRp+ testpub_schemas
+
+\dn+ pub_test1
+\dn+ pub_test2
+
+\d+ pub_test1.test_seq1;
+\d+ pub_test1.test_tbl1;
+
+\d+ pub_test2.test_seq2;
+\d+ pub_test2.test_tbl2;
+
+-- now drop the object type added first
+ALTER PUBLICATION testpub_schemas DROP ALL TABLES IN SCHEMA pub_test2;
+ALTER PUBLICATION testpub_schemas DROP ALL SEQUENCES IN SCHEMA pub_test1;
+
+\dRp+ testpub_schemas
+
+\dn+ pub_test1
+\dn+ pub_test2
+
+\d+ pub_test1.test_seq1;
+\d+ pub_test1.test_tbl1;
+
+\d+ pub_test2.test_seq2;
+\d+ pub_test2.test_tbl2;
+
+-- should fail (publication contains the whole schema)
+ALTER PUBLICATION testpub_schemas ADD TABLE pub_test1.test_tbl1;
+ALTER PUBLICATION testpub_schemas ADD SEQUENCE pub_test2.test_seq2;
+
+-- should work (different schema)
+ALTER PUBLICATION testpub_schemas ADD TABLE pub_test2.test_tbl2;
+ALTER PUBLICATION testpub_schemas ADD SEQUENCE pub_test1.test_seq1;
+
+\dRp+ testpub_schemas
+
+\d+ pub_test1.test_seq1;
+\d+ pub_test1.test_tbl1;
+
+\d+ pub_test2.test_seq2;
+\d+ pub_test2.test_tbl2;
+
+-- now drop the explicitly added objects again
+ALTER PUBLICATION testpub_schemas DROP TABLE pub_test2.test_tbl2;
+ALTER PUBLICATION testpub_schemas DROP SEQUENCE pub_test1.test_seq1;
+
+\dRp+ testpub_schemas
+
+\d+ pub_test1.test_seq1;
+\d+ pub_test1.test_tbl1;
+
+\d+ pub_test2.test_seq2;
+\d+ pub_test2.test_tbl2;
+
+DROP PUBLICATION testpub_schemas;
+DROP TABLE pub_test1.test_tbl1, pub_test2.test_tbl2;
+DROP SEQUENCE pub_test1.test_seq1, pub_test2.test_seq2;
+DROP SCHEMA pub_test1, pub_test2;
+
-- Tests for partitioned tables
SET client_min_messages = 'ERROR';
CREATE PUBLICATION testpub_forparted;
@@ -714,32 +887,51 @@ CREATE SCHEMA sch1;
CREATE SCHEMA sch2;
CREATE TABLE sch1.tbl1 (a int) PARTITION BY RANGE(a);
CREATE TABLE sch2.tbl1_part1 PARTITION OF sch1.tbl1 FOR VALUES FROM (1) to (10);
+CREATE SEQUENCE sch1.seq1;
+CREATE SEQUENCE sch2.seq2;
-- Schema publication that does not include the schema that has the parent table
CREATE PUBLICATION pub FOR ALL TABLES IN SCHEMA sch2 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch2;
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
DROP PUBLICATION pub;
-- Table publication that does not include the parent table
CREATE PUBLICATION pub FOR TABLE sch2.tbl1_part1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
+ALTER PUBLICATION pub ADD SEQUENCE sch2.seq2;
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
DROP PUBLICATION pub;
-- Schema publication that does not include the schema that has the parent table
CREATE PUBLICATION pub FOR ALL TABLES IN SCHEMA sch2 WITH (PUBLISH_VIA_PARTITION_ROOT=0);
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+DROP PUBLICATION pub;
+-- Sequence publication
+CREATE PUBLICATION pub FOR SEQUENCE sch2.seq2;
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
DROP PUBLICATION pub;
-- Table publication that does not include the parent table
CREATE PUBLICATION pub FOR TABLE sch2.tbl1_part1 WITH (PUBLISH_VIA_PARTITION_ROOT=0);
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
-- Table publication that includes both the parent table and the child table
ALTER PUBLICATION pub ADD TABLE sch1.tbl1;
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch2;
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
DROP PUBLICATION pub;
DROP TABLE sch2.tbl1_part1;
@@ -752,10 +944,36 @@ CREATE TABLE sch1.tbl1_part3 (a int) PARTITION BY RANGE(a);
ALTER TABLE sch1.tbl1 ATTACH PARTITION sch1.tbl1_part3 FOR VALUES FROM (20) to (30);
CREATE PUBLICATION pub FOR ALL TABLES IN SCHEMA sch1 WITH (PUBLISH_VIA_PARTITION_ROOT=1);
SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+DROP PUBLICATION pub;
+-- Schema publication
+CREATE PUBLICATION pub FOR SEQUENCE sch2.seq2;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+DROP PUBLICATION pub;
+-- Sequence publication
+CREATE PUBLICATION pub FOR ALL SEQUENCES IN SCHEMA sch2;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+ALTER PUBLICATION pub ADD SEQUENCE sch1.seq1;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+ALTER PUBLICATION pub DROP SEQUENCE sch1.seq1;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
+
+ALTER PUBLICATION pub ADD ALL SEQUENCES IN SCHEMA sch1;
+SELECT * FROM pg_publication_tables;
+SELECT * FROM pg_publication_sequences;
RESET client_min_messages;
DROP PUBLICATION pub;
DROP TABLE sch1.tbl1;
+DROP SEQUENCE sch1.seq1, sch2.seq2;
DROP SCHEMA sch1 cascade;
DROP SCHEMA sch2 cascade;
diff --git a/src/test/subscription/t/028_sequences.pl b/src/test/subscription/t/028_sequences.pl
new file mode 100644
index 00000000000..38434409463
--- /dev/null
+++ b/src/test/subscription/t/028_sequences.pl
@@ -0,0 +1,203 @@
+
+# Copyright (c) 2021, PostgreSQL Global Development Group
+
+# This tests that sequences are replicated correctly by logical replication
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More tests => 6;
+
+# Initialize publisher node
+my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
+$node_publisher->init(allows_streaming => 'logical');
+$node_publisher->start;
+
+# Create subscriber node
+my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
+$node_subscriber->init(allows_streaming => 'logical');
+$node_subscriber->start;
+
+# Create some preexisting content on publisher
+my $ddl = qq(
+ CREATE TABLE seq_test (v BIGINT);
+ CREATE SEQUENCE s;
+);
+
+# Setup structure on the publisher
+$node_publisher->safe_psql('postgres', $ddl);
+
+# Create some the same structure on subscriber, and an extra sequence that
+# we'll create on the publisher later
+$ddl = qq(
+ CREATE TABLE seq_test (v BIGINT);
+ CREATE SEQUENCE s;
+ CREATE SEQUENCE s2;
+);
+
+$node_subscriber->safe_psql('postgres', $ddl);
+
+# Setup logical replication
+my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
+$node_publisher->safe_psql('postgres',
+ "CREATE PUBLICATION seq_pub");
+
+$node_publisher->safe_psql('postgres',
+ "ALTER PUBLICATION seq_pub ADD SEQUENCE s");
+
+$node_subscriber->safe_psql('postgres',
+ "CREATE SUBSCRIPTION seq_sub CONNECTION '$publisher_connstr' PUBLICATION seq_pub WITH (slot_name = seq_sub_slot)"
+);
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Wait for initial sync to finish as well
+my $synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Insert initial test data
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ -- generate a number of values using the sequence
+ INSERT INTO seq_test SELECT nextval('s') FROM generate_series(1,100);
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Check the data on subscriber
+my $result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s;
+));
+
+is( $result, '132|0|t',
+ 'check replicated sequence values on subscriber');
+
+
+# advance the sequence in a rolled-back transaction - the rollback
+# does not wait for the replication, so we could see any intermediate state
+# so do something else after the test, to ensure we wait for everything
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ BEGIN;
+ INSERT INTO seq_test SELECT nextval('s') FROM generate_series(1,100);
+ ROLLBACK;
+ INSERT INTO seq_test VALUES (-1);
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s;
+));
+
+is( $result, '231|0|t',
+ 'check replicated sequence values on subscriber');
+
+
+# create a new sequence and roll it back - should not be replicated, due to
+# the transactional behavior
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ BEGIN;
+ CREATE SEQUENCE s2;
+ ALTER PUBLICATION seq_pub ADD SEQUENCE s2;
+ INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100);
+ ROLLBACK;
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s2;
+));
+
+is( $result, '1|0|f',
+ 'check replicated sequence values on subscriber');
+
+
+# create a new sequence, advance it in a rolled-back transaction, but commit
+# the create - the advance should be replicated nevertheless
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ BEGIN;
+ CREATE SEQUENCE s2;
+ ALTER PUBLICATION seq_pub ADD SEQUENCE s2;
+ SAVEPOINT sp1;
+ INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100);
+ ROLLBACK TO sp1;
+ COMMIT;
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Wait for sync of the second sequence we just added to finish
+$synced_query =
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('s', 'r');";
+$node_subscriber->poll_query_until('postgres', $synced_query)
+ or die "Timed out while waiting for subscriber to synchronize data";
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s2;
+));
+
+is( $result, '132|0|t',
+ 'check replicated sequence values on subscriber');
+
+
+# advance the new sequence in a transaction, and roll it back - the rollback
+# does not wait for the replication, so we could see any intermediate state
+# so do something else after the test, to ensure we wait for everything
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ BEGIN;
+ INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100);
+ ROLLBACK;
+ INSERT INTO seq_test VALUES (-1);
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s2;
+));
+
+is( $result, '231|0|t',
+ 'check replicated sequence values on subscriber');
+
+
+# advance the sequence in a subtransaction - the subtransaction gets rolled
+# back, but commit the main one - the changes should still be replicated
+$node_publisher->safe_psql(
+ 'postgres', qq(
+ BEGIN;
+ SAVEPOINT s1;
+ INSERT INTO seq_test SELECT nextval('s2') FROM generate_series(1,100);
+ ROLLBACK TO s1;
+ COMMIT;
+));
+
+$node_publisher->wait_for_catchup('seq_sub');
+
+# Check the data on subscriber
+$result = $node_subscriber->safe_psql(
+ 'postgres', qq(
+ SELECT * FROM s2;
+));
+
+is( $result, '330|0|t',
+ 'check replicated sequence values on subscriber');
+
+
+$node_subscriber->stop('fast');
+$node_publisher->stop('fast');
--
2.34.1
view thread (47+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: logical decoding and replication of sequences
In-Reply-To: <[email protected]>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox