public inbox for [email protected]
help / color / mirror / Atom feedFrom: Dilip Kumar <[email protected]>
To: Amit Kapila <[email protected]>
Cc: Nisha Moond <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Peter Smith <[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: Sat, 30 May 2026 03:36:59 +0530
Message-ID: <CAFiTN-sx=k+Th=uYsrLcS6YMZbPVi9Wrggn1w2Nzf9MLEU7YRQ@mail.gmail.com> (raw)
In-Reply-To: <CAFiTN-s-tuxar9Dp5He0CFa1pzfy1fmiwcBj6PtwD0hDodE5ng@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>
On Sat, May 30, 2026 at 3:24 AM Dilip Kumar <[email protected]> wrote:
>
> On Fri, May 29, 2026 at 5:11 AM Amit Kapila <[email protected]> wrote:
> >
> > On Thu, May 21, 2026 at 9:51 PM Nisha Moond <[email protected]> wrote:
> > >
> > > On Wed, May 20, 2026 at 3:05 PM vignesh C <[email protected]> wrote:
> > > >
> > > > Rest of the comments were fixed.
> > > > The attached v37 version patch has the changes for the same. Also
> > > > Peter's comments on the documentation patch from [1] and Shveta's
> > > > comments from [2] are addressed in the attached patch.
> > > >
> > >
> > > Here are few comments based on v37 testing:
> > >
> > > 1) Should we consider using TOAST tables for tuple-data columns like
> > > remote_tuple and local_conflicts (the JSON columns)?
> > > This may be a corner case, but if the tuple data becomes too large to
> > > fit into an 8KB heap tuple, then the apply worker keeps failing while
> > > inserting into the CLT with errors like:
> > >
> > > ERROR: row is too big: size 19496, maximum size 8160
> > > LOG: background worker "logical replication apply worker" (PID
> > > 41226) exited with exit code 1
> > >
> >
> > In the docs, it is mentioned: "column_value is the column value. The
> > large column values are truncated to 64 bytes." [1], so I wonder, if
> > we follow this why we need toast entries? Did you tried any case where
> > you are getting above ERROR?
>
> But in this case we are talking about the JSON column of the CLT which
> might contain a full local tuple or even multiple local tuples if a
> remote tuple conflicts with multiple local rows. So, IMHO, we need a
> toast table. Nisha, have you already tested the scenario? If yes, can
> you share your test case?
After putting more thought, I think instead of executing a three-step
process i.e. inserting the pg_subscription tuple, creating the table
with its dependency, and then going back to update the tuple with the
new relation ID, it is much cleaner to do it linearly, i.e. we should
create the conflict log table first to get its OID, insert the
subscription tuple pre-populated with that ID, and then record the
dependency. This achieves the exact same state in a single direct
sequence without the redundant catalog update within the same command.
I agree with that code we would have to keep the record dependency
code in CreateSubscription and AlterSubscription functions, but after
putting more thought I think in thoese function we are already
recording subscription dependencies with other object so wouldn't it
be more natural to add this depednecy as well at the same place?
Anyway I am ready to change that if we have strong opinion against
this approach.
Here is the updated patch and changes are
1. 0003 and 0004 are merged on 0001
2. Merged Amit's v41_amit_1.patch.txt to 0002
3. Fix the dependency order issue (i.e. create dependency after
inserting subscription tuple) and merged in 0002
Open Items:
1. Need to create toast table for CLT after testing with larger JSON row
2. Fixed review comments of Shveta on 0004 and 0005
3. Rebase Vignesh's patch of
"v41-0007-Preserve-conflict-log-destination-and-subscripti" I think we
can do that once we have concensus on whether to create conflict log
table first or insert the subscription row first as based on this
change we would have to rebase this patch again.
4. Once we rebase
"v41-0007-Preserve-conflict-log-destination-and-subscripti" after
dependency order consensus I would rebase doc patch and \dRs+ change
patch of Vignesh.
--
Regards,
Dilip Kumar
Google
Attachments:
[application/octet-stream] v42-0001-Add-configurable-conflict-log-table-for-Logical-.patch (124.5K, 2-v42-0001-Add-configurable-conflict-log-table-for-Logical-.patch)
download | inline diff:
From 19841a561880a8ad0d5010d8f03671f4e14f123e Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Fri, 29 May 2026 15:05:15 +0530
Subject: [PATCH v42 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 | 14 +-
src/backend/catalog/catalog.c | 28 +-
src/backend/catalog/heap.c | 3 +-
src/backend/catalog/namespace.c | 11 +-
src/backend/catalog/pg_publication.c | 11 +-
src/backend/catalog/pg_subscription.c | 7 +
src/backend/commands/subscriptioncmds.c | 262 ++++++++++-
src/backend/commands/tablecmds.c | 6 +-
src/backend/executor/execMain.c | 29 ++
src/backend/replication/logical/conflict.c | 25 ++
src/bin/initdb/initdb.c | 6 +
src/bin/psql/tab-complete.in.c | 8 +-
src/include/catalog/catalog.h | 2 +
src/include/catalog/pg_namespace.dat | 3 +
src/include/catalog/pg_subscription.h | 11 +
src/include/commands/subscriptioncmds.h | 3 +
src/include/replication/conflict.h | 32 ++
src/test/regress/expected/subscription.out | 483 +++++++++++++++++----
src/test/regress/sql/subscription.sql | 254 +++++++++++
src/tools/pgindent/typedefs.list | 2 +
20 files changed, 1094 insertions(+), 106 deletions(-)
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 007ede997c5..84ef5304e22 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -3344,12 +3344,20 @@ pg_class_aclmask_ext(Oid table_oid, Oid roleid, AclMode mask,
* 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.
+ *
+ * For conflict log tables, we allow non-superusers to perform DELETE
+ * and TRUNCATE for maintenance, while still restricting INSERT,
+ * UPDATE, and USAGE.
*/
if ((mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE)) &&
- IsSystemClass(table_oid, classForm) &&
- classForm->relkind != RELKIND_VIEW &&
+ IsConflictClass(classForm) &&
!superuser_arg(roleid))
- mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE);
+ mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_USAGE);
+ else 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.
diff --git a/src/backend/catalog/catalog.c b/src/backend/catalog/catalog.c
index 7be49032934..4578cd07140 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) ||
+ IsConflictClass(reltuple));
}
/*
@@ -230,6 +232,18 @@ IsToastClass(Form_pg_class reltuple)
return IsToastNamespace(relnamespace);
}
+/*
+ * IsConflictClass - Check if the given pg_class tuple belongs to the conflict
+ * namespace.
+ */
+bool
+IsConflictClass(Form_pg_class reltuple)
+{
+ Oid relnamespace = reltuple->relnamespace;
+
+ return IsConflictNamespace(relnamespace);
+}
+
/*
* IsCatalogNamespace
* True iff namespace is pg_catalog.
@@ -264,6 +278,18 @@ IsToastNamespace(Oid namespaceId)
isTempToastNamespace(namespaceId);
}
+/*
+ * IsConflictNamespace
+ * True iff namespace is pg_conflict.
+ *
+ * Does not perform any catalog accesses.
+ */
+bool
+IsConflictNamespace(Oid namespaceId)
+{
+ return namespaceId == PG_CONFLICT_NAMESPACE;
+}
+
/*
* IsReservedName
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 88087654de9..72bd7faaf2d 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -315,7 +315,8 @@ heap_create(const char *relname,
*/
if (!allow_system_table_mods &&
((IsCatalogNamespace(relnamespace) && relkind != RELKIND_INDEX) ||
- IsToastNamespace(relnamespace)) &&
+ IsToastNamespace(relnamespace) ||
+ IsConflictNamespace(relnamespace)) &&
IsNormalProcessingMode())
ereport(ERROR,
(errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
diff --git a/src/backend/catalog/namespace.c b/src/backend/catalog/namespace.c
index 56b87d878e8..c35fcf57fd4 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 CONFLICT schema")));
}
/*
diff --git a/src/backend/catalog/pg_publication.c b/src/backend/catalog/pg_publication.c
index 5c457d9aca8..c680356a10b 100644
--- a/src/backend/catalog/pg_publication.c
+++ b/src/backend/catalog/pg_publication.c
@@ -103,6 +103,13 @@ check_publication_add_relation(PublicationRelInfo *pri)
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg(errormsg, relname),
errdetail("This operation is not supported for unlogged tables.")));
+
+ /* Can't be conflict log table */
+ if (IsConflictNamespace(RelationGetNamespace(targetrel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg(errormsg, relname),
+ errdetail("This operation is not supported for conflict log tables.")));
}
/*
@@ -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) ||
+ IsConflictNamespace(schemaid))
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("cannot add schema \"%s\" to publication",
@@ -157,6 +165,7 @@ is_publishable_class(Oid relid, Form_pg_class reltuple)
reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
reltuple->relkind == RELKIND_SEQUENCE) &&
!IsCatalogRelationOid(relid) &&
+ !IsConflictClass(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/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 523959ba0ce..afa1401f521 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -21,13 +21,16 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
+#include "catalog/heap.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
+#include "catalog/pg_am_d.h"
#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 +38,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 +83,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 +112,7 @@ typedef struct SubOpts
bool retaindeadtuples;
int32 maxretention;
char *origin;
+ ConflictLogDest conflictlogdest;
XLogRecPtr lsn;
char *wal_receiver_timeout;
} SubOpts;
@@ -140,7 +146,7 @@ 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 Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner);
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -196,6 +202,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 +439,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 = GetLogDestination(val);
+ opts->specified_opts |= SUBOPT_CONFLICT_LOG_DEST;
+ }
else
ereport(ERROR,
(errcode(ERRCODE_SYNTAX_ERROR),
@@ -629,6 +649,7 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
uint32 supported_opts;
SubOpts opts = {0};
AclResult aclresult;
+ Oid logrelid = InvalidOid;
/*
* Parse and check options.
@@ -643,7 +664,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 +839,19 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ /* Always set the destination, default will be 'log'. */
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+
+ /* If logging to a table is required, physically create the table. */
+ if (opts.conflictlogdest == CONFLICT_LOG_DEST_TABLE ||
+ opts.conflictlogdest == CONFLICT_LOG_DEST_ALL)
+ 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. */
@@ -1501,7 +1536,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 +1799,64 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
replaces[Anum_pg_subscription_subwalrcvtimeout - 1] = true;
}
+ if (IsSet(opts.specified_opts, SUBOPT_CONFLICT_LOG_DEST))
+ {
+ ConflictLogDest old_dest =
+ GetLogDestination(sub->conflictlogdest);
+
+ if (opts.conflictlogdest != old_dest)
+ {
+ bool want_table = (opts.conflictlogdest == CONFLICT_LOG_DEST_TABLE ||
+ opts.conflictlogdest == CONFLICT_LOG_DEST_ALL);
+ bool has_oldtable = (old_dest == CONFLICT_LOG_DEST_TABLE ||
+ old_dest == CONFLICT_LOG_DEST_ALL);
+
+ values[Anum_pg_subscription_subconflictlogdest - 1] =
+ CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
+ replaces[Anum_pg_subscription_subconflictlogdest - 1] = true;
+
+ if (want_table && !has_oldtable)
+ {
+ Oid relid;
+
+ relid = create_conflict_log_table(subid, sub->name,
+ sub->owner);
+
+ values[Anum_pg_subscription_subconflictlogrelid - 1] =
+ ObjectIdGetDatum(relid);
+ replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
+ true;
+ }
+ else if (!want_table && has_oldtable)
+ {
+ ObjectAddress object;
+
+ /*
+ * Conflict log tables are recorded as internal
+ * dependencies of the subscription. Drop the
+ * table if it is not required anymore to avoid
+ * stale or orphaned relations.
+ *
+ * XXX: At present, only conflict log tables are
+ * managed this way. In future if we introduce
+ * additional internal dependencies, we may need
+ * a targeted deletion to avoid deletion of any
+ * other objects.
+ */
+ ObjectAddressSet(object, SubscriptionRelationId,
+ subid);
+ performDeletion(&object, DROP_CASCADE,
+ PERFORM_DELETION_INTERNAL |
+ PERFORM_DELETION_SKIP_ORIGINAL);
+
+ values[Anum_pg_subscription_subconflictlogrelid - 1] =
+ ObjectIdGetDatum(InvalidOid);
+ replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
+ true;
+ }
+ }
+ }
+
update_tuple = true;
break;
}
@@ -2202,6 +2296,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
Form_pg_subscription form;
List *rstates;
bool must_use_password;
+ ObjectAddress object;
/*
* The launcher may concurrently start a new worker for this subscription.
@@ -2388,6 +2483,19 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
deleteDependencyRecordsFor(SubscriptionRelationId, subid, false);
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
+ /*
+ * Conflict log tables are recorded as internal dependencies of the
+ * subscription. We must drop the dependent objects before the
+ * subscription itself is removed. By using
+ * PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the conflict log
+ * table is reaped while the subscription remains for the final deletion
+ * step.
+ */
+ ObjectAddressSet(object, SubscriptionRelationId, subid);
+ performDeletion(&object, DROP_CASCADE,
+ PERFORM_DELETION_INTERNAL |
+ PERFORM_DELETION_SKIP_ORIGINAL);
+
/* Remove any associated relation synchronization states. */
RemoveSubscriptionRel(subid, InvalidOid);
@@ -2621,6 +2729,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,
@@ -3427,3 +3540,146 @@ defGetStreamingMode(DefElem *def)
def->defname)));
return LOGICALREP_STREAM_OFF; /* keep compiler quiet */
}
+
+/*
+ * Builds the TupleDesc for the conflict log table.
+ */
+static TupleDesc
+create_conflict_log_table_tupdesc(void)
+{
+ TupleDesc tupdesc;
+
+ tupdesc = CreateTemplateTupleDesc(MAX_CONFLICT_ATTR_NUM);
+
+ for (int i = 0; i < MAX_CONFLICT_ATTR_NUM; i++)
+ {
+ Oid type_oid = ConflictLogSchema[i].atttypid;
+
+ /*
+ * Special handling for the JSON array type for proper
+ * TupleDescInitEntry call.
+ */
+ if (type_oid == JSONARRAYOID)
+ type_oid = get_array_type(JSONOID);
+
+ TupleDescInitEntry(tupdesc, i + 1,
+ ConflictLogSchema[i].attname,
+ type_oid,
+ -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.
+ */
+static Oid
+create_conflict_log_table(Oid subid, char *subname, Oid subowner)
+{
+ TupleDesc tupdesc;
+ Oid relid;
+ ObjectAddress myself;
+ ObjectAddress subaddr;
+ char relname[NAMEDATALEN];
+
+ snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
+
+ /*
+ * Check for an existing table with the sname 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.
+ */
+ 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("A table with the same name already exists. "
+ "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 */
+
+ /*
+ * 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().
+ */
+ ObjectAddressSet(myself, RelationRelationId, relid);
+ ObjectAddressSet(subaddr, SubscriptionRelationId, subid);
+ recordDependencyOn(&myself, &subaddr, DEPENDENCY_INTERNAL);
+
+ /* Release tuple descriptor memory. */
+ FreeTupleDesc(tupdesc);
+
+ ereport(NOTICE,
+ (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
+ subname)));
+
+ return relid;
+}
+
+/*
+ * GetLogDestination
+ *
+ * Convert string to enum by comparing against standardized labels.
+ */
+ConflictLogDest
+GetLogDestination(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\".")));
+}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index a1845240a98..ab568f9d1d3 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 these logs to manage disk space.
*/
- if (!allowSystemTableMods && IsSystemClass(relid, reltuple)
+ if (!allowSystemTableMods && IsSystemClass(relid, reltuple) &&
+ !IsConflictClass(reltuple)
&& (!IsBinaryUpgrade ||
(relid != LargeObjectRelationId &&
relid != LargeObjectMetadataRelationId)))
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 4b30f768680..345640fe41d 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 (IsConflictNamespace(RelationGetNamespace(resultRel)) &&
+ operation != CMD_DELETE)
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ 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,17 @@ CheckValidRowMarkRel(Relation rel, RowMarkType markType)
RelationGetRelationName(rel))));
break;
}
+
+ /*
+ * Conflict log tables are managed by the system to record logical
+ * replication conflicts. We do not allow locking rows in CONFLICT
+ * relations.
+ */
+ if (IsConflictNamespace(RelationGetNamespace(rel)))
+ ereport(ERROR,
+ (errcode(ERRCODE_WRONG_OBJECT_TYPE),
+ errmsg("cannot lock rows in conflict log table \"%s\"",
+ RelationGetRelationName(rel))));
}
/*
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 1f8d67fdd90..d038e265ca9 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -24,6 +24,31 @@
#include "storage/lmgr.h"
#include "utils/lsyscache.h"
+const char *const ConflictLogDestNames[] = {
+ [CONFLICT_LOG_DEST_LOG] = "log",
+ [CONFLICT_LOG_DEST_TABLE] = "table",
+ [CONFLICT_LOG_DEST_ALL] = "all"
+};
+
+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 = "replica_identity", .atttypid = JSONOID },
+ { .attname = "remote_tuple", .atttypid = JSONOID },
+ { .attname = "local_conflicts", .atttypid = JSONARRAYOID }
+};
+
+StaticAssertDecl(lengthof(ConflictLogSchema) == MAX_CONFLICT_ATTR_NUM,
+ "ConflictLogSchema length mismatch");
+StaticAssertDecl(lengthof(ConflictLogDestNames) == 3,
+ "ConflictLogDestNames length mismatch");
+
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
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..8193229f2e2 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 IsConflictClass(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 IsConflictNamespace(Oid namespaceId);
extern bool IsReservedName(const char *name);
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..5f214d3586b 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -95,6 +95,7 @@ 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 */
/* Connection string to the publisher */
text subconninfo; /* Set if connecting with connection string */
@@ -111,6 +112,14 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
/* List of publications subscribed to */
text subpublications[1] BKI_FORCE_NOT_NULL;
+ /*
+ * 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;
+
/* Only publish data originating from the specified origin */
text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY);
#endif
@@ -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/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index 63504232a14..a895127f8fe 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,6 +17,7 @@
#include "catalog/objectaddress.h"
#include "parser/parse_node.h"
+#include "replication/conflict.h"
extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
bool isTopLevel);
@@ -36,4 +37,6 @@ extern void CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled,
bool retention_active,
bool max_retention_set);
+extern ConflictLogDest GetLogDestination(const char *dest);
+
#endif /* SUBSCRIPTIONCMDS_H */
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 2d9dbcf4d0d..00a9cbec264 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,37 @@ 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;
+
+/*
+ * Array mapping for converting internal enum to string.
+ */
+extern PGDLLIMPORT const char *const ConflictLogDestNames[];
+
+/* 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;
+
+/* The single source of truth for the conflict log table schema */
+extern PGDLLIMPORT const ConflictLogColumnDef ConflictLogSchema[];
+
+#define MAX_CONFLICT_ATTR_NUM 11
+
extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
TransactionId *xmin,
ReplOriginId *localorigin,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 7e3cabdb93f..08cbc9743d8 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -124,18 +124,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 |
+ 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 | Conflict log destination | Conflict log table
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -200,10 +200,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
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------------+--------------------------+--------------------
+ 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 | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -212,10 +212,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
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
+ 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 | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -231,10 +231,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
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
+ 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 | log | -
(1 row)
-- ok - with lsn = NONE
@@ -243,10 +243,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
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
+ 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 | log | -
(1 row)
BEGIN;
@@ -282,10 +282,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
+ 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 | Conflict log destination | Conflict log table
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
+ 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 | log | -
(1 row)
-- rename back to keep the rest simple
@@ -314,19 +314,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -338,27 +338,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
-- fail - publication already exists
@@ -373,10 +373,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
-- fail - publication used more than once
@@ -391,10 +391,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -430,19 +430,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -452,10 +452,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -468,18 +468,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -492,10 +492,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -509,19 +509,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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | 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 |
+ 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 | Conflict log destination | Conflict log table
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
+ 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 | | log | -
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -576,6 +576,313 @@ 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 relid 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 internal 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 relid 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 | replica_identity
+ 10 | remote_tuple
+ 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_for_subid_' || 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_for_subid_' || 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 internal 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 relid
+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 internal 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 internal 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_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+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
+--
+SET client_min_messages = WARNING;
+-- 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 internal 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 internal 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 internal 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 insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege during INSERT';
+END $$;
+NOTICE: captured expected error: insufficient_privilege 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 insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE';
+END $$;
+NOTICE: captured expected error: insufficient_privilege during UPDATE
+-- Trying to perform TRUNCATE/DELETE on the internal 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: System catalog 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 CONFLICT schema
+DROP TABLE public.test_move;
+-- 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 6c3d9632e8a..7cbe3e1f706 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -431,6 +431,260 @@ 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 relid 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 internal 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 relid 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_for_subid_' || 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_for_subid_' || 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 internal 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 relid
+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 internal 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 internal 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_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+
+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
+--
+
+SET client_min_messages = WARNING;
+-- 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 internal 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 internal 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 insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege 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 insufficient_privilege THEN
+ RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE';
+END $$;
+
+-- Trying to perform TRUNCATE/DELETE on the internal 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;
+
+-- 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.49.0
[application/octet-stream] v42-0002-Review-comment-fixes-for-Add-configurable-confli.patch (122.0K, 3-v42-0002-Review-comment-fixes-for-Add-configurable-confli.patch)
download | inline diff:
From 843825157f1e2fa6122547f445dd0eeee5c7c00b Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 18 May 2026 10:34:43 +0000
Subject: [PATCH v42 2/5] Review comment fixes for "Add configurable conflict
log table for Logical Replication"
Review comment fixes for "Add configurable conflict log table for
Logical Replication"
---
src/backend/catalog/aclchk.c | 61 ++--
src/backend/catalog/catalog.c | 11 +-
src/backend/catalog/heap.c | 33 +-
src/backend/catalog/namespace.c | 6 +-
src/backend/catalog/pg_publication.c | 16 +-
src/backend/commands/subscriptioncmds.c | 346 +++++++++------------
src/backend/commands/tablecmds.c | 2 +-
src/backend/executor/execMain.c | 7 +-
src/backend/replication/logical/conflict.c | 152 ++++++++-
src/include/catalog/catalog.h | 2 +-
src/include/catalog/pg_subscription.h | 16 +-
src/include/commands/subscriptioncmds.h | 3 -
src/include/replication/conflict.h | 19 +-
src/test/regress/expected/subscription.out | 241 +++++++-------
src/test/regress/sql/subscription.sql | 52 ++--
15 files changed, 533 insertions(+), 434 deletions(-)
diff --git a/src/backend/catalog/aclchk.c b/src/backend/catalog/aclchk.c
index 84ef5304e22..19c568f4705 100644
--- a/src/backend/catalog/aclchk.c
+++ b/src/backend/catalog/aclchk.c
@@ -3337,33 +3337,42 @@ 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.
- *
- * For conflict log tables, we allow non-superusers to perform DELETE
- * and TRUNCATE for maintenance, while still restricting INSERT,
- * UPDATE, and USAGE.
- */
- if ((mask & (ACL_INSERT | ACL_UPDATE | ACL_DELETE | ACL_TRUNCATE | ACL_USAGE)) &&
- IsConflictClass(classForm) &&
- !superuser_arg(roleid))
- mask &= ~(ACL_INSERT | ACL_UPDATE | ACL_USAGE);
- else 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. Also allow
+ * INSERT and UPDATE to pass ACL checks so that later checks
+ * in CheckValidResultRel() can raise the specific error instead
+ * of a generic permission denied error. Still restrict USAGE for
+ * non-superusers.
+ */
+ mask &= ~(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 4578cd07140..46d27ed02a9 100644
--- a/src/backend/catalog/catalog.c
+++ b/src/backend/catalog/catalog.c
@@ -88,7 +88,7 @@ IsSystemClass(Oid relid, Form_pg_class reltuple)
/* IsCatalogRelationOid is a bit faster, so test that first */
return (IsCatalogRelationOid(relid) ||
IsToastClass(reltuple) ||
- IsConflictClass(reltuple));
+ IsConflictLogTableClass(reltuple));
}
/*
@@ -233,11 +233,14 @@ IsToastClass(Form_pg_class reltuple)
}
/*
- * IsConflictClass - Check if the given pg_class tuple belongs to the conflict
- * namespace.
+ * IsConflictLogTableClass
+ * True iff Form_pg_class tuple represents a subscription-specific
+ * Conflict Log Table.
+ *
+ * Does not perform any catalog accesses.
*/
bool
-IsConflictClass(Form_pg_class reltuple)
+IsConflictLogTableClass(Form_pg_class reltuple)
{
Oid relnamespace = reltuple->relnamespace;
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 72bd7faaf2d..3b6a163c184 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -305,24 +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) ||
- IsConflictNamespace(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.%s\"",
+ get_namespace_name(relnamespace), relname),
+ errdetail("System catalog modifications are currently disallowed.")));
+
+ if (IsConflictNamespace(relnamespace))
+ ereport(ERROR,
+ (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ errmsg("permission denied to create \"%s.%s\"",
+ get_namespace_name(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 c35fcf57fd4..5bd9be3983e 100644
--- a/src/backend/catalog/namespace.c
+++ b/src/backend/catalog/namespace.c
@@ -3524,7 +3524,7 @@ LookupCreationNamespace(const char *nspname)
* Common checks on switching namespaces.
*
* We complain if either the old or new namespaces is a temporary schema,
- * temporary toast schema, the TOAST schema, or the CONFLICT schema.
+ * temporary toast schema, the TOAST schema, or the conflict schema.
*/
void
CheckSetNamespace(Oid oldNspOid, Oid nspOid)
@@ -3541,11 +3541,11 @@ CheckSetNamespace(Oid oldNspOid, Oid nspOid)
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
errmsg("cannot move objects into or out of TOAST schema")));
- /* similarly for CONFLICT 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 CONFLICT schema")));
+ 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 c680356a10b..93791210e35 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 (IsConflictNamespace(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,
@@ -103,13 +110,6 @@ check_publication_add_relation(PublicationRelInfo *pri)
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg(errormsg, relname),
errdetail("This operation is not supported for unlogged tables.")));
-
- /* Can't be conflict log table */
- if (IsConflictNamespace(RelationGetNamespace(targetrel)))
- ereport(ERROR,
- (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
- errmsg(errormsg, relname),
- errdetail("This operation is not supported for conflict log tables.")));
}
/*
@@ -165,7 +165,7 @@ is_publishable_class(Oid relid, Form_pg_class reltuple)
reltuple->relkind == RELKIND_PARTITIONED_TABLE ||
reltuple->relkind == RELKIND_SEQUENCE) &&
!IsCatalogRelationOid(relid) &&
- !IsConflictClass(reltuple) &&
+ !IsConflictLogTableClass(reltuple) &&
reltuple->relpersistence == RELPERSISTENCE_PERMANENT &&
relid >= FirstNormalObjectId;
}
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index afa1401f521..88f22bbb286 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -21,12 +21,10 @@
#include "access/xact.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
-#include "catalog/heap.h"
#include "catalog/indexing.h"
#include "catalog/namespace.h"
#include "catalog/objectaccess.h"
#include "catalog/objectaddress.h"
-#include "catalog/pg_am_d.h"
#include "catalog/pg_authid_d.h"
#include "catalog/pg_database_d.h"
#include "catalog/pg_foreign_server.h"
@@ -146,7 +144,11 @@ 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 Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner);
+static bool alter_sub_conflictlogdestination(Subscription *sub,
+ ConflictLogDest logdest,
+ Oid *conflicttablerelid);
+static void drop_sub_dependencies(Oid subid, char *subname,
+ Oid subconflictlogrelid);
/*
* Common option parsing function for CREATE and ALTER SUBSCRIPTION commands.
@@ -839,13 +841,15 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
- /* Always set the destination, default will be 'log'. */
values[Anum_pg_subscription_subconflictlogdest - 1] =
CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
- /* If logging to a table is required, physically create the table. */
- if (opts.conflictlogdest == CONFLICT_LOG_DEST_TABLE ||
- opts.conflictlogdest == CONFLICT_LOG_DEST_ALL)
+ /*
+ * If logging to a table is required, physically create it now. We create
+ * the conflict log table here so 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. */
@@ -872,6 +876,25 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
recordDependencyOn(&myself, &referenced, DEPENDENCY_NORMAL);
}
+ /*
+ * If conflicts are logs to table 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 (CONFLICTS_LOGGED_TO_TABLE(opts.conflictlogdest))
+ {
+ ObjectAddress clt;
+
+ ObjectAddressSet(clt, RelationRelationId, logrelid);
+ recordDependencyOn(&clt, &myself, DEPENDENCY_INTERNAL);
+ }
+
/*
* A replication origin is currently created for all subscriptions,
* including those that only contain sequences or are otherwise empty.
@@ -1441,6 +1464,70 @@ 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 logdest,
+ Oid *conflicttablerelid)
+{
+ ConflictLogDest old_dest = GetLogDestination(sub->conflictlogdest);
+ bool want_table;
+ bool has_oldtable;
+ bool update_relid = false;
+ Oid relid = InvalidOid;
+
+ want_table = CONFLICTS_LOGGED_TO_TABLE(logdest);
+ has_oldtable = CONFLICTS_LOGGED_TO_TABLE(old_dest);
+
+ if (has_oldtable)
+ {
+ /* There is a conflict log table already. */
+ if (!want_table)
+ {
+ drop_sub_dependencies(sub->oid, sub->name, sub->conflictlogrelid);
+ update_relid = true;
+ }
+ }
+ else
+ {
+ /* There was no previous conflict log table. */
+ if (want_table)
+ {
+ ObjectAddress clt;
+ 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(clt, RelationRelationId, relid);
+ ObjectAddressSet(subobj, SubscriptionRelationId, sub->oid);
+ recordDependencyOn(&clt, &subobj, DEPENDENCY_INTERNAL);
+ }
+ }
+
+ *conflicttablerelid = relid;
+ return update_relid;
+}
+
/*
* Alter the existing subscription.
*/
@@ -1806,54 +1893,23 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
if (opts.conflictlogdest != old_dest)
{
- bool want_table = (opts.conflictlogdest == CONFLICT_LOG_DEST_TABLE ||
- opts.conflictlogdest == CONFLICT_LOG_DEST_ALL);
- bool has_oldtable = (old_dest == CONFLICT_LOG_DEST_TABLE ||
- old_dest == CONFLICT_LOG_DEST_ALL);
+ bool update_relid;
+ Oid relid = InvalidOid;
values[Anum_pg_subscription_subconflictlogdest - 1] =
CStringGetTextDatum(ConflictLogDestNames[opts.conflictlogdest]);
replaces[Anum_pg_subscription_subconflictlogdest - 1] = true;
- if (want_table && !has_oldtable)
+ update_relid = alter_sub_conflictlogdestination(sub,
+ opts.conflictlogdest,
+ &relid);
+ if (update_relid)
{
- Oid relid;
-
- relid = create_conflict_log_table(subid, sub->name,
- sub->owner);
-
values[Anum_pg_subscription_subconflictlogrelid - 1] =
ObjectIdGetDatum(relid);
replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
true;
}
- else if (!want_table && has_oldtable)
- {
- ObjectAddress object;
-
- /*
- * Conflict log tables are recorded as internal
- * dependencies of the subscription. Drop the
- * table if it is not required anymore to avoid
- * stale or orphaned relations.
- *
- * XXX: At present, only conflict log tables are
- * managed this way. In future if we introduce
- * additional internal dependencies, we may need
- * a targeted deletion to avoid deletion of any
- * other objects.
- */
- ObjectAddressSet(object, SubscriptionRelationId,
- subid);
- performDeletion(&object, DROP_CASCADE,
- PERFORM_DELETION_INTERNAL |
- PERFORM_DELETION_SKIP_ORIGINAL);
-
- values[Anum_pg_subscription_subconflictlogrelid - 1] =
- ObjectIdGetDatum(InvalidOid);
- replaces[Anum_pg_subscription_subconflictlogrelid - 1] =
- true;
- }
}
}
@@ -2272,6 +2328,49 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
return myself;
}
+/*
+ * drop_sub_dependencies
+ *
+ * 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_dependencies(Oid subid, char *subname, Oid subconflictlogrelid)
+{
+ /* Drop any dependent conflict log table */
+ if (OidIsValid(subconflictlogrelid))
+ {
+ ObjectAddress object;
+ char *conflictrelname;
+
+ conflictrelname = get_rel_name(subconflictlogrelid);
+ Assert(conflictrelname);
+
+ /*
+ * 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
*/
@@ -2283,6 +2382,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
HeapTuple tup;
Oid subid;
Oid subowner;
+ Oid subconflictlogrelid;
Datum datum;
bool isnull;
char *subname;
@@ -2296,7 +2396,6 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
Form_pg_subscription form;
List *rstates;
bool must_use_password;
- ObjectAddress object;
/*
* The launcher may concurrently start a new worker for this subscription.
@@ -2329,6 +2428,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 */
@@ -2483,18 +2583,7 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
deleteDependencyRecordsFor(SubscriptionRelationId, subid, false);
deleteSharedDependencyRecordsFor(SubscriptionRelationId, subid, 0);
- /*
- * Conflict log tables are recorded as internal dependencies of the
- * subscription. We must drop the dependent objects before the
- * subscription itself is removed. By using
- * PERFORM_DELETION_SKIP_ORIGINAL, we ensure that only the conflict log
- * table is reaped while the subscription remains for the final deletion
- * step.
- */
- ObjectAddressSet(object, SubscriptionRelationId, subid);
- performDeletion(&object, DROP_CASCADE,
- PERFORM_DELETION_INTERNAL |
- PERFORM_DELETION_SKIP_ORIGINAL);
+ drop_sub_dependencies(subid, subname, subconflictlogrelid);
/* Remove any associated relation synchronization states. */
RemoveSubscriptionRel(subid, InvalidOid);
@@ -3540,146 +3629,3 @@ defGetStreamingMode(DefElem *def)
def->defname)));
return LOGICALREP_STREAM_OFF; /* keep compiler quiet */
}
-
-/*
- * Builds the TupleDesc for the conflict log table.
- */
-static TupleDesc
-create_conflict_log_table_tupdesc(void)
-{
- TupleDesc tupdesc;
-
- tupdesc = CreateTemplateTupleDesc(MAX_CONFLICT_ATTR_NUM);
-
- for (int i = 0; i < MAX_CONFLICT_ATTR_NUM; i++)
- {
- Oid type_oid = ConflictLogSchema[i].atttypid;
-
- /*
- * Special handling for the JSON array type for proper
- * TupleDescInitEntry call.
- */
- if (type_oid == JSONARRAYOID)
- type_oid = get_array_type(JSONOID);
-
- TupleDescInitEntry(tupdesc, i + 1,
- ConflictLogSchema[i].attname,
- type_oid,
- -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.
- */
-static Oid
-create_conflict_log_table(Oid subid, char *subname, Oid subowner)
-{
- TupleDesc tupdesc;
- Oid relid;
- ObjectAddress myself;
- ObjectAddress subaddr;
- char relname[NAMEDATALEN];
-
- snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
-
- /*
- * Check for an existing table with the sname 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.
- */
- 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("A table with the same name already exists. "
- "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 */
-
- /*
- * 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().
- */
- ObjectAddressSet(myself, RelationRelationId, relid);
- ObjectAddressSet(subaddr, SubscriptionRelationId, subid);
- recordDependencyOn(&myself, &subaddr, DEPENDENCY_INTERNAL);
-
- /* Release tuple descriptor memory. */
- FreeTupleDesc(tupdesc);
-
- ereport(NOTICE,
- (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
- get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
- subname)));
-
- return relid;
-}
-
-/*
- * GetLogDestination
- *
- * Convert string to enum by comparing against standardized labels.
- */
-ConflictLogDest
-GetLogDestination(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\".")));
-}
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index ab568f9d1d3..2dcf64fa53b 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -2461,7 +2461,7 @@ truncate_check_rel(Oid relid, Form_pg_class reltuple)
* to permit users to manually prune these logs to manage disk space.
*/
if (!allowSystemTableMods && IsSystemClass(relid, reltuple) &&
- !IsConflictClass(reltuple)
+ !IsConflictLogTableClass(reltuple)
&& (!IsBinaryUpgrade ||
(relid != LargeObjectRelationId &&
relid != LargeObjectMetadataRelationId)))
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 345640fe41d..2c1a4d8ab71 100644
--- a/src/backend/executor/execMain.c
+++ b/src/backend/executor/execMain.c
@@ -1201,7 +1201,7 @@ CheckValidResultRel(ResultRelInfo *resultRelInfo, CmdType operation,
if (IsConflictNamespace(RelationGetNamespace(resultRel)) &&
operation != CMD_DELETE)
ereport(ERROR,
- (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
+ (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.")));
@@ -1279,13 +1279,12 @@ CheckValidRowMarkRel(Relation rel, RowMarkType markType)
/*
* Conflict log tables are managed by the system to record logical
- * replication conflicts. We do not allow locking rows in CONFLICT
- * relations.
+ * replication conflicts.
*/
if (IsConflictNamespace(RelationGetNamespace(rel)))
ereport(ERROR,
(errcode(ERRCODE_WRONG_OBJECT_TYPE),
- errmsg("cannot lock rows in conflict log table \"%s\"",
+ 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 d038e265ca9..51a76bb029d 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -17,6 +17,11 @@
#include "access/commit_ts.h"
#include "access/genam.h"
#include "access/tableam.h"
+#include "catalog/dependency.h"
+#include "catalog/heap.h"
+#include "catalog/pg_am.h"
+#include "catalog/pg_namespace.h"
+#include "commands/subscriptioncmds.h"
#include "executor/executor.h"
#include "pgstat.h"
#include "replication/conflict.h"
@@ -24,13 +29,35 @@
#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"
};
-const ConflictLogColumnDef ConflictLogSchema[] = {
+StaticAssertDecl(lengthof(ConflictLogDestNames) == 3,
+ "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 },
@@ -39,15 +66,12 @@ const ConflictLogColumnDef ConflictLogSchema[] = {
{ .attname = "remote_commit_lsn",.atttypid = LSNOID },
{ .attname = "remote_commit_ts", .atttypid = TIMESTAMPTZOID },
{ .attname = "remote_origin", .atttypid = TEXTOID },
- { .attname = "replica_identity", .atttypid = JSONOID },
{ .attname = "remote_tuple", .atttypid = JSONOID },
+ { .attname = "replica_identity", .atttypid = JSONOID },
{ .attname = "local_conflicts", .atttypid = JSONARRAYOID }
};
-StaticAssertDecl(lengthof(ConflictLogSchema) == MAX_CONFLICT_ATTR_NUM,
- "ConflictLogSchema length mismatch");
-StaticAssertDecl(lengthof(ConflictLogDestNames) == 3,
- "ConflictLogDestNames length mismatch");
+#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
@@ -79,6 +103,122 @@ 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_for_subid_<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_for_subid_%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);
+
+ ereport(NOTICE,
+ (errmsg("created conflict log table \"%s\" for subscription \"%s\"",
+ get_qualified_objname(PG_CONFLICT_NAMESPACE, relname),
+ subname)));
+
+ return relid;
+}
+
+/*
+ * GetLogDestination
+ *
+ * Convert string to enum by comparing against standardized labels.
+ */
+ConflictLogDest
+GetLogDestination(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/include/catalog/catalog.h b/src/include/catalog/catalog.h
index 8193229f2e2..cd05974b86c 100644
--- a/src/include/catalog/catalog.h
+++ b/src/include/catalog/catalog.h
@@ -25,7 +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 IsConflictClass(Form_pg_class reltuple);
+extern bool IsConflictLogTableClass(Form_pg_class reltuple);
extern bool IsCatalogRelationOid(Oid relid);
extern bool IsCatalogTextUniqueIndexOid(Oid relid);
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 5f214d3586b..cc31b4d00bc 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -97,6 +97,14 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
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 */
@@ -112,14 +120,6 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
/* List of publications subscribed to */
text subpublications[1] BKI_FORCE_NOT_NULL;
- /*
- * 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;
-
/* Only publish data originating from the specified origin */
text suborigin BKI_DEFAULT(LOGICALREP_ORIGIN_ANY);
#endif
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index a895127f8fe..63504232a14 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,7 +17,6 @@
#include "catalog/objectaddress.h"
#include "parser/parse_node.h"
-#include "replication/conflict.h"
extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
bool isTopLevel);
@@ -37,6 +36,4 @@ extern void CheckSubDeadTupleRetention(bool check_guc, bool sub_disabled,
bool retention_active,
bool max_retention_set);
-extern ConflictLogDest GetLogDestination(const char *dest);
-
#endif /* SUBSCRIPTIONCMDS_H */
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 00a9cbec264..c92216e2dd5 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -94,23 +94,18 @@ typedef enum ConflictLogDest
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_FILE(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[];
-/* 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;
-
-/* The single source of truth for the conflict log table schema */
-extern PGDLLIMPORT const ConflictLogColumnDef ConflictLogSchema[];
-
-#define MAX_CONFLICT_ATTR_NUM 11
-
+extern Oid create_conflict_log_table(Oid subid, char *subname, Oid subowner);
+extern ConflictLogDest GetLogDestination(const char *dest);
extern bool GetTupleTransactionInfo(TupleTableSlot *localslot,
TransactionId *xmin,
ReplOriginId *localorigin,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 08cbc9743d8..7b68b53c903 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -124,18 +124,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 | Conflict log destination | Conflict log table
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -200,10 +200,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------------+--------------------------+--------------------
- 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 | log | -
+ 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
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -212,10 +212,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
- 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 | log | -
+ 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
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -231,10 +231,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
- 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 | log | -
+ 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
(1 row)
-- ok - with lsn = NONE
@@ -243,10 +243,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
- 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 | log | -
+ 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
(1 row)
BEGIN;
@@ -282,10 +282,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 | Conflict log destination | Conflict log table
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------+--------------------------+--------------------
- 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 | log | -
+ 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
(1 row)
-- rename back to keep the rest simple
@@ -314,19 +314,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -338,27 +338,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
-- fail - publication already exists
@@ -373,10 +373,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
-- fail - publication used more than once
@@ -391,10 +391,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -430,19 +430,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -452,10 +452,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -468,18 +468,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -492,10 +492,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -509,19 +509,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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(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 | Conflict log destination | Conflict log table
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------+--------------------------+--------------------
- 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 | | log | -
+ 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 |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -585,7 +585,7 @@ SET client_min_messages = WARNING;
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 relid is 0 (InvalidOid) for default case
+-- 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.
@@ -607,11 +607,11 @@ FROM pg_subscription WHERE subname = 'regress_conflict_empty_str';
regress_conflict_empty_str | log | 0
(1 row)
--- this should generate an internal conflict log table named pg_conflict_log_$subid$
+-- this should generate an conflict log table named pg_conflict_log_for_subid_$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 relid valid
+-- 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
@@ -623,7 +623,7 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test1';
-- 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_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE s.subname = 'regress_conflict_test1';
nspname | oid_matches
@@ -635,7 +635,7 @@ WHERE s.subname = 'regress_conflict_test1';
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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
ORDER BY a.attnum;
attnum | attname
@@ -648,8 +648,8 @@ WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
6 | remote_commit_lsn
7 | remote_commit_ts
8 | remote_origin
- 9 | replica_identity
- 10 | remote_tuple
+ 9 | remote_tuple
+ 10 | replica_identity
11 | local_conflicts
(11 rows)
@@ -696,7 +696,7 @@ ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
-- expected
--
-- transition from 'log' to 'all'
--- a new internal conflict log table should be created
+-- 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.
@@ -721,7 +721,7 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test2';
(1 row)
-- transition from 'table' to 'log'
--- should drop the table and clear relid
+-- 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';
@@ -733,7 +733,7 @@ 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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_test2';
count
-------
@@ -743,8 +743,8 @@ WHERE s.subname = 'regress_conflict_test2';
--
-- PUBLICATION: Verify conflict log tables are not publishable
--
--- pg_relation_is_publishable should return false for internal conflict log
--- tables to prevent them from being accidentally included in publications
+-- 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
@@ -765,7 +765,7 @@ WHERE s.subname = 'regress_conflict_test1';
--
-- 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 internal conflict log 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
@@ -773,7 +773,7 @@ ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table
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');
+ EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_log_for_subid_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1');
EXCEPTION WHEN insufficient_privilege THEN
RAISE NOTICE 'captured expected error: insufficient_privilege';
END $$;
@@ -782,7 +782,8 @@ NOTICE: captured expected error: insufficient_privilege
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_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+SELECT 'pg_conflict.pg_conflict_log_for_subid_' || 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');
@@ -794,13 +795,12 @@ SELECT to_regclass(:'internal_tablename');
--
-- Additional Namespace and Table Protection Tests
--
-SET client_min_messages = WARNING;
-- 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 internal conflict log 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;
@@ -809,15 +809,15 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
- RAISE NOTICE 'Attempting ALTER TABLE on internal conflict log table';
+ 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 internal conflict log table
+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
@@ -827,14 +827,14 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')';
-EXCEPTION WHEN insufficient_privilege THEN
- RAISE NOTICE 'captured expected error: insufficient_privilege during INSERT';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during INSERT';
END $$;
-NOTICE: captured expected error: insufficient_privilege during INSERT
+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
@@ -843,19 +843,19 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' ';
-EXCEPTION WHEN insufficient_privilege THEN
- RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during UPDATE';
END $$;
-NOTICE: captured expected error: insufficient_privilege during UPDATE
--- Trying to perform TRUNCATE/DELETE on the internal conflict log table
+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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test' \gset
TRUNCATE :conflict_tab;
DELETE FROM :conflict_tab;
@@ -863,13 +863,14 @@ DELETE FROM :conflict_tab;
-- 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: System catalog modifications are currently disallowed.
+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 CONFLICT schema
+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);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 7cbe3e1f706..2a2135767f8 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -442,7 +442,7 @@ 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 relid is 0 (InvalidOid) for default case
+-- 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';
@@ -452,10 +452,10 @@ CREATE SUBSCRIPTION regress_conflict_empty_str CONNECTION 'dbname=regress_doesno
SELECT subname, subconflictlogdest, subconflictlogrelid
FROM pg_subscription WHERE subname = 'regress_conflict_empty_str';
--- this should generate an internal conflict log table named pg_conflict_log_$subid$
+-- this should generate an conflict log table named pg_conflict_log_for_subid_$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 relid valid
+-- 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';
@@ -463,7 +463,7 @@ FROM pg_subscription WHERE subname = 'regress_conflict_test1';
-- 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_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE s.subname = 'regress_conflict_test1';
@@ -471,7 +471,7 @@ WHERE s.subname = 'regress_conflict_test1';
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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_test1' AND a.attnum > 0
ORDER BY a.attnum;
@@ -514,7 +514,7 @@ ALTER SUBSCRIPTION regress_conflict_test1 OWNER TO regress_subscription_user;
-- expected
--
-- transition from 'log' to 'all'
--- a new internal conflict log table should be created
+-- 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');
@@ -530,7 +530,7 @@ 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 relid
+-- 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';
@@ -538,14 +538,14 @@ 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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_test2';
--
-- PUBLICATION: Verify conflict log tables are not publishable
--
--- pg_relation_is_publishable should return false for internal conflict log
--- tables to prevent them from being accidentally included in publications
+-- 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
@@ -563,7 +563,7 @@ WHERE s.subname = 'regress_conflict_test1';
-- 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 internal conflict log 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
@@ -572,7 +572,7 @@ ALTER SUBSCRIPTION regress_conflict_test1 SET (conflict_log_destination = 'table
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');
+ EXECUTE 'DROP TABLE ' || (SELECT 'pg_conflict.pg_conflict_log_for_subid_' || oid FROM pg_subscription WHERE subname = 'regress_conflict_test1');
EXCEPTION WHEN insufficient_privilege THEN
RAISE NOTICE 'captured expected error: insufficient_privilege';
END $$;
@@ -582,8 +582,9 @@ 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_log_' || oid AS internal_tablename FROM pg_subscription WHERE subname = 'regress_conflict_test1' \gset
+SELECT 'pg_conflict.pg_conflict_log_for_subid_' || 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
@@ -593,12 +594,11 @@ SELECT to_regclass(:'internal_tablename');
-- Additional Namespace and Table Protection Tests
--
-SET client_min_messages = WARNING;
-- 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 internal conflict log 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
@@ -608,10 +608,10 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
- RAISE NOTICE 'Attempting ALTER TABLE on internal conflict log table';
+ 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';
@@ -625,12 +625,12 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
EXECUTE 'INSERT INTO ' || tab_name || ' (relname) VALUES (''mytest'')';
-EXCEPTION WHEN insufficient_privilege THEN
- RAISE NOTICE 'captured expected error: insufficient_privilege during INSERT';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during INSERT';
END $$;
-- Test Manual UPDATE on conflict log table
@@ -641,19 +641,19 @@ 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
+ FROM pg_class c JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test';
EXECUTE 'UPDATE ' || tab_name || ' SET relname = ''mytest'' ';
-EXCEPTION WHEN insufficient_privilege THEN
- RAISE NOTICE 'captured expected error: insufficient_privilege during UPDATE';
+EXCEPTION WHEN wrong_object_type THEN
+ RAISE NOTICE 'captured expected error: wrong_object_type during UPDATE';
END $$;
--- Trying to perform TRUNCATE/DELETE on the internal conflict log table
+-- 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
+JOIN pg_subscription s ON c.relname = 'pg_conflict_log_for_subid_' || s.oid
WHERE s.subname = 'regress_conflict_protection_test' \gset
TRUNCATE :conflict_tab;
DELETE FROM :conflict_tab;
@@ -668,6 +668,8 @@ 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);
--
2.49.0
[application/octet-stream] v42-0003-Implement-the-conflict-insertion-infrastructure-.patch (28.6K, 4-v42-0003-Implement-the-conflict-insertion-infrastructure-.patch)
download | inline diff:
From 0daf854403556a8f481e89ef34da823f3213e340 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Mon, 18 May 2026 10:51:02 +0000
Subject: [PATCH v42 3/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_for_subid_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}
---
src/backend/replication/logical/conflict.c | 554 +++++++++++++++++++--
src/backend/replication/logical/launcher.c | 1 +
src/backend/replication/logical/worker.c | 31 +-
src/include/replication/conflict.h | 2 +
src/include/replication/worker_internal.h | 7 +
src/test/subscription/t/035_conflicts.pl | 47 +-
6 files changed, 597 insertions(+), 45 deletions(-)
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 51a76bb029d..01d254baa41 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -16,6 +16,7 @@
#include "access/commit_ts.h"
#include "access/genam.h"
+#include "access/heapam.h"
#include "access/tableam.h"
#include "catalog/dependency.h"
#include "catalog/heap.h"
@@ -23,11 +24,17 @@
#include "catalog/pg_namespace.h"
#include "commands/subscriptioncmds.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/fmgroids.h"
#include "utils/lsyscache.h"
+#include "utils/pg_lsn.h"
+#include "utils/json.h"
/*
* String representations for the supported conflict logging destinations.
@@ -84,6 +91,18 @@ static const char *const ConflictTypeNames[] = {
[CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
};
+/* 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 MAX_LOCAL_CONFLICT_INFO_ATTRS lengthof(LocalConflictSchema)
+
static int errcode_apply_conflict(ConflictType type);
static void errdetail_apply_conflict(EState *estate,
ResultRelInfo *relinfo,
@@ -100,8 +119,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.
@@ -271,30 +309,92 @@ 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_clt = false;
+ 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);
+ if (dest == CONFLICT_LOG_DEST_TABLE || dest == CONFLICT_LOG_DEST_ALL)
+ log_dest_clt = true;
+ if (dest == CONFLICT_LOG_DEST_LOG || dest == CONFLICT_LOG_DEST_ALL)
+ log_dest_logfile = true;
- 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));
+ /* Insert to table if requested. */
+ if (log_dest_clt)
+ {
+ 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.%s\": conflict=%s",
+ get_namespace_name(RelationGetNamespace(localrel)),
+ RelationGetRelationName(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.%s\": conflict=%s",
+ get_namespace_name(RelationGetNamespace(localrel)),
+ RelationGetRelationName(localrel),
+ ConflictTypeNames[type]),
+ errdetail_internal("%s", err_detail.data));
+ }
}
/*
@@ -328,6 +428,58 @@ 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 = GetLogDestination(MySubscription->conflictlogdest);
+
+ /* Quick exit if a conflict log table was not requested. */
+ if (*log_dest == CONFLICT_LOG_DEST_LOG)
+ 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)
+{
+ int options = HEAP_INSERT_NO_LOGICAL;
+
+ /* 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), options, 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.
*/
@@ -761,6 +913,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.
@@ -776,41 +962,323 @@ 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 = 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);
+
+ index_close(indexDesc, NoLock);
+ heap_freetuple(tuple);
+
+ /* Convert to a JSON datum. */
+ return DirectFunctionCall1(row_to_json, datum);
+}
+
+/*
+ * build_conflict_tupledesc
+ *
+ * Build and bless a tuple descriptor for the internal conflict log table
+ * based on the predefined LocalConflictSchema.
+ */
+static TupleDesc
+build_conflict_tupledesc(void)
+{
+ TupleDesc tupdesc;
+
+ tupdesc = CreateTemplateTupleDesc(MAX_LOCAL_CONFLICT_INFO_ATTRS);
+
+ for (int i = 0; i < MAX_LOCAL_CONFLICT_INFO_ATTRS; i++)
+ TupleDescInitEntry(tupdesc, (AttrNumber) (i + 1),
+ LocalConflictSchema[i].attname,
+ LocalConflictSchema[i].atttypid,
+ -1, 0);
+
+ TupleDescFinalize(tupdesc);
+ BlessTupleDesc(tupdesc);
+
+ return tupdesc;
+}
+
+/*
+ * 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;
+
+ /* Build local conflicts tuple descriptor. */
+ tupdesc = build_conflict_tupledesc();
+
+ /* Process local conflict tuple list and prepare an array of JSON. */
+ foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
{
- tableslot = table_slot_create(localrel, &estate->es_tupleTable);
- tableslot = ExecCopySlot(tableslot, slot);
+ Datum values[MAX_LOCAL_CONFLICT_INFO_ATTRS] = {0};
+ bool nulls[MAX_LOCAL_CONFLICT_INFO_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;
+
+ 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 == MAX_LOCAL_CONFLICT_INFO_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);
}
- /*
- * Initialize ecxt_scantuple for potential use in FormIndexDatum when
- * index expressions are present.
- */
- GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+ num_conflicts = list_length(json_datums);
- /*
- * 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);
+ json_datum_array = palloc_array(Datum, num_conflicts);
- index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+ i = 0;
+ foreach(lc, json_datums)
+ {
+ json_datum_array[i] = (Datum) lfirst(lc);
+ i++;
+ }
- index_close(indexDesc, NoLock);
+ /* 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;
+}
- return index_value;
+/*
+ * 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[MAX_CONFLICT_ATTR_NUM] = {0};
+ bool nulls[MAX_CONFLICT_ATTR_NUM] = {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(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;
+
+ if (!TupIsNull(remoteslot))
+ values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
+ else
+ nulls[attno++] = true;
+
+ values[attno] = build_local_conflicts_json_array(estate, rel,
+ conflict_type,
+ conflicttuples);
+
+ Assert(attno + 1 == MAX_CONFLICT_ATTR_NUM);
+
+ 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..469451c736a 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);
@@ -1762,6 +1766,10 @@ apply_handle_stream_start(StringInfo s)
/* extract XID of the top-level transaction */
stream_xid = logicalrep_read_stream_start(s, &first_segment);
+ remote_xid = stream_xid;
+ remote_final_lsn = InvalidXLogRecPtr;
+ remote_commit_ts = 0;
+
if (!TransactionIdIsValid(stream_xid))
ereport(ERROR,
(errcode(ERRCODE_PROTOCOL_VIOLATION),
@@ -5667,6 +5675,27 @@ start_apply(XLogRecPtr origin_startpos)
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid);
+ /*
+ * Insert any pending conflict log tuple under a new transaction.
+ */
+ if (MyLogicalRepWorker->conflict_log_tuple != NULL)
+ {
+ Relation conflictlogrel;
+ ConflictLogDest dest;
+
+ StartTransactionCommand();
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Open conflict log table and insert the tuple. */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
+ Assert(dest != CONFLICT_LOG_DEST_LOG);
+ InsertConflictLogTuple(conflictlogrel);
+ table_close(conflictlogrel, RowExclusiveLock);
+
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+ }
+
PG_RE_THROW();
}
}
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index c92216e2dd5..6dcb7970bb7 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -116,4 +116,6 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
TupleTableSlot *remoteslot,
List *conflicttuples);
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..6a447da6510 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/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index f23fe6af2a5..05c2179b9a8 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.49.0
[application/octet-stream] v42-0004-Review-comment-fixes-for-Implement-the-conflict-.patch (16.8K, 5-v42-0004-Review-comment-fixes-for-Implement-the-conflict-.patch)
download | inline diff:
From 4a6d500b4d2f010cba3081d5ad03b6c7d41b0c2f Mon Sep 17 00:00:00 2001
From: Vignesh C <[email protected]>
Date: Wed, 20 May 2026 10:13:28 +0530
Subject: [PATCH v42 4/5] Review comment fixes for Implement the conflict
insertion infrastructure for the conflict log table
Review comment fixes for Implement the conflict
insertion infrastructure for the conflict log table
---
src/backend/replication/logical/conflict.c | 160 +++++++++++++--------
src/backend/replication/logical/worker.c | 32 +----
src/include/replication/conflict.h | 1 +
src/test/subscription/t/030_origin.pl | 4 +-
src/test/subscription/t/035_conflicts.pl | 4 +-
5 files changed, 116 insertions(+), 85 deletions(-)
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 01d254baa41..15b0ef7f3ca 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -31,10 +31,8 @@
#include "storage/lmgr.h"
#include "utils/array.h"
#include "utils/builtins.h"
-#include "utils/fmgroids.h"
#include "utils/lsyscache.h"
#include "utils/pg_lsn.h"
-#include "utils/json.h"
/*
* String representations for the supported conflict logging destinations.
@@ -48,7 +46,6 @@ const char *const ConflictLogDestNames[] = {
StaticAssertDecl(lengthof(ConflictLogDestNames) == 3,
"ConflictLogDestNames length mismatch");
-
/* Structure to hold metadata for one column of the conflict log table */
typedef struct ConflictLogColumnDef
{
@@ -80,17 +77,6 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
-static const char *const ConflictTypeNames[] = {
- [CT_INSERT_EXISTS] = "insert_exists",
- [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
- [CT_UPDATE_EXISTS] = "update_exists",
- [CT_UPDATE_MISSING] = "update_missing",
- [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
- [CT_UPDATE_DELETED] = "update_deleted",
- [CT_DELETE_MISSING] = "delete_missing",
- [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
-};
-
/* Schema for the elements within the 'local_conflicts' JSON array */
static const ConflictLogColumnDef LocalConflictSchema[] =
{
@@ -101,7 +87,18 @@ static const ConflictLogColumnDef LocalConflictSchema[] =
{ .attname = "tuple", .atttypid = JSONOID }
};
-#define MAX_LOCAL_CONFLICT_INFO_ATTRS lengthof(LocalConflictSchema)
+#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema)
+
+static const char *const ConflictTypeNames[] = {
+ [CT_INSERT_EXISTS] = "insert_exists",
+ [CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
+ [CT_UPDATE_EXISTS] = "update_exists",
+ [CT_UPDATE_MISSING] = "update_missing",
+ [CT_DELETE_ORIGIN_DIFFERS] = "delete_origin_differs",
+ [CT_UPDATE_DELETED] = "update_deleted",
+ [CT_DELETE_MISSING] = "delete_missing",
+ [CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
+};
static int errcode_apply_conflict(ConflictType type);
static void errdetail_apply_conflict(EState *estate,
@@ -312,7 +309,7 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
Relation localrel = relinfo->ri_RelationDesc;
ConflictLogDest dest;
Relation conflictlogrel;
- bool log_dest_clt = false;
+ bool log_dest_table;
bool log_dest_logfile;
pgstat_report_subscription_conflict(MySubscription->oid, type);
@@ -323,13 +320,11 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
*/
conflictlogrel = GetConflictLogDestAndTable(&dest);
- if (dest == CONFLICT_LOG_DEST_TABLE || dest == CONFLICT_LOG_DEST_ALL)
- log_dest_clt = true;
- if (dest == CONFLICT_LOG_DEST_LOG || dest == CONFLICT_LOG_DEST_ALL)
- log_dest_logfile = true;
+ log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest);
+ log_dest_logfile = CONFLICTS_LOGGED_TO_FILE(dest);
/* Insert to table if requested. */
- if (log_dest_clt)
+ if (log_dest_table)
{
Assert(conflictlogrel != NULL);
@@ -358,9 +353,8 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
*/
ereport(elevel,
errcode_apply_conflict(type),
- errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
- get_namespace_name(RelationGetNamespace(localrel)),
- RelationGetRelationName(localrel),
+ 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)));
@@ -389,14 +383,54 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
/* Standard reporting with full internal details. */
ereport(elevel,
errcode_apply_conflict(type),
- errmsg("conflict detected on relation \"%s.%s\": conflict=%s",
- get_namespace_name(RelationGetNamespace(localrel)),
- RelationGetRelationName(localrel),
+ 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;
+
+ StartTransactionCommand();
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /* Open conflict log table and insert the tuple */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
+ Assert(conflictlogrel);
+
+ InsertConflictLogTuple(conflictlogrel);
+
+ table_close(conflictlogrel, RowExclusiveLock);
+
+ PopActiveSnapshot();
+ CommitTransactionCommand();
+}
+
/*
* Find all unique indexes to check for a conflict and store them into
* ResultRelInfo.
@@ -447,7 +481,7 @@ GetConflictLogDestAndTable(ConflictLogDest *log_dest)
*log_dest = GetLogDestination(MySubscription->conflictlogdest);
/* Quick exit if a conflict log table was not requested. */
- if (*log_dest == CONFLICT_LOG_DEST_LOG)
+ if (!CONFLICTS_LOGGED_TO_TABLE(*log_dest))
return NULL;
conflictlogrelid = MySubscription->conflictlogrelid;
@@ -467,13 +501,11 @@ GetConflictLogDestAndTable(ConflictLogDest *log_dest)
void
InsertConflictLogTuple(Relation conflictlogrel)
{
- int options = HEAP_INSERT_NO_LOGICAL;
-
/* 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), options, NULL);
+ GetCurrentCommandId(true), HEAP_INSERT_NO_LOGICAL, NULL);
/* Free conflict log tuple. */
heap_freetuple(MyLogicalRepWorker->conflict_log_tuple);
@@ -1028,7 +1060,7 @@ tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
isnull);
- tupdesc = RelationGetDescr(indexDesc);
+ tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc));
/* Bless the tupdesc so it can be looked up by row_to_json. */
BlessTupleDesc(tupdesc);
@@ -1037,8 +1069,9 @@ tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
tuple = heap_form_tuple(tupdesc, values, isnull);
datum = heap_copy_tuple_as_datum(tuple, tupdesc);
- index_close(indexDesc, NoLock);
heap_freetuple(tuple);
+ FreeTupleDesc(tupdesc);
+ index_close(indexDesc, NoLock);
/* Convert to a JSON datum. */
return DirectFunctionCall1(row_to_json, datum);
@@ -1047,26 +1080,41 @@ tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
/*
* build_conflict_tupledesc
*
- * Build and bless a tuple descriptor for the internal conflict log table
- * based on the predefined LocalConflictSchema.
+ * Build and bless a tuple descriptor for the conflict log table based on the
+ * predefined LocalConflictSchema.
*/
static TupleDesc
build_conflict_tupledesc(void)
{
- TupleDesc tupdesc;
+ static TupleDesc cached_tupdesc = NULL;
- tupdesc = CreateTemplateTupleDesc(MAX_LOCAL_CONFLICT_INFO_ATTRS);
+ if (cached_tupdesc == NULL)
+ {
+ MemoryContext oldcxt;
- for (int i = 0; i < MAX_LOCAL_CONFLICT_INFO_ATTRS; i++)
- TupleDescInitEntry(tupdesc, (AttrNumber) (i + 1),
- LocalConflictSchema[i].attname,
- LocalConflictSchema[i].atttypid,
- -1, 0);
+ oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
- TupleDescFinalize(tupdesc);
- BlessTupleDesc(tupdesc);
+ cached_tupdesc = CreateTemplateTupleDesc(NUM_LOCAL_CONFLICT_ATTRS);
- return tupdesc;
+ 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);
+ }
+
+ return cached_tupdesc;
}
/*
@@ -1098,8 +1146,8 @@ build_local_conflicts_json_array(EState *estate, Relation rel,
/* Process local conflict tuple list and prepare an array of JSON. */
foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
{
- Datum values[MAX_LOCAL_CONFLICT_INFO_ATTRS] = {0};
- bool nulls[MAX_LOCAL_CONFLICT_INFO_ATTRS] = {0};
+ Datum values[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_LOCAL_CONFLICT_ATTRS] = {0};
char *origin_name = NULL;
HeapTuple tuple;
Datum json_datum;
@@ -1149,7 +1197,7 @@ build_local_conflicts_json_array(EState *estate, Relation rel,
else
nulls[attno] = true;
- Assert(attno + 1 == MAX_LOCAL_CONFLICT_INFO_ATTRS);
+ Assert(attno + 1 == NUM_LOCAL_CONFLICT_ATTRS);
tuple = heap_form_tuple(tupdesc, values, nulls);
@@ -1208,8 +1256,8 @@ prepare_conflict_log_tuple(EState *estate, Relation rel,
List *conflicttuples,
TupleTableSlot *remoteslot)
{
- Datum values[MAX_CONFLICT_ATTR_NUM] = {0};
- bool nulls[MAX_CONFLICT_ATTR_NUM] = {0};
+ Datum values[NUM_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_CONFLICT_ATTRS] = {0};
int attno;
char *remote_origin = NULL;
MemoryContext oldctx;
@@ -1247,6 +1295,11 @@ prepare_conflict_log_tuple(EState *estate, Relation rel,
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);
@@ -1266,16 +1319,11 @@ prepare_conflict_log_tuple(EState *estate, Relation rel,
else
nulls[attno++] = true;
- if (!TupIsNull(remoteslot))
- values[attno++] = tuple_table_slot_to_json_datum(remoteslot);
- else
- nulls[attno++] = true;
-
values[attno] = build_local_conflicts_json_array(estate, rel,
conflict_type,
conflicttuples);
- Assert(attno + 1 == MAX_CONFLICT_ATTR_NUM);
+ Assert(attno + 1 == NUM_CONFLICT_ATTRS);
oldctx = MemoryContextSwitchTo(ApplyContext);
MyLogicalRepWorker->conflict_log_tuple =
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 469451c736a..70ae38a7bd1 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -1766,15 +1766,15 @@ apply_handle_stream_start(StringInfo s)
/* extract XID of the top-level transaction */
stream_xid = logicalrep_read_stream_start(s, &first_segment);
- remote_xid = stream_xid;
- remote_final_lsn = InvalidXLogRecPtr;
- remote_commit_ts = 0;
-
if (!TransactionIdIsValid(stream_xid))
ereport(ERROR,
(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. */
@@ -5674,27 +5674,7 @@ start_apply(XLogRecPtr origin_startpos)
*/
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid);
-
- /*
- * Insert any pending conflict log tuple under a new transaction.
- */
- if (MyLogicalRepWorker->conflict_log_tuple != NULL)
- {
- Relation conflictlogrel;
- ConflictLogDest dest;
-
- StartTransactionCommand();
- PushActiveSnapshot(GetTransactionSnapshot());
-
- /* Open conflict log table and insert the tuple. */
- conflictlogrel = GetConflictLogDestAndTable(&dest);
- Assert(dest != CONFLICT_LOG_DEST_LOG);
- InsertConflictLogTuple(conflictlogrel);
- table_close(conflictlogrel, RowExclusiveLock);
-
- PopActiveSnapshot();
- CommitTransactionCommand();
- }
+ ProcessPendingConflictLogTuple();
PG_RE_THROW();
}
@@ -6069,6 +6049,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 6dcb7970bb7..8829f6c6378 100644
--- a/src/include/replication/conflict.h
+++ b/src/include/replication/conflict.h
@@ -115,6 +115,7 @@ 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);
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 05c2179b9a8..4f3880e5b83 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -85,11 +85,11 @@ $node_subscriber->wait_for_log(
$log_offset);
# Verify the contents of the Conflict Log Table (CLT)
-# This section ensures that the clt contains the expected
+# 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";
+my $clt = "pg_conflict.pg_conflict_log_for_subid_$subid";
# Wait for the conflict to be logged in the CLT
my $log_check = $node_subscriber->poll_query_until(
--
2.49.0
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: <CAFiTN-sx=k+Th=uYsrLcS6YMZbPVi9Wrggn1w2Nzf9MLEU7YRQ@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