Movatterモバイル変換

[0]ホーム
URL:

Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly

 Sign up
Appearance settings

Commit8756930

Browse files
author
Amit Kapila
committed
Raise a warning if there is a possibility of data from multiple origins.
This commit raises a warning message for a combination of options('copy_data = true' and 'origin = none') during CREATE/ALTER subscriptionoperations if the publication tables were also replicated from otherpublishers.During replication, we can skip the data from other origins as we have thatinformation in WAL but that is not possible during initial sync so we raisea warning if there is such a possibility.Author: Vignesh CReviewed-By: Peter Smith, Amit Kapila, Jonathan Katz, Shi yu, Wang weiDiscussion:https://www.postgresql.org/message-id/CALDaNm0gwjY_4HFxvvty01BOT01q_fJLKQ3pWP9=9orqubhjcQ@mail.gmail.com
1 parent4b4663f commit8756930

File tree

4 files changed

+258
-29
lines changed

4 files changed

+258
-29
lines changed

‎doc/src/sgml/ref/alter_subscription.sgml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -172,6 +172,11 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
172172
Previously subscribed tables are not copied, even if a table's row
173173
filter <literal>WHERE</literal> clause has since been modified.
174174
</para>
175+
<para>
176+
See <xref linkend="sql-createsubscription-notes"/> for details of
177+
how <literal>copy_data = true</literal> can interact with the
178+
<literal>origin</literal> parameter.
179+
</para>
175180
</listitem>
176181
</varlistentry>
177182
</variablelist></para>

‎doc/src/sgml/ref/create_subscription.sgml‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -213,6 +213,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
213213
will affect what data is copied. Refer to the
214214
<xref linkend="sql-createsubscription-notes" /> for details.
215215
</para>
216+
<para>
217+
See <xref linkend="sql-createsubscription-notes"/> for details of how
218+
<literal>copy_data = true</literal> can interact with the
219+
<literal>origin</literal> parameter.
220+
</para>
216221
</listitem>
217222
</varlistentry>
218223

@@ -315,6 +320,11 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
315320
to <literal>any</literal> means that the publisher sends changes
316321
regardless of their origin. The default is <literal>any</literal>.
317322
</para>
323+
<para>
324+
See <xref linkend="sql-createsubscription-notes"/> for details of how
325+
<literal>copy_data = true</literal> can interact with the
326+
<literal>origin</literal> parameter.
327+
</para>
318328
</listitem>
319329
</varlistentry>
320330
</variablelist></para>
@@ -386,6 +396,31 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
386396
can have non-existent publications.
387397
</para>
388398

399+
<para>
400+
When using a subscription parameter combination of
401+
<literal>copy_data = true</literal> and <literal>origin = NONE</literal>,
402+
the initial sync table data is copied directly from the publisher, meaning
403+
that knowledge of the true origin of that data is not possible. If the
404+
publisher also has subscriptions then the copied table data might have
405+
originated from further upstream. This scenario is detected and a WARNING is
406+
logged to the user, but the warning is only an indication of a potential
407+
problem; it is the user's responsibility to make the necessary checks to
408+
ensure the copied data origins are really as wanted or not.
409+
</para>
410+
411+
<para>
412+
To find which tables might potentially include non-local origins (due to
413+
other subscriptions created on the publisher) try this SQL query:
414+
<programlisting>
415+
# substitute &lt;pub-names&gt; below with your publication name(s) to be queried
416+
SELECT DISTINCT N.nspname AS schemaname, C.relname AS tablename
417+
FROM pg_publication P,
418+
LATERAL pg_get_publication_tables(P.pubname) GPT
419+
JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid),
420+
pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)
421+
WHERE C.oid = GPT.relid AND P.pubname IN (&lt;pub-names&gt;);
422+
</programlisting></para>
423+
389424
</refsect1>
390425

391426
<refsect1>

‎src/backend/commands/subscriptioncmds.c‎

Lines changed: 128 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -92,6 +92,10 @@ typedef struct SubOpts
9292
}SubOpts;
9393

