public inbox for [email protected]
help / color / mirror / Atom feedFrom: vignesh C <[email protected]>
To: Peter Smith <[email protected]>
Cc: Dilip Kumar <[email protected]>
Cc: Nisha Moond <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: shveta malik <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Bharath Rupireddy <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Cc: shveta malik <[email protected]>
Subject: Re: Proposal: Conflict log history table for Logical Replication
Date: Mon, 8 Jun 2026 17:05:33 +0530
Message-ID: <CALDaNm2V3EuSKMaTqDvaiLQW3jwBX90aXTkMST1ft=uJ8J+R5A@mail.gmail.com> (raw)
In-Reply-To: <CAHut+Pt3=rZO+yJqj7o3xH0LcgrztFCvhoayiBJWbhmEio6teQ@mail.gmail.com>
References: <CAFiTN-u5D5o_AGNbHRZHaOqAMWkxLf+hSk_r9X3gv6HbLOB5+g@mail.gmail.com>
<CALj2ACViThGQDYi-yeqUeHqG2Pozn2AiyvtDtjE6zhhbM0KsEA@mail.gmail.com>
<CAA4eK1+44b3vd_OWfiaVNtjf5Njb5cek09pmKRmttBByeg0NoA@mail.gmail.com>
<CAFiTN-v3L0WacCDx5dkOSonaZQbJfstXL4HrCPD1ahRdUsRnSg@mail.gmail.com>
<CALj2ACW63uuxh0fSoxEAF8OMWhz1dJKSkp268WJDzf5BUqCf5g@mail.gmail.com>
<CAFiTN-s9WWLOhW1TO27NtJwGf0bh2+MWyp3NEkZFeN_S5_p_rA@mail.gmail.com>
<CAA4eK1LxnsEx5sMbQkK5MHAgXKPROMQXQ0n=fKMwz+UsfKQaMQ@mail.gmail.com>
<CAD21AoDj+c4LXf2y4ESR-gVyv9d8V0G4R8R9pn-PcmT5zPzYcg@mail.gmail.com>
<CAA4eK1KokmAwNOL6bS-ip_E3F96PiQTjC4j-M+5vD1T6uUyi3Q@mail.gmail.com>
<CAFiTN-vFKE8E_N6h+peX9DP92mxCeFdm5A9Esn4DkLmNcZ-dOA@mail.gmail.com>
<CAFiTN-shLYf-fOTQ_dBf3Xfx05gxs_8d93MHZXyyz6w2Bg5geQ@mail.gmail.com>
<CAFiTN-tEgkKQHUikn6iBFCYf7XOObR7ncUq=OVh7WEk=6P4ymw@mail.gmail.com>
<CAFiTN-tQiakd8m+-d6WN6RpJXSv_JcropZ2oGzme4d1JudQhYg@mail.gmail.com>
<CAJpy0uDKbYWt+YPADj=4fHEvrGEWgnG1n_YsiGT_EZiZf0VSAw@mail.gmail.com>
<CAFiTN-t82BiXen+HfdR9jZyOpuSO92xonnUK=khXsiZWBfOxMA@mail.gmail.com>
<CAJpy0uAu2paxGAEffD=vaBTW9Jqbtxxawb8K8FgiASfeKPnGog@mail.gmail.com>
<CAJpy0uC0ZWgHOivJ102A1fMkppwK3RuSMafRPKyjwkmJrjhVUw@mail.gmail.com>
<CAFiTN-vFV9-zajrwjYHYyFnyQsooOAXW4CpxB5f-iT3APjOtoQ@mail.gmail.com>
<CAJpy0uBeU1dZgaqsSVKc=P=EVUKxRgVuHR8jDXFL-HLibbE-kQ@mail.gmail.com>
<CAA4eK1+FOkOxhzVLAnDymoNjp4i98H-L1+ZsWDgJEv-ndnTzTA@mail.gmail.com>
<CAFiTN-sVK6Bp+BawCJU_WpAXQSTX4OkKmce5EE4YNBgD-XSjZw@mail.gmail.com>
<CAA4eK1LbjV0bctib9wUnBpEkC+2rZFPnGuRtrKuc5AtUAzum+A@mail.gmail.com>
<CAFiTN-vq50N3QP9p3_SH+tJ8Pn=uRDb0X4qEcQZYcGW9AX88rQ@mail.gmail.com>
<CAFiTN-u3+zRGPESP5kUUfa6NxaWh1HL-gd1225KJ0Uvzi1urow@mail.gmail.com>
<CAA4eK1L4iNk6mNTC83PbYrRfUdtivH4U961PkdFfOO7mvc=USg@mail.gmail.com>
<CAFiTN-v+Mh64UfR5zb5rwgyGm6HS80XRSZ_XeaWkg8=+s9o3Kg@mail.gmail.com>
<CAFiTN-s3ZFHteQsiC3H4=AjTWxuwN-w69XQ3xL5X6YOMTua4pA@mail.gmail.com>
<CAJpy0uDe724nY59j-8hMapZ_Fru1Wo-NucF4Ea1B3Jrw=+J+UQ@mail.gmail.com>
<CAFiTN-uR=86L_5tyiA7n73EXCSCuDfQKfL5O=c8n7zZom8_ONQ@mail.gmail.com>
<CAD21AoDfOS-J0M9WbM3D20eGbSPzbfLQ-9XoYkxO4AZ9twqyvg@mail.gmail.com>
<CAFiTN-vMTg2X7vwfHLr5Gvy8ViV63_iaEcpHmM8V5GpA9-u8cg@mail.gmail.com>
<CAA4eK1+b2Ws0e_ZYJsgZAPn7VWndxAK_YM_QMKcfXst3e7F6Jg@mail.gmail.com>
<CAFiTN-v6hFKMPrSyTBsz=AtEETYMbOxrqvhZJsPQqKgQc4WCLw@mail.gmail.com>
<CAA4eK1KV3rYkaxys5fh-PtE9kq5xrFbiaRpOSPoRgQG494ek+g@mail.gmail.com>
<CAFiTN-utvu=QjY1QQ1a_TvkpkpvesMWo9M8wTFYLaOTPdpOJvw@mail.gmail.com>
<CAA4eK1+HoSOEqNwT3twArPNx4_D7hSUoEg2LnYhX8n9iUwhXgQ@mail.gmail.com>
<CAFiTN-tqmsfW0Sk=1RhzuduxqLrf9KEc8VOvBae+4aYxWTJwuA@mail.gmail.com>
<CAA4eK1JmCQ=DHe3HsqpX+P3mGDUd_Z7E7oAxdstK6822W6tuCw@mail.gmail.com>
<CAFiTN-uE4eAUYewuq3c5deAt3TtVork+H6rkUHRv68cOGr5rmQ@mail.gmail.com>
<CAFiTN-sJbhPX+LbA8YuQeYJpfGA2XA+OKXf8jCm04RoJOyzLvw@mail.gmail.com>
<CAJpy0uBPOyWj9itFjHzGXfrUuYS8KGmAvgdcV_9FPjWZ0EZz_w@mail.gmail.com>
<CAFiTN-s=iLE4qM4qmw9yXKqW09R_c_HqaSGeZXJ2EaTVfXss+g@mail.gmail.com>
<CAA4eK1KYo0vZpPSRc_4gVpa06-J39gxjs3tHFyckgkBfYJSfFA@mail.gmail.com>
<CAFiTN-vrKc6OWzrg6yvpwYcj79k=zkrDp3uwiZzjwrWLJAq6tw@mail.gmail.com>
<CAA4eK1LmvrfEgn1NUZZ=E3yMCjQdNZ5=_SBEry73-EmF6jM_PQ@mail.gmail.com>
<CAFiTN-vjfub5b3PqPQzfOw9BSjm8jt28ott+Hoz9CrRxJHzYkg@mail.gmail.com>
<CAFiTN-v=ANapYvRK+SOy2wJb4CSuD6Vb6_bTGuReM9Dv+3tucA@mail.gmail.com>
<CALDaNm1zEYoSdf2Ns-=UJRw95E5sbfpB0oaNUWtRJN27Q1Knhw@mail.gmail.com>
<CALDaNm3USsXVNBsfdpkp60HVgrTV4taWMk1xZYNBa7QUF=V0jg@mail.gmail.com>
<CAFiTN-sNg9ghLNkB2Kn0SwBGOub9acc99XZZU_d5NAcyW-yrEg@mail.gmail.com>
<CAJpy0uAF3EYcYdpTHdKMeXfvaPbNvnWrZUATrSLL1hqjao=33A@mail.gmail.com>
<CAFiTN-uikggCKp2LscTorKY5d3KF9j93DW0xebDcRX86G+ZsSw@mail.gmail.com>
<CAJpy0uDaOoVK8S3_xxTAcTDpfK1AY7tApw7nPOZG_gUz+DMi=Q@mail.gmail.com>
<CAA4eK1+AdeC5B9xrAXSKWGtTh-0d8xdD=fZttmOBm+c8o8thAQ@mail.gmail.com>
<CAFiTN-skBQAeuzuUd+PDK0Gqc8g+4x9ypBMwJhOrmW8ZCFKGSA@mail.gmail.com>
<CAJpy0uCdrsW5T+okq7xTOVxagje7FW3DOeY5B0CGKYa5VqF_tQ@mail.gmail.com>
<CAFiTN-u+_mFj9caYYFO7=_YHFXk5y=vvOm2H2=5hctYktmAVGA@mail.gmail.com>
<CALDaNm1aivk9KgQ5daeF6YZzuE+0wWc2yb7wb6qikNyvfPN0Sg@mail.gmail.com>
<CAJpy0uD6fTEUYJx3+yDbvB=VW7c5AaGoeSd7iwHdYYO=kYGn3g@mail.gmail.com>
<CALDaNm2YOOdJ25X1sJ+DYz37K6Qi4g0ZNFHb_pQMF9UqancnEA@mail.gmail.com>
<CAHut+PtMS5bENS0DVtBj+s3kUEOq61+hSkqLODjFB78egB0imQ@mail.gmail.com>
<CAFiTN-s_M83sfs+MHHbUrMesjsCPN4JWxY5MChCEiY1U-u7=9g@mail.gmail.com>
<CAFiTN-vj8NTm9w_L2XdhxJCub_RZw__YVUgfXa1B1kJzJctRNw@mail.gmail.com>
<CAJpy0uBDLnfhuSiev8W9ZMFNTzUmqhds2dKayUpLoN-z1dtsLA@mail.gmail.com>
<CAFiTN-uL9f0X+=Ep4BbAPvaTJA7S4XHM--G4BsnPJw4uJW7EGQ@mail.gmail.com>
<CAJpy0uDG=t-y_m8t1zpBzfz9viP3K8dyQgkruaraVT85UtTkrg@mail.gmail.com>
<CAFiTN-tR8Rhs8uhfbck0Ac4dd1MopvvYgjK39nWyNXRp9Z3Qww@mail.gmail.com>
<CAA4eK1Kf15UpNmpTTE2XyX=9PE_oTpOoy5xqg3rFWbxwwP4Rbg@mail.gmail.com>
<CAFiTN-tNqb0vjuadDz-as67ksSXa=aEK+JW=4b54RVmkUK1m2Q@mail.gmail.com>
<CAFiTN-vDCxx6ydUFo59L8qNBbierg4as3TGPPiavR7UZjYurzA@mail.gmail.com>
<CAHut+PsWms218ENALnytLEV4NpxjOrAYhChLDaMaeE65-vNgrQ@mail.gmail.com>
<CAFiTN-v9i9RmDvdUmtMUow4=b+nr0k7LKMyEQ+6ZF=EVdfBhBA@mail.gmail.com>
<CALDaNm2YTKwPDjt9OV49RgM0zbkWhMhNu228bj_7f+zzcPb-ew@mail.gmail.com>
<CAFiTN-t_4XvofM3an-WmykqnPE+9wf9U+o2M7p1CWd9eXkN88Q@mail.gmail.com>
<CAHut+PuaqNDfDu_3xkZR4OYxw-B7ew_WjpLXCBvMcSBJz2K6Xg@mail.gmail.com>
<CAFiTN-uqNN9S_hRuda_th5MEpywa15g+XO00yM6tNJ-spGRRJw@mail.gmail.com>
<CABdArM6QxXatkGefTHy__HgaYHBvbKesffeXzT8Vn-kvcvGK4w@mail.gmail.com>
<CAFiTN-tgMWr=TGPhs9BxaPuSC_jhM7sJJ4fHedE5W6=h40jLfA@mail.gmail.com>
<CABdArM5fgzfyC2mH3YGB8t8cJBHWqAG1BS6rJMk7mX-8=9d=Cg@mail.gmail.com>
<CABdArM568KF4WXdFX_aZkCiDK8R71Wpep0gC2a+cV8BMobwkrg@mail.gmail.com>
<CAFiTN-vQ0tu18BD3UmKPb0rzZyFMQAVgGbdpMA8iYLX7PZOqOA@mail.gmail.com>
<CALDaNm20PDtmG2E3qaTC+YuL5twv+c9k573wL3sb=OwgmZphxQ@mail.gmail.com>
<CAA4eK1LhOHa_TEznw+gFoq+w0vMvvsDG2g9Xq8Mwa8xZMY73og@mail.gmail.com>
<CAFiTN-vPDqrQ2rHykNgd+groFxqwBYFQF97R-Co2EmtUkV6MTg@mail.gmail.com>
<CAFiTN-vsd=wNiEPXPQhZnipAb--+mBUC01M-pcjBjbRockgCUA@mail.gmail.com>
<CAJpy0uCjSq_gUCJBfURhqtB6bLvkKSUL-sVXpaGKjEapv5+t+Q@mail.gmail.com>
<CAFiTN-uZ-LaStAY3NuCY-nb7GCB9joiHX7HtHEMseJ0xfnqVSg@mail.gmail.com>
<CAFiTN-vhJxRW5NQ628oidnk0KtHwKt11dW9-+vxqpXLTgjiYiA@mail.gmail.com>
<CALDaNm1cJURibYKY4+DuNosjM72C9oGheUF-roMyff__+AsKBw@mail.gmail.com>
<CAJpy0uD1_77TDAFc4jE-94X-WUus7Q3gGU0pXfC+Tticq1hFvA@mail.gmail.com>
<CAA4eK1LFcSc4XCj4mU-cv27F_6n6=+ehJ=YAsAnyBbz4Sv_tVg@mail.gmail.com>
<CAFiTN-s5ZtjXKrSbam7TNWJ9Ax-kCancXcestAnx2by7dK0-UA@mail.gmail.com>
<CAFiTN-u=Da32mXyz8jocEGtuLSG4ccXXj_aEzUTPp2zkLb3MVA@mail.gmail.com>
<CALDaNm1qY5e0thfsDB2uWXqZn4hgTWTxiUDwcF1hWA-jodsKYg@mail.gmail.com>
<CALDaNm1nFtv3dtdRdbqWo2Rf_av7XbxDfK1Orqjcqs_Su_cLRQ@mail.gmail.com>
<CABdArM7R498qC5Fr42aU_q-2Sc5QsT4dyKgmO_f6Uy=8oCAFXA@mail.gmail.com>
<CAFiTN-sRZ+Z_9B3ue2L4zkbcfmPjjcAjcR1C+px1PyAs+HGsSg@mail.gmail.com>
<CAFiTN-sdcjf9xJ2M-=ab5e4y662tTmFFiP4gHL44tC9PcQozcw@mail.gmail.com>
<CALDaNm2WNjaNxUijVkvT6y69D62rfCu8OMwU-Pf-84un2r_=ig@mail.gmail.com>
<CAHut+PvEP5uUR13xJ3gbNKGU49=Rg32DXMGZ2wL9jTcKHyN_=Q@mail.gmail.com>
<CALDaNm3Jb5AQTsFJFxYZZJCaheT7qToCZkEALfW-vsMMFxjOyQ@mail.gmail.com>
<CAHut+PtQn5U9i00qvBmjo0KBxyb+ZmBb38NzF91KnX4J86Jg_g@mail.gmail.com>
<CALDaNm1a1gzy0L38U394_4OFwGUS8ALgSONYj++VLimY0g9piQ@mail.gmail.com>
<CABdArM5X63AdtS99QKGjVijUd_Q_dV8QUDSo4nTHKJjn3JwtAg@mail.gmail.com>
<CAA4eK1+h=QV4Zi=PW8Zt2D6be5ki5Mu2HgdXcfUophptx6Mt_A@mail.gmail.com>
<CAFiTN-s-tuxar9Dp5He0CFa1pzfy1fmiwcBj6PtwD0hDodE5ng@mail.gmail.com>
<CAFiTN-sx=k+Th=uYsrLcS6YMZbPVi9Wrggn1w2Nzf9MLEU7YRQ@mail.gmail.com>
<CAFiTN-u5pcgAhXyJgj+p7-xmShtp0i8xA000tzjCLFQp_zMXUA@mail.gmail.com>
<CAFiTN-tRpS7b3qFqckqFtHETj0jyzj-8SxC1arjfwf-hQd47PQ@mail.gmail.com>
<CABdArM5Ka_m_GWhL_zZbeDPKmL-Wezwb4A-NHnO-v-fRDuhA-Q@mail.gmail.com>
<CAFiTN-sqEMAbZ2pTt=zMa=918NV7HVeXF4bCOF+swtzKnTy5yQ@mail.gmail.com>
<CALDaNm0WX0Vqoy2UQZh-2TpWraf4OYn28kWe9aGR=vxKwLA+bw@mail.gmail.com>
<CAHut+Pt3=rZO+yJqj7o3xH0LcgrztFCvhoayiBJWbhmEio6teQ@mail.gmail.com>
On Thu, 4 Jun 2026 at 11:59, Peter Smith <[email protected]> wrote:
>
> Some review comments for patch v45-0003.
>
> 3.
> + relid = get_relname_relid(relname, PG_CONFLICT_NAMESPACE);
> + if (OidIsValid(relid))
> + {
> + /* Existing table from upgrade */
> + Assert(IsBinaryUpgrade);
> + }
> + else
>
> Should this be the other way around?
>
> if (IsBinaryUpgrade)
> {
> relid = get_relname_relid(relname, PG_CONFLICT_NAMESPACE);
> Assert(OidIsValid(relid));
> }
This is done intentionally so that the Assert fails if the table
exists in case of non binary upgrade mode.
> 6.
> + if (fout->remoteVersion >= 190000)
> + appendPQExpBufferStr(query,
> + " s.subconflictlogrelid, s.subconflictlogdest\n");
> else
> - appendPQExpBufferStr(query, " NULL AS subservername\n");
> + appendPQExpBufferStr(query,
> + " NULL AS subconflictlogrelid, NULL AS subconflictlogdest\n");
>
> It seems more natural to think of the 'subconflictlogdest' before the
> subconflictlogrelid. Perhaps the SQL should swap those around.
I think the existing is better as it does select in the same order as
it is maintained in the table.
> ~~~
>
> 7.
> + if (PQgetisnull(res, i, i_subconflictlogrelid))
> + subinfo[i].subconflictlogrelid = InvalidOid;
> + else
> + {
> + TableInfo *tableInfo;
> +
> + subinfo[i].subconflictlogrelid =
> + atooid(PQgetvalue(res, i, i_subconflictlogrelid));
> +
> + if (subinfo[i].subconflictlogrelid)
> + {
> + tableInfo = findTableByOid(subinfo[i].subconflictlogrelid);
> + if (!tableInfo)
> + pg_fatal("could not find conflict log table with OID %u",
> + subinfo[i].subconflictlogrelid);
> +
> + addObjectDependency(&subinfo[i].dobj, tableInfo->dobj.dumpId);
> + }
> + }
> +
> + if (PQgetisnull(res, i, i_sublogdestination))
> + subinfo[i].subconflictlogdest = NULL;
> + else
> + subinfo[i].subconflictlogdest =
> + pg_strdup(PQgetvalue(res, i, i_sublogdestination));
> +
>
> 7a.
> Those new attributes come as a pair -- e.g. the relid only makes sense
> depending on the destination value; maybe only look for the CLT relid
> is the destination is not LOG/NULL
I think the existing is better as it does select in the same order as
it is maintained in the table.
> Knowing the destination also means you can also do integrity checks
> for PQgetisnull(res, i, i_subconflictlogrelid) -- e.g. must be valid
> Oid, or must be invalid Oid.
I'm not sure if it is pg_dump's responsibility to do these integrity
checks in this case.
> ~~~
>
> dumpSubscription:
>
> 8.
> + if (subinfo->subconflictlogdest &&
> + (pg_strcasecmp(subinfo->subconflictlogdest, "log") != 0))
> + appendPQExpBuffer(query,
> + "\n\nALTER SUBSCRIPTION %s SET (conflict_log_destination = %s);\n",
> + qsubname,
> + subinfo->subconflictlogdest);
> +
> 8b.
> Not sure why this has a double \n\n prefix instead of just \n. I
> didn't see that pattern elsewhere in this file.
Generally there is 2 line gaps between the create and alter
subscription, it was done to maintain the same line gaps like below:
CREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION
pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);
ALTER SUBSCRIPTION sub3 SET (conflict_log_destination = table);
ALTER SUBSCRIPTION sub3 OWNER TO vignesh;
The rest of the comments were fixed. The attached v47 version patch
has the changes for the same.
Regards,
Vignesh
Attachments:
[application/octet-stream] v47-0001-Add-configurable-conflict-log-table-for-Logical.patch (69.1K, 2-v47-0001-Add-configurable-conflict-log-table-for-Logical.patch)
download | inline diff:
From 0008968e0b516c82d07b77057209eeb450dc31d8 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 8 Jun 2026 06:43:46 +0530
Subject: [PATCH v47 1/5] Add configurable conflict log table for Logical
Replication
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
This patch adds a feature to provide a structured, queryable record of all
logical replication conflicts. The current approach of logging conflicts as
plain text in the server logs makes it difficult to query, analyze, and
use for external monitoring and automation.
This patch addresses these limitations by introducing a configurable
conflict_log_destination=('log'/'table'/'all') option in the CREATE SUBSCRIPTION
command.
If the user chooses to enable logging to a table (by selecting 'table' or 'all'),
an internal logging table named pg_conflict_log_<subid> is automatically
created within a dedicated, system-managed 'pg_conflict' namespace to prevent
users from manually dropping or altering it. This also prevents accidental
name collisions with user-created tables. This table is linked to the
subscription via an internal dependency, ensuring it is automatically dropped
when the subscription is removed
The per-subscription table model was chosen over a single global log to ensure
superior isolation and administrative flexibility by directly aligning table ownership
with the subscription’s lifecycle. This approach allows for granular permission
management, enabling the subscription owner to perform necessary maintenance
tasks like SELECT, DELETE, and TRUNCATE without the security risks or complex
Row-Level Security required by a shared global table.
The conflict details, including the original and remote tuples, are stored in JSON
columns, providing a flexible format to accommodate different table schemas.
The log table captures essential attributes such as local and remote transaction IDs,
LSNs, commit timestamps, and conflict type, providing a complete record for post-mortem
analysis.
This feature will make logical replication conflicts easier to monitor and manage,
significantly improving the overall resilience and operability of replication setups.
The conflict log tables will not be included in a publication, even if the publication
is configured to include ALL TABLES or ALL TABLES IN SCHEMA.
---
src/backend/catalog/aclchk.c | 50 ++--
src/backend/catalog/catalog.c | 29 +-
src/backend/catalog/heap.c | 32 ++-
src/backend/catalog/namespace.c | 11 +-
src/backend/catalog/pg_publication.c | 14 +-
src/backend/catalog/pg_subscription.c | 7 +
src/backend/catalog/system_views.sql | 4 +-
src/backend/commands/subscriptioncmds.c | 212 +++++++++++++-
src/backend/commands/tablecmds.c | 6 +-
src/backend/executor/execMain.c | 28 ++
src/backend/replication/logical/conflict.c | 175 ++++++++++++
src/backend/utils/cache/lsyscache.c | 2 +-
src/bin/initdb/initdb.c | 6 +
src/bin/psql/tab-complete.in.c | 8 +-
src/include/catalog/catalog.h | 2 +
src/include/catalog/catversion.h | 2 +-
src/include/catalog/pg_namespace.dat | 3 +
src/include/catalog/pg_subscription.h | 11 +
src/include/replication/conflict.h | 27 ++
src/include/utils/lsyscache.h | 2 +-
src/test/regress/expected/subscription.out | 308 +++++++++++++++++++++
src/test/regress/sql/subscription.sql | 256 +++++++++++++++++
src/tools/pgindent/typedefs.list | 2 +
23 files changed, 1147 insertions(+), 50 deletions(-)
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 007ede997c5..64e176652d1 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -3337,25 +3337,39 @@ pg_class_aclmask_ext(Oid table_oid, Oid roleid, AclMode mask,
classForm = (Form_pg_class) GETSTRUCT(tuple);
- /*
- * Deny anyone permission to update a system catalog unless
- * pg_authid.rolsuper is set.
- *
- * As of 7.4 we have some updatable system views; those shouldn't be
- * protected in this way. Assume the view rules can take care of
- * themselves. ACL_USAGE is if we ever have system sequences.
- */
- if ((mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE)) &&
- IsSystemClass(table_oid, classForm) &&
- classForm->relkind != RELKIND_VIEW &&
- !superuser_arg(roleid))
- mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE);
-
- /*
- * Otherwise, superusers bypass all permission-checking.
- */
- if (superuser_arg(roleid))
+ if (!superuser_arg(roleid))
+ {
+ if (mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE))
+ {
+ if (IsConflictLogTableClass(classForm))
+ {
+ /*
+ * For conflict log tables, allow non-superusers to perform
+ * DELETE and TRUNCATE for cleanup and maintenance, while still
+ * restricting INSERT, UPDATE, and USAGE.
+ */
+ mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_USAGE);
+ }
+ else if (IsSystemClass(table_oid, classForm) &&
+ classForm->relkind != RELKIND_VIEW)
+ {
+ /*
+ * Deny anyone permission to update a system catalog unless
+ * pg_authid.rolsuper is set.
+ *
+ * As of 7.4 we have some updatable system views; those
+ * shouldn't be protected in this way. Assume the view rules
+ * can take care of themselves. ACL_USAGE is if we ever have
+ * system sequences.
+ */
+ mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE |
+ ACL_USAGE);
+ }
+ }
+ }
+ else
{
+ /* Superusers bypass all permission-checking. */
ReleaseSysCache(tuple);
return mask;
}
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index 7be49032934..be8791af875 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -86,7 +86,9 @@ bool
IsSystemClass(Oid relid, Form_pg_class reltuple)
{
/* IsCatalogRelationOid is a bit faster, so test that first */
- return (IsCatalogRelationOid(relid) || IsToastClass(reltuple));
+ return (IsCatalogRelationOid(relid) ||
+ IsToastClass(reltuple) ||
+ IsConflictLogTableClass(reltuple));
}
/*
@@ -230,6 +232,20 @@ IsToastClass(Form_pg_class reltuple)
return IsToastNamespace(relnamespace);
}
+/*
+ * IsConflictLogTableClass
+ * True iff pg_class tuple represents a Conflict Log Table.
+ *
+ * Does not perform any catalog accesses.
+ */
+bool
+IsConflictLogTableClass(Form_pg_class reltuple)
+{
+ Oid relnamespace = reltuple->relnamespace;
+
+ return IsConflictLogTableNamespace(relnamespace);
+}
+
/*
* IsCatalogNamespace
* True iff namespace is pg_catalog.
@@ -264,6 +280,17 @@ IsToastNamespace(Oid namespaceId)
isTempToastNamespace(namespaceId);
}
+/*
+ * IsConflictLogTableNamespace
+ * True iff namespace is pg_conflict.
+ *
+ * Does not perform any catalog accesses.
+ */
+bool
+IsConflictLogTableNamespace(Oid namespaceId)
+{
+ return namespaceId == PG_CONFLICT_NAMESPACE;
+}
/*
* IsReservedName
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 88087654de9..5d07dcc1c5b 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -305,23 +305,31 @@ heap_create(const char *relname,
Assert(OidIsValid(relid));
/*
- * Don't allow creating relations in pg_catalog directly, even though it
- * is allowed to move user defined relations there. Semantics with search
- * paths including pg_catalog are too confusing for now.
+ * Don't allow creating relations in pg_catalog or pg_conflict directly,
+ * even though it is allowed to move user defined relations there. Semantics
+ * with search paths including pg_catalog are too confusing for now.
*
* But allow creating indexes on relations in pg_catalog even if
* allow_system_table_mods = off, upper layers already guarantee it's on a
* user defined relation, not a system one.
*/
- if (!allow_system_table_mods &&
- ((IsCatalogNamespace(relnamespace) && relkind != RELKIND_INDEX) ||
- IsToastNamespace(relnamespace)) &&
- IsNormalProcessingMode())
- ereport(ERROR,
- (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
- errmsg("permission denied to create \"%s.%s\"",
- get_namespace_name(relnamespace), relname),
- errdetail("System catalog modifications are currently disallowed.")));
+ if (!allow_system_table_mods && IsNormalProcessingMode())
+ {
+ if ((IsCatalogNamespace(relnamespace) && relkind != RELKIND_INDEX) ||
+ IsToastNamespace(relnamespace))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied to create \"%s\"",
+ get_qualified_objname(relnamespace, relname)),
+ errdetail("System catalog modifications are currently disallowed.")));
+
+ if (IsConflictLogTableNamespace(relnamespace))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied to create \"%s\"",
+ get_qualified_objname(relnamespace, relname)),
+ errdetail("Conflict schema modifications are currently disallowed.")));
+ }
*relfrozenxid = InvalidTransactionId;
*relminmxid = InvalidMultiXactId;
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 56b87d878e8..5bd9be3983e 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -3523,9 +3523,8 @@ LookupCreationNamespace(const char *nspname)
/*
* Common checks on switching namespaces.
*
- * We complain if either the old or new namespaces is a temporary schema
- * (or temporary toast schema), or if either the old or new namespaces is the
- * TOAST schema.
+ * We complain if either the old or new namespaces is a temporary schema,
+ * temporary toast schema, the TOAST schema, or the conflict schema.
*/
void
CheckSetNamespace(Oid oldNspOid, Oid nspOid)
@@ -3541,6 +3540,12 @@ CheckSetNamespace(Oid oldNspOid, Oid nspOid)
ereport(ERROR,
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot move objects into or out of TOAST schema")));
+
+ /* similarly for conflict schema */
+ if (nspOid == PG_CONFLICT_NAMESPACE || oldNspOid == PG_CONFLICT_NAMESPACE)
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot move objects into or out of the conflict schema")));
}
/*
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 5c457d9aca8..1ec94c851b2 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -92,6 +92,13 @@ check_publication_add_relation(PublicationRelInfo *pri)
errmsg(errormsg, relname),
errdetail("This operation is not supported for system tables.")));
+ /* Can't be conflict log table */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(targetrel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg(errormsg, relname),
+ errdetail("This operation is not supported for conflict log tables.")));
+
/* UNLOGGED and TEMP relations cannot be part of publication. */
if (targetrel->rd_rel->relpersistence == RELPERSISTENCE_TEMP)
ereport(ERROR,
@@ -113,7 +120,8 @@ static void
check_publication_add_schema(Oid schemaid)
{
/* Can't be system namespace */
- if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid))
+ if (IsCatalogNamespace(schemaid) || IsToastNamespace(schemaid) ||
+ IsConflictLogTableNamespace(schemaid))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("cannot add schema \"%s\" to publication",
@@ -148,7 +156,8 @@ check_publication_add_schema(Oid schemaid)
* is really inadequate for that, since the information_schema could be
* dropped and reloaded and then it'll be considered publishable. The best
* long-term solution may be to add a "relispublishable" bool to pg_class,
- * and depend on that instead of OID checks.
+ * and depend on that instead of OID checks. IsConflictLogTableClass()
+ * excludes tables in conflict schema.
*/
static bool
is_publishable_class(Oid relid, Form_pg_class reltuple)
@@ -157,6 +166,7 @@ is_publishable_class(Oid relid, Form_pg_class reltuple)
reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
reltuple->relkind == RELKIND_SEQUENCE) &&
!IsCatalogRelationOid(relid) &&
+ !IsConflictLogTableClass(reltuple) &&
reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
relid >= FirstNormalObjectId;
}
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 1f1fdc75af6..809818af9ea 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -118,6 +118,7 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
sub->retaindeadtuples = subform->subretaindeadtuples;
sub->maxretention = subform->submaxretention;
sub->retentionactive = subform->subretentionactive;
+ sub->conflictlogrelid = subform->subconflictlogrelid;
/* Get conninfo */
if (OidIsValid(subform->subserver))
@@ -187,6 +188,12 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
Anum_pg_subscription_suborigin);
sub->origin = TextDatumGetCString(datum);
+ /* Get conflict log destination */
+ datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
+ tup,
+ Anum_pg_subscription_subconflictlogdest);
+ sub->conflictlogdest = TextDatumGetCString(datum);
+
/* Is the subscription owner a superuser? */
sub->ownersuperuser = superuser_arg(sub->owner);
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 8f129baec90..5967762e603 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1527,8 +1527,8 @@ GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner, subfailover,
subretaindeadtuples, submaxretention, subretentionactive,
- subserver, subslotname, subsynccommit, subwalrcvtimeout,
- subpublications, suborigin)
+ subserver, subconflictlogrelid, subconflictlogdest, subslotname,
+ subsynccommit, subwalrcvtimeout, subpublications, suborigin)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 523959ba0ce..7d09b80ba5f 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -28,6 +28,7 @@
#include "catalog/pg_authid_d.h"
#include "catalog/pg_database_d.h"
#include "catalog/pg_foreign_server.h"
+#include "catalog/pg_namespace.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
@@ -35,6 +36,7 @@
#include "commands/defrem.h"
#include "commands/event_trigger.h"
#include "commands/subscriptioncmds.h"
+#include "commands/tablecmds.h"
#include "executor/executor.h"
#include "foreign/foreign.h"
#include "miscadmin.h"
@@ -79,6 +81,7 @@
#define SUBOPT_WAL_RECEIVER_TIMEOUT 0x00010000
#define SUBOPT_LSN 0x00020000
#define SUBOPT_ORIGIN 0x00040000
+#define SUBOPT_CONFLICT_LOG_DEST 0x00080000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -107,6 +110,7 @@ typedef struct SubOpts
bool retaindeadtuples;
int32 maxretention;
char *origin;
+ ConflictLogDest conflictlogdest;
XLogRecPtr lsn;
char *wal_receiver_timeout;
} SubOpts;
@@ -140,7 +144,12 @@ static List *merge_publications(List *oldpublist, List *newpublist, bool addpub,
static void ReportSlotConnectionError(List *rstates, Oid subid, char *slotname, char *err);
static void CheckAlterSubOption(Subscription *sub, const char *option,
bool slot_needs_update, bool isTopLevel);
-
+static bool alter_sub_conflictlogdestination(Subscription *sub,
+ ConflictLogDest oldlogdest,
+ ConflictLogDest newlogdest,
+ Oid *conflicttablerelid);
+static void drop_sub_conflict_log_table(Oid subid, char *subname,
+ Oid subconflictlogrelid);
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -196,6 +205,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->maxretention = 0;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_CONFLICT_LOG_DEST))
+ opts->conflictlogdest = CONFLICT_LOG_DEST_LOG;
/* Parse options */
foreach(lc, stmt_options)
@@ -431,6 +442,18 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
PGC_BACKEND, PGC_S_TEST, GUC_ACTION_SET,
false, 0, false);
}
+ else if (IsSet(supported_opts, SUBOPT_CONFLICT_LOG_DEST) &&
+ strcmp(defel->defname, "conflict_log_destination") == 0)
+ {
+ char *val;
+
+ if (IsSet(opts->specified_opts, SUBOPT_CONFLICT_LOG_DEST))
+ errorConflictingDefElem(defel, pstate);
+
+ val = defGetString(defel);
+ opts->conflictlogdest = GetConflictLogDest(val);
+ opts->specified_opts |= SUBOPT_CONFLICT_LOG_DEST;
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -629,6 +652,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
uint32 supported_opts;
SubOpts opts = {0};
AclResult aclresult;
+ Oid logrelid = InvalidOid;
/*
* Parse and check options.
@@ -643,7 +667,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
SUBOPT_RETAIN_DEAD_TUPLES |
SUBOPT_MAX_RETENTION_DURATION |
- SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN);
+ SUBOPT_WAL_RECEIVER_TIMEOUT | SUBOPT_ORIGIN |
+ SUBOPT_CONFLICT_LOG_DEST);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -817,6 +842,20 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+
+ /*
+ * We create the conflict log table here, if required, so that its relation
+ * OID can be stored when inserting the pg_subscription tuple below.
+ */
+ if (CONFLICTS_LOGGED_TO_TABLE(opts.conflictlogdest))
+ logrelid = create_conflict_log_table(subid, stmt->subname, owner);
+
+ /* Store table OID in the catalog. */
+ values[Anum_pg_subscription_subconflictlogrelid - 1] =
+ ObjectIdGetDatum(logrelid);
+
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
/* Insert tuple into catalog. */
@@ -837,6 +876,25 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
}
+ /*
+ * Establish an internal dependency between the conflict log table and
+ * the subscription.
+ *
+ * We use DEPENDENCY_INTERNAL to signify that the table's lifecycle is
+ * strictly tied to the subscription, similar to how a TOAST table relates
+ * to its main table or a sequence relates to an identity column.
+ *
+ * This ensures the conflict log table is automatically reaped during a
+ * DROP SUBSCRIPTION via performDeletion().
+ */
+ if (OidIsValid(logrelid))
+ {
+ ObjectAddress cltaddr;
+
+ ObjectAddressSet(cltaddr, RelationRelationId, logrelid);
+ recordDependencyOn(&cltaddr, &myself, DEPENDENCY_INTERNAL);
+ }
+
/*
* A replication origin is currently created for all subscriptions,
* including those that only contain sequences or are otherwise empty.
@@ -1406,6 +1464,71 @@ CheckAlterSubOption(Subscription *sub, const char *option,
}
}
+/*
+ * alter_sub_conflictlogdestination
+ *
+ * When the subscription's 'conflict_log_destination' is changed, update the
+ * conflict log table if required.
+ *
+ * If the new destination no longer requires a conflict log table, the existing
+ * conflict log table associated with the subscription is removed via internal
+ * dependency cleanup to prevent orphaned relations.
+ *
+ * On success, *conflicttablerelid is set to the OID of the conflict log table
+ * that was created or validated, or to InvalidOid if no table is required.
+ *
+ * Returns true if the subscription's conflict log table reference must be
+ * updated as a result of the destination change; false otherwise.
+ */
+static bool
+alter_sub_conflictlogdestination(Subscription *sub, ConflictLogDest oldlogdest,
+ ConflictLogDest newlogdest,
+ Oid *conflicttablerelid)
+{
+ bool want_table;
+ bool has_oldtable;
+ bool update_relid = false;
+ Oid relid = InvalidOid;
+
+ want_table = CONFLICTS_LOGGED_TO_TABLE(newlogdest);
+ has_oldtable = CONFLICTS_LOGGED_TO_TABLE(oldlogdest);
+
+ if (has_oldtable)
+ {
+ /* There is a conflict log table already. */
+ if (!want_table)
+ {
+ drop_sub_conflict_log_table(sub->oid, sub->name,
+ sub->conflictlogrelid);
+ update_relid = true;
+ }
+ }
+ else
+ {
+ /* There was no previous conflict log table. */
+ if (want_table)
+ {
+ ObjectAddress cltaddr;
+ ObjectAddress subobj;
+
+ relid = create_conflict_log_table(sub->oid, sub->name, sub->owner);
+ update_relid = true;
+
+ /*
+ * Establish an internal dependency between the conflict log table
+ * and the subscription. For details refer comments in
+ * CreateSubscription function.
+ */
+ ObjectAddressSet(cltaddr, RelationRelationId, relid);
+ ObjectAddressSet(subobj, SubscriptionRelationId, sub->oid);
+ recordDependencyOn(&cltaddr, &subobj, DEPENDENCY_INTERNAL);
+ }
+ }
+
+ *conflicttablerelid = relid;
+ return update_relid;
+}
+
/*
* Alter the existing subscription.
*/
@@ -1501,7 +1624,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
SUBOPT_RETAIN_DEAD_TUPLES |
SUBOPT_MAX_RETENTION_DURATION |
SUBOPT_WAL_RECEIVER_TIMEOUT |
- SUBOPT_ORIGIN);
+ SUBOPT_ORIGIN |
+ SUBOPT_CONFLICT_LOG_DEST);
parse_subscription_options(pstate, stmt->options,
supported_opts, &opts);
@@ -1763,6 +1887,34 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_CONFLICT_LOG_DEST))
+ {
+ ConflictLogDest old_dest =
+ GetConflictLogDest(sub->conflictlogdest);
+
+ if (opts.conflictlogdest != old_dest)
+ {
+ bool update_relid;
+ Oid relid = InvalidOid;
+
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+ replaces[Anum_pg_subscription_subconflictlogdest - 1] = true;
+
+ update_relid = alter_sub_conflictlogdestination(sub,
+ old_dest,
+ opts.conflictlogdest,
+ &relid);
+ if (update_relid)
+ {
+ values[Anum_pg_subscription_subconflictlogrelid - 1] =
+ ObjectIdGetDatum(relid);
+ replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
+ true;
+ }
+ }
+ }
+
update_tuple = true;
break;
}
@@ -2178,6 +2330,51 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
return myself;
}
+/*
+ * Drop subscription's conflict log table
+ *
+ * The conflict log table is registered as an internal dependency of the
+ * subscription. This function removes the dependency by performing a
+ * cascading deletion on the subscription object, which in turn drops the
+ * associated conflict log table.
+ *
+ * This is used to clean up conflict log tables that are no longer required,
+ * preventing accumulation of stale or orphaned relations.
+ *
+ * NOTE:
+ * Only conflict log tables are currently managed via this internal dependency
+ * mechanism.
+ */
+static void
+drop_sub_conflict_log_table(Oid subid, char *subname, Oid subconflictlogrelid)
+{
+ /* Drop any dependent conflict log table */
+ if (OidIsValid(subconflictlogrelid))
+ {
+ ObjectAddress object;
+ char *conflictrelname;
+
+ conflictrelname = get_rel_name(subconflictlogrelid);
+ if (conflictrelname == NULL)
+ elog(ERROR, "cache lookup failed for relation %u",
+ subconflictlogrelid);
+
+ /*
+ * By using PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the
+ * conflict log table is deleted while the subscription remains.
+ */
+ ObjectAddressSet(object, SubscriptionRelationId, subid);
+ performDeletion(&object, DROP_CASCADE,
+ PERFORM_DELETION_INTERNAL |
+ PERFORM_DELETION_SKIP_ORIGINAL);
+
+ ereport(NOTICE,
+ errmsg("dropped conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, conflictrelname),
+ subname));
+ }
+}
+
/*
* Drop a subscription
*/
@@ -2189,6 +2386,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
HeapTuple tup;
Oid subid;
Oid subowner;
+ Oid subconflictlogrelid;
Datum datum;
bool isnull;
char *subname;
@@ -2234,6 +2432,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
form = (Form_pg_subscription) GETSTRUCT(tup);
subid = form->oid;
subowner = form->subowner;
+ subconflictlogrelid = form->subconflictlogrelid;
must_use_password = !superuser_arg(subowner) && form->subpasswordrequired;
/* must be owner */
@@ -2388,6 +2587,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
deleteDependencyRecordsFor(SubscriptionRelationId, subid, false);
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
+ drop_sub_conflict_log_table(subid, subname, subconflictlogrelid);
+
/* Remove any associated relation synchronization states. */
RemoveSubscriptionRel(subid, InvalidOid);
@@ -2621,6 +2822,11 @@ AlterSubscriptionOwner_internal(Relation rel, HeapTuple tup, Oid newOwnerId)
form->subowner = newOwnerId;
CatalogTupleUpdate(rel, &tup->t_self, tup);
+ /* Update owner of the conflict log table if it exists. */
+ if (OidIsValid(form->subconflictlogrelid))
+ ATExecChangeOwner(form->subconflictlogrelid, newOwnerId, true,
+ AccessExclusiveLock);
+
/* Update owner dependency reference */
changeDependencyOnOwner(SubscriptionRelationId,
form->oid,
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index a1845240a98..0a0ca2b850e 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -2457,9 +2457,11 @@ truncate_check_rel(Oid relid, Form_pg_class reltuple)
* pg_largeobject and pg_largeobject_metadata to be truncated as part of
* pg_upgrade, because we need to change its relfilenode to match the old
* cluster, and allowing a TRUNCATE command to be executed is the easiest
- * way of doing that.
+ * way of doing that. We also allow TRUNCATE on the conflict log tables,
+ * to permit users to manually prune conflict data to manage disk space.
*/
- if (!allowSystemTableMods && IsSystemClass(relid, reltuple)
+ if (!allowSystemTableMods && IsSystemClass(relid, reltuple) &&
+ !IsConflictLogTableClass(reltuple)
&& (!IsBinaryUpgrade ||
(relid != LargeObjectRelationId &&
relid != LargeObjectMetadataRelationId)))
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 4b30f768680..f87ef65f343 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1187,6 +1187,24 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation,
RelationGetRelationName(resultRel))));
break;
}
+
+ /*
+ * Conflict log tables are managed by the system to record logical
+ * replication conflicts. We allow DELETE and TRUNCATE to permit users to
+ * manually prune these logs, but manual data insertion or modification
+ * (INSERT, UPDATE, MERGE) is prohibited to maintain the integrity of the
+ * system-generated logs.
+ *
+ * Since TRUNCATE is handled as a separate utility command, we only need
+ * to explicitly permit CMD_DELETE here.
+ */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(resultRel)) &&
+ operation != CMD_DELETE)
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot modify or insert data into conflict log table \"%s\"",
+ RelationGetRelationName(resultRel)),
+ errdetail("Conflict log tables are system-managed and only support cleanup via DELETE or TRUNCATE.")));
}
/*
@@ -1258,6 +1276,16 @@ CheckValidRowMarkRel(Relation rel, RowMarkType markType)
RelationGetRelationName(rel))));
break;
}
+
+ /*
+ * Conflict log tables are managed by the system to record logical
+ * replication conflicts.
+ */
+ if (IsConflictLogTableNamespace(RelationGetNamespace(rel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot lock rows in the conflict log table \"%s\"",
+ RelationGetRelationName(rel))));
}
/*
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 1f8d67fdd90..4a647277d1f 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -17,6 +17,10 @@
#include "access/commit_ts.h"
#include "access/genam.h"
#include "access/tableam.h"
+#include "catalog/heap.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_namespace.h"
+#include "catalog/toasting.h"
#include "executor/executor.h"
#include "pgstat.h"
#include "replication/conflict.h"
@@ -24,6 +28,50 @@
#include "storage/lmgr.h"
#include "utils/lsyscache.h"
+/*
+ * String representations for the supported conflict logging destinations.
+ */
+const char *const ConflictLogDestNames[] = {
+ [CONFLICT_LOG_DEST_LOG] = "log",
+ [CONFLICT_LOG_DEST_TABLE] = "table",
+ [CONFLICT_LOG_DEST_ALL] = "all"
+};
+
+StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
+ "ConflictLogDestNames length mismatch");
+
+
+/* Structure to hold metadata for one column of the conflict log table */
+typedef struct ConflictLogColumnDef
+{
+ const char *attname; /* Column name */
+ Oid atttypid; /* Data type OID */
+} ConflictLogColumnDef;
+
+/*
+ * Schema definition for conflict log tables.
+ *
+ * Defines the fixed schema of the per-subscription conflict log table created
+ * in the pg_conflict namespace. Each entry specifies the column name and its
+ * type OID; the table is created in this column order by
+ * create_conflict_log_table().
+ */
+static const ConflictLogColumnDef ConflictLogSchema[] = {
+ { .attname = "relid", .atttypid = OIDOID },
+ { .attname = "schemaname", .atttypid = TEXTOID },
+ { .attname = "relname", .atttypid = TEXTOID },
+ { .attname = "conflict_type", .atttypid = TEXTOID },
+ { .attname = "remote_xid", .atttypid = XIDOID },
+ { .attname = "remote_commit_lsn",.atttypid = LSNOID },
+ { .attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID },
+ { .attname = "remote_origin", .atttypid = TEXTOID },
+ { .attname = "remote_tuple", .atttypid = JSONOID },
+ { .attname = "replica_identity", .atttypid = JSONOID },
+ { .attname = "local_conflicts", .atttypid = JSONARRAYOID }
+};
+
+#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
+
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
@@ -54,6 +102,133 @@ static void get_tuple_desc(EState *estate, ResultRelInfo *relinfo,
static char *build_index_value_desc(EState *estate, Relation localrel,
TupleTableSlot *slot, Oid indexoid);
+/*
+ * Builds the TupleDesc for the conflict log table.
+ */
+static TupleDesc
+create_conflict_log_table_tupdesc(void)
+{
+ TupleDesc tupdesc;
+
+ tupdesc = CreateTemplateTupleDesc(NUM_CONFLICT_ATTRS);
+
+ for (int i = 0; i < NUM_CONFLICT_ATTRS; i++)
+ TupleDescInitEntry(tupdesc, i + 1,
+ ConflictLogSchema[i].attname,
+ ConflictLogSchema[i].atttypid,
+ -1, 0);
+
+ TupleDescFinalize(tupdesc);
+
+ return tupdesc;
+}
+
+/*
+ * Create a structured conflict log table for a subscription.
+ *
+ * The table is created within the system-managed 'pg_conflict' namespace to
+ * prevent users from manually dropping or altering it. This also prevents
+ * accidental name collisions with user-created tables with the same name.
+ *
+ * The table name is generated automatically using the subscription's OID
+ * (e.g., "pg_conflict_log_<subid>") to ensure uniqueness within the
+ * cluster and to avoid collisions during subscription renames.
+ */
+Oid
+create_conflict_log_table(Oid subid, char *subname, Oid subowner)
+{
+ TupleDesc tupdesc;
+ Oid relid;
+ char relname[NAMEDATALEN];
+
+ snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
+
+ /*
+ * Check for an existing table with the same name in the pg_conflict namespace.
+ * A collision should not occur under normal operation, but we must handle cases
+ * where a table has been created manually when allow_system_tables_mods is
+ * ON.
+ */
+ if (OidIsValid(get_relname_relid(relname, PG_CONFLICT_NAMESPACE)))
+ ereport(ERROR,
+ (errcode(ERRCODE_DUPLICATE_TABLE),
+ errmsg("conflict log table pg_conflict.\"%s\" already exists", relname),
+ errhint("To proceed, drop the existing table and retry.")));
+
+ /* Build the tuple descriptor for the new table. */
+ tupdesc = create_conflict_log_table_tupdesc();
+
+ /* Create conflict log table. */
+ relid = heap_create_with_catalog(relname,
+ PG_CONFLICT_NAMESPACE,
+ 0, /* tablespace */
+ InvalidOid, /* relid */
+ InvalidOid, /* reltypeid */
+ InvalidOid, /* reloftypeid */
+ subowner,
+ HEAP_TABLE_AM_OID,
+ tupdesc,
+ NIL,
+ RELKIND_RELATION,
+ RELPERSISTENCE_PERMANENT,
+ false, /* shared_relation */
+ false, /* mapped_relation */
+ ONCOMMIT_NOOP,
+ (Datum) 0, /* reloptions */
+ false, /* use_user_acl */
+ true, /* allow_system_table_mods */
+ true, /* is_internal */
+ InvalidOid, /* relrewrite */
+ NULL); /* typaddress */
+ Assert(OidIsValid(relid));
+
+ /* Release tuple descriptor memory. */
+ FreeTupleDesc(tupdesc);
+
+ /*
+ * We must bump the command counter to make the newly-created relation
+ * tuple visible for opening.
+ */
+ CommandCounterIncrement();
+
+ /*
+ * Create a TOAST table for the conflict log to support out-of-line storage
+ * of large JSON data.
+ */
+ NewRelationCreateToastTable(relid, (Datum) 0);
+
+ ereport(NOTICE,
+ (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
+ subname)));
+
+ return relid;
+}
+
+/*
+ * Convert the string representation of a conflict logging destination to its
+ * corresponding enum value.
+ */
+ConflictLogDest
+GetConflictLogDest(const char *dest)
+{
+ /* Empty string or NULL defaults to LOG. */
+ if (dest == NULL || dest[0] == '\0' || pg_strcasecmp(dest, "log") == 0)
+ return CONFLICT_LOG_DEST_LOG;
+
+ if (pg_strcasecmp(dest, "table") == 0)
+ return CONFLICT_LOG_DEST_TABLE;
+
+ if (pg_strcasecmp(dest, "all") == 0)
+ return CONFLICT_LOG_DEST_ALL;
+
+ /* Unrecognized string. */
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("unrecognized conflict_log_destination value: \"%s\"", dest),
+ errhint("Valid values are \"log\", \"table\", and \"all\".")));
+}
+
/*
* Get the xmin and commit timestamp data (origin and timestamp) associated
* with the provided local row.
diff --git a/src/backend/utils/cache/lsyscache.c b/src/backend/utils/cache/lsyscache.c
index 036086057d7..b60bd8f6c0e 100644
--- a/src/backend/utils/cache/lsyscache.c
+++ b/src/backend/utils/cache/lsyscache.c
@@ -3634,7 +3634,7 @@ get_namespace_name_or_temp(Oid nspid)
* object for the given namespace ID and object name.
*/
char *
-get_qualified_objname(Oid nspid, char *objname)
+get_qualified_objname(Oid nspid, const char *objname)
{
char *nspname;
char *result;
diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c
index 14cb79c26be..803ca4112d4 100644
--- a/src/bin/initdb/initdb.c
+++ b/src/bin/initdb/initdb.c
@@ -1839,6 +1839,12 @@ setup_privileges(FILE *cmdfd)
" AND relacl IS NULL;\n\n",
escape_quotes(username));
PG_CMD_PUTS("GRANT USAGE ON SCHEMA pg_catalog, public TO PUBLIC;\n\n");
+
+ /*
+ * Allow non-superuser subscription owners to access their associated
+ * conflict log tables in the pg_conflict schema.
+ */
+ PG_CMD_PUTS("GRANT USAGE ON SCHEMA pg_conflict TO PUBLIC;\n\n");
PG_CMD_PUTS("REVOKE ALL ON pg_largeobject FROM PUBLIC;\n\n");
PG_CMD_PUTS("INSERT INTO pg_init_privs "
" (objoid, classoid, objsubid, initprivs, privtype)"
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index de547a8cb37..66ee1b6c2a4 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2376,8 +2376,8 @@ match_previous_words(int pattern_id,
COMPLETE_WITH("(", "PUBLICATION");
/* ALTER SUBSCRIPTION <name> SET ( */
else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN, "SET", "("))
- COMPLETE_WITH("binary", "disable_on_error", "failover",
- "max_retention_duration", "origin",
+ COMPLETE_WITH("binary", "conflict_log_destination", "disable_on_error",
+ "failover", "max_retention_duration", "origin",
"password_required", "retain_dead_tuples",
"run_as_owner", "slot_name", "streaming",
"synchronous_commit", "two_phase");
@@ -3960,8 +3960,8 @@ match_previous_words(int pattern_id,
COMPLETE_WITH("WITH (");
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "WITH", "("))
- COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "failover",
+ COMPLETE_WITH("binary", "conflict_log_destination", "connect", "copy_data",
+ "create_slot", "disable_on_error", "enabled", "failover",
"max_retention_duration", "origin",
"password_required", "retain_dead_tuples",
"run_as_owner", "slot_name", "streaming",
diff --git a/src/include/catalog/catalog.h b/src/include/catalog/catalog.h
index a9d6e8ea986..feab3982cf5 100644
--- a/src/include/catalog/catalog.h
+++ b/src/include/catalog/catalog.h
@@ -25,6 +25,7 @@ extern bool IsInplaceUpdateRelation(Relation relation);
extern bool IsSystemClass(Oid relid, Form_pg_class reltuple);
extern bool IsToastClass(Form_pg_class reltuple);
+extern bool IsConflictLogTableClass(Form_pg_class reltuple);
extern bool IsCatalogRelationOid(Oid relid);
extern bool IsCatalogTextUniqueIndexOid(Oid relid);
@@ -32,6 +33,7 @@ extern bool IsInplaceUpdateOid(Oid relid);
extern bool IsCatalogNamespace(Oid namespaceId);
extern bool IsToastNamespace(Oid namespaceId);
+extern bool IsConflictLogTableNamespace(Oid namespaceId);
extern bool IsReservedName(const char *name);
diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h
index 87007e725c1..09874a4a7d6 100644
--- a/src/include/catalog/catversion.h
+++ b/src/include/catalog/catversion.h
@@ -57,6 +57,6 @@
*/
/* yyyymmddN */
-#define CATALOG_VERSION_NO 202606051
+#define CATALOG_VERSION_NO 202606081
#endif
diff --git a/src/include/catalog/pg_namespace.dat b/src/include/catalog/pg_namespace.dat
index 3075e142c73..b45cb9383a8 100644
--- a/src/include/catalog/pg_namespace.dat
+++ b/src/include/catalog/pg_namespace.dat
@@ -18,6 +18,9 @@
{ oid => '99', oid_symbol => 'PG_TOAST_NAMESPACE',
descr => 'reserved schema for TOAST tables',
nspname => 'pg_toast', nspacl => '_null_' },
+{ oid => '1382', oid_symbol => 'PG_CONFLICT_NAMESPACE',
+ descr => 'reserved schema for subscription-specific conflict log tables',
+ nspname => 'pg_conflict', nspacl => '_null_' },
# update dumpNamespace() if changing this descr
{ oid => '2200', oid_symbol => 'PG_PUBLIC_NAMESPACE',
descr => 'standard public schema',
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index a6a2ad1e49c..89d2300abe1 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -95,7 +95,16 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
Oid subserver BKI_LOOKUP_OPT(pg_foreign_server); /* If connection uses
* server */
+ Oid subconflictlogrelid; /* Relid of the conflict log table. */
#ifdef CATALOG_VARLEN /* variable-length fields start here */
+ /*
+ * Strategy for logging replication conflicts:
+ * 'log' - server log only,
+ * 'table' - conflict log table only,
+ * 'all' - both log and table.
+ */
+ text subconflictlogdest BKI_FORCE_NOT_NULL;
+
/* Connection string to the publisher */
text subconninfo; /* Set if connecting with connection string */
@@ -164,6 +173,7 @@ typedef struct Subscription
* and the retention duration has not
* exceeded max_retention_duration, when
* defined */
+ Oid conflictlogrelid; /* conflict log table Oid */
char *conninfo; /* Connection string to the publisher */
char *slotname; /* Name of the replication slot */
char *synccommit; /* Synchronous commit setting for worker */
@@ -171,6 +181,7 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
char *origin; /* Only publish data originating from the
* specified origin */
+ char *conflictlogdest; /* Conflict log destination */
} Subscription;
#ifdef EXPOSE_TO_CLIENT_CODE
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 2d9dbcf4d0d..a017e1e6cb5 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -10,6 +10,7 @@
#define CONFLICT_H
#include "access/xlogdefs.h"
+#include "catalog/pg_type.h"
#include "datatype/timestamp.h"
#include "nodes/pg_list.h"
@@ -79,6 +80,32 @@ typedef struct ConflictTupleInfo
* conflicting local row occurred */
} ConflictTupleInfo;
+/*
+ * Defines where logical replication conflict details are recorded.
+ *
+ * While stored as a text-based array/string in
+ * pg_subscription.subconflictlogdest for user readability and extensibility,
+ * we map these to an internal enum to allow for efficient checks.
+ */
+typedef enum ConflictLogDest
+{
+ CONFLICT_LOG_DEST_LOG = 0, /* Emit to server logs */
+ CONFLICT_LOG_DEST_TABLE, /* Insert into the conflict log table */
+ CONFLICT_LOG_DEST_ALL /* Both log and table */
+} ConflictLogDest;
+
+#define CONFLICTS_LOGGED_TO_TABLE(dest) \
+ ((dest == CONFLICT_LOG_DEST_TABLE) || (dest == CONFLICT_LOG_DEST_ALL))
+#define CONFLICTS_LOGGED_TO_LOG(dest) \
+ ((dest == CONFLICT_LOG_DEST_LOG) || (dest == CONFLICT_LOG_DEST_ALL))
+
+/*
+ * Array mapping for converting internal enum to string.
+ */
+extern PGDLLIMPORT const char *const ConflictLogDestNames[];
+
+extern Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner);
+extern ConflictLogDest GetConflictLogDest(const char *dest);
extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
TransactionId *xmin,
ReplOriginId *localorigin,
diff --git a/src/include/utils/lsyscache.h b/src/include/utils/lsyscache.h
index 8545e67a632..60b2d5b6e3e 100644
--- a/src/include/utils/lsyscache.h
+++ b/src/include/utils/lsyscache.h
@@ -200,7 +200,7 @@ extern bool get_attstatsslot(AttStatsSlot *sslot, HeapTuple statstuple,
extern void free_attstatsslot(AttStatsSlot *sslot);
extern char *get_namespace_name(Oid nspid);
extern char *get_namespace_name_or_temp(Oid nspid);
-extern char *get_qualified_objname(Oid nspid, char *objname);
+extern char *get_qualified_objname(Oid nspid, const char *objname);
extern Oid get_range_subtype(Oid rangeOid);
extern Oid get_range_collation(Oid rangeOid);
extern Oid get_range_constructor2(Oid rangeOid);
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 8481056a702..9a693ed010a 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -591,6 +591,314 @@ COMMIT;
-- ok, owning it is enough for this stuff
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+--
+-- CONFLICT LOG DESTINATION TESTS
+--
+SET SESSION AUTHORIZATION 'regress_subscription_user';
+SET client_min_messages = WARNING;
+-- fail - unrecognized parameter value
+CREATE SUBSCRIPTION regress_conflict_fail CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'invalid');
+ERROR: unrecognized conflict_log_destination value: "invalid"
+HINT: Valid values are "log", "table", and "all".
+-- verify subconflictlogdest is 'log' and subconflictlogrelid is 0 (InvalidOid) for default case
+CREATE SUBSCRIPTION regress_conflict_log_default CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+SELECT subname, subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_log_default';
+ subname | subconflictlogdest | subconflictlogrelid
+------------------------------+--------------------+---------------------
+ regress_conflict_log_default | log | 0
+(1 row)
+
+-- verify empty string defaults to 'log'
+CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = '');
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+SELECT subname, subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_empty_str';
+ subname | subconflictlogdest | subconflictlogrelid
+----------------------------+--------------------+---------------------
+ regress_conflict_empty_str | log | 0
+(1 row)
+
+-- this should generate an conflict log table named pg_conflict_log_$subid$
+CREATE SUBSCRIPTION regress_conflict_test1 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table');
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+-- check metadata in pg_subscription: destination should be 'table' and subconflictlogrelid valid
+SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid
+FROM pg_subscription WHERE subname = 'regress_conflict_test1';
+ subname | subconflictlogdest | has_relid
+------------------------+--------------------+-----------
+ regress_conflict_test1 | table | t
+(1 row)
+
+-- verify the physical table exists, its OID matches subconflictlogrelid,
+-- and it is located in the 'pg_conflict' namespace
+SELECT n.nspname, (c.oid = s.subconflictlogrelid) AS "oid_matches"
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+JOIN pg_namespace n ON c.relnamespace = n.oid
+WHERE s.subname = 'regress_conflict_test1';
+ nspname | oid_matches
+-------------+-------------
+ pg_conflict | t
+(1 row)
+
+-- check if the conflict log table has the correct schema
+SELECT a.attnum, a.attname
+FROM pg_attribute a
+JOIN pg_class c ON a.attrelid = c.oid
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
+ ORDER BY a.attnum;
+ attnum | attname
+--------+-------------------
+ 1 | relid
+ 2 | schemaname
+ 3 | relname
+ 4 | conflict_type
+ 5 | remote_xid
+ 6 | remote_commit_lsn
+ 7 | remote_commit_ts
+ 8 | remote_origin
+ 9 | remote_tuple
+ 10 | replica_identity
+ 11 | local_conflicts
+(11 rows)
+
+-- Changing the subscription owner should also update the owner
+-- of the associated conflict log table.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user2;
+SELECT pg_catalog.pg_get_userbyid(c.relowner) AS owner
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1';
+ owner
+----------------------------
+ regress_subscription_user2
+(1 row)
+
+-- Verify that a non-superuser subscription owner can truncate,
+-- delete from, and select from the associated conflict log table.
+SET ROLE 'regress_subscription_user2';
+SELECT format('%I.%I', n.nspname, c.relname) AS conflict_log_table
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_namespace n
+ ON n.oid = c.relnamespace
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1'
+\gset
+TRUNCATE TABLE :conflict_log_table;
+DELETE FROM :conflict_log_table;
+SELECT COUNT(*) FROM :conflict_log_table;
+ count
+-------
+ 0
+(1 row)
+
+RESET ROLE;
+-- Restore the original subscription owner.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
+--
+-- ALTER SUBSCRIPTION - conflict_log_destination state transitions
+--
+-- These tests verify the transition logic between different logging
+-- destinations, ensuring conflict log tables are created or dropped as
+-- expected
+--
+-- transition from 'log' to 'all'
+-- a new conflict log table should be created
+CREATE SUBSCRIPTION regress_conflict_test2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'log');
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'all');
+-- verify metadata after ALTER (destination should be 'all')
+SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+ subname | subconflictlogdest | has_relid
+------------------------+--------------------+-----------
+ regress_conflict_test2 | all | t
+(1 row)
+
+-- transition from 'all' to 'table'
+-- should NOT drop the table, only change destination string
+SELECT subconflictlogrelid AS old_relid FROM pg_subscription WHERE subname = 'regress_conflict_test2' \gset
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table');
+SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+ subconflictlogdest | relid_unchanged
+--------------------+-----------------
+ table | t
+(1 row)
+
+-- transition from 'table' to 'log'
+-- should drop the table and clear subconflictlogrelid
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log');
+SELECT subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+ subconflictlogdest | subconflictlogrelid
+--------------------+---------------------
+ log | 0
+(1 row)
+
+-- verify the physical table is gone
+SELECT count(*)
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test2';
+ count
+-------
+ 0
+(1 row)
+
+--
+-- PUBLICATION: Verify conflict log tables are not publishable
+--
+-- pg_relation_is_publishable should return false for conflict log tables to
+-- prevent them from being accidentally included in publications
+--
+SELECT n.nspname, pg_relation_is_publishable(c.oid)
+FROM pg_class c
+JOIN pg_namespace n ON c.relnamespace = n.oid
+JOIN pg_subscription s ON s.subconflictlogrelid = c.oid
+WHERE s.subname = 'regress_conflict_test1';
+ nspname | pg_relation_is_publishable
+-------------+----------------------------
+ pg_conflict | f
+(1 row)
+
+--
+-- Table Protection and Lifecycle Management
+--
+-- These tests verify that:
+-- Manual DROP TABLE is disallowed
+-- DROP SUBSCRIPTION automatically reaps the table
+--
+-- re-enable table logging for verification
+ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table');
+-- We use a DO block with dynamic SQL because the conflict log table
+-- name contains the subscription OID, which is non-deterministic. This
+-- approach allows us to attempt the DROP and capture the expected error
+-- without hard-coding a specific OID in the expected output
+-- fail - drop table not allowed due to internal dependency
+SET client_min_messages = NOTICE;
+DO $$
+BEGIN
+ EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_log_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1');
+EXCEPTION WHEN insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege';
+END $$;
+NOTICE: captured expected error: insufficient_privilege
+-- CLEANUP: DROP SUBSCRIPTION reaps the table
+ALTER SUBSCRIPTION regress_conflict_test1 DISABLE;
+ALTER SUBSCRIPTION regress_conflict_test1 SET (slot_name = NONE);
+-- Verify the table OID for reap check
+SELECT 'pg_conflict.pg_conflict_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+SET client_min_messages = WARNING;
+DROP SUBSCRIPTION regress_conflict_test1;
+-- should return NULL, meaning the conflict log table was reaped via dependency
+SELECT to_regclass(:'internal_tablename');
+ to_regclass
+-------------
+
+(1 row)
+
+--
+-- Additional Namespace and Table Protection Tests
+--
+-- Setup: Ensure we have a subscription with a conflict log table
+CREATE SUBSCRIPTION regress_conflict_protection_test CONNECTION 'dbname=regress_doesnotexist'
+ PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table');
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
+-- Trying to ALTER the conflict log table
+-- This should fail because the table is system-managed
+-- As mentioned in previous test cases, we use a DO block to hide dynamic OIDs
+SET client_min_messages = NOTICE;
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ RAISE NOTICE 'Attempting ALTER TABLE on conflict log table';
+ EXECUTE 'ALTER TABLE ' || tab_name || ' ADD COLUMN extra_info text';
+EXCEPTION WHEN insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege during ALTER';
+END $$;
+NOTICE: Attempting ALTER TABLE on conflict log table
+NOTICE: captured expected error: insufficient_privilege during ALTER
+-- Test Manual INSERT on conflict log table
+-- This should fail because the table is system-managed
+-- Hiding the OID in the error message by catching the exception
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during INSERT';
+END $$;
+NOTICE: captured expected error: wrong_object_type during INSERT
+-- Test Manual UPDATE on conflict log table
+-- This should fail because the table is system-managed
+-- Hiding the OID in the error message by catching the exception
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' ';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during UPDATE';
+END $$;
+NOTICE: captured expected error: wrong_object_type during UPDATE
+-- Trying to perform TRUNCATE/DELETE on the conflict log table
+-- This should be allowed so that user can perform cleanup
+SELECT 'pg_conflict.' || relname AS conflict_tab
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_protection_test' \gset
+TRUNCATE :conflict_tab;
+DELETE FROM :conflict_tab;
+-- Trying to create a new table manually in the pg_conflict namespace
+-- This should fail as the namespace is reserved for conflict logs tables
+CREATE TABLE pg_conflict.manual_table (id int);
+ERROR: permission denied to create "pg_conflict.manual_table"
+DETAIL: Conflict schema modifications are currently disallowed.
+-- Moving an existing table into the pg_conflict namespace
+-- Users should not be able to move their own tables within this namespace
+CREATE TABLE public.test_move (id int);
+ALTER TABLE public.test_move SET SCHEMA pg_conflict;
+ERROR: cannot move objects into or out of the conflict schema
+DROP TABLE public.test_move;
+SET client_min_messages = WARNING;
+-- Clean up remaining test subscription
+ALTER SUBSCRIPTION regress_conflict_log_default DISABLE;
+ALTER SUBSCRIPTION regress_conflict_log_default SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_log_default;
+ALTER SUBSCRIPTION regress_conflict_empty_str DISABLE;
+ALTER SUBSCRIPTION regress_conflict_empty_str SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_empty_str;
+ALTER SUBSCRIPTION regress_conflict_test2 DISABLE;
+ALTER SUBSCRIPTION regress_conflict_test2 SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_test2;
+ALTER SUBSCRIPTION regress_conflict_protection_test DISABLE;
+ALTER SUBSCRIPTION regress_conflict_protection_test SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_protection_test;
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 374fad6aa7b..f209eae881e 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -442,6 +442,262 @@ COMMIT;
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+--
+-- CONFLICT LOG DESTINATION TESTS
+--
+
+SET SESSION AUTHORIZATION 'regress_subscription_user';
+
+SET client_min_messages = WARNING;
+
+-- fail - unrecognized parameter value
+CREATE SUBSCRIPTION regress_conflict_fail CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'invalid');
+
+-- verify subconflictlogdest is 'log' and subconflictlogrelid is 0 (InvalidOid) for default case
+CREATE SUBSCRIPTION regress_conflict_log_default CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false);
+SELECT subname, subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_log_default';
+
+-- verify empty string defaults to 'log'
+CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = '');
+SELECT subname, subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_empty_str';
+
+-- this should generate an conflict log table named pg_conflict_log_$subid$
+CREATE SUBSCRIPTION regress_conflict_test1 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table');
+
+-- check metadata in pg_subscription: destination should be 'table' and subconflictlogrelid valid
+SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid
+FROM pg_subscription WHERE subname = 'regress_conflict_test1';
+
+-- verify the physical table exists, its OID matches subconflictlogrelid,
+-- and it is located in the 'pg_conflict' namespace
+SELECT n.nspname, (c.oid = s.subconflictlogrelid) AS "oid_matches"
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+JOIN pg_namespace n ON c.relnamespace = n.oid
+WHERE s.subname = 'regress_conflict_test1';
+
+-- check if the conflict log table has the correct schema
+SELECT a.attnum, a.attname
+FROM pg_attribute a
+JOIN pg_class c ON a.attrelid = c.oid
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
+ ORDER BY a.attnum;
+
+-- Changing the subscription owner should also update the owner
+-- of the associated conflict log table.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user2;
+SELECT pg_catalog.pg_get_userbyid(c.relowner) AS owner
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1';
+
+-- Verify that a non-superuser subscription owner can truncate,
+-- delete from, and select from the associated conflict log table.
+SET ROLE 'regress_subscription_user2';
+
+SELECT format('%I.%I', n.nspname, c.relname) AS conflict_log_table
+FROM pg_catalog.pg_class c
+JOIN pg_catalog.pg_namespace n
+ ON n.oid = c.relnamespace
+JOIN pg_catalog.pg_subscription s
+ ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test1'
+\gset
+
+TRUNCATE TABLE :conflict_log_table;
+DELETE FROM :conflict_log_table;
+SELECT COUNT(*) FROM :conflict_log_table;
+
+RESET ROLE;
+
+-- Restore the original subscription owner.
+ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
+
+--
+-- ALTER SUBSCRIPTION - conflict_log_destination state transitions
+--
+-- These tests verify the transition logic between different logging
+-- destinations, ensuring conflict log tables are created or dropped as
+-- expected
+--
+-- transition from 'log' to 'all'
+-- a new conflict log table should be created
+CREATE SUBSCRIPTION regress_conflict_test2 CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'log');
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'all');
+
+-- verify metadata after ALTER (destination should be 'all')
+SELECT subname, subconflictlogdest, subconflictlogrelid > 0 AS has_relid
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+
+-- transition from 'all' to 'table'
+-- should NOT drop the table, only change destination string
+SELECT subconflictlogrelid AS old_relid FROM pg_subscription WHERE subname = 'regress_conflict_test2' \gset
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'table');
+SELECT subconflictlogdest, subconflictlogrelid = :old_relid AS relid_unchanged
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+
+-- transition from 'table' to 'log'
+-- should drop the table and clear subconflictlogrelid
+ALTER SUBSCRIPTION regress_conflict_test2 SET (conflict_log_destination = 'log');
+SELECT subconflictlogdest, subconflictlogrelid
+FROM pg_subscription WHERE subname = 'regress_conflict_test2';
+
+-- verify the physical table is gone
+SELECT count(*)
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_test2';
+
+--
+-- PUBLICATION: Verify conflict log tables are not publishable
+--
+-- pg_relation_is_publishable should return false for conflict log tables to
+-- prevent them from being accidentally included in publications
+--
+SELECT n.nspname, pg_relation_is_publishable(c.oid)
+FROM pg_class c
+JOIN pg_namespace n ON c.relnamespace = n.oid
+JOIN pg_subscription s ON s.subconflictlogrelid = c.oid
+WHERE s.subname = 'regress_conflict_test1';
+
+--
+-- Table Protection and Lifecycle Management
+--
+-- These tests verify that:
+-- Manual DROP TABLE is disallowed
+-- DROP SUBSCRIPTION automatically reaps the table
+--
+-- re-enable table logging for verification
+ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table');
+
+-- We use a DO block with dynamic SQL because the conflict log table
+-- name contains the subscription OID, which is non-deterministic. This
+-- approach allows us to attempt the DROP and capture the expected error
+-- without hard-coding a specific OID in the expected output
+
+-- fail - drop table not allowed due to internal dependency
+SET client_min_messages = NOTICE;
+DO $$
+BEGIN
+ EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_log_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1');
+EXCEPTION WHEN insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege';
+END $$;
+
+-- CLEANUP: DROP SUBSCRIPTION reaps the table
+ALTER SUBSCRIPTION regress_conflict_test1 DISABLE;
+ALTER SUBSCRIPTION regress_conflict_test1 SET (slot_name = NONE);
+
+-- Verify the table OID for reap check
+SELECT 'pg_conflict.pg_conflict_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+
+SET client_min_messages = WARNING;
+DROP SUBSCRIPTION regress_conflict_test1;
+
+-- should return NULL, meaning the conflict log table was reaped via dependency
+SELECT to_regclass(:'internal_tablename');
+
+--
+-- Additional Namespace and Table Protection Tests
+--
+
+-- Setup: Ensure we have a subscription with a conflict log table
+CREATE SUBSCRIPTION regress_conflict_protection_test CONNECTION 'dbname=regress_doesnotexist'
+ PUBLICATION testpub WITH (connect = false, conflict_log_destination = 'table');
+
+-- Trying to ALTER the conflict log table
+-- This should fail because the table is system-managed
+-- As mentioned in previous test cases, we use a DO block to hide dynamic OIDs
+
+SET client_min_messages = NOTICE;
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ RAISE NOTICE 'Attempting ALTER TABLE on conflict log table';
+ EXECUTE 'ALTER TABLE ' || tab_name || ' ADD COLUMN extra_info text';
+EXCEPTION WHEN insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege during ALTER';
+END $$;
+
+-- Test Manual INSERT on conflict log table
+-- This should fail because the table is system-managed
+-- Hiding the OID in the error message by catching the exception
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during INSERT';
+END $$;
+
+-- Test Manual UPDATE on conflict log table
+-- This should fail because the table is system-managed
+-- Hiding the OID in the error message by catching the exception
+DO $$
+DECLARE
+ tab_name text;
+BEGIN
+ SELECT 'pg_conflict.' || relname INTO tab_name
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+ WHERE s.subname = 'regress_conflict_protection_test';
+
+ EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' ';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during UPDATE';
+END $$;
+
+-- Trying to perform TRUNCATE/DELETE on the conflict log table
+-- This should be allowed so that user can perform cleanup
+SELECT 'pg_conflict.' || relname AS conflict_tab
+FROM pg_class c
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_' || s.oid
+WHERE s.subname = 'regress_conflict_protection_test' \gset
+TRUNCATE :conflict_tab;
+DELETE FROM :conflict_tab;
+
+-- Trying to create a new table manually in the pg_conflict namespace
+-- This should fail as the namespace is reserved for conflict logs tables
+CREATE TABLE pg_conflict.manual_table (id int);
+
+-- Moving an existing table into the pg_conflict namespace
+-- Users should not be able to move their own tables within this namespace
+CREATE TABLE public.test_move (id int);
+ALTER TABLE public.test_move SET SCHEMA pg_conflict;
+DROP TABLE public.test_move;
+
+SET client_min_messages = WARNING;
+
+-- Clean up remaining test subscription
+ALTER SUBSCRIPTION regress_conflict_log_default DISABLE;
+ALTER SUBSCRIPTION regress_conflict_log_default SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_log_default;
+
+ALTER SUBSCRIPTION regress_conflict_empty_str DISABLE;
+ALTER SUBSCRIPTION regress_conflict_empty_str SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_empty_str;
+
+ALTER SUBSCRIPTION regress_conflict_test2 DISABLE;
+ALTER SUBSCRIPTION regress_conflict_test2 SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_test2;
+
+ALTER SUBSCRIPTION regress_conflict_protection_test DISABLE;
+ALTER SUBSCRIPTION regress_conflict_protection_test SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_conflict_protection_test;
+
RESET SESSION AUTHORIZATION;
DROP ROLE regress_subscription_user;
DROP ROLE regress_subscription_user2;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 8cf40c87043..203959e5018 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -515,6 +515,8 @@ ConditionalStack
ConditionalStackData
ConfigData
ConfigVariable
+ConflictLogColumnDef
+ConflictLogDest
ConflictTupleInfo
ConflictType
ConnCacheEntry
--
2.53.0
[application/octet-stream] v47-0002-Implement-the-conflict-insertion-infrastructure-.patch (34.8K, 3-v47-0002-Implement-the-conflict-insertion-infrastructure-.patch)
download | inline diff:
From 62a4eaba13e434405685c3ac89da2d61cc712dc7 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Sat, 30 May 2026 13:35:04 +0530
Subject: [PATCH v47 2/5] Implement the conflict insertion infrastructure for
the conflict log table
This patch introduces the core logic to populate the conflict log table whenever
a logical replication conflict is detected. It captures the remote transaction
details along with the corresponding local state at the time of the conflict.
Handling Multi-row Conflicts: A single remote tuple may conflict with multiple
local tuples (e.g., in the case of multiple_unique_conflicts). To handle this,
the infrastructure creates a single row in the conflict log table for each
remote tuple. The details of all conflicting local rows are aggregated into a
single JSON array in the local_conflicts column.
The JSON array uses the following structured format:
[ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1",
"key": {"id": 1}, "tuple": {"id": 1, "val": "old_data"} }, ... ]
Example of querying the structured conflict data:
SELECT remote_xid, relname, remote_origin, local_conflicts[1] ->> 'xid' AS local_xid,
local_conflicts[1] ->> 'tuple' AS local_tuple
FROM pg_conflict.pg_conflict_log_16396;
remote_xid | relname | remote_origin | local_xid | local_tuple
------------+----------+---------------+-----------+---------------------
760 | test | pg_16406 | 771 | {"a":1,"b":10}
765 | conf_tab | pg_16406 | 775 | {"a":2,"b":2,"c":2}
---
.../replication/logical/applyparallelworker.c | 37 +-
src/backend/replication/logical/conflict.c | 647 ++++++++++++++++--
src/backend/replication/logical/launcher.c | 1 +
src/backend/replication/logical/worker.c | 16 +-
src/include/replication/conflict.h | 3 +
src/include/replication/worker_internal.h | 7 +
src/test/subscription/t/030_origin.pl | 4 +-
src/test/subscription/t/035_conflicts.pl | 47 +-
8 files changed, 713 insertions(+), 49 deletions(-)
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 012d55e9d3d..d24c52d43e6 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -986,7 +986,42 @@ ParallelApplyWorkerMain(Datum main_arg)
set_apply_error_context_origin(originname);
- LogicalParallelApplyLoop(mqh);
+ PG_TRY();
+ {
+ LogicalParallelApplyLoop(mqh);
+ }
+ PG_CATCH();
+ {
+ MemoryContext oldcontext;
+ ErrorData *edata;
+
+ /*
+ * Copy the ErrorData before doing any further work. The error may
+ * have been raised while running under ErrorContext, so switch to
+ * a safe context (TopMemoryContext) to avoid assertions and ensure
+ * the error data survives subsequent cleanup.
+ */
+ oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ edata = CopyErrorData();
+ MemoryContextSwitchTo(oldcontext);
+
+ FlushErrorState();
+
+ /*
+ * Reset the origin state to prevent the advancement of origin
+ * progress if we fail to apply. Otherwise, this will result in
+ * transaction loss as that transaction won't be sent again by the
+ * server.
+ */
+ replorigin_xact_clear(true);
+
+ AbortOutOfAnyTransaction();
+ ProcessPendingConflictLogTuple();
+
+ /* Re-throw the original error. */
+ ReThrowError(edata);
+ }
+ PG_END_TRY();
/*
* The parallel apply worker must not get here because the parallel apply
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 4a647277d1f..6bf3d6d5a44 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -16,17 +16,24 @@
#include "access/commit_ts.h"
#include "access/genam.h"
+#include "access/heapam.h"
#include "access/tableam.h"
+#include "access/xact.h"
#include "catalog/heap.h"
#include "catalog/pg_am.h"
#include "catalog/pg_namespace.h"
#include "catalog/toasting.h"
#include "executor/executor.h"
+#include "funcapi.h"
#include "pgstat.h"
#include "replication/conflict.h"
#include "replication/worker_internal.h"
#include "storage/lmgr.h"
+#include "utils/array.h"
+#include "utils/builtins.h"
#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/pg_lsn.h"
/*
* String representations for the supported conflict logging destinations.
@@ -40,7 +47,6 @@ const char *const ConflictLogDestNames[] = {
StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
"ConflictLogDestNames length mismatch");
-
/* Structure to hold metadata for one column of the conflict log table */
typedef struct ConflictLogColumnDef
{
@@ -72,6 +78,18 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
+/* Schema for the elements within the 'local_conflicts' JSON array */
+static const ConflictLogColumnDef LocalConflictSchema[] =
+{
+ { .attname = "xid", .atttypid = XIDOID },
+ { .attname = "commit_ts", .atttypid = TIMESTAMPTZOID },
+ { .attname = "origin", .atttypid = TEXTOID },
+ { .attname = "key", .atttypid = JSONOID },
+ { .attname = "tuple", .atttypid = JSONOID }
+};
+
+#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema)
+
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
@@ -99,8 +117,27 @@ static void get_tuple_desc(EState *estate, ResultRelInfo *relinfo,
TupleTableSlot *remoteslot, char **remote_desc,
TupleTableSlot *searchslot, char **search_desc,
Oid indexoid);
+static void build_index_datums_from_slot(EState *estate, Relation localrel,
+ TupleTableSlot *slot,
+ Relation indexDesc, Datum *values,
+ bool *isnull);
static char *build_index_value_desc(EState *estate, Relation localrel,
TupleTableSlot *slot, Oid indexoid);
+static Datum tuple_table_slot_to_json_datum(TupleTableSlot *slot);
+static Datum tuple_table_slot_to_indextup_json(EState *estate,
+ Relation localrel,
+ Oid replica_index,
+ TupleTableSlot *slot);
+static TupleDesc build_conflict_tupledesc(void);
+static Datum build_local_conflicts_json_array(EState *estate, Relation rel,
+ ConflictType conflict_type,
+ List *conflicttuples);
+static void prepare_conflict_log_tuple(EState *estate, Relation rel,
+ Relation conflictlogrel,
+ ConflictType conflict_type,
+ TupleTableSlot *searchslot,
+ List *conflicttuples,
+ TupleTableSlot *remoteslot);
/*
* Builds the TupleDesc for the conflict log table.
@@ -281,30 +318,170 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
ConflictType type, TupleTableSlot *searchslot,
TupleTableSlot *remoteslot, List *conflicttuples)
{
- Relation localrel = relinfo->ri_RelationDesc;
- StringInfoData err_detail;
+ Relation localrel = relinfo->ri_RelationDesc;
+ ConflictLogDest dest;
+ Relation conflictlogrel;
+ bool log_dest_table;
+ bool log_dest_logfile;
- initStringInfo(&err_detail);
+ pgstat_report_subscription_conflict(MySubscription->oid, type);
- /* Form errdetail message by combining conflicting tuples information. */
- foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
- errdetail_apply_conflict(estate, relinfo, type, searchslot,
- conflicttuple->slot, remoteslot,
- conflicttuple->indexoid,
- conflicttuple->xmin,
- conflicttuple->origin,
- conflicttuple->ts,
- &err_detail);
+ /*
+ * Get the conflict log destination. Also, (if there is one) return the
+ * CLT relation already opened and ready for insertion.
+ */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
- pgstat_report_subscription_conflict(MySubscription->oid, type);
+ log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest);
+ log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest);
+
+ /* Insert to table if requested. */
+ if (log_dest_table)
+ {
+ Assert(conflictlogrel != NULL);
+
+ /*
+ * Prepare the conflict log tuple. If the error level is below ERROR,
+ * insert it immediately. Otherwise, defer the insertion to a new
+ * transaction after the current one aborts, ensuring the insertion of
+ * the log tuple is not rolled back.
+ */
+ prepare_conflict_log_tuple(estate,
+ relinfo->ri_RelationDesc,
+ conflictlogrel,
+ type,
+ searchslot,
+ conflicttuples,
+ remoteslot);
+ if (elevel < ERROR)
+ InsertConflictLogTuple(conflictlogrel);
+
+ if (!log_dest_logfile)
+ {
+ /*
+ * Not logging conflict details to the server log; Report the error
+ * msg but omit raw tuple data from server logs since it's already
+ * captured in the conflict log table.
+ */
+ ereport(elevel,
+ errcode_apply_conflict(type),
+ errmsg("conflict detected on relation \"%s\": conflict=%s",
+ RelationGetQualifiedRelationName(localrel),
+ ConflictTypeNames[type]),
+ errdetail("Conflict details are logged to the conflict log table: %s",
+ RelationGetRelationName(conflictlogrel)));
+ }
+
+ table_close(conflictlogrel, RowExclusiveLock);
+ }
+
+ /* Log into the server log if requested. */
+ if (log_dest_logfile)
+ {
+ StringInfoData err_detail;
+
+ initStringInfo(&err_detail);
+
+ /* Form errdetail message by combining conflicting tuples information. */
+ foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+ errdetail_apply_conflict(estate, relinfo, type, searchslot,
+ conflicttuple->slot, remoteslot,
+ conflicttuple->indexoid,
+ conflicttuple->xmin,
+ conflicttuple->origin,
+ conflicttuple->ts,
+ &err_detail);
+
+ /* Standard reporting with full internal details. */
+ ereport(elevel,
+ errcode_apply_conflict(type),
+ errmsg("conflict detected on relation \"%s\": conflict=%s",
+ RelationGetQualifiedRelationName(localrel),
+ ConflictTypeNames[type]),
+ errdetail_internal("%s", err_detail.data));
+ }
+}
+
+/*
+ * ProcessPendingConflictLogTuple
+ * Insert any deferred conflict log tuple in a separate transaction.
+ *
+ * For conflicts raised at ERROR level, the conflict log tuple cannot be
+ * inserted immediately because the surrounding transaction will abort.
+ * To ensure that conflict information is not lost, such tuples are prepared
+ * during error processing (see prepare_conflict_log_tuple()) but their
+ * insertion is deferred.
+ *
+ * This function is responsible for completing that deferred insertion after
+ * the failing transaction has been aborted and the system has returned to an
+ * idle state. It executes the insertion in a new, independent transaction,
+ * ensuring that the conflict log entry is durable and not rolled back
+ * together with the failed apply transaction.
+ */
+void
+ProcessPendingConflictLogTuple(void)
+{
+ Relation conflictlogrel;
+ ConflictLogDest dest;
+
+ /* Nothing to do */
+ if (MyLogicalRepWorker->conflict_log_tuple == NULL)
+ return;
+
+ PG_TRY();
+ {
+ StartTransactionCommand();
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Open conflict log table and insert the tuple */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
+ Assert(conflictlogrel);
+
+ InsertConflictLogTuple(conflictlogrel);
+
+ table_close(conflictlogrel, RowExclusiveLock);
+
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ }
+ PG_CATCH();
+ {
+ ErrorData *edata;
+ MemoryContext oldctx;
+
+ /* Save error info in our memory context */
+ oldctx = MemoryContextSwitchTo(TopMemoryContext);
+ edata = CopyErrorData();
+ MemoryContextSwitchTo(oldctx);
+
+ /* Clear the error state so we can continue */
+ FlushErrorState();
- ereport(elevel,
- errcode_apply_conflict(type),
- errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
- get_namespace_name(RelationGetNamespace(localrel)),
- RelationGetRelationName(localrel),
- ConflictTypeNames[type]),
- errdetail_internal("%s", err_detail.data));
+ /* Abort the transaction we started above */
+ AbortOutOfAnyTransaction();
+
+ /*
+ * Report the error as a warning. We use WARNING because we don't want
+ * this to be a fatal error for the worker, and we want to allow the
+ * caller's original error to remain primary.
+ */
+ ereport(WARNING,
+ (errmsg("could not log conflict to table for subscription \"%s\": %s",
+ MySubscription->name, edata->message)));
+
+ FreeErrorData(edata);
+
+ /*
+ * Free the conflict log tuple and set it to NULL. This ensures we
+ * don't try to insert the same problematic tuple again.
+ */
+ if (MyLogicalRepWorker->conflict_log_tuple != NULL)
+ {
+ heap_freetuple(MyLogicalRepWorker->conflict_log_tuple);
+ MyLogicalRepWorker->conflict_log_tuple = NULL;
+ }
+ }
+ PG_END_TRY();
}
/*
@@ -338,6 +515,56 @@ InitConflictIndexes(ResultRelInfo *relInfo)
relInfo->ri_onConflictArbiterIndexes = uniqueIndexes;
}
+/*
+ * GetConflictLogDestAndTable
+ *
+ * Fetches conflict logging metadata from the cached MySubscription pointer.
+ * Sets the destination enum in *log_dest and, if applicable, opens and
+ * returns the relation handle for the conflict log table.
+ */
+Relation
+GetConflictLogDestAndTable(ConflictLogDest *log_dest)
+{
+ Oid conflictlogrelid;
+
+ /*
+ * Convert the text log destination to the internal enum. MySubscription
+ * already contains the data from pg_subscription.
+ */
+ *log_dest = GetConflictLogDest(MySubscription->conflictlogdest);
+
+ /* Quick exit if a conflict log table was not requested. */
+ if (!CONFLICTS_LOGGED_TO_TABLE(*log_dest))
+ return NULL;
+
+ conflictlogrelid = MySubscription->conflictlogrelid;
+
+ Assert(OidIsValid(conflictlogrelid));
+
+ return table_open(conflictlogrelid, RowExclusiveLock);
+}
+
+/*
+ * InsertConflictLogTuple
+ *
+ * Insert conflict log tuple into the conflict log table. It uses
+ * HEAP_INSERT_NO_LOGICAL to explicitly block logical decoding of the tuple
+ * inserted into the conflict log table.
+ */
+void
+InsertConflictLogTuple(Relation conflictlogrel)
+{
+ /* A valid tuple must be prepared and stored in MyLogicalRepWorker. */
+ Assert(MyLogicalRepWorker->conflict_log_tuple != NULL);
+
+ heap_insert(conflictlogrel, MyLogicalRepWorker->conflict_log_tuple,
+ GetCurrentCommandId(true), HEAP_INSERT_NO_LOGICAL, NULL);
+
+ /* Free conflict log tuple. */
+ heap_freetuple(MyLogicalRepWorker->conflict_log_tuple);
+ MyLogicalRepWorker->conflict_log_tuple = NULL;
+}
+
/*
* Add SQLSTATE error code to the current conflict report.
*/
@@ -771,6 +998,40 @@ get_tuple_desc(EState *estate, ResultRelInfo *relinfo, ConflictType type,
}
}
+/*
+ * Helper function to extract the "raw" index key Datums and their null flags
+ * from a TupleTableSlot, given an already open index descriptor.
+ * This is the reusable core logic.
+ */
+static void
+build_index_datums_from_slot(EState *estate, Relation localrel,
+ TupleTableSlot *slot,
+ Relation indexDesc, Datum *values,
+ bool *isnull)
+{
+ TupleTableSlot *tableslot = slot;
+
+ /*
+ * If the slot is a virtual slot, copy it into a heap tuple slot as
+ * FormIndexDatum only works with heap tuple slots.
+ */
+ if (TTS_IS_VIRTUAL(slot))
+ {
+ /* Slot is created within the EState's tuple table */
+ tableslot = table_slot_create(localrel, &estate->es_tupleTable);
+ tableslot = ExecCopySlot(tableslot, slot);
+ }
+
+ /*
+ * Initialize ecxt_scantuple for potential use in FormIndexDatum
+ */
+ GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+
+ /* Form the index datums */
+ FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values,
+ isnull);
+}
+
/*
* Helper functions to construct a string describing the contents of an index
* entry. See BuildIndexValueDescription for details.
@@ -786,41 +1047,339 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
Relation indexDesc;
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
- TupleTableSlot *tableslot = slot;
- if (!tableslot)
+ if (!slot)
return NULL;
Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
indexDesc = index_open(indexoid, NoLock);
- /*
- * If the slot is a virtual slot, copy it into a heap tuple slot as
- * FormIndexDatum only works with heap tuple slots.
- */
- if (TTS_IS_VIRTUAL(slot))
+ build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+ isnull);
+
+ index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+
+ index_close(indexDesc, NoLock);
+
+ return index_value;
+}
+
+/*
+ * tuple_table_slot_to_json_datum
+ *
+ * Helper function to convert a TupleTableSlot to JSON.
+ */
+static Datum
+tuple_table_slot_to_json_datum(TupleTableSlot *slot)
+{
+ HeapTuple tuple;
+ Datum datum;
+ Datum json;
+
+ Assert(slot != NULL);
+
+ tuple = ExecCopySlotHeapTuple(slot);
+ datum = heap_copy_tuple_as_datum(tuple, slot->tts_tupleDescriptor);
+
+ json = DirectFunctionCall1(row_to_json, datum);
+ heap_freetuple(tuple);
+
+ return json;
+}
+
+/*
+ * tuple_table_slot_to_indextup_json
+ *
+ * Fetch replica identity key from the tuple table slot and convert into a
+ * JSON datum.
+ */
+static Datum
+tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
+ Oid indexid, TupleTableSlot *slot)
+{
+ Relation indexDesc;
+ Datum values[INDEX_MAX_KEYS];
+ bool isnull[INDEX_MAX_KEYS];
+ HeapTuple tuple;
+ TupleDesc tupdesc;
+ Datum datum;
+
+ Assert(slot != NULL);
+
+ Assert(CheckRelationOidLockedByMe(indexid, RowExclusiveLock, true));
+
+ indexDesc = index_open(indexid, NoLock);
+
+ build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+ isnull);
+ tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc));
+
+ /* Bless the tupdesc so it can be looked up by row_to_json. */
+ BlessTupleDesc(tupdesc);
+
+ /* Form the replica identity tuple. */
+ tuple = heap_form_tuple(tupdesc, values, isnull);
+ datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+ heap_freetuple(tuple);
+ FreeTupleDesc(tupdesc);
+ index_close(indexDesc, NoLock);
+
+ /* Convert to a JSON datum. */
+ return DirectFunctionCall1(row_to_json, datum);
+}
+
+/*
+ * build_conflict_tupledesc
+ *
+ * Build and bless a tuple descriptor for the conflict log table based on the
+ * predefined LocalConflictSchema.
+ */
+static TupleDesc
+build_conflict_tupledesc(void)
+{
+ static TupleDesc cached_tupdesc = NULL;
+
+ if (cached_tupdesc == NULL)
{
- tableslot = table_slot_create(localrel, &estate->es_tupleTable);
- tableslot = ExecCopySlot(tableslot, slot);
+ MemoryContext oldcxt;
+
+ oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+
+ cached_tupdesc = CreateTemplateTupleDesc(NUM_LOCAL_CONFLICT_ATTRS);
+
+ for (int i = 0; i < NUM_LOCAL_CONFLICT_ATTRS; i++)
+ TupleDescInitEntry(cached_tupdesc,
+ (AttrNumber) (i + 1),
+ LocalConflictSchema[i].attname,
+ LocalConflictSchema[i].atttypid,
+ -1, 0);
+
+ TupleDescFinalize(cached_tupdesc);
+
+ /*
+ * Bless once so it can be used as a RECORD type (e.g. for
+ * row_to_json or other record-based operations).
+ */
+ BlessTupleDesc(cached_tupdesc);
+
+ MemoryContextSwitchTo(oldcxt);
}
- /*
- * Initialize ecxt_scantuple for potential use in FormIndexDatum when
- * index expressions are present.
- */
- GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+ return cached_tupdesc;
+}
- /*
- * The values/nulls arrays passed to BuildIndexValueDescription should be
- * the results of FormIndexDatum, which are the "raw" input to the index
- * AM.
- */
- FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull);
+/*
+ * Builds the local conflicts JSON array column from the list of
+ * ConflictTupleInfo objects.
+ *
+ * Example output structure:
+ * [ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ]
+ */
+static Datum
+build_local_conflicts_json_array(EState *estate, Relation rel,
+ ConflictType conflict_type,
+ List *conflicttuples)
+{
+ ListCell *lc;
+ List *json_datums = NIL;
+ Datum *json_datum_array;
+ Datum json_array_datum;
+ int num_conflicts;
+ int i;
+ int16 typlen;
+ bool typbyval;
+ char typalign;
+ TupleDesc tupdesc;
- index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+ /* Build local conflicts tuple descriptor. */
+ tupdesc = build_conflict_tupledesc();
- index_close(indexDesc, NoLock);
+ /* Process local conflict tuple list and prepare an array of JSON. */
+ foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+ {
+ Datum values[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+ char *origin_name = NULL;
+ HeapTuple tuple;
+ Datum json_datum;
+ int attno;
+
+ attno = 0;
+ values[attno++] = TransactionIdGetDatum(conflicttuple->xmin);
+
+ if (conflicttuple->ts)
+ values[attno++] = TimestampTzGetDatum(conflicttuple->ts);
+ else
+ nulls[attno++] = true;
- return index_value;
+ if (conflicttuple->origin != InvalidReplOriginId)
+ replorigin_by_oid(conflicttuple->origin, true, &origin_name);
+
+ /* Store empty string if origin name for the tuple is NULL. */
+ if (origin_name != NULL)
+ values[attno++] = CStringGetTextDatum(origin_name);
+ else
+ nulls[attno++] = true;
+
+ /*
+ * Add the conflicting key values in the case of a unique constraint
+ * violation.
+ */
+ if (conflict_type == CT_INSERT_EXISTS ||
+ conflict_type == CT_UPDATE_EXISTS ||
+ conflict_type == CT_MULTIPLE_UNIQUE_CONFLICTS)
+ {
+ Oid indexoid = conflicttuple->indexoid;
+
+ Assert(OidIsValid(indexoid) && conflicttuple->slot &&
+ CheckRelationOidLockedByMe(indexoid, RowExclusiveLock,
+ true));
+ values[attno++] =
+ tuple_table_slot_to_indextup_json(estate, rel,
+ indexoid,
+ conflicttuple->slot);
+ }
+ else
+ nulls[attno++] = true;
+
+ /* Convert conflicting tuple to JSON datum. */
+ if (conflicttuple->slot)
+ values[attno] = tuple_table_slot_to_json_datum(conflicttuple->slot);
+ else
+ nulls[attno] = true;
+
+ Assert(attno + 1 == NUM_LOCAL_CONFLICT_ATTRS);
+
+ tuple = heap_form_tuple(tupdesc, values, nulls);
+
+ json_datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+ /*
+ * Build the higher level JSON datum in format described in function
+ * header.
+ */
+ json_datum = DirectFunctionCall1(row_to_json, json_datum);
+
+ /* Done with the temporary tuple. */
+ heap_freetuple(tuple);
+
+ /* Add to the array element. */
+ json_datums = lappend(json_datums, (void *) json_datum);
+ }
+
+ num_conflicts = list_length(json_datums);
+
+ json_datum_array = palloc_array(Datum, num_conflicts);
+
+ i = 0;
+ foreach(lc, json_datums)
+ {
+ json_datum_array[i] = (Datum) lfirst(lc);
+ i++;
+ }
+
+ /* Construct the JSON array Datum. */
+ get_typlenbyvalalign(JSONOID, &typlen, &typbyval, &typalign);
+ json_array_datum = PointerGetDatum(construct_array(json_datum_array,
+ num_conflicts,
+ JSONOID,
+ typlen,
+ typbyval,
+ typalign));
+ pfree(json_datum_array);
+
+ return json_array_datum;
+}
+
+/*
+ * prepare_conflict_log_tuple
+ *
+ * This routine prepares a tuple detailing a conflict encountered during
+ * logical replication. The prepared tuple will be stored in
+ * MyLogicalRepWorker->conflict_log_tuple which should be inserted into the
+ * conflict log table by calling InsertConflictLogTuple.
+ */
+static void
+prepare_conflict_log_tuple(EState *estate, Relation rel,
+ Relation conflictlogrel,
+ ConflictType conflict_type,
+ TupleTableSlot *searchslot,
+ List *conflicttuples,
+ TupleTableSlot *remoteslot)
+{
+ Datum values[NUM_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_CONFLICT_ATTRS] = {0};
+ int attno;
+ char *remote_origin = NULL;
+ MemoryContext oldctx;
+
+ Assert(MyLogicalRepWorker->conflict_log_tuple == NULL);
+
+ /* Populate the values and nulls arrays. */
+ attno = 0;
+ values[attno++] = ObjectIdGetDatum(RelationGetRelid(rel));
+
+ values[attno++] =
+ CStringGetTextDatum(get_namespace_name(RelationGetNamespace(rel)));
+
+ values[attno++] = CStringGetTextDatum(RelationGetRelationName(rel));
+
+ values[attno++] = CStringGetTextDatum(ConflictTypeNames[conflict_type]);
+
+ if (TransactionIdIsValid(remote_xid))
+ values[attno++] = TransactionIdGetDatum(remote_xid);
+ else
+ nulls[attno++] = true;
+
+ values[attno++] = LSNGetDatum(remote_final_lsn);
+
+ if (remote_commit_ts > 0)
+ values[attno++] = TimestampTzGetDatum(remote_commit_ts);
+ else
+ nulls[attno++] = true;
+
+ if (replorigin_xact_state.origin != InvalidReplOriginId)
+ replorigin_by_oid(replorigin_xact_state.origin, true, &remote_origin);
+
+ if (remote_origin != NULL)
+ values[attno++] = CStringGetTextDatum(remote_origin);
+ else
+ nulls[attno++] = true;
+
+ if (!TupIsNull(remoteslot))
+ values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
+ else
+ nulls[attno++] = true;
+
+ if (!TupIsNull(searchslot))
+ {
+ Oid replica_index = GetRelationIdentityOrPK(rel);
+
+ /*
+ * If the table has a valid replica identity index, build the index
+ * JSON datum from key value. Otherwise, construct it from the complete
+ * tuple in REPLICA IDENTITY FULL cases.
+ */
+ if (OidIsValid(replica_index))
+ values[attno++] = tuple_table_slot_to_indextup_json(estate, rel,
+ replica_index,
+ searchslot);
+ else
+ values[attno++] = tuple_table_slot_to_json_datum(searchslot);
+ }
+ else
+ nulls[attno++] = true;
+
+ values[attno] = build_local_conflicts_json_array(estate, rel,
+ conflict_type,
+ conflicttuples);
+
+ Assert(attno + 1 == NUM_CONFLICT_ATTRS);
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+ MyLogicalRepWorker->conflict_log_tuple =
+ heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls);
+ MemoryContextSwitchTo(oldctx);
}
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 313e31ff2e3..05a30342f69 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -487,6 +487,7 @@ retry:
worker->oldest_nonremovable_xid = retain_dead_tuples
? MyReplicationSlot->data.xmin
: InvalidTransactionId;
+ worker->conflict_log_tuple = NULL;
worker->last_lsn = InvalidXLogRecPtr;
TIMESTAMP_NOBEGIN(worker->last_send_time);
TIMESTAMP_NOBEGIN(worker->last_recv_time);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a3f2406ed83..7b462904c9b 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -487,7 +487,9 @@ static bool MySubscriptionValid = false;
static List *on_commit_wakeup_workers_subids = NIL;
bool in_remote_transaction = false;
-static XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+XLogRecPtr remote_final_lsn = InvalidXLogRecPtr;
+TransactionId remote_xid = InvalidTransactionId;
+TimestampTz remote_commit_ts = 0;
/* fields valid only when processing streamed transaction */
static bool in_streamed_transaction = false;
@@ -1236,6 +1238,8 @@ apply_handle_begin(StringInfo s)
set_apply_error_context_xact(begin_data.xid, begin_data.final_lsn);
remote_final_lsn = begin_data.final_lsn;
+ remote_commit_ts = begin_data.committime;
+ remote_xid = begin_data.xid;
maybe_start_skipping_changes(begin_data.final_lsn);
@@ -1767,6 +1771,10 @@ apply_handle_stream_start(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
+ remote_xid = stream_xid;
+ remote_final_lsn = InvalidXLogRecPtr;
+ remote_commit_ts = 0;
+
set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
/* Try to allocate a worker for the streaming transaction. */
@@ -2440,6 +2448,9 @@ apply_handle_stream_commit(StringInfo s)
/* Unlink the files with serialized changes and subxact info. */
stream_cleanup_files(MyLogicalRepWorker->subid, xid);
+ /* Set remote_commit_ts for conflict logging. */
+ remote_commit_ts = commit_data.committime;
+
elog(DEBUG1, "finished processing the STREAM COMMIT command");
break;
@@ -5666,6 +5677,7 @@ start_apply(XLogRecPtr origin_startpos)
*/
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid);
+ ProcessPendingConflictLogTuple();
PG_RE_THROW();
}
@@ -6040,6 +6052,8 @@ DisableSubscriptionAndExit(void)
*/
pgstat_report_subscription_error(MyLogicalRepWorker->subid);
+ ProcessPendingConflictLogTuple();
+
/* Disable the subscription */
StartTransactionCommand();
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index a017e1e6cb5..b727cc285e9 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -115,5 +115,8 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
TupleTableSlot *searchslot,
TupleTableSlot *remoteslot,
List *conflicttuples);
+extern void ProcessPendingConflictLogTuple(void);
extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern Relation GetConflictLogDestAndTable(ConflictLogDest *log_dest);
+extern void InsertConflictLogTuple(Relation conflictlogrel);
#endif
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 745b7d9e969..6b6525dc2e2 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -100,6 +100,9 @@ typedef struct LogicalRepWorker
*/
TransactionId oldest_nonremovable_xid;
+ /* A conflict log tuple that is prepared but not yet inserted. */
+ HeapTuple conflict_log_tuple;
+
/* Stats. */
XLogRecPtr last_lsn;
TimestampTz last_send_time;
@@ -255,6 +258,10 @@ extern PGDLLIMPORT bool InitializingApplyWorker;
extern PGDLLIMPORT List *table_states_not_ready;
+extern XLogRecPtr remote_final_lsn;
+extern TimestampTz remote_commit_ts;
+extern TransactionId remote_xid;
+
extern void logicalrep_worker_attach(int slot);
extern LogicalRepWorker *logicalrep_worker_find(LogicalRepWorkerType wtype,
Oid subid, Oid relid,
diff --git a/src/test/subscription/t/030_origin.pl b/src/test/subscription/t/030_origin.pl
index 6bc6b7874c2..5f4d00bdd33 100644
--- a/src/test/subscription/t/030_origin.pl
+++ b/src/test/subscription/t/030_origin.pl
@@ -166,7 +166,7 @@ is($result, qq(32), 'The node_A data replicated to node_B');
$node_C->safe_psql('postgres', "UPDATE $tab SET a = 33 WHERE a = 32;");
$node_B->wait_for_log(
- qr/conflict detected on relation "public.$tab_unquoted": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./
+ qr/conflict detected on relation "public.$tab": conflict=update_origin_differs.*\n.*DETAIL:.* Updating the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(32\), remote row \(33\), replica identity \(a\)=\(32\)./
);
$node_B->safe_psql('postgres', "DELETE FROM $tab;");
@@ -182,7 +182,7 @@ is($result, qq(33), 'The node_A data replicated to node_B');
$node_C->safe_psql('postgres', "DELETE FROM $tab WHERE a = 33;");
$node_B->wait_for_log(
- qr/conflict detected on relation "public.$tab_unquoted": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/
+ qr/conflict detected on relation "public.$tab": conflict=delete_origin_differs.*\n.*DETAIL:.* Deleting the row that was modified by a different origin ".*" in transaction [0-9]+ at .*: local row \(33\), replica identity \(a\)=\(33\).*/
);
# The remaining tests no longer test conflict detection.
diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index f23fe6af2a5..de2d37c8754 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -50,7 +50,7 @@ $node_subscriber->safe_psql(
'postgres',
"CREATE SUBSCRIPTION sub_tab
CONNECTION '$publisher_connstr application_name=$appname'
- PUBLICATION pub_tab;");
+ PUBLICATION pub_tab WITH (conflict_log_destination=all)");
# Wait for initial table sync to finish
$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);
@@ -84,10 +84,35 @@ $node_subscriber->wait_for_log(
.*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(4\), local row \(4, 4, 4\)./,
$log_offset);
+# Verify the contents of the Conflict Log Table (CLT)
+# This section ensures that the CLT contains the expected
+# type and specific key data.
+my $subid = $node_subscriber->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
+my $clt = "pg_conflict.pg_conflict_log_$subid";
+
+# Wait for the conflict to be logged in the CLT
+my $log_check = $node_subscriber->poll_query_until(
+ 'postgres',
+ "SELECT count(*) > 0 FROM $clt;"
+);
+
+my $conflict_check = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) >= 1 FROM $clt WHERE conflict_type = 'multiple_unique_conflicts';");
+is($conflict_check, 't', 'Verified multiple_unique_conflicts logged into conflict log table');
+
+my $json_query = "SELECT local_conflicts FROM $clt;";
+my $raw_json = $node_subscriber->safe_psql('postgres', $json_query);
+
+# Verify that '2' is present inside the JSON structure using a regex
+# This matches the key/value pattern for "a": 2
+like($raw_json, qr/\\"a\\":2/, 'Verified that key 2 exists in the local_conflicts');
+
pass('multiple_unique_conflicts detected during insert');
# Truncate table to get rid of the error
$node_subscriber->safe_psql('postgres', "TRUNCATE conf_tab;");
+$node_subscriber->safe_psql('postgres', "DELETE FROM $clt");
##################################################
# Test multiple_unique_conflicts due to UPDATE
@@ -114,6 +139,26 @@ $node_subscriber->wait_for_log(
.*Key already exists in unique index \"conf_tab_c_key\", modified in transaction .*: key \(c\)=\(8\), local row \(8, 8, 8\)./,
$log_offset);
+# Verify the contents of the Conflict Log Table (CLT)
+# This section ensures that the CLT contains the expected
+# type and specific key data.
+
+# Wait for the conflict to be logged in the CLT
+$log_check = $node_subscriber->poll_query_until(
+ 'postgres',
+ "SELECT count(*) > 0 FROM $clt;"
+);
+
+$conflict_check = $node_subscriber->safe_psql('postgres',
+ "SELECT count(*) >= 1 FROM $clt WHERE conflict_type = 'multiple_unique_conflicts';");
+is($conflict_check, 't', 'Verified multiple_unique_conflicts logged into conflict log table');
+
+$raw_json = $node_subscriber->safe_psql('postgres', $json_query);
+
+# Verify that '6' is present inside the JSON structure using a regex
+# This matches the key/value pattern for "a": 6
+like($raw_json, qr/\\"a\\":6/, 'Verified that key 6 exists in the local_conflicts');
+
pass('multiple_unique_conflicts detected during update');
# Truncate table to get rid of the error
--
2.53.0
[application/octet-stream] v47-0003-Preserve-conflict-log-destination-and-subscripti.patch (23.6K, 4-v47-0003-Preserve-conflict-log-destination-and-subscripti.patch)
download | inline diff:
From 30ca196266c751482fd03f4b1b83886e1e21a968 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Fri, 5 Jun 2026 15:51:22 +0530
Subject: [PATCH v47 3/5] Preserve conflict log destination and subscription
OID for subscriptions
Support pg_dump to dump and restore the conflict_log_destination setting for
subscriptions.
During a normal CREATE SUBSCRIPTION, a conflict log table is created
automatically when required. However, during binary upgrade, the conflict
log table will already exist and must be reused rather than recreated, and
the subscription must retain its original OID to correctly re-establish
catalog relationships.
To ensure correct behavior, pg_dump now emits an ALTER SUBSCRIPTION command
after subscription creation to restore the conflict_log_destination setting.
---
src/backend/catalog/heap.c | 4 +-
src/backend/commands/subscriptioncmds.c | 56 +++++++-
src/backend/replication/logical/conflict.c | 2 +-
src/backend/utils/adt/pg_upgrade_support.c | 10 ++
src/bin/pg_dump/pg_dump.c | 125 +++++++++++++++++-
src/bin/pg_dump/pg_dump.h | 2 +
src/bin/pg_dump/pg_dump_sort.c | 31 +++++
src/bin/pg_dump/t/002_pg_dump.pl | 5 +-
src/bin/pg_upgrade/pg_upgrade.c | 4 +
src/bin/pg_upgrade/t/004_subscription.pl | 14 +-
src/include/catalog/binary_upgrade.h | 1 +
src/include/catalog/pg_proc.dat | 4 +
src/include/catalog/pg_subscription.h | 2 +
.../expected/spgist_name_ops.out | 6 +-
14 files changed, 249 insertions(+), 17 deletions(-)
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 5d07dcc1c5b..1d9bd903a51 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -312,6 +312,8 @@ heap_create(const char *relname,
* But allow creating indexes on relations in pg_catalog even if
* allow_system_table_mods = off, upper layers already guarantee it's on a
* user defined relation, not a system one.
+ *
+ * Allow creation of the conflict log table in binary-upgrade mode.
*/
if (!allow_system_table_mods && IsNormalProcessingMode())
{
@@ -323,7 +325,7 @@ heap_create(const char *relname,
get_qualified_objname(relnamespace, relname)),
errdetail("System catalog modifications are currently disallowed.")));
- if (IsConflictLogTableNamespace(relnamespace))
+ if (!IsBinaryUpgrade && IsConflictLogTableNamespace(relnamespace))
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
errmsg("permission denied to create \"%s\"",
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 7d09b80ba5f..3b1c802e7a6 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -19,6 +19,7 @@
#include "access/table.h"
#include "access/twophase.h"
#include "access/xact.h"
+#include "catalog/binary_upgrade.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
@@ -86,6 +87,12 @@
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
+/*
+ * This will be set by the pg_upgrade_support function --
+ * binary_upgrade_set_next_pg_subscription_oid().
+ */
+Oid binary_upgrade_next_pg_subscription_oid = InvalidOid;
+
/*
* Structure to hold a bitmap representing the user-provided CREATE/ALTER
* SUBSCRIPTION command options and the parsed/default values of each of them.
@@ -797,8 +804,21 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
- subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
- Anum_pg_subscription_oid);
+ /* Use binary-upgrade override for pg_subscription.oid? */
+ if (IsBinaryUpgrade)
+ {
+ if (!OidIsValid(binary_upgrade_next_pg_subscription_oid))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("pg_subscription OID value not set when in binary upgrade mode")));
+
+ subid = binary_upgrade_next_pg_subscription_oid;
+ binary_upgrade_next_pg_subscription_oid = InvalidOid;
+ }
+ else
+ subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
+ Anum_pg_subscription_oid);
+
values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(subid);
values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
@@ -1510,18 +1530,40 @@ alter_sub_conflictlogdestination(Subscription *sub, ConflictLogDest oldlogdest,
{
ObjectAddress cltaddr;
ObjectAddress subobj;
+ char relname[NAMEDATALEN];
- relid = create_conflict_log_table(sub->oid, sub->name, sub->owner);
- update_relid = true;
+ snprintf(relname, NAMEDATALEN, CONFLICT_LOG_RELATION_NAME_FMT, sub->oid);
/*
- * Establish an internal dependency between the conflict log table
- * and the subscription. For details refer comments in
- * CreateSubscription function.
+ * In upgrade scenarios, old_dest reflects the default behavior of
+ * a newly created subscription (i.e., no conflict logging to
+ * table) However, the conflict log table will already exist from
+ * the upgraded cluster. In such cases, we need to detect the
+ * pre-existing table and update the catalog state to associate it
+ * with the subscription instead of creating a new one.
+ */
+ relid = get_relname_relid(relname, PG_CONFLICT_NAMESPACE);
+ if (OidIsValid(relid))
+ {
+ /* Existing table from upgrade */
+ Assert(IsBinaryUpgrade);
+ }
+ else
+ {
+ relid = create_conflict_log_table(sub->oid, sub->name,
+ sub->owner);
+ }
+
+ /*
+ * Establish an internal dependency between the conflict log
+ * table and the subscription. Refer to comments in the
+ * CreateSubscription function for details.
*/
ObjectAddressSet(cltaddr, RelationRelationId, relid);
ObjectAddressSet(subobj, SubscriptionRelationId, sub->oid);
recordDependencyOn(&cltaddr, &subobj, DEPENDENCY_INTERNAL);
+
+ update_relid = true;
}
}
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 6bf3d6d5a44..4cad0eaf70c 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -178,7 +178,7 @@ create_conflict_log_table(Oid subid, char *subname, Oid subowner)
Oid relid;
char relname[NAMEDATALEN];
- snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
+ snprintf(relname, NAMEDATALEN, CONFLICT_LOG_RELATION_NAME_FMT, subid);
/*
* Check for an existing table with the same name in the pg_conflict namespace.
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index b505a6b4fee..59c3e7f0146 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -181,6 +181,16 @@ binary_upgrade_set_next_pg_authid_oid(PG_FUNCTION_ARGS)
PG_RETURN_VOID();
}
+Datum
+binary_upgrade_set_next_pg_subscription_oid(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+
+ CHECK_IS_BINARY_UPGRADE;
+ binary_upgrade_next_pg_subscription_oid = subid;
+ PG_RETURN_VOID();
+}
+
Datum
binary_upgrade_create_empty_extension(PG_FUNCTION_ARGS)
{
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index a0f7f8e2168..39d5673109b 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -1981,6 +1981,8 @@ checkExtensionMembership(DumpableObject *dobj, Archive *fout)
static void
selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
{
+ DumpOptions *dopt = fout->dopt;
+
/*
* DUMP_COMPONENT_DEFINITION typically implies a CREATE SCHEMA statement
* and (for --clean) a DROP SCHEMA statement. (In the absence of
@@ -2010,6 +2012,32 @@ selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
*/
nsinfo->dobj.dump_contains = nsinfo->dobj.dump = DUMP_COMPONENT_ACL;
}
+ else if (strcmp(nsinfo->dobj.name, "pg_conflict") == 0)
+ {
+ if (dopt->binary_upgrade)
+ {
+ /*
+ * The pg_conflict schema is a strange beast that sits in a sort
+ * of no-mans-land between being a system object and a user
+ * object. CREATE SCHEMA would fail, so its
+ * DUMP_COMPONENT_DEFINITION is just a comment.
+ */
+ nsinfo->create = false;
+ nsinfo->dobj.dump = DUMP_COMPONENT_ALL;
+ nsinfo->dobj.dump &= ~DUMP_COMPONENT_DEFINITION;
+ nsinfo->dobj.dump_contains = DUMP_COMPONENT_ALL;
+
+ /*
+ * Also, make like it has a comment even if it doesn't; this is so
+ * that we'll emit a command to drop the comment, if appropriate.
+ * (Without this, we'd not call dumpCommentExtended for it.)
+ */
+ nsinfo->dobj.components |= DUMP_COMPONENT_COMMENT;
+ }
+ else
+ nsinfo->dobj.dump_contains = nsinfo->dobj.dump =
+ DUMP_COMPONENT_NONE;
+ }
else if (strncmp(nsinfo->dobj.name, "pg_", 3) == 0 ||
strcmp(nsinfo->dobj.name, "information_schema") == 0)
{
@@ -2067,9 +2095,38 @@ selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
static void
selectDumpableTable(TableInfo *tbinfo, Archive *fout)
{
+ DumpOptions *dopt = fout->dopt;
+
if (checkExtensionMembership(&tbinfo->dobj, fout))
return; /* extension membership overrides all else */
+ if (strcmp(tbinfo->dobj.namespace->dobj.name, "pg_conflict") == 0)
+ {
+ /*
+ * Dump pg_conflict tables only during binary upgrade. The schema is
+ * assumed to already exist.
+ */
+ if (dopt->binary_upgrade)
+ {
+ tbinfo->dobj.dump = DUMP_COMPONENT_DEFINITION;
+
+ /*
+ * Suppress the "ALTER TABLE ... OWNER TO ..." command for this
+ * table. This prevents pg_dump from outputting the owner change.
+ */
+ tbinfo->rolname = NULL;
+ }
+ else
+ tbinfo->dobj.dump = DUMP_COMPONENT_NONE;
+
+ /*
+ * Do not apply table selection or exclusion rules to pg_conflict
+ * tables. During binary upgrade, these tables are dumped to preserve
+ * catalog state; otherwise, they are not dumped at all.
+ */
+ return;
+ }
+
/*
* If specific tables are being dumped, dump just those tables; else, dump
* according to the parent namespace's dump flag.
@@ -5184,6 +5241,8 @@ getSubscriptions(Archive *fout)
int i_subfailover;
int i_subretaindeadtuples;
int i_submaxretention;
+ int i_subconflictlogrelid;
+ int i_sublogdestination;
int i,
ntups;
@@ -5282,9 +5341,16 @@ getSubscriptions(Archive *fout)
" '-1' AS subwalrcvtimeout,\n");
if (fout->remoteVersion >= 190000)
- appendPQExpBufferStr(query, " fs.srvname AS subservername\n");
+ appendPQExpBufferStr(query, " fs.srvname AS subservername,\n");
+ else
+ appendPQExpBufferStr(query, " NULL AS subservername,\n");
+
+ if (fout->remoteVersion >= 190000)
+ appendPQExpBufferStr(query,
+ " s.subconflictlogrelid, s.subconflictlogdest\n");
else
- appendPQExpBufferStr(query, " NULL AS subservername\n");
+ appendPQExpBufferStr(query,
+ " NULL AS subconflictlogrelid, NULL AS subconflictlogdest\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -5333,6 +5399,8 @@ getSubscriptions(Archive *fout)
i_subpublications = PQfnumber(res, "subpublications");
i_suborigin = PQfnumber(res, "suborigin");
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
+ i_subconflictlogrelid = PQfnumber(res, "subconflictlogrelid");
+ i_sublogdestination = PQfnumber(res, "subconflictlogdest");
subinfo = pg_malloc_array(SubscriptionInfo, ntups);
@@ -5391,6 +5459,32 @@ getSubscriptions(Archive *fout)
subinfo[i].suboriginremotelsn =
pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn));
+ if (PQgetisnull(res, i, i_subconflictlogrelid))
+ subinfo[i].subconflictlogrelid = InvalidOid;
+ else
+ {
+ subinfo[i].subconflictlogrelid =
+ atooid(PQgetvalue(res, i, i_subconflictlogrelid));
+
+ if (subinfo[i].subconflictlogrelid)
+ {
+ TableInfo *tableInfo;
+
+ tableInfo = findTableByOid(subinfo[i].subconflictlogrelid);
+ if (!tableInfo)
+ pg_fatal("could not find conflict log table with OID %u",
+ subinfo[i].subconflictlogrelid);
+
+ addObjectDependency(&subinfo[i].dobj, tableInfo->dobj.dumpId);
+ }
+ }
+
+ if (PQgetisnull(res, i, i_sublogdestination))
+ subinfo[i].subconflictlogdest = NULL;
+ else
+ subinfo[i].subconflictlogdest =
+ pg_strdup(PQgetvalue(res, i, i_sublogdestination));
+
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
}
@@ -5583,6 +5677,14 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBuffer(delq, "DROP SUBSCRIPTION %s;\n",
qsubname);
+ if (dopt->binary_upgrade)
+ {
+ appendPQExpBufferStr(query, "\n-- For binary upgrade, must preserve pg_subscription.oid\n");
+ appendPQExpBuffer(query,
+ "SELECT pg_catalog.binary_upgrade_set_next_pg_subscription_oid('%u'::pg_catalog.oid);\n\n",
+ subinfo->dobj.catId.oid);
+ }
+
appendPQExpBuffer(query, "CREATE SUBSCRIPTION %s ",
qsubname);
if (subinfo->subservername)
@@ -5656,6 +5758,25 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBufferStr(query, ");\n");
+ /*
+ * Restore the conflict log destination separately from CREATE SUBSCRIPTION.
+ * Setting conflict_log_destination to 'table' may create and associate
+ * a conflict log table whose name is derived from the subscription OID.
+ * Since the subscription must already exist before that OID is known,
+ * emit this as a separate ALTER SUBSCRIPTION so that any required conflict
+ * log table can be created and linked after the subscription itself has
+ * been restored.
+ *
+ * We skip the default value ('log') to match the handling of other
+ * default subscription options.
+ */
+ if (subinfo->subconflictlogdest &&
+ (pg_strcasecmp(subinfo->subconflictlogdest, "log") != 0))
+ appendPQExpBuffer(query,
+ "\n\nALTER SUBSCRIPTION %s SET (conflict_log_destination = %s);\n",
+ qsubname,
+ subinfo->subconflictlogdest);
+
/*
* In binary-upgrade mode, we allow the replication to continue after the
* upgrade.
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 5a6726d8b12..a43a3049343 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -722,6 +722,7 @@ typedef struct _SubscriptionInfo
bool subfailover;
bool subretaindeadtuples;
int submaxretention;
+ Oid subconflictlogrelid;
char *subservername;
char *subconninfo;
char *subslotname;
@@ -730,6 +731,7 @@ typedef struct _SubscriptionInfo
char *subpublications;
char *suborigin;
char *suboriginremotelsn;
+ char *subconflictlogdest;
} SubscriptionInfo;
/*
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 03e5c1c1116..c27b232e799 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -1131,6 +1131,19 @@ repairTableAttrDefMultiLoop(DumpableObject *tableobj,
addObjectDependency(attrdefobj, tableobj->dumpId);
}
+/*
+ * Because we make subscriptions depend on their conflict log tables, while
+ * there is an automatic dependency in the other direction, we need to break
+ * the loop. Remove the automatic dependency, allowing the table to be created
+ * first.
+ */
+static void
+repairSubscriptionTableLoop(DumpableObject *subobj, DumpableObject *tableobj)
+{
+ /* Remove table's dependency on subscription */
+ removeObjectDependency(tableobj, subobj->dumpId);
+}
+
/*
* CHECK, NOT NULL constraints on domains work just like those on tables ...
*/
@@ -1361,6 +1374,24 @@ repairDependencyLoop(DumpableObject **loop,
return;
}
+ /*
+ * Subscription and its Conflict Log Table
+ */
+ if (nLoop == 2 &&
+ loop[0]->objType == DO_TABLE &&
+ loop[1]->objType == DO_SUBSCRIPTION)
+ {
+ repairSubscriptionTableLoop(loop[1], loop[0]);
+ return;
+ }
+ if (nLoop == 2 &&
+ loop[0]->objType == DO_SUBSCRIPTION &&
+ loop[1]->objType == DO_TABLE)
+ {
+ repairSubscriptionTableLoop(loop[0], loop[1]);
+ return;
+ }
+
/* index on partitioned table and corresponding index on partition */
if (nLoop == 2 &&
loop[0]->objType == DO_INDEX &&
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 3ee9fda50e4..3ff50dd50ee 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -3276,9 +3276,10 @@ my %tests = (
create_order => 50,
create_sql => 'CREATE SUBSCRIPTION sub3
CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1
- WITH (connect = false, origin = any, streaming = on);',
+ WITH (connect = false, origin = any, streaming = on, conflict_log_destination= table);',
regexp => qr/^
- \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E
+ \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E\n\n\n
+ \QALTER SUBSCRIPTION sub3 SET (conflict_log_destination = table);\E
/xm,
like => { %full_runs, section_post_data => 1, },
unlike => {
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 2127d297bfe..135ef658c2c 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -35,6 +35,10 @@
*
* We control all assignments of pg_database.oid because we want the directory
* names to match between the old and new cluster.
+ *
+ * We control assignment of pg_subscription.oid because we want the oid to
+ * match between the old and new cluster to make use of subscription's
+ * conflict log table which is named using the subscription oid.
*/
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index c94a82deae0..73f00d2426c 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -290,7 +290,7 @@ $publisher->safe_psql(
$old_sub->safe_psql(
'postgres', qq[
CREATE TABLE tab_upgraded2(id int);
- CREATE SUBSCRIPTION regress_sub5 CONNECTION '$connstr' PUBLICATION regress_pub5;
+ CREATE SUBSCRIPTION regress_sub5 CONNECTION '$connstr' PUBLICATION regress_pub5 with (conflict_log_destination = 'table');
]);
# The table tab_upgraded2 will be in the init state as the subscriber's
@@ -312,7 +312,10 @@ my $tab_upgraded1_oid = $old_sub->safe_psql('postgres',
"SELECT oid FROM pg_class WHERE relname = 'tab_upgraded1'");
my $tab_upgraded2_oid = $old_sub->safe_psql('postgres',
"SELECT oid FROM pg_class WHERE relname = 'tab_upgraded2'");
-
+my $sub5_oid = $old_sub->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription where subname = 'regress_sub5'");
+my $sub_clt_relid = $old_sub->safe_psql('postgres',
+ "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'regress_sub5'");
$old_sub->stop;
# Change configuration so that initial table sync does not get started
@@ -394,6 +397,13 @@ $result = $new_sub->safe_psql('postgres',
);
is($result, qq(t), "conflict detection slot exists");
+# The subscription oid and the subscription conflict log table relid should be preserved
+$result = $new_sub->safe_psql('postgres', "SELECT oid FROM pg_subscription WHERE subname = 'regress_sub5'");
+is($result, qq($sub5_oid), "subscription oid should have been preserved");
+
+$result = $new_sub->safe_psql('postgres', "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'regress_sub5'");
+is($result, qq($sub_clt_relid), "subscription conflict log table relid should have been preserved");
+
# Resume the initial sync and wait until all tables of subscription
# 'regress_sub5' are synchronized
$new_sub->append_conf('postgresql.conf',
diff --git a/src/include/catalog/binary_upgrade.h b/src/include/catalog/binary_upgrade.h
index 7bf7ae44385..b15b18e7dc9 100644
--- a/src/include/catalog/binary_upgrade.h
+++ b/src/include/catalog/binary_upgrade.h
@@ -32,6 +32,7 @@ extern PGDLLIMPORT RelFileNumber binary_upgrade_next_toast_pg_class_relfilenumbe
extern PGDLLIMPORT Oid binary_upgrade_next_pg_enum_oid;
extern PGDLLIMPORT Oid binary_upgrade_next_pg_authid_oid;
+extern PGDLLIMPORT Oid binary_upgrade_next_pg_subscription_oid;
extern PGDLLIMPORT bool binary_upgrade_record_init_privs;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index be157a5fbe9..3b555415cbc 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11969,6 +11969,10 @@
proisstrict => 'f', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => '',
prosrc => 'binary_upgrade_create_conflict_detection_slot' },
+{ oid => '8407', descr => 'for use by pg_upgrade',
+ proname => 'binary_upgrade_set_next_pg_subscription_oid', provolatile => 'v',
+ proparallel => 'r', prorettype => 'void', proargtypes => 'oid',
+ prosrc => 'binary_upgrade_set_next_pg_subscription_oid' },
# conversion functions
{ oid => '4310', descr => 'internal conversion function for KOI8R to WIN1251',
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 89d2300abe1..6a5dc0f1e31 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -221,6 +221,8 @@ typedef struct Subscription
*/
#define LOGICALREP_STREAM_PARALLEL 'p'
+#define CONFLICT_LOG_RELATION_NAME_FMT "pg_conflict_log_%u"
+
#endif /* EXPOSE_TO_CLIENT_CODE */
extern Subscription *GetSubscription(Oid subid, bool missing_ok,
diff --git a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
index 1ee65ede243..39d43368c42 100644
--- a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
+++ b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
@@ -59,11 +59,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
-- Verify clean failure when INCLUDE'd columns result in overlength tuple
-- The error message details are platform-dependent, so show only SQLSTATE
@@ -108,11 +109,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
\set VERBOSITY sqlstate
insert into t values(repeat('xyzzy', 12), 42, repeat('xyzzy', 4000));
--
2.53.0
[application/octet-stream] v47-0005-Documentation-patch.patch (42.3K, 5-v47-0005-Documentation-patch.patch)
download | inline diff:
From b7909b0ecff7112cb3be4a748255cf7009e40350 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Sun, 5 Apr 2026 17:02:01 +0530
Subject: [PATCH v47 5/5] Documentation patch
---
doc/src/sgml/ddl.sgml | 20 +
doc/src/sgml/glossary.sgml | 12 +
doc/src/sgml/logical-replication.sgml | 669 +++++++++++++---------
doc/src/sgml/ref/alter_subscription.sgml | 13 +-
doc/src/sgml/ref/create_subscription.sgml | 51 ++
5 files changed, 494 insertions(+), 271 deletions(-)
diff --git a/doc/src/sgml/ddl.sgml b/doc/src/sgml/ddl.sgml
index 747f929aee3..c16e35ddf59 100644
--- a/doc/src/sgml/ddl.sgml
+++ b/doc/src/sgml/ddl.sgml
@@ -3759,6 +3759,26 @@ REVOKE CREATE ON SCHEMA public FROM PUBLIC;
</para>
</sect2>
+ <sect2 id="ddl-schemas-conflict">
+ <title>The Conflict Schema</title>
+
+ <indexterm zone="ddl-schemas-conflict">
+ <primary>conflict</primary>
+ <secondary>schema</secondary>
+ </indexterm>
+
+ <para>
+ Similarly, the <literal>pg_conflict</literal> schema (sometimes referred to
+ as the <emphasis>conflict schema</emphasis>) contains system managed
+ conflict log tables used for logical replication conflict tracking. These
+ tables are created and maintained by the system and are not intended for
+ direct user manipulation. Unlike <literal>pg_catalog</literal>, the
+ <literal>pg_catalog</literal> schema is not implicitly included in the
+ search path, so objects within it must be referenced explicitly or by
+ adjusting the search path.
+ </para>
+ </sect2>
+
<sect2 id="ddl-schemas-patterns">
<title>Usage Patterns</title>
diff --git a/doc/src/sgml/glossary.sgml b/doc/src/sgml/glossary.sgml
index b881ae71198..57b6970ad9f 100644
--- a/doc/src/sgml/glossary.sgml
+++ b/doc/src/sgml/glossary.sgml
@@ -506,6 +506,18 @@
</glossdef>
</glossentry>
+ <glossentry id="glossary-conflict-schema">
+ <glossterm>conflict schema</glossterm>
+ <glossdef>
+ <para>
+ The <literal>pg_conflict</literal> schema that contains system-managed
+ conflict log tables for logical replication. These tables are created
+ and maintained automatically by the system and are not intended for
+ direct user manipulation. See <xref linkend="ddl-schemas-conflict"/>.
+ </para>
+ </glossdef>
+ </glossentry>
+
<glossentry id="glossary-connection">
<glossterm>Connection</glossterm>
<glossdef>
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index 9e7868487de..fb85cb26296 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -293,6 +293,21 @@
option of <command>CREATE SUBSCRIPTION</command> for details.
</para>
+ <para>
+ <link linkend="logical-replication-conflicts">Conflicts</link> that occur
+ during replication are, by default, logged as plain text in the server log,
+ which can make automated monitoring and analysis difficult. The
+ <command>CREATE SUBSCRIPTION</command> command provides the
+ <link linkend="sql-createsubscription-params-with-conflict-log-destination">
+ <literal>conflict_log_destination</literal></link> option to record detailed
+ conflict information in a structured, queryable format. When this parameter
+ is set to <literal>table</literal> or <literal>all</literal>, the system
+ automatically manages a dedicated <firstterm>conflict log table</firstterm>,
+ which is created an dropped along with the subscription. This significantly
+ improves post-mortem analysis and operational visibility of the replication
+ setup.
+ </para>
+
<sect2 id="logical-replication-subscription-slot">
<title>Logical Replication Slot Management</title>
@@ -2009,120 +2024,225 @@ Included in publications:
operations will simply be skipped.
</para>
- <para>
- Additional logging is triggered, and the conflict statistics are collected (displayed in the
- <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view)
- in the following <firstterm>conflict</firstterm> cases:
- <variablelist>
- <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
- <term><literal>insert_exists</literal></term>
- <listitem>
- <para>
- Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
- unique constraint. Note that to log the origin and commit
- timestamp details of the conflicting key,
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- should be enabled on the subscriber. In this case, an error will be
- raised until the conflict is resolved manually.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
- <term><literal>update_origin_differs</literal></term>
- <listitem>
- <para>
- Updating a row that was previously modified by another origin.
- Note that this conflict can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Currently, the update is always applied
- regardless of the origin of the local row.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-exists" xreflabel="update_exists">
- <term><literal>update_exists</literal></term>
- <listitem>
- <para>
- The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
- unique constraint. Note that to log the origin and commit
- timestamp details of the conflicting key,
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- should be enabled on the subscriber. In this case, an error will be
- raised until the conflict is resolved manually. Note that when updating a
- partitioned table, if the updated row value satisfies another partition
- constraint resulting in the row being inserted into a new partition, the
- <literal>insert_exists</literal> conflict may arise if the new row
- violates a <literal>NOT DEFERRABLE</literal> unique constraint.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
- <term><literal>update_deleted</literal></term>
- <listitem>
- <para>
- The tuple to be updated was concurrently deleted by another origin. The
- update will simply be skipped in this scenario. Note that this conflict
- can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
- are enabled. Note that if a tuple cannot be found due to the table being
- truncated, only a <literal>update_missing</literal> conflict will
- arise. Additionally, if the tuple was deleted by the same origin, an
- <literal>update_missing</literal> conflict will arise.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-missing" xreflabel="update_missing">
- <term><literal>update_missing</literal></term>
- <listitem>
- <para>
- The row to be updated was not found. The update will simply be
- skipped in this scenario.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
- <term><literal>delete_origin_differs</literal></term>
- <listitem>
- <para>
- Deleting a row that was previously modified by another origin. Note that
- this conflict can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Currently, the delete is always applied
- regardless of the origin of the local row.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
- <term><literal>delete_missing</literal></term>
- <listitem>
- <para>
- The row to be deleted was not found. The delete will simply be
- skipped in this scenario.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
- <term><literal>multiple_unique_conflicts</literal></term>
- <listitem>
- <para>
- Inserting or updating a row violates multiple
- <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
- the origin and commit timestamp details of conflicting keys, ensure
- that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. In this case, an error will be raised until
- the conflict is resolved manually.
- </para>
- </listitem>
- </varlistentry>
- </variablelist>
- Note that there are other conflict scenarios, such as exclusion constraint
- violations. Currently, we do not provide additional details for them in the
- log.
- </para>
+ <sect2 id="logical-replication-conflict-logging">
+ <title>Conflict logging</title>
+ <para>
+ Additional logging is triggered, and the conflict statistics are collected (displayed in the
+ <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view)
+ in the following <firstterm>conflict</firstterm> cases:
+ <variablelist>
+ <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
+ <term><literal>insert_exists</literal></term>
+ <listitem>
+ <para>
+ Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
+ unique constraint. Note that to log the origin and commit
+ timestamp details of the conflicting key,
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ should be enabled on the subscriber. In this case, an error will be
+ raised until the conflict is resolved manually.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
+ <term><literal>update_origin_differs</literal></term>
+ <listitem>
+ <para>
+ Updating a row that was previously modified by another origin.
+ Note that this conflict can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Currently, the update is always applied
+ regardless of the origin of the local row.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-exists" xreflabel="update_exists">
+ <term><literal>update_exists</literal></term>
+ <listitem>
+ <para>
+ The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
+ unique constraint. Note that to log the origin and commit
+ timestamp details of the conflicting key,
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ should be enabled on the subscriber. In this case, an error will be
+ raised until the conflict is resolved manually. Note that when updating a
+ partitioned table, if the updated row value satisfies another partition
+ constraint resulting in the row being inserted into a new partition, the
+ <literal>insert_exists</literal> conflict may arise if the new row
+ violates a <literal>NOT DEFERRABLE</literal> unique constraint.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
+ <term><literal>update_deleted</literal></term>
+ <listitem>
+ <para>
+ The tuple to be updated was concurrently deleted by another origin. The
+ update will simply be skipped in this scenario. Note that this conflict
+ can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
+ are enabled. Note that if a tuple cannot be found due to the table being
+ truncated, only a <literal>update_missing</literal> conflict will
+ arise. Additionally, if the tuple was deleted by the same origin, an
+ <literal>update_missing</literal> conflict will arise.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-missing" xreflabel="update_missing">
+ <term><literal>update_missing</literal></term>
+ <listitem>
+ <para>
+ The row to be updated was not found. The update will simply be
+ skipped in this scenario.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
+ <term><literal>delete_origin_differs</literal></term>
+ <listitem>
+ <para>
+ Deleting a row that was previously modified by another origin. Note that
+ this conflict can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Currently, the delete is always applied
+ regardless of the origin of the local row.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
+ <term><literal>delete_missing</literal></term>
+ <listitem>
+ <para>
+ The row to be deleted was not found. The delete will simply be
+ skipped in this scenario.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
+ <term><literal>multiple_unique_conflicts</literal></term>
+ <listitem>
+ <para>
+ Inserting or updating a row violates multiple
+ <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
+ the origin and commit timestamp details of conflicting keys, ensure
+ that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. In this case, an error will be raised until
+ the conflict is resolved manually.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ Note that there are other conflict scenarios, such as exclusion constraint
+ violations. Currently, we do not provide additional details for them in the
+ log.
+ </para>
+ </sect2>
- <para>
- The log format for logical replication conflicts is as follows:
+ <sect2 id="logical-replication-conflict-table-based-logging">
+ <title>Table-based logging</title>
+ <para>
+ If <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ parameter is set to <literal>table</literal> or <literal>all</literal> then
+ a dedicated conflict log table will be automatically created. This table is
+ created in the <literal>pg_conflict</literal> namespace. The name of the
+ conflict log table is
+ <literal>pg_conflict_log_for_subid_<subid></literal>. The predefined
+ schema of this table is detailed in
+ <xref linkend="logical-replication-conflict-log-schema"/>.
+ </para>
+
+ <table id="logical-replication-conflict-log-schema">
+ <title>Conflict Log Table Schema</title>
+ <tgroup cols="3">
+ <thead>
+ <row>
+ <entry>Column</entry>
+ <entry>Type</entry>
+ <entry>Description</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry><literal>relid</literal></entry>
+ <entry><type>oid</type></entry>
+ <entry>The OID of the local table where the conflict occurred.</entry>
+ </row>
+ <row>
+ <entry><literal>schemaname</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The schema name of the conflicting table.</entry>
+ </row>
+ <row>
+ <entry><literal>relname</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The name of the conflicting table.</entry>
+ </row>
+ <row>
+ <entry><literal>conflict_type</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The type of conflict that occurred (e.g., <literal>insert_exists</literal>).</entry>
+ </row>
+ <row>
+ <entry><literal>remote_xid</literal></entry>
+ <entry><type>xid</type></entry>
+ <entry>The remote transaction ID that caused the conflict.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_commit_lsn</literal></entry>
+ <entry><type>pg_lsn</type></entry>
+ <entry>The final LSN of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_commit_ts</literal></entry>
+ <entry><type>timestamptz</type></entry>
+ <entry>The remote commit timestamp of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_origin</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The origin of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_tuple</literal></entry>
+ <entry><type>json</type></entry>
+ <entry>The JSON representation of the incoming remote row that caused
+ the conflict.</entry>
+ </row>
+ <row>
+ <entry><literal>replica_identity</literal></entry>
+ <entry><type>json</type></entry>
+ <entry>The JSON representation of the replica identity.</entry>
+ </row>
+ <row>
+ <entry><literal>local_conflicts</literal></entry>
+ <entry><type>json[]</type></entry>
+ <entry>
+ An array of JSON objects representing the state of existing local
+ row(s) that caused the conflict. Each object includes the local
+ transaction ID (<literal>xid</literal>), commit timestamp
+ (<literal>commit_ts</literal>), origin (<literal>origin</literal>),
+ conflicting key data (<literal>key</literal>), and the full local row
+ image (<literal>tuple</literal>).
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+
+ <para>
+ The conflicting row data, including the incoming remote row (<literal>remote_tuple</literal>)
+ and the associated local conflict details (<literal>local_conflicts</literal>), is stored in
+ <type>JSON</type> formats for flexible querying and analysis.
+ </para>
+ </sect2>
+
+ <sect2 id="logical-replication-conflict-file-based-logging">
+ <title>File-based logging</title>
+ <para>
+ If <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ is set to <literal>log</literal> or <literal>all</literal> then conflicts
+ are logged to the server using the following format:
<synopsis>
LOG: conflict detected on relation "<replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>": conflict=<replaceable>conflict_type</replaceable>
DETAIL: <replaceable class="parameter">detailed_explanation</replaceable>[: <replaceable class="parameter">detail_values</replaceable> [, ... ]].
@@ -2135,182 +2255,185 @@ DETAIL: <replaceable class="parameter">detailed_explanation</replaceable>[: <re
<literal>replica identity</literal> {(<replaceable>column_name</replaceable> <optional>, ...</optional>)=(<replaceable>column_value</replaceable> <optional>, ...</optional>) | full <optional>(<replaceable>column_name</replaceable> <optional>, ...</optional>)=</optional>(<replaceable>column_value</replaceable> <optional>, ...</optional>)}
</synopsis>
- The log provides the following information:
- <variablelist>
- <varlistentry>
- <term><literal>LOG</literal></term>
- <listitem>
+ The log provides the following information:
+ <variablelist>
+ <varlistentry>
+ <term><literal>LOG</literal></term>
+ <listitem>
+ <itemizedlist>
+ <listitem>
+ <para>
+ <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
+ identifies the local relation involved in the conflict.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <replaceable>conflict_type</replaceable> is the type of conflict that occurred
+ (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+ </para>
+ </listitem>
+ </itemizedlist>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><literal>DETAIL</literal></term>
+ <listitem>
<itemizedlist>
<listitem>
<para>
- <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
- identifies the local relation involved in the conflict.
+ <replaceable class="parameter">detailed_explanation</replaceable> includes
+ the origin, transaction ID, and commit timestamp of the transaction that
+ modified the local row, if available.
</para>
</listitem>
<listitem>
<para>
- <replaceable>conflict_type</replaceable> is the type of conflict that occurred
- (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+ The <literal>key</literal> section includes the key values of the local
+ row that violated a unique constraint for
+ <literal>insert_exists</literal>, <literal>update_exists</literal> or
+ <literal>multiple_unique_conflicts</literal> conflicts.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>local row</literal> section includes the local row if its
+ origin differs from the remote row for
+ <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
+ conflicts, or if the key value conflicts with the remote row for
+ <literal>insert_exists</literal>, <literal>update_exists</literal> or
+ <literal>multiple_unique_conflicts</literal> conflicts.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>remote row</literal> section includes the new row from
+ the remote insert or update operation that caused the conflict. Note that
+ for an update operation, the column value of the new row will be null
+ if the value is unchanged and toasted.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>replica identity</literal> section includes the replica
+ identity key values that were used to search for the existing local
+ row to be updated or deleted. This may include the full row value
+ if the local relation is marked with
+ <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <replaceable class="parameter">column_name</replaceable> is the column name.
+ For <literal>local row</literal>, <literal>remote row</literal>, and
+ <literal>replica identity full</literal> cases, column names are
+ logged only if the user lacks the privilege to access all columns of
+ the table. If column names are present, they appear in the same order
+ as the corresponding column values.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <replaceable class="parameter">column_value</replaceable> is the column value.
+ The large column values are truncated to 64 bytes.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
+ multiple <replaceable class="parameter">detailed_explanation</replaceable>
+ and <replaceable class="parameter">detail_values</replaceable> lines
+ will be generated, each detailing the conflict information associated
+ with distinct unique constraints.
</para>
</listitem>
</itemizedlist>
</listitem>
- </varlistentry>
-
- <varlistentry>
- <term><literal>DETAIL</literal></term>
- <listitem>
- <itemizedlist>
- <listitem>
- <para>
- <replaceable class="parameter">detailed_explanation</replaceable> includes
- the origin, transaction ID, and commit timestamp of the transaction that
- modified the local row, if available.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>key</literal> section includes the key values of the local
- row that violated a unique constraint for
- <literal>insert_exists</literal>, <literal>update_exists</literal> or
- <literal>multiple_unique_conflicts</literal> conflicts.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>local row</literal> section includes the local row if its
- origin differs from the remote row for
- <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
- conflicts, or if the key value conflicts with the remote row for
- <literal>insert_exists</literal>, <literal>update_exists</literal> or
- <literal>multiple_unique_conflicts</literal> conflicts.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>remote row</literal> section includes the new row from
- the remote insert or update operation that caused the conflict. Note that
- for an update operation, the column value of the new row will be null
- if the value is unchanged and toasted.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>replica identity</literal> section includes the replica
- identity key values that were used to search for the existing local
- row to be updated or deleted. This may include the full row value
- if the local relation is marked with
- <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
- </para>
- </listitem>
- <listitem>
- <para>
- <replaceable class="parameter">column_name</replaceable> is the column name.
- For <literal>local row</literal>, <literal>remote row</literal>, and
- <literal>replica identity full</literal> cases, column names are
- logged only if the user lacks the privilege to access all columns of
- the table. If column names are present, they appear in the same order
- as the corresponding column values.
- </para>
- </listitem>
- <listitem>
- <para>
- <replaceable class="parameter">column_value</replaceable> is the column value.
- The large column values are truncated to 64 bytes.
- </para>
- </listitem>
- <listitem>
- <para>
- Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
- multiple <replaceable class="parameter">detailed_explanation</replaceable>
- and <replaceable class="parameter">detail_values</replaceable> lines
- will be generated, each detailing the conflict information associated
- with distinct unique
- constraints.
- </para>
- </listitem>
- </itemizedlist>
- </listitem>
- </varlistentry>
- </variablelist>
- </para>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </sect2>
- <para>
- Logical replication operations are performed with the privileges of the role
- which owns the subscription. Permissions failures on target tables will
- cause replication conflicts, as will enabled
- <link linkend="ddl-rowsecurity">row-level security</link> on target tables
- that the subscription owner is subject to, without regard to whether any
- policy would ordinarily reject the <command>INSERT</command>,
- <command>UPDATE</command>, <command>DELETE</command> or
- <command>TRUNCATE</command> which is being replicated. This restriction on
- row-level security may be lifted in a future version of
- <productname>PostgreSQL</productname>.
- </para>
+ <sect2 id="logical-replication-conflict-notes">
+ <title>Notes</title>
+ <para>
+ Logical replication operations are performed with the privileges of the role
+ which owns the subscription. Permissions failures on target tables will
+ cause replication conflicts, as will enabled
+ <link linkend="ddl-rowsecurity">row-level security</link> on target tables
+ that the subscription owner is subject to, without regard to whether any
+ policy would ordinarily reject the <command>INSERT</command>,
+ <command>UPDATE</command>, <command>DELETE</command> or
+ <command>TRUNCATE</command> which is being replicated. This restriction on
+ row-level security may be lifted in a future version of
+ <productname>PostgreSQL</productname>.
+ </para>
- <para>
- A conflict that produces an error will stop the replication; it must be
- resolved manually by the user. Details about the conflict can be found in
- the subscriber's server log.
- </para>
+ <para>
+ A conflict that produces an error will stop the replication; it must be
+ resolved manually by the user. Details about the conflict can be found in
+ the subscriber's server log.
+ </para>
- <para>
- The resolution can be done either by changing data or permissions on the subscriber so
- that it does not conflict with the incoming change or by skipping the
- transaction that conflicts with the existing data. When a conflict produces
- an error, the replication won't proceed, and the logical replication worker will
- emit the following kind of message to the subscriber's server log:
+ <para>
+ The resolution can be done either by changing data or permissions on the subscriber so
+ that it does not conflict with the incoming change or by skipping the
+ transaction that conflicts with the existing data. When a conflict produces
+ an error, the replication won't proceed, and the logical replication worker will
+ emit the following kind of message to the subscriber's server log:
<screen>
ERROR: conflict detected on relation "public.test": conflict=insert_exists
DETAIL: Could not apply remote change: remote row (1, 'remote').
Key already exists in unique index "test_pkey", modified locally in transaction 800 at 2026-01-16 18:15:25.652759+09: key (c)=(1), local row (1, 'local').
CONTEXT: processing remote data for replication origin "pg_16395" during "INSERT" for replication target relation "public.test" in transaction 725 finished at 0/014C0378
</screen>
- The LSN of the transaction that contains the change violating the constraint and
- the replication origin name can be found from the server log (LSN 0/014C0378 and
- replication origin <literal>pg_16395</literal> in the above case). The
- transaction that produced the conflict can be skipped by using
- <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
- with the finish LSN
- (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction
- is committed or prepared on the publisher. Alternatively, the transaction can
- also be skipped by calling the <link linkend="pg-replication-origin-advance">
- <function>pg_replication_origin_advance()</function></link> function.
- Before using this function, the subscription needs to be disabled temporarily
- either by <link linkend="sql-altersubscription-params-disable">
- <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
- subscription can be used with the
- <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
- option. Then, you can use <function>pg_replication_origin_advance()</function>
- function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
- and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of
- origins can be seen in the <link linkend="view-pg-replication-origin-status">
- <structname>pg_replication_origin_status</structname></link> system view.
- Please note that skipping the whole transaction includes skipping changes that
- might not violate any constraint. This can easily make the subscriber
- inconsistent.
- The additional details regarding conflicting rows, such as their origin and
- commit timestamp can be seen in the <literal>DETAIL</literal> line of the
- log. But note that this information is only available when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Users can use this information to decide
- whether to retain the local change or adopt the remote alteration. For
- instance, the <literal>DETAIL</literal> line in the above log indicates that
- the existing row was modified locally. Users can manually perform a
- remote-change-win.
- </para>
-
- <para>
- When the
- <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
- mode is <literal>parallel</literal>, the finish LSN of failed transactions
- may not be logged. In that case, it may be necessary to change the streaming
- mode to <literal>on</literal> or <literal>off</literal> and cause the same
- conflicts again so the finish LSN of the failed transaction will be written
- to the server log. For the usage of finish LSN, please refer to <link
- linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
- SKIP</command></link>.
- </para>
+ The LSN of the transaction that contains the change violating the constraint and
+ the replication origin name can be found from the server log (LSN 0/014C0378 and
+ replication origin <literal>pg_16395</literal> in the above case). The
+ transaction that produced the conflict can be skipped by using
+ <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
+ with the finish LSN
+ (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction
+ is committed or prepared on the publisher. Alternatively, the transaction can
+ also be skipped by calling the <link linkend="pg-replication-origin-advance">
+ <function>pg_replication_origin_advance()</function></link> function.
+ Before using this function, the subscription needs to be disabled temporarily
+ either by <link linkend="sql-altersubscription-params-disable">
+ <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
+ subscription can be used with the
+ <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
+ option. Then, you can use <function>pg_replication_origin_advance()</function>
+ function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
+ and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of
+ origins can be seen in the <link linkend="view-pg-replication-origin-status">
+ <structname>pg_replication_origin_status</structname></link> system view.
+ Please note that skipping the whole transaction includes skipping changes that
+ might not violate any constraint. This can easily make the subscriber
+ inconsistent.
+ The additional details regarding conflicting rows, such as their origin and
+ commit timestamp can be seen in the <literal>DETAIL</literal> line of the
+ log. But note that this information is only available when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Users can use this information to decide
+ whether to retain the local change or adopt the remote alteration. For
+ instance, the <literal>DETAIL</literal> line in the above log indicates that
+ the existing row was modified locally. Users can manually perform a
+ remote-change-win.
+ </para>
+
+ <para>
+ When the
+ <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
+ mode is <literal>parallel</literal>, the finish LSN of failed transactions
+ may not be logged. In that case, it may be necessary to change the streaming
+ mode to <literal>on</literal> or <literal>off</literal> and cause the same
+ conflicts again so the finish LSN of the failed transaction will be written
+ to the server log. For the usage of finish LSN, please refer to <link
+ linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
+ SKIP</command></link>.
+ </para>
+ </sect2>
</sect1>
<sect1 id="logical-replication-restrictions">
@@ -2415,6 +2538,14 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
key or replica identity defined for it.
</para>
</listitem>
+
+ <listitem>
+ <para>
+ Conflict log tables (see <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link> parameter)
+ are never published, even when using <literal>FOR ALL TABLES</literal> in a
+ publication.
+ </para>
+ </listitem>
</itemizedlist>
</sect1>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index e4f0b6b16c7..13b413d142f 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -293,8 +293,9 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>,
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>,
<link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>,
- <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>, and
- <link linkend="sql-createsubscription-params-with-wal-receiver-timeout"><literal>wal_receiver_timeout</literal></link>.
+ <link linkend="sql-createsubscription-params-with-max-retention-duration"><literal>max_retention_duration</literal></link>,
+ <link linkend="sql-createsubscription-params-with-wal-receiver-timeout"><literal>wal_receiver_timeout</literal></link>, and
+ <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>.
Only a superuser can set <literal>password_required = false</literal>.
</para>
@@ -352,6 +353,14 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
<quote><literal>pg_conflict_detection</literal></quote>, created to retain
dead tuples for conflict detection, will be dropped.
</para>
+
+ <para>
+ When the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ parameter is set to <literal>table</literal> or <literal>all</literal>, the system
+ automatically creates the conflict log table. Conversely, if the destination is changed to
+ <literal>log</literal>, logging to the table stops and the conflict log
+ table is automatically dropped.
+ </para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 07d5b1bd77c..527f5ab874a 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -261,6 +261,57 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</listitem>
</varlistentry>
+ <varlistentry id="sql-createsubscription-params-with-conflict-log-destination">
+ <term><literal>conflict_log_destination</literal> (<type>enum</type>)</term>
+ <listitem>
+ <para>
+ Specifies the destination for recording logical replication conflicts.
+ </para>
+ <para>
+ The available destinations are:
+ <itemizedlist>
+ <listitem>
+ <para>
+ <literal>log</literal>: Conflict details are recorded in the server log.
+ This is the default behavior. See
+ <xref linkend="logical-replication-conflict-file-based-logging"/>
+ for details.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <literal>table</literal>: The system automatically creates a structured table
+ named <literal>pg_conflict_log_for_subid_<subid></literal>
+ in the <literal>pg_conflict</literal> schema. This allows for easy
+ querying and analysis of conflicts. See
+ <xref linkend="logical-replication-conflict-table-based-logging"/>
+ for details.
+ </para>
+ <caution>
+ <para>
+ The conflict log table is strictly tied to the lifecycle of the
+ subscription or the <literal>conflict_log_destination</literal> setting. If
+ the subscription is dropped, or if the destination is changed to
+ <literal>log</literal>, the table and all its recorded conflict data are
+ <emphasis>permanently deleted</emphasis>.
+ </para>
+ <para>
+ If conflict history may be needed later, back up the conflict log
+ table before it gets removed.
+ </para>
+ </caution>
+ </listitem>
+ <listitem>
+ <para>
+ <literal>all</literal>: Records conflict details to both destinations
+ <literal>log</literal> and <literal>table</literal>.
+ </para>
+ </listitem>
+ </itemizedlist>
+ </para>
+ </listitem>
+ </varlistentry>
+
<varlistentry id="sql-createsubscription-params-with-copy-data">
<term><literal>copy_data</literal> (<type>boolean</type>)</term>
<listitem>
--
2.53.0
[application/octet-stream] v47-0004-Add-conflict-log-table-information-to-describe-s.patch (78.2K, 6-v47-0004-Add-conflict-log-table-information-to-describe-s.patch)
download | inline diff:
From abff3c1d35d5c92af7df1623755992cc7be69948 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 8 Jun 2026 12:42:40 +0530
Subject: [PATCH v47 4/5] Add conflict log table information to describe
subscription output
Display the associated conflict log table as a footer in \dRs+
output when conflict logging to table/all is enabled for a
subscription.
Previously, subscriptions were displayed using a single tabular
output format. Since the conflict log table information is specific
to each subscription and is better suited as auxiliary information,
change the output to display each subscription individually in a
row-wise table format and show the conflict log table as a footer
when applicable.
This approach was chosen based on suggestions at:
https://www.postgresql.org/message-id/CAA4eK1KdKqKkaTqcj3in6ehD_hg6oOaCF_-JsVfd8N6nS8oV9g%40mail.gmail.com
---
src/bin/psql/command.c | 5 +-
src/bin/psql/describe.c | 404 +++++++++++++++++----
src/bin/psql/describe.h | 5 +-
src/test/regress/expected/subscription.out | 176 ++++-----
4 files changed, 421 insertions(+), 169 deletions(-)
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index 01b8f11aadd..777d0553246 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -1220,7 +1220,10 @@ exec_command_d(PsqlScanState scan_state, bool active_branch, const char *cmd)
success = listPublications(pattern);
break;
case 's':
- success = describeSubscriptions(pattern, show_verbose);
+ if (show_verbose)
+ success = describeSubscriptions(pattern);
+ else
+ success = listSubscriptions(pattern);
break;
default:
status = PSQL_CMD_UNKNOWN;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index e1449654f96..3bd1dfcc3af 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -23,6 +23,7 @@
#include "catalog/pg_collation_d.h"
#include "catalog/pg_constraint_d.h"
#include "catalog/pg_default_acl_d.h"
+#include "catalog/pg_namespace_d.h"
#include "catalog/pg_proc_d.h"
#include "catalog/pg_propgraph_element_d.h"
#include "catalog/pg_publication_d.h"
@@ -7081,19 +7082,17 @@ error_return:
/*
* \dRs
- * Describes subscriptions.
+ * Lists subscriptions.
*
* Takes an optional regexp to select particular subscriptions
*/
bool
-describeSubscriptions(const char *pattern, bool verbose)
+listSubscriptions(const char *pattern)
{
PQExpBufferData buf;
PGresult *res;
printQueryOpt myopt = pset.popt;
- static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false, false, false,
- false, false, false, false, false, false, false};
+ static const bool translate_columns[] = {false, false, false, false};
if (pset.sversion < 100000)
{
@@ -7108,101 +7107,206 @@ describeSubscriptions(const char *pattern, bool verbose)
initPQExpBuffer(&buf);
printfPQExpBuffer(&buf, "/* %s */\n", _("Get matching subscriptions"));
+
+ /* Only display subscriptions in current database. */
appendPQExpBuffer(&buf,
"SELECT subname AS \"%s\"\n"
", pg_catalog.pg_get_userbyid(subowner) AS \"%s\"\n"
", subenabled AS \"%s\"\n"
- ", subpublications AS \"%s\"\n",
+ ", subpublications AS \"%s\"\n"
+ "FROM pg_catalog.pg_subscription\n"
+ "WHERE subdbid = (SELECT oid\n"
+ " FROM pg_catalog.pg_database\n"
+ " WHERE datname = pg_catalog.current_database())",
gettext_noop("Name"),
gettext_noop("Owner"),
gettext_noop("Enabled"),
gettext_noop("Publication"));
- if (verbose)
+ if (!validateSQLNamePattern(&buf, pattern, true, false,
+ NULL, "subname", NULL,
+ NULL,
+ NULL, 1))
{
- /* Binary mode and streaming are only supported in v14 and higher */
- if (pset.sversion >= 140000)
- {
- appendPQExpBuffer(&buf,
- ", subbinary AS \"%s\"\n",
- gettext_noop("Binary"));
+ termPQExpBuffer(&buf);
+ return false;
+ }
- if (pset.sversion >= 160000)
- appendPQExpBuffer(&buf,
- ", (CASE substream\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_OFF) " THEN 'off'\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_ON) " THEN 'on'\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_PARALLEL) " THEN 'parallel'\n"
- " END) AS \"%s\"\n",
- gettext_noop("Streaming"));
- else
- appendPQExpBuffer(&buf,
- ", substream AS \"%s\"\n",
- gettext_noop("Streaming"));
- }
+ appendPQExpBufferStr(&buf, "ORDER BY 1;");
- /* Two_phase and disable_on_error are only supported in v15 and higher */
- if (pset.sversion >= 150000)
- appendPQExpBuffer(&buf,
- ", subtwophasestate AS \"%s\"\n"
- ", subdisableonerr AS \"%s\"\n",
- gettext_noop("Two-phase commit"),
- gettext_noop("Disable on error"));
+ res = PSQLexec(buf.data);
+ termPQExpBuffer(&buf);
+ if (!res)
+ return false;
- if (pset.sversion >= 160000)
- appendPQExpBuffer(&buf,
- ", suborigin AS \"%s\"\n"
- ", subpasswordrequired AS \"%s\"\n"
- ", subrunasowner AS \"%s\"\n",
- gettext_noop("Origin"),
- gettext_noop("Password required"),
- gettext_noop("Run as owner?"));
+ myopt.title = _("List of subscriptions");
+ myopt.translate_header = true;
+ myopt.translate_columns = translate_columns;
+ myopt.n_translate_columns = lengthof(translate_columns);
- if (pset.sversion >= 170000)
+ printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+
+ PQclear(res);
+
+ return true;
+}
+
+/*
+ * \dRs+
+ * Describes subscriptions.
+ *
+ * Takes an optional regexp to select particular subscriptions
+ */
+bool
+describeSubscriptions(const char *pattern)
+{
+ PQExpBufferData buf;
+ int i;
+ PGresult *res;
+ int ncols;
+ int nrows = 1;
+
+ PQExpBufferData title;
+ printTableContent cont;
+
+ if (pset.sversion < 100000)
+ {
+ char sverbuf[32];
+
+ pg_log_error("The server (version %s) does not support subscriptions.",
+ formatPGVersionNumber(pset.sversion, false,
+ sverbuf, sizeof(sverbuf)));
+ return true;
+ }
+
+ initPQExpBuffer(&buf);
+
+ printfPQExpBuffer(&buf, "/* %s */\n", _("Get matching subscriptions"));
+ appendPQExpBuffer(&buf,
+ "SELECT oid, subname AS \"%s\"\n"
+ ", (SELECT nspname FROM pg_namespace WHERE oid = " CppAsString2(PG_CONFLICT_NAMESPACE) ") AS \"%s\"\n"
+ ", pg_catalog.pg_get_userbyid(subowner) AS \"%s\"\n"
+ ", subenabled AS \"%s\"\n"
+ ", subpublications AS \"%s\"\n",
+ gettext_noop("Name"),
+ gettext_noop("Conflict_schema"),
+ gettext_noop("Owner"),
+ gettext_noop("Enabled"),
+ gettext_noop("Publication"));
+
+ /*
+ * oid, subname and conflict_schema columns are internal and not displayed,
+ * so only 3 visible columns.
+ */
+ ncols = 3;
+
+ /* Binary mode and streaming are only supported in v14 and higher */
+ if (pset.sversion >= 140000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subbinary AS \"%s\"\n",
+ gettext_noop("Binary"));
+ ncols++;
+
+ if (pset.sversion >= 160000)
appendPQExpBuffer(&buf,
- ", subfailover AS \"%s\"\n",
- gettext_noop("Failover"));
- if (pset.sversion >= 190000)
- {
+ ", (CASE substream\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_OFF) " THEN 'off'\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_ON) " THEN 'on'\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_PARALLEL) " THEN 'parallel'\n"
+ " END) AS \"%s\"\n",
+ gettext_noop("Streaming"));
+ else
appendPQExpBuffer(&buf,
- ", (select srvname from pg_foreign_server where oid=subserver) AS \"%s\"\n",
- gettext_noop("Server"));
+ ", substream AS \"%s\"\n",
+ gettext_noop("Streaming"));
- appendPQExpBuffer(&buf,
- ", subretaindeadtuples AS \"%s\"\n",
- gettext_noop("Retain dead tuples"));
+ ncols++;
+ }
- appendPQExpBuffer(&buf,
- ", submaxretention AS \"%s\"\n",
- gettext_noop("Max retention duration"));
+ /* Two_phase and disable_on_error are only supported in v15 and higher */
+ if (pset.sversion >= 150000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subtwophasestate AS \"%s\"\n"
+ ", subdisableonerr AS \"%s\"\n",
+ gettext_noop("Two-phase commit"),
+ gettext_noop("Disable on error"));
+ ncols += 2;
+ }
- appendPQExpBuffer(&buf,
- ", subretentionactive AS \"%s\"\n",
- gettext_noop("Retention active"));
- }
+ if (pset.sversion >= 160000)
+ {
+ appendPQExpBuffer(&buf,
+ ", suborigin AS \"%s\"\n"
+ ", subpasswordrequired AS \"%s\"\n"
+ ", subrunasowner AS \"%s\"\n",
+ gettext_noop("Origin"),
+ gettext_noop("Password required"),
+ gettext_noop("Run as owner?"));
+ ncols += 3;
+ }
+ if (pset.sversion >= 170000)
+ {
appendPQExpBuffer(&buf,
- ", subsynccommit AS \"%s\"\n"
- ", subconninfo AS \"%s\"\n",
- gettext_noop("Synchronous commit"),
- gettext_noop("Conninfo"));
+ ", subfailover AS \"%s\"\n",
+ gettext_noop("Failover"));
+ ncols++;
+ }
- if (pset.sversion >= 190000)
- appendPQExpBuffer(&buf,
- ", subwalrcvtimeout AS \"%s\"\n",
- gettext_noop("Receiver timeout"));
+ if (pset.sversion >= 190000)
+ {
+ appendPQExpBuffer(&buf,
+ ", (select srvname from pg_foreign_server where oid=subserver) AS \"%s\"\n"
+ ", subretaindeadtuples AS \"%s\"\n"
+ ", submaxretention AS \"%s\"\n"
+ ", subretentionactive AS \"%s\"\n",
+ gettext_noop("Server"),
+ gettext_noop("Retain dead tuples"),
+ gettext_noop("Max retention duration"),
+ gettext_noop("Retention active"));
+ ncols += 4;
+ }
- /* Skip LSN is only supported in v15 and higher */
- if (pset.sversion >= 150000)
- appendPQExpBuffer(&buf,
- ", subskiplsn AS \"%s\"\n",
- gettext_noop("Skip LSN"));
+ appendPQExpBuffer(&buf,
+ ", subsynccommit AS \"%s\"\n"
+ ", subconninfo AS \"%s\"\n",
+ gettext_noop("Synchronous commit"),
+ gettext_noop("Conninfo"));
+ ncols += 2;
+ if (pset.sversion >= 190000)
+ {
appendPQExpBuffer(&buf,
- ", pg_catalog.obj_description(oid, 'pg_subscription') AS \"%s\"\n",
- gettext_noop("Description"));
+ ", subwalrcvtimeout AS \"%s\"\n",
+ gettext_noop("Receiver timeout"));
+ ncols++;
}
+ /* Skip LSN is only supported in v15 and higher */
+ if (pset.sversion >= 150000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subskiplsn AS \"%s\"\n",
+ gettext_noop("Skip LSN"));
+ ncols++;
+ }
+
+ /* Conflict log destination is supported in v19 and higher */
+ if (pset.sversion >= 190000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subconflictlogdest AS \"%s\"\n",
+ gettext_noop("Conflict log destination"));
+ ncols++;
+ }
+
+ appendPQExpBuffer(&buf,
+ ", pg_catalog.obj_description(oid, 'pg_subscription') AS \"%s\"\n",
+ gettext_noop("Description"));
+ ncols++;
+
/* Only display subscriptions in current database. */
appendPQExpBufferStr(&buf,
"FROM pg_catalog.pg_subscription\n"
@@ -7219,20 +7323,162 @@ describeSubscriptions(const char *pattern, bool verbose)
return false;
}
- appendPQExpBufferStr(&buf, "ORDER BY 1;");
+ appendPQExpBufferStr(&buf, "ORDER BY subname;");
res = PSQLexec(buf.data);
termPQExpBuffer(&buf);
if (!res)
return false;
- myopt.title = _("List of subscriptions");
- myopt.translate_header = true;
- myopt.translate_columns = translate_columns;
- myopt.n_translate_columns = lengthof(translate_columns);
+ if (PQntuples(res) == 0)
+ {
+ if (!pset.quiet)
+ {
+ if (pattern)
+ pg_log_error("Did not find any subscription named \"%s\".",
+ pattern);
+ else
+ pg_log_error("Did not find any subscriptions.");
+ }
- printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+ termPQExpBuffer(&buf);
+ PQclear(res);
+ return false;
+ }
+
+ for (i = 0; i < PQntuples(res); i++)
+ {
+ const char align = 'l';
+ Oid subid = atooid(PQgetvalue(res, i, 0));
+ char *subname = PQgetvalue(res, i, 1);
+ char *conflict_schema = PQgetvalue(res, i, 2);
+ int current_col = 3;
+ char *logdest;
+ printTableOpt myopt = pset.popt.topt;
+
+ initPQExpBuffer(&title);
+ printfPQExpBuffer(&title, _("Subscription %s"), subname);
+ printTableInit(&cont, &myopt, title.data, ncols, nrows);
+
+ printTableAddHeader(&cont, gettext_noop("Owner"), true, align);
+ printTableAddHeader(&cont, gettext_noop("Enabled"), true, align);
+ printTableAddHeader(&cont, gettext_noop("Publication"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ if (pset.sversion >= 140000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Binary"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Streaming"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 150000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Two-phase commit"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Disable on error"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 160000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Origin"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Password required"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Run as owner?"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 170000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Failover"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Server"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Retain dead tuples"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Max retention duration"),
+ true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Retention active"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ printTableAddHeader(&cont, gettext_noop("Synchronous commit"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Conninfo"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Receiver timeout"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 150000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Skip LSN"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Conflict log destination"),
+ true, align);
+
+ logdest = PQgetvalue(res, i, current_col++);
+
+ printTableAddCell(&cont, logdest, false, false);
+ }
+
+ printTableAddHeader(&cont, gettext_noop("Description"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ if (pset.sversion >= 190000)
+ {
+ if (strcmp(logdest, "table") == 0 || strcmp(logdest, "all") == 0)
+ {
+ /*
+ * The size accounts for schema name (NAMEDATALEN), relation
+ * name (NAMEDATALEN), separator '.' and null terminator.
+ */
+ char conflictlogtable[NAMEDATALEN + NAMEDATALEN + 2];
+
+ snprintf(conflictlogtable,
+ sizeof(conflictlogtable),
+ "%s." CONFLICT_LOG_RELATION_NAME_FMT,
+ conflict_schema, subid);
+
+ printTableAddFooter(&cont, _("Conflict log table:"));
+ printTableAddFooter(&cont,
+ psprintf(" \"%s\"", conflictlogtable));
+ }
+ }
+
+ printTable(&cont, pset.queryFout, false, pset.logfile);
+ printTableCleanup(&cont);
+
+ termPQExpBuffer(&title);
+ }
+
+ termPQExpBuffer(&buf);
PQclear(res);
return true;
}
diff --git a/src/bin/psql/describe.h b/src/bin/psql/describe.h
index 47fae5ceafb..15c6c685323 100644
--- a/src/bin/psql/describe.h
+++ b/src/bin/psql/describe.h
@@ -126,7 +126,10 @@ bool listPublications(const char *pattern);
bool describePublications(const char *pattern);
/* \dRs */
-bool describeSubscriptions(const char *pattern, bool verbose);
+bool listSubscriptions(const char *pattern);
+
+/* \dRs+ */
+bool describeSubscriptions(const char *pattern);
/* \dAc */
extern bool listOperatorClasses(const char *access_method_pattern,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 9a693ed010a..9e44ffccfd1 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -139,18 +139,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub4
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub4
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -215,10 +215,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log | test subscription
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -227,10 +227,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | log | test subscription
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -246,10 +246,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | log | test subscription
(1 row)
-- ok - with lsn = NONE
@@ -258,10 +258,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | log | test subscription
(1 row)
BEGIN;
@@ -297,10 +297,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s');
ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar');
ERROR: invalid value for parameter "wal_receiver_timeout": "foobar"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription
+ Subscription regress_testsub_foo
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | log | test subscription
(1 row)
-- rename back to keep the rest simple
@@ -329,19 +329,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -353,27 +353,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- fail - publication already exists
@@ -388,10 +388,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- fail - publication used more than once
@@ -406,10 +406,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -445,19 +445,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- we can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -467,10 +467,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -483,18 +483,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -507,10 +507,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -524,19 +524,19 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- ok
ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
--
2.53.0
view thread (100+ 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]
Subject: Re: Proposal: Conflict log history table for Logical Replication
In-Reply-To: <CALDaNm2V3EuSKMaTqDvaiLQW3jwBX90aXTkMST1ft=uJ8J+R5A@mail.gmail.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