public inbox for [email protected]
help / color / mirror / Atom feedFrom: [email protected] <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Andres Freund <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Takamichi Osumi (Fujitsu) <[email protected]>
Cc: [email protected] <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Amit Langote <[email protected]>
Cc: Peter Eisentraut <[email protected]>
Cc: [email protected] <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Greg Nancarrow <[email protected]>
Subject: RE: Data is copied twice when specifying both child and parent table in publication
Date: Fri, 17 Mar 2023 06:28:00 +0000
Message-ID: <OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAA4eK1LCzvPLVHHXn+VcGmFHApKtbpaNF1UFed3qE8=GhPzPGw@mail.gmail.com>
References: <OS3PR01MB62751414E4DABF144F5E481C9E529@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PtTN1Udug3x1eZnDdh8Z6PL5VN=xz1b44VbB9T1z_PtcQ@mail.gmail.com>
<OS3PR01MB6275A9B8C65C381C6828DF9D9E549@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+PvaSDuZ7PhzZOm4c2u-sbN4DUsZr22qyYQ8dncUpqMpXg@mail.gmail.com>
<OS3PR01MB6275843B2BBE92870F7881C19E299@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAHut+Pt+1PNx6VsZ-xKzAU-18HmNXhjCC1TGakKX46Wg7YNT1Q@mail.gmail.com>
<OS3PR01MB6275FB5397C6A647F262A3A69E009@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CALDaNm3ZzNJ6Y1iX=PBYCcCU+EPm4=9LVXFin5gvQ9XkEzc8UQ@mail.gmail.com>
<OS3PR01MB6275C62FF1D1CDE4FCBF61339E079@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<[email protected]>
<OS3PR01MB62758DFA4FE5BFA7CA0815329ED89@OS3PR01MB6275.jpnprd01.prod.outlook.com>
<CAA4eK1LCzvPLVHHXn+VcGmFHApKtbpaNF1UFed3qE8=GhPzPGw@mail.gmail.com>
On Thu, Mar 16, 2023 at 20:25 PM Amit Kapila <[email protected]> wrote:
>
Thanks for your comments.
> + if (server_version >= 160000)
> + {
> + appendStringInfo(&cmd, "SELECT DISTINCT N.nspname, C.relname,\n"
> + " ( SELECT array_agg(a.attname ORDER BY a.attnum)\n"
> + " FROM pg_attribute a\n"
> + " WHERE a.attrelid = GPT.relid AND a.attnum > 0 AND\n"
> + " NOT a.attisdropped AND\n"
> + " (a.attnum = ANY(GPT.attrs) OR GPT.attrs IS NULL)\n"
> + " ) AS attnames\n"
> + " FROM pg_class C\n"
> + " JOIN pg_namespace N ON N.oid = C.relnamespace\n"
> + " JOIN ( SELECT (pg_get_publication_tables(VARIADIC
> array_agg(pubname::text))).*\n"
> + " FROM pg_publication\n"
> + " WHERE pubname IN ( %s )) as GPT\n"
> + " ON GPT.relid = C.oid\n",
> + pub_names.data);
>
> The function pg_get_publication_tables() has already handled dropped
> columns, so we don't need it here in this query. Also, the part to
> build attnames should be the same as it is in view
> pg_publication_tables.
Agree. Changed.
> Can we directly try to pass the list of
> pubnames to the function pg_get_publication_tables() instead of
> joining it with pg_publication?
Changed.
I think the aim of joining it with pg_publication before is to exclude
non-existing publications. Otherwise, we would get an error because of the call
to function GetPublicationByName (with 'missing_ok = false') in function
pg_get_publication_tables. So, I changed "missing_ok" to true. If anyone doesn't
like this change, I'll reconsider this in the next version.
> Can we keep the changes in the else part (fix when publisher < 16) the
> same as HEAD and move the proposed change to a separate patch?
> Basically, for the HEAD patch, let's just try to fix this when
> publisher >=16. I am slightly worried that as this is a corner case
> bug and we didn't see any user complaints for this, so introducing a
> complex fix for back branches may not be required or at least we can
> discuss that separately.
Split the patch as suggested.
Attach the new patch set.
Regards,
Wang Wei
Attachments:
[application/octet-stream] HEAD-v17-0001-Fix-data-replicated-twice-when-specifying-publis.patch (21.7K, ../OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com/2-HEAD-v17-0001-Fix-data-replicated-twice-when-specifying-publis.patch)
download | inline diff:
From da970df0aaef60e4b456ce43335a7a8dcc60393b Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Tue, 14 Jun 2022 13:44:09 +0800
Subject: [PATCH v17 1/3] Fix data replicated twice when specifying
publish_via_partition_root option.
If there are two publications - one of them publishing a parent table
(using publish_via_partition_root = true) and the other is publishing one of
the parent's child tables - then subscribing to both publications from one
subscription results in the same initial child data being copied twice. It
should only be copied once.
To fix this, the API function pg_get_publication_tables has been extended to
take a publication list. Now, when getting the table information, if the
publish_via_partition_root is true, the function can exclude a partition table
whose ancestor is also published by the same publication list.
---
src/backend/catalog/pg_publication.c | 209 ++++++++++++++-------
src/backend/commands/subscriptioncmds.c | 57 +++++-
src/include/catalog/pg_proc.dat | 12 +-
src/test/regress/expected/rules.out | 2 +-
src/test/subscription/t/013_partition.pl | 18 +-
src/test/subscription/t/028_row_filter.pl | 13 +-
src/test/subscription/t/031_column_list.pl | 5 +-
src/tools/pgindent/typedefs.list | 1 +
8 files changed, 211 insertions(+), 106 deletions(-)
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index a98fcad421..d76758d8aa 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -45,6 +45,14 @@
#include "utils/rel.h"
#include "utils/syscache.h"
+/* Records association between publication and published table */
+typedef struct
+{
+ Oid relid; /* OID of published table */
+ Oid pubid; /* OID of publication that publishes this
+ * table. */
+} published_rel;
+
static void publication_translate_columns(Relation targetrel, List *columns,
int *natts, AttrNumber **attrs);
@@ -175,39 +183,45 @@ pg_relation_is_publishable(PG_FUNCTION_ARGS)
* Filter out the partitions whose parent tables were also specified in
* the publication.
*/
-static List *
-filter_partitions(List *relids)
+static void
+filter_partitions(List *table_infos)
{
- List *result = NIL;
ListCell *lc;
- ListCell *lc2;
- foreach(lc, relids)
+ foreach(lc, table_infos)
{
- bool skip = false;
- List *ancestors = NIL;
- Oid relid = lfirst_oid(lc);
+ bool skip = false;
+ List *ancestors = NIL;
+ ListCell *lc2;
+ published_rel *table_info = (published_rel *) lfirst(lc);
- if (get_rel_relispartition(relid))
- ancestors = get_partition_ancestors(relid);
+ if (get_rel_relispartition(table_info->relid))
+ ancestors = get_partition_ancestors(table_info->relid);
foreach(lc2, ancestors)
{
Oid ancestor = lfirst_oid(lc2);
+ ListCell *lc3;
/* Check if the parent table exists in the published table list. */
- if (list_member_oid(relids, ancestor))
+ foreach(lc3, table_infos)
{
- skip = true;
- break;
+ Oid relid = ((published_rel *) lfirst(lc3))->relid;
+
+ if (relid == ancestor)
+ {
+ skip = true;
+ break;
+ }
}
+
+ if (skip)
+ break;
}
- if (!skip)
- result = lappend_oid(result, relid);
+ if (skip)
+ table_infos = foreach_delete_current(table_infos, lc);
}
-
- return result;
}
/*
@@ -1026,22 +1040,28 @@ GetPublicationByName(const char *pubname, bool missing_ok)
}
/*
- * Returns information of tables in a publication.
+ * Get information of the tables in the given publication array.
+ *
+ * Returns pubid, relid, column list, row filter for each table.
*/
Datum
pg_get_publication_tables(PG_FUNCTION_ARGS)
{
-#define NUM_PUBLICATION_TABLES_ELEM 3
+#define NUM_PUBLICATION_TABLES_ELEM 4
FuncCallContext *funcctx;
- char *pubname = text_to_cstring(PG_GETARG_TEXT_PP(0));
- Publication *publication;
- List *tables;
+ List *table_infos = NIL;
/* stuff done only on the first call of the function */
if (SRF_IS_FIRSTCALL())
{
TupleDesc tupdesc;
MemoryContext oldcontext;
+ ArrayType *arr;
+ Datum *elems;
+ int nelems,
+ i;
+ bool viaroot = false;
+ ListCell *lc;
/* create a function context for cross-call persistence */
funcctx = SRF_FIRSTCALL_INIT();
@@ -1049,68 +1069,112 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
/* 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.
+ * Deconstruct the parameter into elements where each element is a
+ * publication name.
*/
- if (publication->alltables)
- {
- tables = GetAllTablesPublicationRelations(publication->pubviaroot);
- }
- else
+ arr = PG_GETARG_ARRAYTYPE_P(0);
+ deconstruct_array(arr, TEXTOID, -1, false, TYPALIGN_INT,
+ &elems, NULL, &nelems);
+
+ /* Get Oids of tables from each publication. */
+ for (i = 0; i < nelems; i++)
{
- List *relids,
- *schemarelids;
-
- relids = GetPublicationRelations(publication->oid,
- publication->pubviaroot ?
- PUBLICATION_PART_ROOT :
- PUBLICATION_PART_LEAF);
- schemarelids = GetAllSchemaPublicationRelations(publication->oid,
- publication->pubviaroot ?
- PUBLICATION_PART_ROOT :
- PUBLICATION_PART_LEAF);
- tables = list_concat_unique_oid(relids, schemarelids);
+ Publication *pub_elem;
+ List *pub_elem_tables = NIL;
+
+ pub_elem = GetPublicationByName(TextDatumGetCString(elems[i]), true);
+
+ if (!pub_elem)
+ continue;
/*
- * If the publication publishes partition changes via their
- * respective root partitioned tables, we must exclude partitions
- * in favor of including the root partitioned tables. Otherwise,
- * the function could return both the child and parent tables
- * which could cause data of the child table to be
- * double-published on the subscriber side.
+ * Publications support partitioned tables. If
+ * publish_via_partition_root is false, all changes are replicated
+ * using leaf partition identity and schema, so we only need those.
+ * Otherwise, get the partitioned table itself.
*/
- if (publication->pubviaroot)
- tables = filter_partitions(tables);
+ if (pub_elem->alltables)
+ pub_elem_tables = GetAllTablesPublicationRelations(pub_elem->pubviaroot);
+ else
+ {
+ List *relids,
+ *schemarelids;
+
+ relids = GetPublicationRelations(pub_elem->oid,
+ pub_elem->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ schemarelids = GetAllSchemaPublicationRelations(pub_elem->oid,
+ pub_elem->pubviaroot ?
+ PUBLICATION_PART_ROOT :
+ PUBLICATION_PART_LEAF);
+ pub_elem_tables = list_concat_unique_oid(relids, schemarelids);
+ }
+
+ /*
+ * Record the published table and the corresponding publication so
+ * that we can get row filters and column list later.
+ *
+ * When a table is published by multiple publications, to obtain
+ * all row filters and column list, the structure related to this
+ * table will be recorded multiple times.
+ */
+ foreach(lc, pub_elem_tables)
+ {
+ published_rel *table_info = (published_rel *) malloc(sizeof(published_rel));
+
+ table_info->relid = lfirst_oid(lc);
+ table_info->pubid = pub_elem->oid;
+ table_infos = lappend(table_infos, table_info);
+ }
+
+ /* At least one publication is using publish_via_partition_root. */
+ if (pub_elem->pubviaroot)
+ viaroot = true;
}
+ pfree(elems);
+
+ /*
+ * If the publication publishes partition changes via their respective
+ * root partitioned tables, we must exclude partitions in favor of
+ * including the root partitioned tables. Otherwise, the function
+ * could return both the child and parent tables which could cause
+ * data of the child table to be double-published on the subscriber
+ * side.
+ */
+ if (viaroot)
+ filter_partitions(table_infos);
+
/* Construct a tuple descriptor for the result rows. */
tupdesc = CreateTemplateTupleDesc(NUM_PUBLICATION_TABLES_ELEM);
- TupleDescInitEntry(tupdesc, (AttrNumber) 1, "relid",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 1, "pubid",
+ OIDOID, -1, 0);
+ TupleDescInitEntry(tupdesc, (AttrNumber) 2, "relid",
OIDOID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 2, "attrs",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 3, "attrs",
INT2VECTOROID, -1, 0);
- TupleDescInitEntry(tupdesc, (AttrNumber) 3, "qual",
+ TupleDescInitEntry(tupdesc, (AttrNumber) 4, "qual",
PG_NODE_TREEOID, -1, 0);
funcctx->tuple_desc = BlessTupleDesc(tupdesc);
- funcctx->user_fctx = (void *) tables;
+ funcctx->user_fctx = (void *) table_infos;
MemoryContextSwitchTo(oldcontext);
}
/* stuff done on every call of the function */
funcctx = SRF_PERCALL_SETUP();
- tables = (List *) funcctx->user_fctx;
+ table_infos = (List *) funcctx->user_fctx;
- if (funcctx->call_cntr < list_length(tables))
+ if (funcctx->call_cntr < list_length(table_infos))
{
HeapTuple pubtuple = NULL;
HeapTuple rettuple;
- Oid relid = list_nth_oid(tables, funcctx->call_cntr);
+ Publication *pub;
+ published_rel *table_info = (published_rel *) list_nth(table_infos, funcctx->call_cntr);
+ Oid relid = table_info->relid;
Oid schemaid = get_rel_namespace(relid);
Datum values[NUM_PUBLICATION_TABLES_ELEM] = {0};
bool nulls[NUM_PUBLICATION_TABLES_ELEM] = {0};
@@ -1119,42 +1183,43 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
* Form tuple with appropriate data.
*/
- publication = GetPublicationByName(pubname, false);
+ pub = GetPublication(table_info->pubid);
- values[0] = ObjectIdGetDatum(relid);
+ values[0] = ObjectIdGetDatum(pub->oid);
+ values[1] = ObjectIdGetDatum(relid);
/*
* We don't consider row filters or column lists for FOR ALL TABLES or
* FOR TABLES IN SCHEMA publications.
*/
- if (!publication->alltables &&
+ if (!pub->alltables &&
!SearchSysCacheExists2(PUBLICATIONNAMESPACEMAP,
ObjectIdGetDatum(schemaid),
- ObjectIdGetDatum(publication->oid)))
+ ObjectIdGetDatum(pub->oid)))
pubtuple = SearchSysCacheCopy2(PUBLICATIONRELMAP,
ObjectIdGetDatum(relid),
- ObjectIdGetDatum(publication->oid));
+ ObjectIdGetDatum(pub->oid));
if (HeapTupleIsValid(pubtuple))
{
/* Lookup the column list attribute. */
- values[1] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
+ values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
Anum_pg_publication_rel_prattrs,
- &(nulls[1]));
+ &(nulls[2]));
/* Null indicates no filter. */
- values[2] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
+ values[3] = SysCacheGetAttr(PUBLICATIONRELMAP, pubtuple,
Anum_pg_publication_rel_prqual,
- &(nulls[2]));
+ &(nulls[3]));
}
else
{
- nulls[1] = true;
nulls[2] = true;
+ nulls[3] = true;
}
/* Show all columns when the column list is not specified. */
- if (nulls[1] == true)
+ if (nulls[2] == true)
{
Relation rel = table_open(relid, AccessShareLock);
int nattnums = 0;
@@ -1176,8 +1241,8 @@ pg_get_publication_tables(PG_FUNCTION_ARGS)
if (nattnums > 0)
{
- values[1] = PointerGetDatum(buildint2vector(attnums, nattnums));
- nulls[1] = false;
+ values[2] = PointerGetDatum(buildint2vector(attnums, nattnums));
+ nulls[2] = false;
}
table_close(rel, AccessShareLock);
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 464db6d247..0d87c75803 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1936,25 +1936,62 @@ static List *
fetch_table_list(WalReceiverConn *wrconn, List *publications)
{
WalRcvExecResult *res;
- StringInfoData cmd;
+ StringInfoData cmd,
+ pub_names;
TupleTableSlot *slot;
Oid tableRow[3] = {TEXTOID, TEXTOID, NAMEARRAYOID};
List *tablelist = NIL;
- bool check_columnlist = (walrcv_server_version(wrconn) >= 150000);
+ int server_version = walrcv_server_version(wrconn);
+ bool check_columnlist = (server_version >= 150000);
+
+ initStringInfo(&pub_names);
+ get_publications_str(publications, &pub_names, true);
initStringInfo(&cmd);
- appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename \n");
- /* Get column lists for each relation if the publisher supports it */
- if (check_columnlist)
- appendStringInfoString(&cmd, ", t.attnames\n");
+ /*
+ * Get namespace, relname and column list (if supported) of the tables
+ * belonging to the specified publications.
+ *
+ * Get the list of tables from the publisher. The partition table whose
+ * ancestor is also in this list will be ignored, otherwise the initial
+ * data in the partition table would be replicated twice.
+ *
+ * From version 16, the parameter of the function
+ * pg_get_publication_tables can be an array of publications. The
+ * partition table whose ancestor is also published in this publication
+ * array will be filtered out in this function.
+ */
+ if (server_version >= 160000)
+ {
+ appendStringInfo(&cmd, "SELECT DISTINCT N.nspname, C.relname,\n"
+ " ( SELECT array_agg(a.attname ORDER BY a.attnum)\n"
+ " FROM pg_attribute a\n"
+ " WHERE a.attrelid = GPT.relid AND\n"
+ " a.attnum = ANY(GPT.attrs)\n"
+ " ) AS attnames\n"
+ " FROM pg_class C\n"
+ " JOIN pg_namespace N ON N.oid = C.relnamespace,\n"
+ " LATERAL pg_get_publication_tables( %s ) GPT\n"
+ " WHERE GPT.relid = C.oid;\n",
+ pub_names.data);
+ }
+ else
+ {
+ appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename \n");
- appendStringInfoString(&cmd, "FROM pg_catalog.pg_publication_tables t\n"
- " WHERE t.pubname IN (");
- get_publications_str(publications, &cmd, true);
- appendStringInfoChar(&cmd, ')');
+ /* Get column lists for each relation if the publisher supports it */
+ if (check_columnlist)
+ appendStringInfoString(&cmd, ", t.attnames\n");
+
+ appendStringInfoString(&cmd, "FROM pg_catalog.pg_publication_tables t\n"
+ " WHERE t.pubname IN (");
+ get_publications_str(publications, &cmd, true);
+ appendStringInfoChar(&cmd, ')');
+ }
res = walrcv_exec(wrconn, cmd.data, check_columnlist ? 3 : 2, tableRow);
+ pfree(pub_names.data);
pfree(cmd.data);
if (res->status != WALRCV_OK_TUPLES)
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fbc4aade49..68e25c6506 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11725,11 +11725,13 @@
prosrc => 'pg_show_replication_origin_status' },
# publications
-{ oid => '6119', descr => 'get information of tables in a publication',
- proname => 'pg_get_publication_tables', prorows => '1000', proretset => 't',
- provolatile => 's', prorettype => 'record', proargtypes => 'text',
- proallargtypes => '{text,oid,int2vector,pg_node_tree}',
- proargmodes => '{i,o,o,o}', proargnames => '{pubname,relid,attrs,qual}',
+{ oid => '6119',
+ descr => 'get information of the tables in the given publication array',
+ proname => 'pg_get_publication_tables', prorows => '1000',
+ provariadic => 'text', proretset => 't', provolatile => 's',
+ prorettype => 'record', proargtypes => '_text',
+ proallargtypes => '{_text,oid,oid,int2vector,pg_node_tree}',
+ proargmodes => '{v,o,o,o,o}', proargnames => '{pubname,pubid,relid,attrs,qual}',
prosrc => 'pg_get_publication_tables' },
{ oid => '6121',
descr => 'returns whether a relation can be part of a publication',
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index e953d1f515..6d8f200c0b 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1449,7 +1449,7 @@ pg_publication_tables| SELECT p.pubname,
WHERE ((a.attrelid = gpt.relid) AND (a.attnum = ANY ((gpt.attrs)::smallint[])))) AS attnames,
pg_get_expr(gpt.qual, gpt.relid) AS rowfilter
FROM pg_publication p,
- LATERAL pg_get_publication_tables((p.pubname)::text) gpt(relid, attrs, qual),
+ LATERAL pg_get_publication_tables(VARIADIC ARRAY[(p.pubname)::text]) gpt(pubid, relid, attrs, qual),
(pg_class c
JOIN pg_namespace n ON ((n.oid = c.relnamespace)))
WHERE (c.oid = gpt.relid);
diff --git a/src/test/subscription/t/013_partition.pl b/src/test/subscription/t/013_partition.pl
index 11a5c3c03e..58e28d51d0 100644
--- a/src/test/subscription/t/013_partition.pl
+++ b/src/test/subscription/t/013_partition.pl
@@ -473,12 +473,12 @@ $node_subscriber2->safe_psql('postgres',
"CREATE TABLE tab3_1 (a int PRIMARY KEY, c text DEFAULT 'sub2_tab3_1', b text)"
);
-# Note: We create two separate tables, not a partitioned one, so that we can
-# easily identity through which relation were the changes replicated.
+# Note: We only create one table (tab4) here. We specified
+# publish_via_partition_root = true (see pub_all and pub_lower_level above), so
+# all data will be replicated to that table.
$node_subscriber2->safe_psql('postgres',
"CREATE TABLE tab4 (a int PRIMARY KEY)");
-$node_subscriber2->safe_psql('postgres',
- "CREATE TABLE tab4_1 (a int PRIMARY KEY)");
+
# Publication that sub2 points to now publishes via root, so must update
# subscription target relations. We set the list of publications so that
# the FOR ALL TABLES publication is second (the list order matters).
@@ -550,11 +550,6 @@ $result =
$node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4 ORDER BY 1");
is($result, qq(0), 'inserts into tab4 replicated');
-$result =
- $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4_1 ORDER BY 1");
-is($result, qq(), 'inserts into tab4_1 replicated');
-
-
# now switch the order of publications in the list, try again, the result
# should be the same (no dependence on order of pulications)
$node_subscriber2->safe_psql('postgres',
@@ -577,11 +572,6 @@ $result =
is( $result, qq(0
1), 'inserts into tab4 replicated');
-$result =
- $node_subscriber2->safe_psql('postgres', "SELECT a FROM tab4_1 ORDER BY 1");
-is($result, qq(), 'inserts into tab4_1 replicated');
-
-
# update (replicated as update)
$node_publisher->safe_psql('postgres', "UPDATE tab1 SET a = 6 WHERE a = 5");
$node_publisher->safe_psql('postgres', "UPDATE tab2 SET a = 6 WHERE a = 5");
diff --git a/src/test/subscription/t/028_row_filter.pl b/src/test/subscription/t/028_row_filter.pl
index def3fbc54b..d8972aeb93 100644
--- a/src/test/subscription/t/028_row_filter.pl
+++ b/src/test/subscription/t/028_row_filter.pl
@@ -386,6 +386,10 @@ $node_publisher->safe_psql('postgres',
"INSERT INTO tab_rowfilter_child(a, b) VALUES(0,'0'),(30,'30'),(40,'40')"
);
+# insert data into partitioned table.
+$node_publisher->safe_psql('postgres',
+ "INSERT INTO tab_rowfilter_viaroot_part(a) VALUES(13), (17)");
+
$node_subscriber->safe_psql('postgres',
"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub_1, tap_pub_2, tap_pub_3, tap_pub_4a, tap_pub_4b, tap_pub_5a, tap_pub_5b, tap_pub_toast, tap_pub_inherits, tap_pub_viaroot_2, tap_pub_viaroot_1"
);
@@ -707,13 +711,18 @@ is($result, qq(t|1), 'check replicated rows to tab_rowfilter_toast');
# the row filter for the top-level ancestor:
#
# tab_rowfilter_viaroot_part filter is: (a > 15)
+# - INSERT (13) NO, 13 < 15
# - INSERT (14) NO, 14 < 15
# - INSERT (15) NO, 15 = 15
# - INSERT (16) YES, 16 > 15
+# - INSERT (17) YES, 17 > 15
$result =
$node_subscriber->safe_psql('postgres',
- "SELECT a FROM tab_rowfilter_viaroot_part");
-is($result, qq(16), 'check replicated rows to tab_rowfilter_viaroot_part');
+ "SELECT a FROM tab_rowfilter_viaroot_part ORDER BY 1");
+is($result, qq(16
+17),
+ 'check replicated rows to tab_rowfilter_viaroot_part'
+);
# Check there is no data in tab_rowfilter_viaroot_part_1 because rows are
# replicated via the top most parent table tab_rowfilter_viaroot_part
diff --git a/src/test/subscription/t/031_column_list.pl b/src/test/subscription/t/031_column_list.pl
index fd77757352..1cdc679c56 100644
--- a/src/test/subscription/t/031_column_list.pl
+++ b/src/test/subscription/t/031_column_list.pl
@@ -959,7 +959,8 @@ $node_publisher->safe_psql(
CREATE TABLE test_root_1 PARTITION OF test_root FOR VALUES FROM (1) TO (10);
CREATE TABLE test_root_2 PARTITION OF test_root FOR VALUES FROM (10) TO (20);
- CREATE PUBLICATION pub_root_true FOR TABLE test_root (a) WITH (publish_via_partition_root = true);
+ CREATE PUBLICATION pub_root_true_1 FOR TABLE test_root (a) WITH (publish_via_partition_root = true);
+ CREATE PUBLICATION pub_root_true_2 FOR TABLE test_root_1 (a, b) WITH (publish_via_partition_root = true);
-- initial data
INSERT INTO test_root VALUES (1, 2, 3);
@@ -968,7 +969,7 @@ $node_publisher->safe_psql(
$node_subscriber->safe_psql(
'postgres', qq(
- CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub_root_true;
+ CREATE SUBSCRIPTION sub1 CONNECTION '$publisher_connstr' PUBLICATION pub_root_true_1, pub_root_true_2;
));
$node_subscriber->wait_for_subscription_sync;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 097f42e1b3..6d99372eeb 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -3601,6 +3601,7 @@ pthread_mutex_t
pthread_once_t
pthread_t
ptrdiff_t
+published_rel
pull_var_clause_context
pull_varattnos_context
pull_varnos_context
--
2.39.1.windows.1
[application/octet-stream] HEAD-v17-0002-Fix-this-problem-for-back-branches.patch (2.5K, ../OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com/3-HEAD-v17-0002-Fix-this-problem-for-back-branches.patch)
download | inline diff:
From a962ca61ecbdf196318e2311654e5ebbec1b4c6e Mon Sep 17 00:00:00 2001
From: Wang Wei <[email protected]>
Date: Fri, 17 Mar 2023 13:16:00 +0800
Subject: [PATCH v17 2/3] Fix this problem for back branches
---
src/backend/commands/subscriptioncmds.c | 31 ++++++++++++++++++++-----
1 file changed, 25 insertions(+), 6 deletions(-)
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 0d87c75803..788e463f53 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1978,16 +1978,35 @@ fetch_table_list(WalReceiverConn *wrconn, List *publications)
}
else
{
- appendStringInfoString(&cmd, "SELECT DISTINCT t.schemaname, t.tablename \n");
+ appendStringInfoString(&cmd, "WITH pub_tabs AS(\n"
+ " SELECT DISTINCT N.nspname, C.oid, C.relname, C.relispartition\n");
/* Get column lists for each relation if the publisher supports it */
if (check_columnlist)
- appendStringInfoString(&cmd, ", t.attnames\n");
+ appendStringInfoString(&cmd, ", ( SELECT array_agg(a.attname ORDER BY a.attnum)\n"
+ " FROM pg_attribute a\n"
+ " WHERE a.attrelid = GPT.relid AND a.attnum > 0 AND\n"
+ " NOT a.attisdropped AND\n"
+ " (a.attnum = ANY(GPT.attrs) OR GPT.attrs IS NULL)\n"
+ " ) AS attnames\n");
+
+ appendStringInfo(&cmd, " FROM pg_publication P,\n"
+ " LATERAL pg_get_publication_tables(P.pubname) GPT,\n"
+ " pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
+ " WHERE C.oid = GPT.relid AND P.pubname IN ( %s )\n"
+ ")\n"
+ "SELECT DISTINCT pub_tabs.nspname, pub_tabs.relname\n",
+ pub_names.data);
+
+ /* Get column lists for each relation if the publisher supports it */
+ if (check_columnlist)
+ appendStringInfoString(&cmd, ", pub_tabs.attnames\n");
- appendStringInfoString(&cmd, "FROM pg_catalog.pg_publication_tables t\n"
- " WHERE t.pubname IN (");
- get_publications_str(publications, &cmd, true);
- appendStringInfoChar(&cmd, ')');
+ appendStringInfoString(&cmd, "FROM pub_tabs\n"
+ " WHERE (pub_tabs.relispartition IS FALSE\n"
+ " OR NOT EXISTS (SELECT 1 FROM pg_partition_ancestors(pub_tabs.oid) as PA\n"
+ " WHERE PA.relid IN (SELECT pub_tabs.oid FROM pub_tabs)\n"
+ " AND PA.relid != pub_tabs.oid))\n");
}
res = walrcv_exec(wrconn, cmd.data, check_columnlist ? 3 : 2, tableRow);
--
2.39.1.windows.1
[application/octet-stream] HEAD-v17-0003-Add-clarification-for-the-behaviour-of-the-publi.patch (2.4K, ../OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com/4-HEAD-v17-0003-Add-clarification-for-the-behaviour-of-the-publi.patch)
download | inline diff:
From de244323822ab7ef6595a2196e70c22bdceca13b Mon Sep 17 00:00:00 2001
From: wangw <[email protected]>
Date: Wed, 28 Sep 2022 14:35:41 +0800
Subject: [PATCH v17 3/3] Add clarification for the behaviour of the
publication parameter publish_via_partition_root
Assume a subscription is subscribing to multiple publications, and these
publications publish a partitioned table and its partitions respectively:
[publisher-side]
create table parent (a int primary key) partition by range (a);
create table child partition of parent default;
create publication pub1 for table parent;
create publication pub2 for table child;
[subscriber-side]
create subscription sub connection 'xxxx' publication pub1, pub2;
The manual does not clearly describe the behaviour when the user had
specified the parameter 'publish_via_partition_root' on just one of
the publications. This patch modifies documentation to clarify the
following rules:
- If the parameter publish_via_partition_root is specified only in pub1,
changes will be published using the identity and schema of the table 'parent'.
- If the parameter publish_via_partition_root is specified only in pub2,
changes will be published using the identity and schema of the table 'child'.
---
doc/src/sgml/ref/create_publication.sgml | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/doc/src/sgml/ref/create_publication.sgml b/doc/src/sgml/ref/create_publication.sgml
index a2946feaa3..523c495a13 100644
--- a/doc/src/sgml/ref/create_publication.sgml
+++ b/doc/src/sgml/ref/create_publication.sgml
@@ -201,6 +201,16 @@ CREATE PUBLICATION <replaceable class="parameter">name</replaceable>
consisting of a different set of partitions.
</para>
+ <para>
+ There can be a case where a subscription combines multiple
+ publications. If a root partitioned table is published by any
+ subscribed publications which set
+ <literal>publish_via_partition_root</literal> = true, changes on this
+ root partitioned table (or on its partitions) will be published using
+ the identity and schema of this root partitioned table rather than
+ that of the individual partitions.
+ </para>
+
<para>
This parameter also affects how row filters and column lists are
chosen for partitions; see below for details.
--
2.39.1.windows.1
[application/octet-stream] REL14_v17-0001-Fix-data-replicated-twice-when-specifying-publis_patch (5.3K, ../OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com/5-REL14_v17-0001-Fix-data-replicated-twice-when-specifying-publis_patch)
download
[application/octet-stream] REL15_v17-0001-Fix-data-replicated-twice-when-specifying-publis_patch (8.6K, ../OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com/6-REL15_v17-0001-Fix-data-replicated-twice-when-specifying-publis_patch)
download
view thread (23+ 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], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Data is copied twice when specifying both child and parent table in publication
In-Reply-To: <OSZPR01MB62783EDCE9DDC21D35B789209EBD9@OSZPR01MB6278.jpnprd01.prod.outlook.com>
* 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