9494
staticList*fetch_table_list(WalReceiverConn*wrconn,List*publications);
95+
staticvoidcheck_publications_origin(WalReceiverConn*wrconn,
96+
List*publications,boolcopydata,
97+
char*origin,Oid*subrel_local_oids,
98+
intsubrel_count,char*subname);
9599
staticvoidcheck_duplicates_in_publist(List*publist,Datum*datums);
96100
staticList*merge_publications(List*oldpublist,List*newpublist,booladdpub,constchar*subname);
97101
staticvoidReportSlotConnectionError(List*rstates,Oidsubid,char*slotname,char*err);
@@ -680,6 +684,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
680684
PG_TRY();
681685
{
682686
check_publications(wrconn,publications);
687+
check_publications_origin(wrconn,publications,opts.copy_data,
688+
opts.origin,NULL,0,stmt->subname);
683689

684690
/*
685691
* Set sync state based on if we were asked to do data copy or
@@ -786,6 +792,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
786792
ListCell*lc;
787793
intoff;
788794
intremove_rel_len;
795+
intsubrel_count;
789796
Relationrel=NULL;
790797
typedefstructSubRemoveRels
791798
{
@@ -815,28 +822,33 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
815822

816823
/* Get local table list. */
817824
subrel_states=GetSubscriptionRelations(sub->oid, false);
825+
subrel_count=list_length(subrel_states);
818826

819827
/*
820828
* Build qsorted array of local table oids for faster lookup. This can
821829
* potentially contain all tables in the database so speed of lookup
822830
* is important.
823831
*/
824-
subrel_local_oids=palloc(list_length(subrel_states)*sizeof(Oid));
832+
subrel_local_oids=palloc(subrel_count*sizeof(Oid));
825833
off=0;
826834
foreach(lc,subrel_states)
827835
{
828836
SubscriptionRelState*relstate= (SubscriptionRelState*)lfirst(lc);
829837

830838
subrel_local_oids[off++]=relstate->relid;
831839
}
832-
qsort(subrel_local_oids,list_length(subrel_states),
840+
qsort(subrel_local_oids,subrel_count,
833841
sizeof(Oid),oid_cmp);
834842

843+
check_publications_origin(wrconn,sub->publications,copy_data,
844+
sub->origin,subrel_local_oids,
845+
subrel_count,sub->name);
846+
835847
/*
836848
* Rels that we want to remove from subscription and drop any slots
837849
* and origins corresponding to them.
838850
*/
839-
sub_remove_rels=palloc(list_length(subrel_states)*sizeof(SubRemoveRels));
851+
sub_remove_rels=palloc(subrel_count*sizeof(SubRemoveRels));
840852

841853
/*
842854
* Walk over the remote tables and try to match them to locally known
@@ -862,7 +874,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
862874
pubrel_local_oids[off++]=relid;
863875

864876
if (!bsearch(&relid,subrel_local_oids,
865-
list_length(subrel_states),sizeof(Oid),oid_cmp))
877+
subrel_count,sizeof(Oid),oid_cmp))
866878
{
867879
AddSubscriptionRelState(sub->oid,relid,
868880
copy_data ?SUBREL_STATE_INIT :SUBREL_STATE_READY,
@@ -881,7 +893,7 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
881893
sizeof(Oid),oid_cmp);
882894

883895
remove_rel_len=0;
884-
for (off=0;off<list_length(subrel_states);off++)
896+
for (off=0;off<subrel_count;off++)
885897
{
886898
Oidrelid=subrel_local_oids[off];
887899

@@ -1784,6 +1796,117 @@ AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId)
17841796
table_close(rel,RowExclusiveLock);
17851797
}
17861798

1799+
/*
1800+
* Check and log a warning if the publisher has subscribed to the same table
1801+
* from some other publisher. This check is required only if "copy_data = true"
1802+
* and "origin = none" for CREATE SUBSCRIPTION and
1803+
* ALTER SUBSCRIPTION ... REFRESH statements to notify the user that data
1804+
* having origin might have been copied.
1805+
*
1806+
* This check need not be performed on the tables that are already added
1807+
* because incremental sync for those tables will happen through WAL and the
1808+
* origin of the data can be identified from the WAL records.
1809+
*
1810+
* subrel_local_oids contains the list of relation oids that are already
1811+
* present on the subscriber.
1812+
*/
1813+
staticvoid
1814+
check_publications_origin(WalReceiverConn*wrconn,List*publications,
1815+
boolcopydata,char*origin,Oid*subrel_local_oids,
1816+
intsubrel_count,char*subname)
1817+
{
1818+
WalRcvExecResult*res;
1819+
StringInfoDatacmd;
1820+
TupleTableSlot*slot;
1821+
OidtableRow[1]= {TEXTOID};
1822+
List*publist=NIL;
1823+
inti;
1824+
1825+
if (!copydata|| !origin||
1826+
(pg_strcasecmp(origin,LOGICALREP_ORIGIN_NONE)!=0))
1827+
return;
1828+
1829+
initStringInfo(&cmd);
1830+
appendStringInfoString(&cmd,
1831+
"SELECT DISTINCT P.pubname AS pubname\n"
1832+
"FROM pg_publication P,\n"
1833+
" LATERAL pg_get_publication_tables(P.pubname) GPT\n"
1834+
" JOIN pg_subscription_rel PS ON (GPT.relid = PS.srrelid),\n"
1835+
" pg_class C JOIN pg_namespace N ON (N.oid = C.relnamespace)\n"
1836+
"WHERE C.oid = GPT.relid AND P.pubname IN (");
1837+
get_publications_str(publications,&cmd, true);
1838+
appendStringInfoString(&cmd,")\n");
1839+
1840+
/*
1841+
* In case of ALTER SUBSCRIPTION ... REFRESH, subrel_local_oids contains
1842+
* the list of relation oids that are already present on the subscriber.
1843+
* This check should be skipped for these tables.
1844+
*/
1845+
for (i=0;i<subrel_count;i++)
1846+
{
1847+
Oidrelid=subrel_local_oids[i];
1848+
char*schemaname=get_namespace_name(get_rel_namespace(relid));
1849+
char*tablename=get_rel_name(relid);
1850+
1851+
appendStringInfo(&cmd,"AND NOT (N.nspname = '%s' AND C.relname = '%s')\n",
1852+
schemaname,tablename);
1853+
}
1854+
1855+
res=walrcv_exec(wrconn,cmd.data,1,tableRow);
1856+
pfree(cmd.data);
1857+
1858+
if (res->status!=WALRCV_OK_TUPLES)
1859+
ereport(ERROR,
1860+
(errcode(ERRCODE_CONNECTION_FAILURE),
1861+
errmsg("could not receive list of replicated tables from the publisher: %s",
1862+
res->err)));
1863+
1864+
/* Process tables. */
1865+
slot=MakeSingleTupleTableSlot(res->tupledesc,&TTSOpsMinimalTuple);
1866+
while (tuplestore_gettupleslot(res->tuplestore, true, false,slot))
1867+
{
1868+
char*pubname;
1869+
boolisnull;
1870+
1871+
pubname=TextDatumGetCString(slot_getattr(slot,1,&isnull));
1872+
Assert(!isnull);
1873+
1874+
ExecClearTuple(slot);
1875+
publist=list_append_unique(publist,makeString(pubname));
1876+
}
1877+
1878+
/*
1879+
* Log a warning if the publisher has subscribed to the same table from
1880+
* some other publisher. We cannot know the origin of data during the
1881+
* initial sync. Data origins can be found only from the WAL by looking at
1882+
* the origin id.
1883+
*
1884+
* XXX: For simplicity, we don't check whether the table has any data or
1885+
* not. If the table doesn't have any data then we don't need to
1886+
* distinguish between data having origin and data not having origin so we
1887+
* can avoid logging a warning in that case.
1888+
*/
1889+
if (publist)
1890+
{
1891+
StringInfopubnames=makeStringInfo();
1892+
1893+
/* Prepare the list of publication(s) for warning message. */
1894+
get_publications_str(publist,pubnames, false);
1895+
ereport(WARNING,
1896+
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1897+
errmsg("subscription \"%s\" requested copy_data with origin = NONE but might copy data that had a different origin",
1898+
subname),
1899+
errdetail_plural("Subscribed publication %s is subscribing to other publications.",
1900+
"Subscribed publications %s are subscribing to other publications.",
1901+
list_length(publist),pubnames->data),
1902+
errhint("Verify that initial data copied from the publisher tables did not come from other origins."));
1903+
}
1904+
1905+
ExecDropSingleTupleTableSlot(slot);
1906+
1907+
walrcv_clear_result(res);
1908+
}
1909+
17871910
/*
17881911
* Get the list of tables which belong to specified publications on the
17891912
* publisher connection.

0 commit comments

Comments
 (0)
[8]ページ先頭
©2009-2025 Movatter.jp