public inbox for [email protected]
help / color / mirror / Atom feedFrom: Dilip Kumar <[email protected]>
To: shveta malik <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: vignesh C <[email protected]>
Cc: Nisha Moond <[email protected]>
Cc: Masahiko Sawada <[email protected]>
Cc: Bharath Rupireddy <[email protected]>
Cc: PostgreSQL Hackers <[email protected]>
Subject: Re: Proposal: Conflict log history table for Logical Replication
Date: Thu, 2 Jul 2026 13:39:12 +0530
Message-ID: <CAFiTN-s45_nOdvE8mBquZO=vyKbVY1ww_q3pejUz7DC29NcSxQ@mail.gmail.com> (raw)
In-Reply-To: <CAFiTN-tV0CSFpkUvDOrCdFob6uVFGj+qxDwY-kw8JKfXqm8mzg@mail.gmail.com>
References: <CAFiTN-u5D5o_AGNbHRZHaOqAMWkxLf+hSk_r9X3gv6HbLOB5+g@mail.gmail.com>
<CALj2ACViThGQDYi-yeqUeHqG2Pozn2AiyvtDtjE6zhhbM0KsEA@mail.gmail.com>
<CAA4eK1+44b3vd_OWfiaVNtjf5Njb5cek09pmKRmttBByeg0NoA@mail.gmail.com>
<CAFiTN-v3L0WacCDx5dkOSonaZQbJfstXL4HrCPD1ahRdUsRnSg@mail.gmail.com>
<CALj2ACW63uuxh0fSoxEAF8OMWhz1dJKSkp268WJDzf5BUqCf5g@mail.gmail.com>
<CAFiTN-s9WWLOhW1TO27NtJwGf0bh2+MWyp3NEkZFeN_S5_p_rA@mail.gmail.com>
<CAA4eK1LxnsEx5sMbQkK5MHAgXKPROMQXQ0n=fKMwz+UsfKQaMQ@mail.gmail.com>
<CAD21AoDj+c4LXf2y4ESR-gVyv9d8V0G4R8R9pn-PcmT5zPzYcg@mail.gmail.com>
<CAA4eK1KokmAwNOL6bS-ip_E3F96PiQTjC4j-M+5vD1T6uUyi3Q@mail.gmail.com>
<CAFiTN-vFKE8E_N6h+peX9DP92mxCeFdm5A9Esn4DkLmNcZ-dOA@mail.gmail.com>
<CAFiTN-shLYf-fOTQ_dBf3Xfx05gxs_8d93MHZXyyz6w2Bg5geQ@mail.gmail.com>
<CAFiTN-tEgkKQHUikn6iBFCYf7XOObR7ncUq=OVh7WEk=6P4ymw@mail.gmail.com>
<CAFiTN-tQiakd8m+-d6WN6RpJXSv_JcropZ2oGzme4d1JudQhYg@mail.gmail.com>
<CAJpy0uDKbYWt+YPADj=4fHEvrGEWgnG1n_YsiGT_EZiZf0VSAw@mail.gmail.com>
<CAFiTN-t82BiXen+HfdR9jZyOpuSO92xonnUK=khXsiZWBfOxMA@mail.gmail.com>
<CAJpy0uAu2paxGAEffD=vaBTW9Jqbtxxawb8K8FgiASfeKPnGog@mail.gmail.com>
<CAJpy0uC0ZWgHOivJ102A1fMkppwK3RuSMafRPKyjwkmJrjhVUw@mail.gmail.com>
<CAFiTN-vFV9-zajrwjYHYyFnyQsooOAXW4CpxB5f-iT3APjOtoQ@mail.gmail.com>
<CAJpy0uBeU1dZgaqsSVKc=P=EVUKxRgVuHR8jDXFL-HLibbE-kQ@mail.gmail.com>
<CAA4eK1+FOkOxhzVLAnDymoNjp4i98H-L1+ZsWDgJEv-ndnTzTA@mail.gmail.com>
<CAFiTN-sVK6Bp+BawCJU_WpAXQSTX4OkKmce5EE4YNBgD-XSjZw@mail.gmail.com>
<CAA4eK1LbjV0bctib9wUnBpEkC+2rZFPnGuRtrKuc5AtUAzum+A@mail.gmail.com>
<CAFiTN-vq50N3QP9p3_SH+tJ8Pn=uRDb0X4qEcQZYcGW9AX88rQ@mail.gmail.com>
<CAFiTN-u3+zRGPESP5kUUfa6NxaWh1HL-gd1225KJ0Uvzi1urow@mail.gmail.com>
<CAA4eK1L4iNk6mNTC83PbYrRfUdtivH4U961PkdFfOO7mvc=USg@mail.gmail.com>
<CAFiTN-v+Mh64UfR5zb5rwgyGm6HS80XRSZ_XeaWkg8=+s9o3Kg@mail.gmail.com>
<CAFiTN-s3ZFHteQsiC3H4=AjTWxuwN-w69XQ3xL5X6YOMTua4pA@mail.gmail.com>
<CAJpy0uDe724nY59j-8hMapZ_Fru1Wo-NucF4Ea1B3Jrw=+J+UQ@mail.gmail.com>
<CAFiTN-uR=86L_5tyiA7n73EXCSCuDfQKfL5O=c8n7zZom8_ONQ@mail.gmail.com>
<CAD21AoDfOS-J0M9WbM3D20eGbSPzbfLQ-9XoYkxO4AZ9twqyvg@mail.gmail.com>
<CAFiTN-vMTg2X7vwfHLr5Gvy8ViV63_iaEcpHmM8V5GpA9-u8cg@mail.gmail.com>
<CAA4eK1+b2Ws0e_ZYJsgZAPn7VWndxAK_YM_QMKcfXst3e7F6Jg@mail.gmail.com>
<CAFiTN-v6hFKMPrSyTBsz=AtEETYMbOxrqvhZJsPQqKgQc4WCLw@mail.gmail.com>
<CAA4eK1KV3rYkaxys5fh-PtE9kq5xrFbiaRpOSPoRgQG494ek+g@mail.gmail.com>
<CAFiTN-utvu=QjY1QQ1a_TvkpkpvesMWo9M8wTFYLaOTPdpOJvw@mail.gmail.com>
<CAA4eK1+HoSOEqNwT3twArPNx4_D7hSUoEg2LnYhX8n9iUwhXgQ@mail.gmail.com>
<CAFiTN-tqmsfW0Sk=1RhzuduxqLrf9KEc8VOvBae+4aYxWTJwuA@mail.gmail.com>
<CAA4eK1JmCQ=DHe3HsqpX+P3mGDUd_Z7E7oAxdstK6822W6tuCw@mail.gmail.com>
<CAFiTN-uE4eAUYewuq3c5deAt3TtVork+H6rkUHRv68cOGr5rmQ@mail.gmail.com>
<CAFiTN-sJbhPX+LbA8YuQeYJpfGA2XA+OKXf8jCm04RoJOyzLvw@mail.gmail.com>
<CAJpy0uBPOyWj9itFjHzGXfrUuYS8KGmAvgdcV_9FPjWZ0EZz_w@mail.gmail.com>
<CAFiTN-s=iLE4qM4qmw9yXKqW09R_c_HqaSGeZXJ2EaTVfXss+g@mail.gmail.com>
<CAA4eK1KYo0vZpPSRc_4gVpa06-J39gxjs3tHFyckgkBfYJSfFA@mail.gmail.com>
<CAFiTN-vrKc6OWzrg6yvpwYcj79k=zkrDp3uwiZzjwrWLJAq6tw@mail.gmail.com>
<CAA4eK1LmvrfEgn1NUZZ=E3yMCjQdNZ5=_SBEry73-EmF6jM_PQ@mail.gmail.com>
<CAFiTN-vjfub5b3PqPQzfOw9BSjm8jt28ott+Hoz9CrRxJHzYkg@mail.gmail.com>
<CAFiTN-v=ANapYvRK+SOy2wJb4CSuD6Vb6_bTGuReM9Dv+3tucA@mail.gmail.com>
<CALDaNm1zEYoSdf2Ns-=UJRw95E5sbfpB0oaNUWtRJN27Q1Knhw@mail.gmail.com>
<CALDaNm3USsXVNBsfdpkp60HVgrTV4taWMk1xZYNBa7QUF=V0jg@mail.gmail.com>
<CAFiTN-sNg9ghLNkB2Kn0SwBGOub9acc99XZZU_d5NAcyW-yrEg@mail.gmail.com>
<CAJpy0uAF3EYcYdpTHdKMeXfvaPbNvnWrZUATrSLL1hqjao=33A@mail.gmail.com>
<CAFiTN-uikggCKp2LscTorKY5d3KF9j93DW0xebDcRX86G+ZsSw@mail.gmail.com>
<CAJpy0uDaOoVK8S3_xxTAcTDpfK1AY7tApw7nPOZG_gUz+DMi=Q@mail.gmail.com>
<CAA4eK1+AdeC5B9xrAXSKWGtTh-0d8xdD=fZttmOBm+c8o8thAQ@mail.gmail.com>
<CAFiTN-skBQAeuzuUd+PDK0Gqc8g+4x9ypBMwJhOrmW8ZCFKGSA@mail.gmail.com>
<CAJpy0uCdrsW5T+okq7xTOVxagje7FW3DOeY5B0CGKYa5VqF_tQ@mail.gmail.com>
<CAFiTN-u+_mFj9caYYFO7=_YHFXk5y=vvOm2H2=5hctYktmAVGA@mail.gmail.com>
<CALDaNm1aivk9KgQ5daeF6YZzuE+0wWc2yb7wb6qikNyvfPN0Sg@mail.gmail.com>
<CAJpy0uD6fTEUYJx3+yDbvB=VW7c5AaGoeSd7iwHdYYO=kYGn3g@mail.gmail.com>
<CALDaNm2YOOdJ25X1sJ+DYz37K6Qi4g0ZNFHb_pQMF9UqancnEA@mail.gmail.com>
<CAHut+PtMS5bENS0DVtBj+s3kUEOq61+hSkqLODjFB78egB0imQ@mail.gmail.com>
<CAFiTN-s_M83sfs+MHHbUrMesjsCPN4JWxY5MChCEiY1U-u7=9g@mail.gmail.com>
<CAFiTN-vj8NTm9w_L2XdhxJCub_RZw__YVUgfXa1B1kJzJctRNw@mail.gmail.com>
<CAJpy0uBDLnfhuSiev8W9ZMFNTzUmqhds2dKayUpLoN-z1dtsLA@mail.gmail.com>
<CAFiTN-uL9f0X+=Ep4BbAPvaTJA7S4XHM--G4BsnPJw4uJW7EGQ@mail.gmail.com>
<CAJpy0uDG=t-y_m8t1zpBzfz9viP3K8dyQgkruaraVT85UtTkrg@mail.gmail.com>
<CAFiTN-tR8Rhs8uhfbck0Ac4dd1MopvvYgjK39nWyNXRp9Z3Qww@mail.gmail.com>
<CAA4eK1Kf15UpNmpTTE2XyX=9PE_oTpOoy5xqg3rFWbxwwP4Rbg@mail.gmail.com>
<CAFiTN-tNqb0vjuadDz-as67ksSXa=aEK+JW=4b54RVmkUK1m2Q@mail.gmail.com>
<CAFiTN-vDCxx6ydUFo59L8qNBbierg4as3TGPPiavR7UZjYurzA@mail.gmail.com>
<CAHut+PsWms218ENALnytLEV4NpxjOrAYhChLDaMaeE65-vNgrQ@mail.gmail.com>
<CAFiTN-v9i9RmDvdUmtMUow4=b+nr0k7LKMyEQ+6ZF=EVdfBhBA@mail.gmail.com>
<CALDaNm2YTKwPDjt9OV49RgM0zbkWhMhNu228bj_7f+zzcPb-ew@mail.gmail.com>
<CAFiTN-t_4XvofM3an-WmykqnPE+9wf9U+o2M7p1CWd9eXkN88Q@mail.gmail.com>
<CAHut+PuaqNDfDu_3xkZR4OYxw-B7ew_WjpLXCBvMcSBJz2K6Xg@mail.gmail.com>
<CAFiTN-uqNN9S_hRuda_th5MEpywa15g+XO00yM6tNJ-spGRRJw@mail.gmail.com>
<CABdArM6QxXatkGefTHy__HgaYHBvbKesffeXzT8Vn-kvcvGK4w@mail.gmail.com>
<CAFiTN-tgMWr=TGPhs9BxaPuSC_jhM7sJJ4fHedE5W6=h40jLfA@mail.gmail.com>
<CABdArM5fgzfyC2mH3YGB8t8cJBHWqAG1BS6rJMk7mX-8=9d=Cg@mail.gmail.com>
<CABdArM568KF4WXdFX_aZkCiDK8R71Wpep0gC2a+cV8BMobwkrg@mail.gmail.com>
<CAFiTN-vQ0tu18BD3UmKPb0rzZyFMQAVgGbdpMA8iYLX7PZOqOA@mail.gmail.com>
<CALDaNm20PDtmG2E3qaTC+YuL5twv+c9k573wL3sb=OwgmZphxQ@mail.gmail.com>
<CAA4eK1LhOHa_TEznw+gFoq+w0vMvvsDG2g9Xq8Mwa8xZMY73og@mail.gmail.com>
<CAFiTN-vPDqrQ2rHykNgd+groFxqwBYFQF97R-Co2EmtUkV6MTg@mail.gmail.com>
<CAFiTN-vsd=wNiEPXPQhZnipAb--+mBUC01M-pcjBjbRockgCUA@mail.gmail.com>
<CAJpy0uCjSq_gUCJBfURhqtB6bLvkKSUL-sVXpaGKjEapv5+t+Q@mail.gmail.com>
<CAFiTN-uZ-LaStAY3NuCY-nb7GCB9joiHX7HtHEMseJ0xfnqVSg@mail.gmail.com>
<CAFiTN-vhJxRW5NQ628oidnk0KtHwKt11dW9-+vxqpXLTgjiYiA@mail.gmail.com>
<CALDaNm1cJURibYKY4+DuNosjM72C9oGheUF-roMyff__+AsKBw@mail.gmail.com>
<CAJpy0uD1_77TDAFc4jE-94X-WUus7Q3gGU0pXfC+Tticq1hFvA@mail.gmail.com>
<CAA4eK1LFcSc4XCj4mU-cv27F_6n6=+ehJ=YAsAnyBbz4Sv_tVg@mail.gmail.com>
<CAFiTN-s5ZtjXKrSbam7TNWJ9Ax-kCancXcestAnx2by7dK0-UA@mail.gmail.com>
<CAFiTN-u=Da32mXyz8jocEGtuLSG4ccXXj_aEzUTPp2zkLb3MVA@mail.gmail.com>
<CALDaNm1qY5e0thfsDB2uWXqZn4hgTWTxiUDwcF1hWA-jodsKYg@mail.gmail.com>
<CALDaNm1nFtv3dtdRdbqWo2Rf_av7XbxDfK1Orqjcqs_Su_cLRQ@mail.gmail.com>
<CABdArM7R498qC5Fr42aU_q-2Sc5QsT4dyKgmO_f6Uy=8oCAFXA@mail.gmail.com>
<CAFiTN-sRZ+Z_9B3ue2L4zkbcfmPjjcAjcR1C+px1PyAs+HGsSg@mail.gmail.com>
<CAFiTN-sdcjf9xJ2M-=ab5e4y662tTmFFiP4gHL44tC9PcQozcw@mail.gmail.com>
<CALDaNm2WNjaNxUijVkvT6y69D62rfCu8OMwU-Pf-84un2r_=ig@mail.gmail.com>
<CAHut+PvEP5uUR13xJ3gbNKGU49=Rg32DXMGZ2wL9jTcKHyN_=Q@mail.gmail.com>
<CALDaNm3Jb5AQTsFJFxYZZJCaheT7qToCZkEALfW-vsMMFxjOyQ@mail.gmail.com>
<CAHut+PtQn5U9i00qvBmjo0KBxyb+ZmBb38NzF91KnX4J86Jg_g@mail.gmail.com>
<CALDaNm1a1gzy0L38U394_4OFwGUS8ALgSONYj++VLimY0g9piQ@mail.gmail.com>
<CABdArM5X63AdtS99QKGjVijUd_Q_dV8QUDSo4nTHKJjn3JwtAg@mail.gmail.com>
<CAA4eK1+h=QV4Zi=PW8Zt2D6be5ki5Mu2HgdXcfUophptx6Mt_A@mail.gmail.com>
<CAFiTN-s-tuxar9Dp5He0CFa1pzfy1fmiwcBj6PtwD0hDodE5ng@mail.gmail.com>
<CAFiTN-sx=k+Th=uYsrLcS6YMZbPVi9Wrggn1w2Nzf9MLEU7YRQ@mail.gmail.com>
<CAFiTN-u5pcgAhXyJgj+p7-xmShtp0i8xA000tzjCLFQp_zMXUA@mail.gmail.com>
<CAFiTN-tRpS7b3qFqckqFtHETj0jyzj-8SxC1arjfwf-hQd47PQ@mail.gmail.com>
<CABdArM5Ka_m_GWhL_zZbeDPKmL-Wezwb4A-NHnO-v-fRDuhA-Q@mail.gmail.com>
<CAFiTN-sqEMAbZ2pTt=zMa=918NV7HVeXF4bCOF+swtzKnTy5yQ@mail.gmail.com>
<CALDaNm0WX0Vqoy2UQZh-2TpWraf4OYn28kWe9aGR=vxKwLA+bw@mail.gmail.com>
<CAJpy0uAA_XszCAcoBuCUM3VobD39DbMDwCPUT+XW7wFfE+_E8w@mail.gmail.com>
<CAFiTN-urKwsdyFvwz1go_C7jJFtLA8TJEhnaECAjuqkRk_1cXA@mail.gmail.com>
<CAJpy0uBhJGqD+OyA9Uk8bHyk61XWHEf3Le1QxkotwOLcQCqaZA@mail.gmail.com>
<CAFiTN-sbAXS6deUarPDunj2U+A6Dvhw_TASy4oMv4Tc63-_T9g@mail.gmail.com>
<CAJpy0uBaW7ziTsHOu_z37-epihxx3qGMjqdV+_-Z-RFjq5EOGg@mail.gmail.com>
<CAFiTN-sxVPbuqHjz99NGTz5UU0xgegsvpRa6=NkbP8_iW+X6-A@mail.gmail.com>
<CAJpy0uBs6D9ojMpz4MWgrxDbvRxqnvN+B+JnMNezBtuvhk_j9A@mail.gmail.com>
<CAFiTN-urmbTtk40RfLU8UMYtzk-_DLXwui3_G+TWp6XYKBphjw@mail.gmail.com>
<CABdArM5dtMEUw14-aDSht2Vh1tsgO67t_ZU8VQp=Ut7MwK5aEQ@mail.gmail.com>
<CAJpy0uBSY7zTH=4TvAOS=kj9vivBUc9NO+Vp6KNw-Na9RiAsMg@mail.gmail.com>
<CAFiTN-vqyo2=pgA_jUyQoWrOwUXxw1s3aYHcZBr_EHsztZN9jg@mail.gmail.com>
<CAJpy0uDGuaEvBym5VnE4mXJ_meo3jR7uG=c6wVbDAiuTjHcwrg@mail.gmail.com>
<CAA4eK1JhZYRMP_YYa1j3uAK6L4v057JDuM0+YLABOgAOYuwM8Q@mail.gmail.com>
<CABdArM6r0u6e2HQY3CQ15uZwOD2iTtVndecsNyK4R5tX9+eaNQ@mail.gmail.com>
<CAFiTN-tnBhCU8opWAjp1xYkjqGeXtQ1tfDZHbYBS7TifymXEvw@mail.gmail.com>
<CALDaNm0siWVCuTg6S8Xbk5SMHc91AX8Y+fAtOyV+cjwcpiFDWQ@mail.gmail.com>
<CAFiTN-vvJ42f4Bs679gVxBH8BpqmOskW1bnOUTUpy-x0L2wz_g@mail.gmail.com>
<CAA4eK1KsF+v3v6wOoWDWzj3fb_1kDi9dpf+aHdjz=2-k_C3mrg@mail.gmail.com>
<CALDaNm2GJYd6OBO9h=nTHuLt-Vg66nsGMv0GrvfX=S=nE3qqEg@mail.gmail.com>
<CAFiTN-tni=ADm_PrC2-H_6-4OdAqYNfUgQaFORdR6bnRNWrB1g@mail.gmail.com>
<CALDaNm3QaSpS6Er=9Q1Cj1Ry1QJoM8XaomyaFpp=_GKEp2MegQ@mail.gmail.com>
<CAA4eK1L7tSG6wEErAPd4MGO7yTXd-7fcWG74dgMqQ0Wz2Gw+hQ@mail.gmail.com>
<CAA4eK1L9XbR5Ecma-7cb_xChPMd=8Cy==yebPtY8SEVXUMhxtA@mail.gmail.com>
<CAFiTN-v=-hNgJ4LSHBO9XCGzYe1gBeKEyNM4j2B4XV7HayMrsA@mail.gmail.com>
<CAHut+PtjCycd1aZjaarptdfu-v+ZXrC_ws6gcdC61K2utNQiDg@mail.gmail.com>
<CAFiTN-v9gak1G0u5qzO2Q3Towcvm0JQR-EC5BZhewg4mcReXWg@mail.gmail.com>
<CAA4eK1+y4P+uL+keDZ1pD6EFKvEXvGX8k57mmM4SiLFd1H2xMQ@mail.gmail.com>
<CAA4eK1Kirm_rVGkuKyV_b=cP00C=L61VpeceqQo8+BPGHmv6vg@mail.gmail.com>
<CAFiTN-u2ZSuDdLNUCFCnt5VL3nqp2vs9GdJUN-DfdftsLgijGw@mail.gmail.com>
<CAJpy0uDyHrZndi2qA=t_Jiyg8EcGwSCVZAddigZcKu==qtP3kg@mail.gmail.com>
<CAFiTN-s-7Dse7=BgVv6JKTTU_oy6_jufw9OJTk5y4M09fBLC-w@mail.gmail.com>
<CAJpy0uCiTX6Epnn-Kb2ff3317wKe-0x3+2edMKT6S40+wFPXuw@mail.gmail.com>
On Wed, Jul 1, 2026 at 6:55 PM Dilip Kumar <[email protected]> wrote:
>
> On Wed, Jul 1, 2026 at 1:16 PM shveta malik <[email protected]> wrote:
> >
> > On Wed, Jul 1, 2026 at 12:45 PM Dilip Kumar <[email protected]> wrote:
> > >
> > > Thanks I have fixed these comments and also merged the docs.
> >
> > Thanks. Sharing the version with pgindent run on v61.
>
> Here is the updated patch version. It fixes offlist review comments
> provided by Amit and Shveta, bumps the catversion and adds authors and
> reviewers details.
I rebased the remaining patches on top of HEAD. So far, I have run
pgindent and completed the doc merge for 0001. The 0002 and 0003 are
just rebased.
--
Regards,
Dilip Kumar
Google
Attachments:
[application/octet-stream] v63-0001-Implement-the-conflict-insertion-infrastructure-.patch (94.2K, ../CAFiTN-s45_nOdvE8mBquZO=vyKbVY1ww_q3pejUz7DC29NcSxQ@mail.gmail.com/2-v63-0001-Implement-the-conflict-insertion-infrastructure-.patch)
download | inline diff:
From 4028b33496732bf7ee4e3140134b49e98d7cffcc Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Thu, 2 Jul 2026 09:17:51 +0530
Subject: [PATCH v63 1/3] Implement the conflict insertion infrastructure for
the conflict log table
This patch introduces the core logic to populate the conflict log table whenever
a logical replication conflict is detected. It captures the remote transaction
details along with the corresponding local state at the time of the conflict.
Handling Multi-row Conflicts: A single remote tuple may conflict with multiple
local tuples (e.g., in the case of multiple_unique_conflicts). To handle this,
the infrastructure creates a single row in the conflict log table for each
remote tuple. The details of all conflicting local rows are aggregated into a
single JSON array in the local_conflicts column.
The JSON array uses the following structured format:
[ { "xid": "1001", "commit_ts": "2025-12-25 10:00:00+05:30", "origin": "node_1",
"key": {"id": 1}, "tuple": {"id": 1, "val": "old_data"} }, ... ]
Example of querying the structured conflict data:
SELECT remote_xid, relname, remote_origin, local_conflicts[1] ->> 'xid' AS local_xid,
local_conflicts[1] ->> 'tuple' AS local_tuple
FROM pg_conflict.pg_conflict_log_16396;
remote_xid | relname | remote_origin | local_xid | local_tuple
------------+----------+---------------+-----------+---------------------
760 | test | pg_16406 | 771 | {"a":1,"b":10}
765 | conf_tab | pg_16406 | 775 | {"a":2,"b":2,"c":2}
---
doc/src/sgml/logical-replication.sgml | 659 +++++++++------
doc/src/sgml/ref/alter_subscription.sgml | 3 +-
doc/src/sgml/ref/create_subscription.sgml | 17 +-
.../replication/logical/applyparallelworker.c | 85 +-
src/backend/replication/logical/conflict.c | 777 +++++++++++++++++-
src/backend/replication/logical/worker.c | 67 +-
src/include/replication/conflict.h | 4 +
src/include/replication/worker_internal.h | 7 +
src/include/utils/rel.h | 13 +-
src/test/subscription/meson.build | 1 +
src/test/subscription/t/030_origin.pl | 4 +-
src/test/subscription/t/035_conflicts.pl | 49 +-
.../t/039_pa_conflict_log_lock_wait.pl | 203 +++++
src/tools/pgindent/typedefs.list | 1 +
14 files changed, 1554 insertions(+), 336 deletions(-)
create mode 100644 src/test/subscription/t/039_pa_conflict_log_lock_wait.pl
diff --git a/doc/src/sgml/logical-replication.sgml b/doc/src/sgml/logical-replication.sgml
index f21239af6b8..e0399eed6b7 100644
--- a/doc/src/sgml/logical-replication.sgml
+++ b/doc/src/sgml/logical-replication.sgml
@@ -2024,120 +2024,230 @@ Included in publications:
operations will simply be skipped.
</para>
- <para>
- Additional logging is triggered, and the conflict statistics are collected (displayed in the
- <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view)
- in the following <firstterm>conflict</firstterm> cases:
- <variablelist>
- <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
- <term><literal>insert_exists</literal></term>
- <listitem>
- <para>
- Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
- unique constraint. Note that to log the origin and commit
- timestamp details of the conflicting key,
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- should be enabled on the subscriber. In this case, an error will be
- raised until the conflict is resolved manually.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
- <term><literal>update_origin_differs</literal></term>
- <listitem>
- <para>
- Updating a row that was previously modified by another origin.
- Note that this conflict can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Currently, the update is always applied
- regardless of the origin of the local row.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-exists" xreflabel="update_exists">
- <term><literal>update_exists</literal></term>
- <listitem>
- <para>
- The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
- unique constraint. Note that to log the origin and commit
- timestamp details of the conflicting key,
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- should be enabled on the subscriber. In this case, an error will be
- raised until the conflict is resolved manually. Note that when updating a
- partitioned table, if the updated row value satisfies another partition
- constraint resulting in the row being inserted into a new partition, the
- <literal>insert_exists</literal> conflict may arise if the new row
- violates a <literal>NOT DEFERRABLE</literal> unique constraint.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
- <term><literal>update_deleted</literal></term>
- <listitem>
- <para>
- The tuple to be updated was concurrently deleted by another origin. The
- update will simply be skipped in this scenario. Note that this conflict
- can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
- are enabled. Note that if a tuple cannot be found due to the table being
- truncated, only a <literal>update_missing</literal> conflict will
- arise. Additionally, if the tuple was deleted by the same origin, an
- <literal>update_missing</literal> conflict will arise.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-update-missing" xreflabel="update_missing">
- <term><literal>update_missing</literal></term>
- <listitem>
- <para>
- The row to be updated was not found. The update will simply be
- skipped in this scenario.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
- <term><literal>delete_origin_differs</literal></term>
- <listitem>
- <para>
- Deleting a row that was previously modified by another origin. Note that
- this conflict can only be detected when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Currently, the delete is always applied
- regardless of the origin of the local row.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
- <term><literal>delete_missing</literal></term>
- <listitem>
- <para>
- The row to be deleted was not found. The delete will simply be
- skipped in this scenario.
- </para>
- </listitem>
- </varlistentry>
- <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
- <term><literal>multiple_unique_conflicts</literal></term>
- <listitem>
- <para>
- Inserting or updating a row violates multiple
- <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
- the origin and commit timestamp details of conflicting keys, ensure
- that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. In this case, an error will be raised until
- the conflict is resolved manually.
- </para>
- </listitem>
- </varlistentry>
- </variablelist>
- Note that there are other conflict scenarios, such as exclusion constraint
- violations. Currently, we do not provide additional details for them in the
- log.
- </para>
+ <sect2 id="logical-replication-conflict-logging">
+ <title>Conflict logging</title>
+ <para>
+ Additional logging is triggered, and the conflict statistics are collected (displayed in the
+ <link linkend="monitoring-pg-stat-subscription-stats"><structname>pg_stat_subscription_stats</structname></link> view)
+ in the following <firstterm>conflict</firstterm> cases:
+ <variablelist>
+ <varlistentry id="conflict-insert-exists" xreflabel="insert_exists">
+ <term><literal>insert_exists</literal></term>
+ <listitem>
+ <para>
+ Inserting a row that violates a <literal>NOT DEFERRABLE</literal>
+ unique constraint. Note that to log the origin and commit
+ timestamp details of the conflicting key,
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ should be enabled on the subscriber. In this case, an error will be
+ raised until the conflict is resolved manually.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-origin-differs" xreflabel="update_origin_differs">
+ <term><literal>update_origin_differs</literal></term>
+ <listitem>
+ <para>
+ Updating a row that was previously modified by another origin.
+ Note that this conflict can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Currently, the update is always applied
+ regardless of the origin of the local row.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-exists" xreflabel="update_exists">
+ <term><literal>update_exists</literal></term>
+ <listitem>
+ <para>
+ The updated value of a row violates a <literal>NOT DEFERRABLE</literal>
+ unique constraint. Note that to log the origin and commit
+ timestamp details of the conflicting key,
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ should be enabled on the subscriber. In this case, an error will be
+ raised until the conflict is resolved manually. Note that when updating a
+ partitioned table, if the updated row value satisfies another partition
+ constraint resulting in the row being inserted into a new partition, the
+ <literal>insert_exists</literal> conflict may arise if the new row
+ violates a <literal>NOT DEFERRABLE</literal> unique constraint.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-deleted" xreflabel="update_deleted">
+ <term><literal>update_deleted</literal></term>
+ <listitem>
+ <para>
+ The tuple to be updated was concurrently deleted by another origin. The
+ update will simply be skipped in this scenario. Note that this conflict
+ can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ and <link linkend="sql-createsubscription-params-with-retain-dead-tuples"><literal>retain_dead_tuples</literal></link>
+ are enabled. Note that if a tuple cannot be found due to the table being
+ truncated, only a <literal>update_missing</literal> conflict will
+ arise. Additionally, if the tuple was deleted by the same origin, an
+ <literal>update_missing</literal> conflict will arise.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-update-missing" xreflabel="update_missing">
+ <term><literal>update_missing</literal></term>
+ <listitem>
+ <para>
+ The row to be updated was not found. The update will simply be
+ skipped in this scenario.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-delete-origin-differs" xreflabel="delete_origin_differs">
+ <term><literal>delete_origin_differs</literal></term>
+ <listitem>
+ <para>
+ Deleting a row that was previously modified by another origin. Note that
+ this conflict can only be detected when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Currently, the delete is always applied
+ regardless of the origin of the local row.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-delete-missing" xreflabel="delete_missing">
+ <term><literal>delete_missing</literal></term>
+ <listitem>
+ <para>
+ The row to be deleted was not found. The delete will simply be
+ skipped in this scenario.
+ </para>
+ </listitem>
+ </varlistentry>
+ <varlistentry id="conflict-multiple-unique-conflicts" xreflabel="multiple_unique_conflicts">
+ <term><literal>multiple_unique_conflicts</literal></term>
+ <listitem>
+ <para>
+ Inserting or updating a row violates multiple
+ <literal>NOT DEFERRABLE</literal> unique constraints. Note that to log
+ the origin and commit timestamp details of conflicting keys, ensure
+ that <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. In this case, an error will be raised until
+ the conflict is resolved manually.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+ Note that there are other conflict scenarios, such as exclusion constraint
+ violations. Currently, we do not provide additional details for them in the
+ log.
+ </para>
+ </sect2>
- <para>
- The log format for logical replication conflicts is as follows:
+ <sect2 id="logical-replication-conflict-table-based-logging">
+ <title>Table-based logging</title>
+ <para>
+ If the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ parameter is set to <literal>table</literal> or <literal>all</literal>,
+ a dedicated conflict log table will be automatically created. This table is
+ created in the <literal>pg_conflict</literal> namespace. The name of the
+ conflict log table is
+ <literal>pg_conflict_log_<subid></literal>. The predefined
+ schema of this table is detailed in
+ <xref linkend="logical-replication-conflict-log-schema"/>.
+ </para>
+
+ <table id="logical-replication-conflict-log-schema">
+ <title>Conflict Log Table Schema</title>
+ <tgroup cols="3">
+ <thead>
+ <row>
+ <entry>Column</entry>
+ <entry>Type</entry>
+ <entry>Description</entry>
+ </row>
+ </thead>
+ <tbody>
+ <row>
+ <entry><literal>relid</literal></entry>
+ <entry><type>oid</type></entry>
+ <entry>The OID of the local table where the conflict occurred.</entry>
+ </row>
+ <row>
+ <entry><literal>schemaname</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The schema name of the conflicting table.</entry>
+ </row>
+ <row>
+ <entry><literal>relname</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The name of the conflicting table.</entry>
+ </row>
+ <row>
+ <entry><literal>conflict_type</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The type of conflict that occurred (e.g., <literal>insert_exists</literal>).</entry>
+ </row>
+ <row>
+ <entry><literal>remote_xid</literal></entry>
+ <entry><type>xid</type></entry>
+ <entry>The remote transaction ID that caused the conflict.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_commit_lsn</literal></entry>
+ <entry><type>pg_lsn</type></entry>
+ <entry>The final LSN of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_commit_ts</literal></entry>
+ <entry><type>timestamptz</type></entry>
+ <entry>The remote commit timestamp of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_origin</literal></entry>
+ <entry><type>text</type></entry>
+ <entry>The origin of the remote transaction.</entry>
+ </row>
+ <row>
+ <entry><literal>replica_identity_full</literal></entry>
+ <entry><type>boolean</type></entry>
+ <entry>Indicates whether <literal>replica_identity</literal> represents a full tuple (<literal>true</literal>) or key values of a replica identity index (<literal>false</literal>).</entry>
+ </row>
+ <row>
+ <entry><literal>replica_identity</literal></entry>
+ <entry><type>json</type></entry>
+ <entry>The JSON representation of the replica identity key values or full tuple.</entry>
+ </row>
+ <row>
+ <entry><literal>remote_tuple</literal></entry>
+ <entry><type>json</type></entry>
+ <entry>The JSON representation of the incoming remote row that caused
+ the conflict.</entry>
+ </row>
+ <row>
+ <entry><literal>local_conflicts</literal></entry>
+ <entry><type>json[]</type></entry>
+ <entry>
+ An array of JSON objects representing the state of existing local
+ row(s) that caused the conflict. Each object includes the local
+ transaction ID (<literal>xid</literal>), commit timestamp
+ (<literal>commit_ts</literal>), origin (<literal>origin</literal>),
+ conflicting key data (<literal>key</literal>), and the full local row
+ image (<literal>tuple</literal>).
+ </entry>
+ </row>
+ </tbody>
+ </tgroup>
+ </table>
+
+ <para>
+ The conflicting row data, including the incoming remote row (<literal>remote_tuple</literal>)
+ and the associated local conflict details (<literal>local_conflicts</literal>), is stored in
+ <type>JSON</type> formats for flexible querying and analysis.
+ </para>
+ </sect2>
+
+ <sect2 id="logical-replication-conflict-file-based-logging">
+ <title>File-based logging</title>
+ <para>
+ If the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
+ parameter is set to <literal>log</literal> or <literal>all</literal>, conflicts
+ are logged to the server log using the following format:
<synopsis>
LOG: conflict detected on relation "<replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>": conflict=<replaceable>conflict_type</replaceable>
DETAIL: <replaceable class="parameter">detailed_explanation</replaceable>[: <replaceable class="parameter">detail_values</replaceable> [, ... ]].
@@ -2150,182 +2260,185 @@ DETAIL: <replaceable class="parameter">detailed_explanation</replaceable>[: <re
<literal>replica identity</literal> {(<replaceable>column_name</replaceable> <optional>, ...</optional>)=(<replaceable>column_value</replaceable> <optional>, ...</optional>) | full <optional>(<replaceable>column_name</replaceable> <optional>, ...</optional>)=</optional>(<replaceable>column_value</replaceable> <optional>, ...</optional>)}
</synopsis>
- The log provides the following information:
- <variablelist>
- <varlistentry>
- <term><literal>LOG</literal></term>
- <listitem>
+ The log provides the following information:
+ <variablelist>
+ <varlistentry>
+ <term><literal>LOG</literal></term>
+ <listitem>
+ <itemizedlist>
+ <listitem>
+ <para>
+ <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
+ identifies the local relation involved in the conflict.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <replaceable>conflict_type</replaceable> is the type of conflict that occurred
+ (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+ </para>
+ </listitem>
+ </itemizedlist>
+ </listitem>
+ </varlistentry>
+
+ <varlistentry>
+ <term><literal>DETAIL</literal></term>
+ <listitem>
<itemizedlist>
<listitem>
<para>
- <replaceable>schemaname</replaceable>.<replaceable>tablename</replaceable>
- identifies the local relation involved in the conflict.
+ <replaceable class="parameter">detailed_explanation</replaceable> includes
+ the origin, transaction ID, and commit timestamp of the transaction that
+ modified the local row, if available.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>key</literal> section includes the key values of the local
+ row that violated a unique constraint for
+ <literal>insert_exists</literal>, <literal>update_exists</literal> or
+ <literal>multiple_unique_conflicts</literal> conflicts.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>local row</literal> section includes the local row if its
+ origin differs from the remote row for
+ <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
+ conflicts, or if the key value conflicts with the remote row for
+ <literal>insert_exists</literal>, <literal>update_exists</literal> or
+ <literal>multiple_unique_conflicts</literal> conflicts.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>remote row</literal> section includes the new row from
+ the remote insert or update operation that caused the conflict. Note that
+ for an update operation, the column value of the new row will be null
+ if the value is unchanged and toasted.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ The <literal>replica identity</literal> section includes the replica
+ identity key values that were used to search for the existing local
+ row to be updated or deleted. This may include the full row value
+ if the local relation is marked with
+ <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
</para>
</listitem>
<listitem>
<para>
- <replaceable>conflict_type</replaceable> is the type of conflict that occurred
- (e.g., <literal>insert_exists</literal>, <literal>update_exists</literal>).
+ <replaceable class="parameter">column_name</replaceable> is the column name.
+ For <literal>local row</literal>, <literal>remote row</literal>, and
+ <literal>replica identity full</literal> cases, column names are
+ logged only if the user lacks the privilege to access all columns of
+ the table. If column names are present, they appear in the same order
+ as the corresponding column values.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ <replaceable class="parameter">column_value</replaceable> is the column value.
+ The large column values are truncated to 64 bytes.
+ </para>
+ </listitem>
+ <listitem>
+ <para>
+ Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
+ multiple <replaceable class="parameter">detailed_explanation</replaceable>
+ and <replaceable class="parameter">detail_values</replaceable> lines
+ will be generated, each detailing the conflict information associated
+ with distinct unique constraints.
</para>
</listitem>
</itemizedlist>
</listitem>
- </varlistentry>
-
- <varlistentry>
- <term><literal>DETAIL</literal></term>
- <listitem>
- <itemizedlist>
- <listitem>
- <para>
- <replaceable class="parameter">detailed_explanation</replaceable> includes
- the origin, transaction ID, and commit timestamp of the transaction that
- modified the local row, if available.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>key</literal> section includes the key values of the local
- row that violated a unique constraint for
- <literal>insert_exists</literal>, <literal>update_exists</literal> or
- <literal>multiple_unique_conflicts</literal> conflicts.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>local row</literal> section includes the local row if its
- origin differs from the remote row for
- <literal>update_origin_differs</literal> or <literal>delete_origin_differs</literal>
- conflicts, or if the key value conflicts with the remote row for
- <literal>insert_exists</literal>, <literal>update_exists</literal> or
- <literal>multiple_unique_conflicts</literal> conflicts.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>remote row</literal> section includes the new row from
- the remote insert or update operation that caused the conflict. Note that
- for an update operation, the column value of the new row will be null
- if the value is unchanged and toasted.
- </para>
- </listitem>
- <listitem>
- <para>
- The <literal>replica identity</literal> section includes the replica
- identity key values that were used to search for the existing local
- row to be updated or deleted. This may include the full row value
- if the local relation is marked with
- <link linkend="sql-altertable-replica-identity-full"><literal>REPLICA IDENTITY FULL</literal></link>.
- </para>
- </listitem>
- <listitem>
- <para>
- <replaceable class="parameter">column_name</replaceable> is the column name.
- For <literal>local row</literal>, <literal>remote row</literal>, and
- <literal>replica identity full</literal> cases, column names are
- logged only if the user lacks the privilege to access all columns of
- the table. If column names are present, they appear in the same order
- as the corresponding column values.
- </para>
- </listitem>
- <listitem>
- <para>
- <replaceable class="parameter">column_value</replaceable> is the column value.
- The large column values are truncated to 64 bytes.
- </para>
- </listitem>
- <listitem>
- <para>
- Note that in case of <literal>multiple_unique_conflicts</literal> conflict,
- multiple <replaceable class="parameter">detailed_explanation</replaceable>
- and <replaceable class="parameter">detail_values</replaceable> lines
- will be generated, each detailing the conflict information associated
- with distinct unique
- constraints.
- </para>
- </listitem>
- </itemizedlist>
- </listitem>
- </varlistentry>
- </variablelist>
- </para>
+ </varlistentry>
+ </variablelist>
+ </para>
+ </sect2>
- <para>
- Logical replication operations are performed with the privileges of the role
- which owns the subscription. Permissions failures on target tables will
- cause replication conflicts, as will enabled
- <link linkend="ddl-rowsecurity">row-level security</link> on target tables
- that the subscription owner is subject to, without regard to whether any
- policy would ordinarily reject the <command>INSERT</command>,
- <command>UPDATE</command>, <command>DELETE</command> or
- <command>TRUNCATE</command> which is being replicated. This restriction on
- row-level security may be lifted in a future version of
- <productname>PostgreSQL</productname>.
- </para>
+ <sect2 id="logical-replication-conflict-notes">
+ <title>Notes</title>
+ <para>
+ Logical replication operations are performed with the privileges of the role
+ which owns the subscription. Permissions failures on target tables will
+ cause replication conflicts, as will enabled
+ <link linkend="ddl-rowsecurity">row-level security</link> on target tables
+ that the subscription owner is subject to, without regard to whether any
+ policy would ordinarily reject the <command>INSERT</command>,
+ <command>UPDATE</command>, <command>DELETE</command> or
+ <command>TRUNCATE</command> which is being replicated. This restriction on
+ row-level security may be lifted in a future version of
+ <productname>PostgreSQL</productname>.
+ </para>
- <para>
- A conflict that produces an error will stop the replication; it must be
- resolved manually by the user. Details about the conflict can be found in
- the subscriber's server log.
- </para>
+ <para>
+ A conflict that produces an error will stop the replication; it must be
+ resolved manually by the user. Details about the conflict can be found in
+ the subscriber's server log.
+ </para>
- <para>
- The resolution can be done either by changing data or permissions on the subscriber so
- that it does not conflict with the incoming change or by skipping the
- transaction that conflicts with the existing data. When a conflict produces
- an error, the replication won't proceed, and the logical replication worker will
- emit the following kind of message to the subscriber's server log:
+ <para>
+ The resolution can be done either by changing data or permissions on the subscriber so
+ that it does not conflict with the incoming change or by skipping the
+ transaction that conflicts with the existing data. When a conflict produces
+ an error, the replication won't proceed, and the logical replication worker will
+ emit the following kind of message to the subscriber's server log:
<screen>
ERROR: conflict detected on relation "public.test": conflict=insert_exists
DETAIL: Could not apply remote change: remote row (1, 'remote').
Key already exists in unique index "test_pkey", modified locally in transaction 800 at 2026-01-16 18:15:25.652759+09: key (c)=(1), local row (1, 'local').
CONTEXT: processing remote data for replication origin "pg_16395" during "INSERT" for replication target relation "public.test" in transaction 725 finished at 0/014C0378
</screen>
- The LSN of the transaction that contains the change violating the constraint and
- the replication origin name can be found from the server log (LSN 0/014C0378 and
- replication origin <literal>pg_16395</literal> in the above case). The
- transaction that produced the conflict can be skipped by using
- <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
- with the finish LSN
- (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction
- is committed or prepared on the publisher. Alternatively, the transaction can
- also be skipped by calling the <link linkend="pg-replication-origin-advance">
- <function>pg_replication_origin_advance()</function></link> function.
- Before using this function, the subscription needs to be disabled temporarily
- either by <link linkend="sql-altersubscription-params-disable">
- <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
- subscription can be used with the
- <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
- option. Then, you can use <function>pg_replication_origin_advance()</function>
- function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
- and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of
- origins can be seen in the <link linkend="view-pg-replication-origin-status">
- <structname>pg_replication_origin_status</structname></link> system view.
- Please note that skipping the whole transaction includes skipping changes that
- might not violate any constraint. This can easily make the subscriber
- inconsistent.
- The additional details regarding conflicting rows, such as their origin and
- commit timestamp can be seen in the <literal>DETAIL</literal> line of the
- log. But note that this information is only available when
- <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
- is enabled on the subscriber. Users can use this information to decide
- whether to retain the local change or adopt the remote alteration. For
- instance, the <literal>DETAIL</literal> line in the above log indicates that
- the existing row was modified locally. Users can manually perform a
- remote-change-win.
- </para>
-
- <para>
- When the
- <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
- mode is <literal>parallel</literal>, the finish LSN of failed transactions
- may not be logged. In that case, it may be necessary to change the streaming
- mode to <literal>on</literal> or <literal>off</literal> and cause the same
- conflicts again so the finish LSN of the failed transaction will be written
- to the server log. For the usage of finish LSN, please refer to <link
- linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
- SKIP</command></link>.
- </para>
+ The LSN of the transaction that contains the change violating the constraint and
+ the replication origin name can be found from the server log (LSN 0/014C0378 and
+ replication origin <literal>pg_16395</literal> in the above case). The
+ transaction that produced the conflict can be skipped by using
+ <link linkend="sql-altersubscription-params-skip"><command>ALTER SUBSCRIPTION ... SKIP</command></link>
+ with the finish LSN
+ (i.e., LSN 0/014C0378). The finish LSN could be an LSN at which the transaction
+ is committed or prepared on the publisher. Alternatively, the transaction can
+ also be skipped by calling the <link linkend="pg-replication-origin-advance">
+ <function>pg_replication_origin_advance()</function></link> function.
+ Before using this function, the subscription needs to be disabled temporarily
+ either by <link linkend="sql-altersubscription-params-disable">
+ <command>ALTER SUBSCRIPTION ... DISABLE</command></link> or, the
+ subscription can be used with the
+ <link linkend="sql-createsubscription-params-with-disable-on-error"><literal>disable_on_error</literal></link>
+ option. Then, you can use <function>pg_replication_origin_advance()</function>
+ function with the <parameter>node_name</parameter> (i.e., <literal>pg_16395</literal>)
+ and the next LSN of the finish LSN (i.e., 0/014C0379). The current position of
+ origins can be seen in the <link linkend="view-pg-replication-origin-status">
+ <structname>pg_replication_origin_status</structname></link> system view.
+ Please note that skipping the whole transaction includes skipping changes that
+ might not violate any constraint. This can easily make the subscriber
+ inconsistent.
+ The additional details regarding conflicting rows, such as their origin and
+ commit timestamp can be seen in the <literal>DETAIL</literal> line of the
+ log. But note that this information is only available when
+ <link linkend="guc-track-commit-timestamp"><varname>track_commit_timestamp</varname></link>
+ is enabled on the subscriber. Users can use this information to decide
+ whether to retain the local change or adopt the remote alteration. For
+ instance, the <literal>DETAIL</literal> line in the above log indicates that
+ the existing row was modified locally. Users can manually perform a
+ remote-change-win.
+ </para>
+
+ <para>
+ When the
+ <link linkend="sql-createsubscription-params-with-streaming"><literal>streaming</literal></link>
+ mode is <literal>parallel</literal>, the finish LSN of failed transactions
+ may not be logged. In that case, it may be necessary to change the streaming
+ mode to <literal>on</literal> or <literal>off</literal> and cause the same
+ conflicts again so the finish LSN of the failed transaction will be written
+ to the server log. For the usage of finish LSN, please refer to <link
+ linkend="sql-altersubscription"><command>ALTER SUBSCRIPTION ...
+ SKIP</command></link>.
+ </para>
+ </sect2>
</sect1>
<sect1 id="logical-replication-restrictions">
@@ -2430,6 +2543,14 @@ CONTEXT: processing remote data for replication origin "pg_16395" during "INSER
key or replica identity defined for it.
</para>
</listitem>
+
+ <listitem>
+ <para>
+ Conflict log tables (see <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link> parameter)
+ are never published, even when using <literal>FOR ALL TABLES</literal> in a
+ publication.
+ </para>
+ </listitem>
</itemizedlist>
</sect1>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index fad1f90956a..13b413d142f 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -358,7 +358,8 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
When the <link linkend="sql-createsubscription-params-with-conflict-log-destination"><literal>conflict_log_destination</literal></link>
parameter is set to <literal>table</literal> or <literal>all</literal>, the system
automatically creates the conflict log table. Conversely, if the destination is changed to
- <literal>log</literal>, the conflict log table is automatically dropped.
+ <literal>log</literal>, logging to the table stops and the conflict log
+ table is automatically dropped.
</para>
</listitem>
</varlistentry>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index 81fbf3487a4..0be1aa4698d 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -272,8 +272,10 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<term><literal>log</literal></term>
<listitem>
<para>
- Conflict details are recorded in the server log.
- This is the default behavior.
+ Conflict details are recorded in the server log. This is
+ the default behavior. See
+ <xref linkend="logical-replication-conflict-file-based-logging"/>
+ for details.
</para>
</listitem>
</varlistentry>
@@ -284,14 +286,17 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
The system automatically creates a structured table named
<literal>pg_conflict_log_<subid></literal> in the
<literal>pg_conflict</literal> schema. This allows for easy
- querying and analysis of conflicts.
+ querying and analysis of conflicts. See
+ <xref linkend="logical-replication-conflict-table-based-logging"/>
+ for details.
</para>
<caution>
<para>
The conflict log table is strictly tied to the lifecycle of the
subscription or the <literal>conflict_log_destination</literal> setting. If
the subscription is dropped, or if the destination is changed to
- <literal>log</literal>, the table is <emphasis>dropped</emphasis>.
+ <literal>log</literal>, the table and all its recorded conflict data are
+ <emphasis>permanently deleted</emphasis>.
</para>
</caution>
</listitem>
@@ -300,8 +305,8 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
<term><literal>all</literal></term>
<listitem>
<para>
- This is equivalent to configuring both destinations <literal>log</literal>
- and <literal>table</literal> together.
+ Records conflict details to both destinations <literal>log</literal>
+ and <literal>table</literal>.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 012d55e9d3d..55a0ee4831a 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -986,7 +986,66 @@ ParallelApplyWorkerMain(Datum main_arg)
set_apply_error_context_origin(originname);
- LogicalParallelApplyLoop(mqh);
+ PG_TRY();
+ {
+ LogicalParallelApplyLoop(mqh);
+ }
+ PG_CATCH();
+ {
+ MemoryContext oldcontext;
+ ErrorData *edata;
+
+ /*
+ * Reset the origin state to prevent the advancement of origin
+ * progress if we fail to apply. Otherwise, this will result in
+ * transaction loss as that transaction won't be sent again by the
+ * server.
+ */
+ replorigin_xact_clear(true);
+
+ /*
+ * Copy the error and recover to an idle state so we can insert the
+ * deferred conflict log tuple (if any) before re-throwing. Copy the
+ * error into a longer-lived context first, as it may have been raised
+ * under ErrorContext. Also reset the error context stack: the
+ * callbacks in effect when the error was thrown belong to unwound
+ * stack frames, and the deferred insert installs its own.
+ */
+ oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ edata = CopyErrorData();
+ MemoryContextSwitchTo(oldcontext);
+
+ FlushErrorState();
+ error_context_stack = NULL;
+
+ /*
+ * Tell the leader we failed and are about to report the error and log
+ * the conflict. This must be set before AbortOutOfAnyTransaction()
+ * below releases the transaction lock that the leader waits on in
+ * pa_wait_for_xact_finish(); otherwise the leader would see a
+ * non-finished state, assume the connection was lost, and tear this
+ * worker down while it is still writing the conflict log tuple.
+ */
+ pa_set_xact_state(MyParallelShared, PARALLEL_TRANS_ERROR);
+
+ AbortOutOfAnyTransaction();
+
+ /*
+ * Insert the deferred conflict log tuple before re-throwing.
+ * Re-throwing is what reports the error to the leader (via the error
+ * queue set up above), so the insertion must happen first: otherwise
+ * the leader could start tearing down this worker while it is still
+ * writing the conflict log tuple. If the insertion itself fails,
+ * that error (annotated with the conflict context, see
+ * InsertConflictLogTuple) propagates to the leader instead of the
+ * original.
+ */
+ ProcessPendingConflictLogTuple();
+
+ /* Re-throw the original error, which reports it to the leader. */
+ ReThrowError(edata);
+ }
+ PG_END_TRY();
/*
* The parallel apply worker must not get here because the parallel apply
@@ -1314,9 +1373,33 @@ pa_wait_for_xact_finish(ParallelApplyWorkerInfo *winfo)
* released.
*/
if (pa_get_xact_state(winfo->shared) != PARALLEL_TRANS_FINISHED)
+ {
+ /*
+ * If the worker signalled that it errored (PARALLEL_TRANS_ERROR), it
+ * is logging the conflict and will report the actual error via the
+ * error queue before exiting. Wait for that rather than reporting a
+ * generic lost connection: CHECK_FOR_INTERRUPTS() drives
+ * ProcessParallelApplyMessages(), which raises the real error on the
+ * worker's ErrorResponse (or "lost connection" if the worker died
+ * without reporting). Waiting here also keeps the worker alive long
+ * enough to finish writing the conflict log tuple.
+ */
+ while (pa_get_xact_state(winfo->shared) == PARALLEL_TRANS_ERROR)
+ {
+ CHECK_FOR_INTERRUPTS();
+
+ (void) WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 10L,
+ WAIT_EVENT_LOGICAL_PARALLEL_APPLY_STATE_CHANGE);
+
+ ResetLatch(MyLatch);
+ }
+
ereport(ERROR,
(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
errmsg("lost connection to the logical replication parallel apply worker")));
+ }
}
/*
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 5ea8c455cd4..5551d87d1bc 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -16,17 +16,25 @@
#include "access/commit_ts.h"
#include "access/genam.h"
+#include "access/heapam.h"
#include "access/tableam.h"
+#include "access/xact.h"
#include "catalog/heap.h"
#include "catalog/pg_am.h"
#include "catalog/pg_namespace.h"
#include "catalog/toasting.h"
#include "executor/executor.h"
+#include "funcapi.h"
#include "pgstat.h"
#include "replication/conflict.h"
#include "replication/worker_internal.h"
#include "storage/lmgr.h"
+#include "utils/array.h"
+#include "utils/builtins.h"
+#include "utils/injection_point.h"
#include "utils/lsyscache.h"
+#include "utils/memutils.h"
+#include "utils/pg_lsn.h"
/*
* String representations for the supported conflict logging destinations.
@@ -40,7 +48,6 @@ const char *const ConflictLogDestNames[] = {
StaticAssertDecl(lengthof(ConflictLogDestNames) == CONFLICT_LOG_DEST_ALL + 1,
"ConflictLogDestNames length mismatch");
-
/* Structure to hold metadata for one column of the conflict log table */
typedef struct ConflictLogColumnDef
{
@@ -81,6 +88,18 @@ static const ConflictLogColumnDef ConflictLogSchema[] = {
#define NUM_CONFLICT_ATTRS lengthof(ConflictLogSchema)
+/* Schema for the elements within the 'local_conflicts' JSON array */
+static const ConflictLogColumnDef LocalConflictSchema[] =
+{
+ {.attname = "xid", .atttypid = XIDOID},
+ {.attname = "commit_ts", .atttypid = TIMESTAMPTZOID},
+ {.attname = "origin", .atttypid = TEXTOID},
+ {.attname = "key", .atttypid = JSONOID},
+ {.attname = "tuple", .atttypid = JSONOID}
+};
+
+#define NUM_LOCAL_CONFLICT_ATTRS lengthof(LocalConflictSchema)
+
static const char *const ConflictTypeNames[] = {
[CT_INSERT_EXISTS] = "insert_exists",
[CT_UPDATE_ORIGIN_DIFFERS] = "update_origin_differs",
@@ -92,6 +111,25 @@ static const char *const ConflictTypeNames[] = {
[CT_MULTIPLE_UNIQUE_CONFLICTS] = "multiple_unique_conflicts"
};
+/*
+ * Worker-private state for a conflict that has been prepared for the conflict
+ * log table but not yet inserted. It carries the prepared tuple, and a
+ * description of the conflict used for error context, from
+ * prepare_conflict_log_tuple() across the apply error boundary to
+ * ProcessPendingConflictLogTuple()/InsertConflictLogTuple(). Both pointers
+ * reference memory allocated in ApplyContext.
+ *
+ * This is purely process-local state, so it lives here rather than in the
+ * shared LogicalRepWorker struct.
+ */
+typedef struct PendingConflictLogData
+{
+ HeapTuple tuple; /* prepared, not-yet-inserted conflict tuple */
+ char *errcontext_str; /* conflict description for error context */
+} PendingConflictLogData;
+
+static PendingConflictLogData pending_conflict_log = {0};
+
static int errcode_apply_conflict(ConflictType type);
static void errdetail_apply_conflict(EState *estate,
ResultRelInfo *relinfo,
@@ -108,8 +146,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.
@@ -279,29 +336,219 @@ ReportApplyConflict(EState *estate, ResultRelInfo *relinfo, int elevel,
TupleTableSlot *remoteslot, List *conflicttuples)
{
Relation localrel = relinfo->ri_RelationDesc;
- StringInfoData err_detail;
+ ConflictLogDest dest;
+ Relation conflictlogrel;
+ bool log_dest_table;
+ bool log_dest_logfile;
- initStringInfo(&err_detail);
+ pgstat_report_subscription_conflict(MySubscription->oid, type);
- /* Form errdetail message by combining conflicting tuples information. */
- foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
- errdetail_apply_conflict(estate, relinfo, type, searchslot,
- conflicttuple->slot, remoteslot,
- conflicttuple->indexoid,
- conflicttuple->xmin,
- conflicttuple->origin,
- conflicttuple->ts,
- &err_detail);
+ /*
+ * Get the conflict log destination. Also, if a table is one of the
+ * destinations, return the CLT relation already opened and ready for
+ * insertion -- or NULL if it has been dropped concurrently.
+ */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
- pgstat_report_subscription_conflict(MySubscription->oid, type);
+ log_dest_table = CONFLICTS_LOGGED_TO_TABLE(dest);
+ log_dest_logfile = CONFLICTS_LOGGED_TO_LOG(dest);
+
+ /*
+ * If a conflict log table was requested but it has been dropped
+ * concurrently (e.g. a concurrent ALTER SUBSCRIPTION changed
+ * conflict_log_destination), GetConflictLogDestAndTable() returned NULL.
+ * Fall back to logging the full conflict details to the server log so
+ * that the conflict is not lost.
+ */
+ if (log_dest_table && conflictlogrel == NULL)
+ {
+ log_dest_table = false;
+ log_dest_logfile = true;
+ }
+
+ /*
+ * Prepare the conflict log tuple first when the destination includes the
+ * table. This must happen before the ereport() below, because for an
+ * ERROR-level conflict that ereport() raises the error and defers the
+ * actual insertion to ProcessPendingConflictLogTuple(), which relies on
+ * the tuple having been prepared.
+ */
+ if (log_dest_table)
+ {
+ Assert(conflictlogrel != NULL);
+ prepare_conflict_log_tuple(estate,
+ relinfo->ri_RelationDesc,
+ conflictlogrel,
+ type,
+ searchslot,
+ conflicttuples,
+ remoteslot);
+ }
+
+ /*
+ * Report the conflict to the server log before inserting it into the
+ * conflict log table. Emitting it first guarantees the conflict is
+ * recorded even if the table insert below fails; it is also what raises
+ * the error for ERROR-level conflicts. When the server log is one of the
+ * destinations we emit the full details, otherwise (table-only) we emit a
+ * shorter message since the details are captured in the table.
+ */
+ if (log_dest_logfile)
+ {
+ StringInfoData err_detail;
+
+ initStringInfo(&err_detail);
+
+ /* Form errdetail message by combining conflicting tuples information. */
+ foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+ errdetail_apply_conflict(estate, relinfo, type, searchslot,
+ conflicttuple->slot, remoteslot,
+ conflicttuple->indexoid,
+ conflicttuple->xmin,
+ conflicttuple->origin,
+ conflicttuple->ts,
+ &err_detail);
+
+ /* Standard reporting with full internal details. */
+ ereport(elevel,
+ errcode_apply_conflict(type),
+ errmsg("conflict detected on relation \"%s\": conflict=%s",
+ RelationGetQualifiedRelationName(localrel),
+ ConflictTypeNames[type]),
+ errdetail_internal("%s", err_detail.data));
+ }
+ else if (log_dest_table)
+ {
+ /*
+ * Not logging conflict details to the server log; report the conflict
+ * but omit raw tuple data since it is captured in the conflict log
+ * table.
+ */
+ ereport(elevel,
+ errcode_apply_conflict(type),
+ errmsg("conflict detected on relation \"%s\": conflict=%s",
+ RelationGetQualifiedRelationName(localrel),
+ ConflictTypeNames[type]),
+ errdetail("Conflict details are logged to the conflict log table: %s",
+ RelationGetRelationName(conflictlogrel)));
+ }
+
+ /*
+ * Insert into the conflict log table if requested. For conflicts below
+ * ERROR the apply transaction continues, so insert immediately; for
+ * ERROR-level conflicts the ereport() above already raised the error and
+ * the insertion is deferred to a new transaction
+ * (ProcessPendingConflictLogTuple) so that it is not rolled back.
+ */
+ if (log_dest_table)
+ {
+ if (elevel < ERROR)
+ {
+ PG_TRY();
+ {
+ InsertConflictLogTuple(conflictlogrel);
+ }
+ PG_CATCH();
+ {
+ /*
+ * The insert failed, so the apply transaction will abort and
+ * the error will propagate to the worker's error handler. The
+ * conflict was already reported to the server log above, so
+ * it is not lost. Discard the prepared tuple so that the
+ * deferred insertion path (ProcessPendingConflictLogTuple)
+ * does not retry this same failing insert.
+ */
+ if (pending_conflict_log.tuple != NULL)
+ {
+ heap_freetuple(pending_conflict_log.tuple);
+ pending_conflict_log.tuple = NULL;
+ }
+ if (pending_conflict_log.errcontext_str != NULL)
+ {
+ pfree(pending_conflict_log.errcontext_str);
+ pending_conflict_log.errcontext_str = NULL;
+ }
+ PG_RE_THROW();
+ }
+ PG_END_TRY();
+ }
+
+ table_close(conflictlogrel, RowExclusiveLock);
+ }
+}
+
+/*
+ * 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 (pending_conflict_log.tuple == NULL)
+ return;
+
+ /*
+ * Insert the deferred conflict log tuple in its own transaction. A
+ * failure here (e.g. an out-of-disk-space error) is treated like any
+ * other apply error and raises an ERROR; such failures are expected to be
+ * rare and persistent. Callers must therefore have already reported (and
+ * cleared) any in-progress apply error before calling this, so that this
+ * error does not mask the original one.
+ */
+ StartTransactionCommand();
+ PushActiveSnapshot(GetTransactionSnapshot());
+
+ /*
+ * Test hook: pause here so a TAP test can take a conflicting lock on the
+ * conflict log table before this transaction tries to open it. See
+ * src/test/subscription/t/039_pa_conflict_log_lock_wait.pl.
+ */
+ INJECTION_POINT("clt-pending-flush-before-open", NULL);
+
+ /* Open the conflict log table and insert the tuple. */
+ conflictlogrel = GetConflictLogDestAndTable(&dest);
- 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));
+ if (conflictlogrel != NULL)
+ {
+ InsertConflictLogTuple(conflictlogrel);
+ table_close(conflictlogrel, RowExclusiveLock);
+ }
+ else
+ {
+ /*
+ * The conflict log table was dropped concurrently (e.g. by an ALTER
+ * SUBSCRIPTION that changed conflict_log_destination) after the
+ * conflict was already reported to the server log by
+ * ReportApplyConflict(). Nothing more to do; just discard the
+ * prepared tuple and its context string.
+ */
+ heap_freetuple(pending_conflict_log.tuple);
+ pending_conflict_log.tuple = NULL;
+ if (pending_conflict_log.errcontext_str)
+ {
+ pfree(pending_conflict_log.errcontext_str);
+ pending_conflict_log.errcontext_str = NULL;
+ }
+ }
+
+ PopActiveSnapshot();
+ CommitTransactionCommand();
}
/*
@@ -335,6 +582,98 @@ 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 a table is one of the
+ * destinations, opens and returns the relation handle for the conflict log
+ * table.
+ *
+ * The table is opened with try_table_open(), so NULL is returned if the
+ * conflict log table has been dropped concurrently (e.g. by an ALTER
+ * SUBSCRIPTION that changed conflict_log_destination). Callers must treat a
+ * NULL result for a table destination as "table unavailable" and fall back to
+ * server-log reporting rather than failing.
+ */
+Relation
+GetConflictLogDestAndTable(ConflictLogDest *log_dest)
+{
+ Oid conflictlogrelid;
+
+ /*
+ * Convert the text log destination to the internal enum. MySubscription
+ * already contains the data from pg_subscription.
+ */
+ *log_dest = GetConflictLogDest(MySubscription->conflictlogdest);
+
+ /* Quick exit if a conflict log table was not requested. */
+ if (!CONFLICTS_LOGGED_TO_TABLE(*log_dest))
+ return NULL;
+
+ conflictlogrelid = MySubscription->conflictlogrelid;
+
+ Assert(OidIsValid(conflictlogrelid));
+
+ /*
+ * Use try_table_open(): the table may have been dropped concurrently by
+ * an ALTER SUBSCRIPTION that changed conflict_log_destination. Returning
+ * NULL lets the caller fall back to the server log instead of failing.
+ */
+ return try_table_open(conflictlogrelid, RowExclusiveLock);
+}
+
+/*
+ * Error context callback for failures while inserting into the conflict log
+ * table. Adds a line identifying the conflict that was being logged.
+ */
+static void
+conflict_log_insert_errcontext(void *arg)
+{
+ char *ctx = (char *) arg;
+
+ if (ctx)
+ errcontext("%s", ctx);
+}
+
+/*
+ * InsertConflictLogTuple
+ *
+ * Insert conflict log tuple into the conflict log table.
+ */
+void
+InsertConflictLogTuple(Relation conflictlogrel)
+{
+ ErrorContextCallback errcallback;
+
+ /* A valid tuple must be prepared and stored in pending_conflict_log. */
+ Assert(pending_conflict_log.tuple != NULL);
+
+ /*
+ * Set up an error context so that a failure to insert (e.g. the conflict
+ * log table was dropped, or an out-of-space error) carries information
+ * identifying the conflict we were trying to log.
+ */
+ errcallback.callback = conflict_log_insert_errcontext;
+ errcallback.arg = pending_conflict_log.errcontext_str;
+ errcallback.previous = error_context_stack;
+ error_context_stack = &errcallback;
+
+ heap_insert(conflictlogrel, pending_conflict_log.tuple,
+ GetCurrentCommandId(true), 0, NULL);
+
+ error_context_stack = errcallback.previous;
+
+ /* Free the conflict log tuple and its context string. */
+ heap_freetuple(pending_conflict_log.tuple);
+ pending_conflict_log.tuple = NULL;
+ if (pending_conflict_log.errcontext_str)
+ {
+ pfree(pending_conflict_log.errcontext_str);
+ pending_conflict_log.errcontext_str = NULL;
+ }
+}
+
/*
* Add SQLSTATE error code to the current conflict report.
*/
@@ -768,6 +1107,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.
@@ -783,41 +1156,359 @@ build_index_value_desc(EState *estate, Relation localrel, TupleTableSlot *slot,
Relation indexDesc;
Datum values[INDEX_MAX_KEYS];
bool isnull[INDEX_MAX_KEYS];
- TupleTableSlot *tableslot = slot;
- if (!tableslot)
+ if (!slot)
return NULL;
Assert(CheckRelationOidLockedByMe(indexoid, RowExclusiveLock, true));
indexDesc = index_open(indexoid, NoLock);
- /*
- * If the slot is a virtual slot, copy it into a heap tuple slot as
- * FormIndexDatum only works with heap tuple slots.
- */
- if (TTS_IS_VIRTUAL(slot))
+ build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+ isnull);
+
+ index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+
+ index_close(indexDesc, NoLock);
+
+ return index_value;
+}
+
+/*
+ * tuple_table_slot_to_json_datum
+ *
+ * Helper function to convert a TupleTableSlot to JSON.
+ */
+static Datum
+tuple_table_slot_to_json_datum(TupleTableSlot *slot)
+{
+ HeapTuple tuple;
+ Datum datum;
+ Datum json;
+
+ Assert(slot != NULL);
+
+ tuple = ExecCopySlotHeapTuple(slot);
+ datum = heap_copy_tuple_as_datum(tuple, slot->tts_tupleDescriptor);
+
+ json = DirectFunctionCall1(row_to_json, datum);
+ heap_freetuple(tuple);
+
+ return json;
+}
+
+/*
+ * tuple_table_slot_to_indextup_json
+ *
+ * Fetch replica identity key from the tuple table slot and convert into a
+ * JSON datum.
+ */
+static Datum
+tuple_table_slot_to_indextup_json(EState *estate, Relation localrel,
+ Oid indexid, TupleTableSlot *slot)
+{
+ Relation indexDesc;
+ Datum values[INDEX_MAX_KEYS];
+ bool isnull[INDEX_MAX_KEYS];
+ HeapTuple tuple;
+ TupleDesc tupdesc;
+ Datum datum;
+
+ Assert(slot != NULL);
+
+ Assert(CheckRelationOidLockedByMe(indexid, RowExclusiveLock, true));
+
+ indexDesc = index_open(indexid, NoLock);
+
+ build_index_datums_from_slot(estate, localrel, slot, indexDesc, values,
+ isnull);
+ tupdesc = CreateTupleDescCopy(RelationGetDescr(indexDesc));
+
+ /* Bless the tupdesc so it can be looked up by row_to_json. */
+ BlessTupleDesc(tupdesc);
+
+ /* Form the replica identity tuple. */
+ tuple = heap_form_tuple(tupdesc, values, isnull);
+ datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+ heap_freetuple(tuple);
+ FreeTupleDesc(tupdesc);
+ index_close(indexDesc, NoLock);
+
+ /* Convert to a JSON datum. */
+ return DirectFunctionCall1(row_to_json, datum);
+}
+
+/*
+ * build_conflict_tupledesc
+ *
+ * Build and bless a tuple descriptor for the conflict log table based on the
+ * predefined LocalConflictSchema.
+ */
+static TupleDesc
+build_conflict_tupledesc(void)
+{
+ static TupleDesc cached_tupdesc = NULL;
+
+ if (cached_tupdesc == NULL)
{
- tableslot = table_slot_create(localrel, &estate->es_tupleTable);
- tableslot = ExecCopySlot(tableslot, slot);
+ MemoryContext oldcxt;
+
+ oldcxt = MemoryContextSwitchTo(CacheMemoryContext);
+
+ cached_tupdesc = CreateTemplateTupleDesc(NUM_LOCAL_CONFLICT_ATTRS);
+
+ for (int i = 0; i < NUM_LOCAL_CONFLICT_ATTRS; i++)
+ TupleDescInitEntry(cached_tupdesc,
+ (AttrNumber) (i + 1),
+ LocalConflictSchema[i].attname,
+ LocalConflictSchema[i].atttypid,
+ -1, 0);
+
+ TupleDescFinalize(cached_tupdesc);
+
+ /*
+ * Bless once so it can be used as a RECORD type (e.g. for row_to_json
+ * or other record-based operations).
+ */
+ BlessTupleDesc(cached_tupdesc);
+
+ MemoryContextSwitchTo(oldcxt);
}
- /*
- * Initialize ecxt_scantuple for potential use in FormIndexDatum when
- * index expressions are present.
- */
- GetPerTupleExprContext(estate)->ecxt_scantuple = tableslot;
+ return cached_tupdesc;
+}
- /*
- * The values/nulls arrays passed to BuildIndexValueDescription should be
- * the results of FormIndexDatum, which are the "raw" input to the index
- * AM.
- */
- FormIndexDatum(BuildIndexInfo(indexDesc), tableslot, estate, values, isnull);
+/*
+ * Builds the local conflicts JSON array column from the list of
+ * ConflictTupleInfo objects.
+ *
+ * Example output structure:
+ * [ { "xid": "1001", "commit_ts": "...", "origin": "...", "tuple": {...} }, ... ]
+ */
+static Datum
+build_local_conflicts_json_array(EState *estate, Relation rel,
+ ConflictType conflict_type,
+ List *conflicttuples)
+{
+ ListCell *lc;
+ List *json_datums = NIL;
+ Datum *json_datum_array;
+ Datum json_array_datum;
+ int num_conflicts;
+ int i;
+ int16 typlen;
+ bool typbyval;
+ char typalign;
+ TupleDesc tupdesc;
- index_value = BuildIndexValueDescription(indexDesc, values, isnull);
+ /* Build local conflicts tuple descriptor. */
+ tupdesc = build_conflict_tupledesc();
- index_close(indexDesc, NoLock);
+ /* Process local conflict tuple list and prepare an array of JSON. */
+ foreach_ptr(ConflictTupleInfo, conflicttuple, conflicttuples)
+ {
+ Datum values[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_LOCAL_CONFLICT_ATTRS] = {0};
+ char *origin_name = NULL;
+ HeapTuple tuple;
+ Datum json_datum;
+ int attno;
+
+ attno = 0;
+ values[attno++] = TransactionIdGetDatum(conflicttuple->xmin);
+
+ if (conflicttuple->ts)
+ values[attno++] = TimestampTzGetDatum(conflicttuple->ts);
+ else
+ nulls[attno++] = true;
- return index_value;
+ if (conflicttuple->origin != InvalidReplOriginId)
+ replorigin_by_oid(conflicttuple->origin, true, &origin_name);
+
+ /* Store empty string if origin name for the tuple is NULL. */
+ if (origin_name != NULL)
+ values[attno++] = CStringGetTextDatum(origin_name);
+ else
+ nulls[attno++] = true;
+
+ /*
+ * Add the conflicting key values in the case of a unique constraint
+ * violation.
+ */
+ if (conflict_type == CT_INSERT_EXISTS ||
+ conflict_type == CT_UPDATE_EXISTS ||
+ conflict_type == CT_MULTIPLE_UNIQUE_CONFLICTS)
+ {
+ Oid indexoid = conflicttuple->indexoid;
+
+ Assert(OidIsValid(indexoid) && conflicttuple->slot &&
+ CheckRelationOidLockedByMe(indexoid, RowExclusiveLock,
+ true));
+ values[attno++] =
+ tuple_table_slot_to_indextup_json(estate, rel,
+ indexoid,
+ conflicttuple->slot);
+ }
+ else
+ nulls[attno++] = true;
+
+ /* Convert conflicting tuple to JSON datum. */
+ if (conflicttuple->slot)
+ values[attno] = tuple_table_slot_to_json_datum(conflicttuple->slot);
+ else
+ nulls[attno] = true;
+
+ Assert(attno + 1 == NUM_LOCAL_CONFLICT_ATTRS);
+
+ tuple = heap_form_tuple(tupdesc, values, nulls);
+
+ json_datum = heap_copy_tuple_as_datum(tuple, tupdesc);
+
+ /*
+ * Build the higher level JSON datum in format described in function
+ * header.
+ */
+ json_datum = DirectFunctionCall1(row_to_json, json_datum);
+
+ /* Done with the temporary tuple. */
+ heap_freetuple(tuple);
+
+ /* Add to the array element. */
+ json_datums = lappend(json_datums, (void *) json_datum);
+ }
+
+ num_conflicts = list_length(json_datums);
+
+ json_datum_array = palloc_array(Datum, num_conflicts);
+
+ i = 0;
+ foreach(lc, json_datums)
+ {
+ json_datum_array[i] = (Datum) lfirst(lc);
+ i++;
+ }
+
+ /* Construct the JSON array Datum. */
+ get_typlenbyvalalign(JSONOID, &typlen, &typbyval, &typalign);
+ json_array_datum = PointerGetDatum(construct_array(json_datum_array,
+ num_conflicts,
+ JSONOID,
+ typlen,
+ typbyval,
+ typalign));
+ pfree(json_datum_array);
+
+ return json_array_datum;
+}
+
+/*
+ * prepare_conflict_log_tuple
+ *
+ * This routine prepares a tuple detailing a conflict encountered during
+ * logical replication. The prepared tuple will be stored in
+ * pending_conflict_log.tuple which should be inserted into the
+ * conflict log table by calling InsertConflictLogTuple.
+ */
+static void
+prepare_conflict_log_tuple(EState *estate, Relation rel,
+ Relation conflictlogrel,
+ ConflictType conflict_type,
+ TupleTableSlot *searchslot,
+ List *conflicttuples,
+ TupleTableSlot *remoteslot)
+{
+ Datum values[NUM_CONFLICT_ATTRS] = {0};
+ bool nulls[NUM_CONFLICT_ATTRS] = {0};
+ int attno;
+ char *remote_origin = NULL;
+ MemoryContext oldctx;
+
+ Assert(pending_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++] = BoolGetDatum(false);
+ values[attno++] = tuple_table_slot_to_indextup_json(estate, rel,
+ replica_index,
+ searchslot);
+ }
+ else
+ {
+ values[attno++] = BoolGetDatum(true);
+ values[attno++] = tuple_table_slot_to_json_datum(searchslot);
+ }
+ }
+ else
+ {
+ nulls[attno++] = true;
+ 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 == NUM_CONFLICT_ATTRS);
+
+ oldctx = MemoryContextSwitchTo(ApplyContext);
+ pending_conflict_log.tuple =
+ heap_form_tuple(RelationGetDescr(conflictlogrel), values, nulls);
+
+ /*
+ * Stash a context string describing this conflict, so that if inserting
+ * the tuple into the conflict log table fails, the resulting error
+ * carries enough context to identify the conflict (see
+ * InsertConflictLogTuple).
+ */
+ pending_conflict_log.errcontext_str =
+ psprintf("while logging conflict \"%s\" detected on relation \"%s\"",
+ ConflictTypeNames[conflict_type],
+ RelationGetRelationName(rel));
+ MemoryContextSwitchTo(oldctx);
}
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 7799266c614..042c7203619 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);
@@ -1296,6 +1300,8 @@ apply_handle_begin_prepare(StringInfo s)
set_apply_error_context_xact(begin_data.xid, begin_data.prepare_lsn);
remote_final_lsn = begin_data.prepare_lsn;
+ remote_xid = begin_data.xid;
+ remote_commit_ts = 0;
maybe_start_skipping_changes(begin_data.prepare_lsn);
@@ -1767,6 +1773,10 @@ apply_handle_stream_start(StringInfo s)
(errcode(ERRCODE_PROTOCOL_VIOLATION),
errmsg_internal("invalid transaction ID in streamed replication transaction")));
+ remote_xid = stream_xid;
+ remote_final_lsn = InvalidXLogRecPtr;
+ remote_commit_ts = 0;
+
set_apply_error_context_xact(stream_xid, InvalidXLogRecPtr);
/* Try to allocate a worker for the streaming transaction. */
@@ -2428,6 +2438,9 @@ apply_handle_stream_commit(StringInfo s)
{
case TRANS_LEADER_APPLY:
+ /* Set remote_commit_ts for conflict logging. */
+ remote_commit_ts = commit_data.committime;
+
/*
* The transaction has been serialized to file, so replay all the
* spooled operations.
@@ -5647,6 +5660,9 @@ start_apply(XLogRecPtr origin_startpos)
}
PG_CATCH();
{
+ MemoryContext oldcontext;
+ ErrorData *edata;
+
/*
* Reset the origin state to prevent the advancement of origin
* progress if we fail to apply. Otherwise, this will result in
@@ -5660,14 +5676,34 @@ start_apply(XLogRecPtr origin_startpos)
else
{
/*
- * Report the worker failed while applying changes. Abort the
- * current transaction so that the stats message is sent in an
- * idle state.
+ * Save the error and recover to an idle state so we can insert
+ * the deferred conflict log tuple (if any) before re-throwing.
+ * Copy the error into a long-lived context first, as it may have
+ * been raised under ErrorContext. Also reset the error context
+ * stack: the callbacks in effect when the error was thrown belong
+ * to unwound stack frames, and the deferred insert installs its
+ * own.
*/
+ oldcontext = MemoryContextSwitchTo(TopMemoryContext);
+ edata = CopyErrorData();
+ MemoryContextSwitchTo(oldcontext);
+
+ FlushErrorState();
+ error_context_stack = NULL;
AbortOutOfAnyTransaction();
pgstat_report_subscription_error(MySubscription->oid);
- PG_RE_THROW();
+ /*
+ * Insert the deferred conflict log tuple in its own transaction.
+ * If this fails, that error (annotated with the conflict context,
+ * see InsertConflictLogTuple) propagates instead of the original;
+ * such failures are expected to be rare and persistent (e.g. out
+ * of disk space).
+ */
+ ProcessPendingConflictLogTuple();
+
+ /* Re-throw the original error. */
+ ReThrowError(edata);
}
}
PG_END_TRY();
@@ -6034,6 +6070,14 @@ DisableSubscriptionAndExit(void)
RESUME_INTERRUPTS();
+ /*
+ * The error context callbacks in effect when the error was thrown belong
+ * to now-unwound stack frames; reset the stack before running further
+ * code (including the deferred conflict log insertion, which installs its
+ * own).
+ */
+ error_context_stack = NULL;
+
/*
* Report the worker failed during sequence synchronization, table
* synchronization, or apply.
@@ -6062,6 +6106,19 @@ DisableSubscriptionAndExit(void)
errmsg("subscription \"%s\" has been disabled because of an error",
MySubscription->name));
+ /*
+ * Insert the deferred conflict log tuple (if any) now that the
+ * subscription has been disabled and committed. Doing it after the
+ * disable means a failure to log the conflict (treated as a hard error,
+ * see ProcessPendingConflictLogTuple) cannot prevent the subscription
+ * from being disabled and so cannot leave the worker restarting and
+ * failing forever. Do it before the dead-tuple retention check below:
+ * that check only warns today, but it takes an elevel and could raise an
+ * error, which must not prevent the conflict from being recorded. The
+ * original error was already reported above.
+ */
+ ProcessPendingConflictLogTuple();
+
/*
* Skip the track_commit_timestamp check when disabling the worker due to
* an error, as verifying commit timestamps is unnecessary in this
diff --git a/src/include/replication/conflict.h b/src/include/replication/conflict.h
index 326570ed639..4db3a36d4df 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 "access/genam.h"
#include "datatype/timestamp.h"
#include "nodes/pg_list.h"
@@ -114,5 +115,8 @@ extern void ReportApplyConflict(EState *estate, ResultRelInfo *relinfo,
TupleTableSlot *searchslot,
TupleTableSlot *remoteslot,
List *conflicttuples);
+extern void ProcessPendingConflictLogTuple(void);
extern void InitConflictIndexes(ResultRelInfo *relInfo);
+extern Relation GetConflictLogDestAndTable(ConflictLogDest *log_dest);
+extern void InsertConflictLogTuple(Relation conflictlogrel);
#endif
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 745b7d9e969..394f4c6265e 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -121,6 +121,9 @@ typedef enum ParallelTransState
PARALLEL_TRANS_UNKNOWN,
PARALLEL_TRANS_STARTED,
PARALLEL_TRANS_FINISHED,
+ PARALLEL_TRANS_ERROR, /* worker failed; it will report the error
+ * (and log the conflict, if any) before
+ * exiting */
} ParallelTransState;
/*
@@ -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/include/utils/rel.h b/src/include/utils/rel.h
index fa07ebf8ff7..55279edfda6 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -713,16 +713,19 @@ RelationCloseSmgr(Relation relation)
*
* We don't log information for unlogged tables (since they don't WAL log
* anyway), for foreign tables (since they don't WAL log, either),
- * and for system tables (their content is hard to make sense of, and
- * it would complicate decoding slightly for little gain). Note that we *do*
- * log information for user defined catalog tables since they presumably are
- * interesting to the user...
+ * for system tables (their content is hard to make sense of, and
+ * it would complicate decoding slightly for little gain), and for conflict
+ * log tables in the pg_conflict namespace (which are system-managed tables
+ * used internally on subscribers). Note that we *do* log information for
+ * user defined catalog tables since they presumably are interesting to the
+ * user...
*/
#define RelationIsLogicallyLogged(relation) \
(XLogLogicalInfoActive() && \
RelationNeedsWAL(relation) && \
(relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \
- !IsCatalogRelation(relation))
+ !IsCatalogRelation(relation) && \
+ !IsConflictLogTableNamespace(RelationGetNamespace(relation)))
/* routines in utils/cache/relcache.c */
extern void RelationIncrementReferenceCount(Relation rel);
diff --git a/src/test/subscription/meson.build b/src/test/subscription/meson.build
index e71e95c6297..225f90a37b3 100644
--- a/src/test/subscription/meson.build
+++ b/src/test/subscription/meson.build
@@ -48,6 +48,7 @@ tests += {
't/036_sequences.pl',
't/037_except.pl',
't/038_walsnd_shutdown_timeout.pl',
+ 't/039_pa_conflict_log_lock_wait.pl',
't/100_bugs.pl',
],
},
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 3910a49c0aa..6a7d094712d 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -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
@@ -674,10 +719,6 @@ ok( $node_A->poll_query_until(
# A conflict log table is system-managed and cannot be altered directly, so
# moving it to another tablespace must be rejected.
###############################################################################
-my $subid = $node_subscriber->safe_psql('postgres',
- "SELECT oid FROM pg_subscription WHERE subname = 'sub_tab';");
-my $clt = "pg_conflict.pg_conflict_log_$subid";
-
(undef, undef, $stderr) = $node_subscriber->psql('postgres',
"ALTER TABLE $clt SET TABLESPACE pg_default");
like(
diff --git a/src/test/subscription/t/039_pa_conflict_log_lock_wait.pl b/src/test/subscription/t/039_pa_conflict_log_lock_wait.pl
new file mode 100644
index 00000000000..f80047dcb50
--- /dev/null
+++ b/src/test/subscription/t/039_pa_conflict_log_lock_wait.pl
@@ -0,0 +1,203 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+# Test that a parallel apply (PA) worker correctly inserts a deferred
+# conflict-log tuple even when, by the time it reaches
+# ProcessPendingConflictLogTuple(), the conflict log table is held under
+# ACCESS EXCLUSIVE by another session.
+#
+# The window we want to exercise is narrow: PA must already be past
+# ReportApplyConflict() (so the error has fired and PA is in PG_CATCH),
+# and the locker must take the CLT lock *before* PA reaches the second
+# CLT open inside ProcessPendingConflictLogTuple(). An injection point
+# pauses PA at exactly that point so the locker can grab the lock first;
+# without it, the locker either takes the lock too early (PA blocks in
+# the inline CLT open inside ReportApplyConflict, before the error has
+# fired) or too late (PA inserts before the lock is taken).
+
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+ plan skip_all => 'Injection points not supported by this build';
+}
+
+# ---------------------------------------------------------------------
+# Set up publisher and subscriber. Force every transaction to stream so
+# the conflict is handled by a parallel apply worker rather than the
+# leader.
+# ---------------------------------------------------------------------
+my $node_pub = PostgreSQL::Test::Cluster->new('publisher');
+$node_pub->init(allows_streaming => 'logical');
+$node_pub->append_conf('postgresql.conf', q{
+debug_logical_replication_streaming = immediate
+logical_decoding_work_mem = 64kB
+});
+$node_pub->start;
+
+my $node_sub = PostgreSQL::Test::Cluster->new('subscriber');
+$node_sub->init;
+$node_sub->append_conf('postgresql.conf', q{
+shared_preload_libraries = 'injection_points'
+max_logical_replication_workers = 4
+max_parallel_apply_workers_per_subscription = 2
+});
+$node_sub->start;
+
+# Replicated table; the pre-existing row on the subscriber is what makes
+# the publisher's INSERT (id=1) trigger an INSERT_EXISTS conflict.
+$node_pub->safe_psql('postgres', q{
+ CREATE TABLE t (id int PRIMARY KEY, val text);
+ ALTER TABLE t REPLICA IDENTITY FULL;
+ CREATE PUBLICATION p FOR TABLE t;
+});
+
+$node_sub->safe_psql('postgres', q{
+ CREATE TABLE t (id int PRIMARY KEY, val text);
+ INSERT INTO t VALUES (1, 'pre-existing');
+ CREATE EXTENSION injection_points;
+});
+
+my $pub_connstr = $node_pub->connstr . ' dbname=postgres';
+$node_sub->safe_psql('postgres', qq{
+ CREATE SUBSCRIPTION s
+ CONNECTION '$pub_connstr'
+ PUBLICATION p
+ WITH (streaming = parallel,
+ conflict_log_destination = 'all',
+ disable_on_error = true);
+});
+
+$node_sub->wait_for_subscription_sync($node_pub, 's');
+
+# ---------------------------------------------------------------------
+# Send a non-conflicting INSERT and then wait until pg_subscription_rel
+# reaches 'r' (ready) on every relation. pa_can_start() requires
+# AllTablesyncsReady(), which returns true only when every
+# pg_subscription_rel row is 'r'. The 's' (syncdone) -> 'r' transition
+# fires inside ProcessSyncingTablesForApply, which only flips the state
+# when the apply worker's last_received LSN has advanced past the
+# tablesync end LSN -- so we need a triggering commit on the publisher
+# to drive last_received forward. Without this step, the conflict txn
+# below would arrive while the table is still 's', pa_can_start() would
+# return false, the leader would spool to file and apply serially, and
+# no parallel apply worker would ever spawn.
+# ---------------------------------------------------------------------
+$node_pub->safe_psql('postgres', "INSERT INTO t VALUES (1000, 'warmup');");
+$node_sub->poll_query_until('postgres',
+ "SELECT count(1) = 0 FROM pg_subscription_rel WHERE srsubstate NOT IN ('r');"
+) or die "subscription tables did not reach READY state";
+
+# ---------------------------------------------------------------------
+# Look up the per-subscription CLT name.
+# ---------------------------------------------------------------------
+my $sub_oid = $node_sub->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription WHERE subname = 's'");
+my $clt = "pg_conflict.pg_conflict_log_$sub_oid";
+note "conflict log table for subscription s: $clt";
+
+# ---------------------------------------------------------------------
+# Arm the injection point. This pauses the PA worker inside
+# ProcessPendingConflictLogTuple() — i.e. *after* the error has fired
+# and the PG_CATCH has run, *before* the second open of the CLT. This
+# is the exact window the deferred-insert path needs to be tested in.
+# ---------------------------------------------------------------------
+$node_sub->safe_psql('postgres',
+ "SELECT injection_points_attach('clt-pending-flush-before-open', 'wait');");
+
+# ---------------------------------------------------------------------
+# Drive the conflict. PA receives the streamed txn, hits INSERT_EXISTS
+# inside ReportApplyConflict (which opens/closes the CLT cleanly while
+# preparing the deferred tuple), then ereport(ERROR) fires, PG_CATCH
+# runs, and PA enters ProcessPendingConflictLogTuple — where it pauses
+# at the injection point.
+# ---------------------------------------------------------------------
+my $log_offset = -s $node_sub->logfile;
+
+$node_pub->safe_psql('postgres', q{
+ BEGIN;
+ INSERT INTO t SELECT g, repeat('x', 1000) FROM generate_series(2, 200) g;
+ INSERT INTO t VALUES (1, 'conflict');
+ COMMIT;
+});
+
+# Wait until PA is parked at the injection point.
+$node_sub->wait_for_event('logical replication parallel worker',
+ 'clt-pending-flush-before-open');
+
+# ---------------------------------------------------------------------
+# Now take ACCESS EXCLUSIVE on the CLT. TRUNCATE is permitted on CLTs;
+# At this point the CLT is empty, so the TRUNCATE is effectively a no-op
+# that just acquires the lock.
+# Because PA is paused at the injection point, this lock is guaranteed
+# to be acquired *before* PA tries to open the CLT.
+# ---------------------------------------------------------------------
+my $locker = $node_sub->background_psql('postgres');
+$locker->query_safe(qq{
+ BEGIN;
+ TRUNCATE $clt;
+});
+
+# ---------------------------------------------------------------------
+# Wake the PA from the injection point. It will now try to open the
+# CLT inside ProcessPendingConflictLogTuple and block on the lock the
+# locker session holds.
+# ---------------------------------------------------------------------
+$node_sub->safe_psql('postgres',
+ "SELECT injection_points_wakeup('clt-pending-flush-before-open');
+ SELECT injection_points_detach('clt-pending-flush-before-open');");
+
+# Confirm the PA worker is actually parked waiting on the CLT lock —
+# this verifies we are exercising the deferred-insert lock-wait path,
+# not racing past it.
+my $clt_oid = $node_sub->safe_psql('postgres',
+ "SELECT '$clt'::regclass::oid");
+ok( $node_sub->poll_query_until(
+ 'postgres', qq{
+ SELECT EXISTS (
+ SELECT 1
+ FROM pg_locks l
+ JOIN pg_stat_activity a ON l.pid = a.pid
+ WHERE NOT l.granted
+ AND l.relation = $clt_oid
+ AND a.backend_type = 'logical replication parallel worker'
+ );
+ }, 't'),
+ 'PA worker is blocked on the CLT lock inside ProcessPendingConflictLogTuple');
+
+# ---------------------------------------------------------------------
+# Release the lock. PA wakes, inserts the deferred row, commits its
+# CLT txn, re-throws the original error to the leader, and the leader
+# disables the subscription (disable_on_error = true).
+# ---------------------------------------------------------------------
+$locker->query_safe('COMMIT;');
+ok($locker->quit, 'locker session closed cleanly');
+
+ok( $node_sub->poll_query_until(
+ 'postgres',
+ "SELECT subenabled = false FROM pg_subscription WHERE subname = 's'",
+ 't'),
+ 'subscription disabled after the conflict');
+
+# ---------------------------------------------------------------------
+# Verify the deferred conflict log tuple survived the lock wait.
+# ---------------------------------------------------------------------
+my $rows = $node_sub->safe_psql('postgres',
+ "SELECT count(*) FROM $clt WHERE conflict_type = 'insert_exists'");
+is($rows, '1',
+ 'deferred CLT insert by PA worker succeeded after lock release');
+
+# ---------------------------------------------------------------------
+# Also verify the conflict was reported in the server log
+# (conflict_log_destination = 'all' logs to both the table and the log).
+# ---------------------------------------------------------------------
+my $log_contents = slurp_file($node_sub->logfile, $log_offset);
+like(
+ $log_contents,
+ qr/ERROR:\s+conflict detected on relation "public\.t": conflict=insert_exists/,
+ 'conflict reported in server log');
+
+done_testing();
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 117e7379f10..fa55410860d 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -2249,6 +2249,7 @@ PatternInfo
PatternInfoArray
Pattern_Prefix_Status
Pattern_Type
+PendingConflictLogData
PendingFsyncEntry
PendingListenAction
PendingListenEntry
--
2.49.0
[application/octet-stream] v63-0003-Add-conflict-log-table-information-to-describe.patch (78.7K, ../CAFiTN-s45_nOdvE8mBquZO=vyKbVY1ww_q3pejUz7DC29NcSxQ@mail.gmail.com/3-v63-0003-Add-conflict-log-table-information-to-describe.patch)
download | inline diff:
From bd259cdc9e727427ee9ec497c5790772d04a476e Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Thu, 2 Jul 2026 13:21:53 +0530
Subject: [PATCH v63 3/3] Add conflict log table information to describe
subscription output
Display the associated conflict log table as a footer in \dRs+
output when conflict logging to table/all is enabled for a
subscription.
Previously, subscriptions were displayed using a single tabular
output format. Since the conflict log table information is specific
to each subscription and is better suited as auxiliary information,
change the output to display each subscription individually in a
row-wise table format and show the conflict log table as a footer
when applicable.
TODO: Ensure all 190000 series references are updated to 200000
when the version number is bumped.
This approach was chosen based on suggestions at:
https://www.postgresql.org/message-id/CAA4eK1KdKqKkaTqcj3in6ehD_hg6oOaCF_-JsVfd8N6nS8oV9g%40mail.gmail.com
---
src/bin/psql/command.c | 5 +-
src/bin/psql/describe.c | 410 +++++++++++++++++----
src/bin/psql/describe.h | 5 +-
src/test/regress/expected/subscription.out | 176 ++++-----
4 files changed, 429 insertions(+), 167 deletions(-)
diff --git a/src/bin/psql/command.c b/src/bin/psql/command.c
index 5dde9bdd0e8..2ae43889eda 100644
--- a/src/bin/psql/command.c
+++ b/src/bin/psql/command.c
@@ -1220,7 +1220,10 @@ exec_command_d(PsqlScanState scan_state, bool active_branch, const char *cmd)
success = listPublications(pattern);
break;
case 's':
- success = describeSubscriptions(pattern, show_verbose);
+ if (show_verbose)
+ success = describeSubscriptions(pattern);
+ else
+ success = listSubscriptions(pattern);
break;
default:
status = PSQL_CMD_UNKNOWN;
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index a1a440a6469..c655d2f523d 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -23,6 +23,7 @@
#include "catalog/pg_collation_d.h"
#include "catalog/pg_constraint_d.h"
#include "catalog/pg_default_acl_d.h"
+#include "catalog/pg_namespace_d.h"
#include "catalog/pg_proc_d.h"
#include "catalog/pg_propgraph_element_d.h"
#include "catalog/pg_publication_d.h"
@@ -7081,19 +7082,17 @@ error_return:
/*
* \dRs
- * Describes subscriptions.
+ * Lists subscriptions.
*
* Takes an optional regexp to select particular subscriptions
*/
bool
-describeSubscriptions(const char *pattern, bool verbose)
+listSubscriptions(const char *pattern)
{
PQExpBufferData buf;
PGresult *res;
printQueryOpt myopt = pset.popt;
- static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false, false, false,
- false, false, false, false, false, false, false};
+ static const bool translate_columns[] = {false, false, false, false};
if (pset.sversion < 100000)
{
@@ -7108,101 +7107,235 @@ describeSubscriptions(const char *pattern, bool verbose)
initPQExpBuffer(&buf);
printfPQExpBuffer(&buf, "/* %s */\n", _("Get matching subscriptions"));
+
+ /* Only display subscriptions in current database. */
appendPQExpBuffer(&buf,
"SELECT subname AS \"%s\"\n"
", pg_catalog.pg_get_userbyid(subowner) AS \"%s\"\n"
", subenabled AS \"%s\"\n"
- ", subpublications AS \"%s\"\n",
+ ", subpublications AS \"%s\"\n"
+ "FROM pg_catalog.pg_subscription\n"
+ "WHERE subdbid = (SELECT oid\n"
+ " FROM pg_catalog.pg_database\n"
+ " WHERE datname = pg_catalog.current_database())",
gettext_noop("Name"),
gettext_noop("Owner"),
gettext_noop("Enabled"),
gettext_noop("Publication"));
- if (verbose)
+ if (!validateSQLNamePattern(&buf, pattern, true, false,
+ NULL, "subname", NULL,
+ NULL,
+ NULL, 1))
{
- /* Binary mode and streaming are only supported in v14 and higher */
- if (pset.sversion >= 140000)
+ termPQExpBuffer(&buf);
+ return false;
+ }
+
+ appendPQExpBufferStr(&buf, "ORDER BY 1;");
+
+ res = PSQLexec(buf.data);
+ termPQExpBuffer(&buf);
+ if (!res)
+ return false;
+
+ myopt.title = _("List of subscriptions");
+ myopt.translate_header = true;
+ myopt.translate_columns = translate_columns;
+ myopt.n_translate_columns = lengthof(translate_columns);
+
+ printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+
+ PQclear(res);
+
+ return true;
+}
+
+/*
+ * Add footer to subscription description.
+ */
+static void
+addFooterToSubscriptionDesc(char *logdest, char *conflict_schema, Oid subid,
+ printTableContent *const cont)
+{
+ if (pset.sversion >= 190000)
+ {
+ if (strcmp(logdest, "table") == 0 || strcmp(logdest, "all") == 0)
{
- appendPQExpBuffer(&buf,
- ", subbinary AS \"%s\"\n",
- gettext_noop("Binary"));
+ /*
+ * The size accounts for schema name (NAMEDATALEN), relation
+ * name (NAMEDATALEN), separator '.' and null terminator.
+ */
+ char conflictlogtable[NAMEDATALEN + NAMEDATALEN + 2];
- if (pset.sversion >= 160000)
- appendPQExpBuffer(&buf,
- ", (CASE substream\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_OFF) " THEN 'off'\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_ON) " THEN 'on'\n"
- " WHEN " CppAsString2(LOGICALREP_STREAM_PARALLEL) " THEN 'parallel'\n"
- " END) AS \"%s\"\n",
- gettext_noop("Streaming"));
- else
- appendPQExpBuffer(&buf,
- ", substream AS \"%s\"\n",
- gettext_noop("Streaming"));
+ snprintf(conflictlogtable,
+ sizeof(conflictlogtable),
+ "%s." CONFLICT_LOG_RELATION_NAME_FMT,
+ conflict_schema, subid);
+
+ printTableAddFooter(cont, _("Conflict log table:"));
+ printTableAddFooter(cont,
+ psprintf(" \"%s\"", conflictlogtable));
}
+ }
+}
- /* Two_phase and disable_on_error are only supported in v15 and higher */
- if (pset.sversion >= 150000)
- appendPQExpBuffer(&buf,
- ", subtwophasestate AS \"%s\"\n"
- ", subdisableonerr AS \"%s\"\n",
- gettext_noop("Two-phase commit"),
- gettext_noop("Disable on error"));
+/*
+ * \dRs+
+ * Describes subscriptions.
+ *
+ * Takes an optional regexp to select particular subscriptions
+ */
+bool
+describeSubscriptions(const char *pattern)
+{
+ PQExpBufferData buf;
+ int i;
+ PGresult *res;
+ int ncols;
+ int nrows = 1;
- if (pset.sversion >= 160000)
- appendPQExpBuffer(&buf,
- ", suborigin AS \"%s\"\n"
- ", subpasswordrequired AS \"%s\"\n"
- ", subrunasowner AS \"%s\"\n",
- gettext_noop("Origin"),
- gettext_noop("Password required"),
- gettext_noop("Run as owner?"));
+ PQExpBufferData title;
+ printTableContent cont;
- if (pset.sversion >= 170000)
+ if (pset.sversion < 100000)
+ {
+ char sverbuf[32];
+
+ pg_log_error("The server (version %s) does not support subscriptions.",
+ formatPGVersionNumber(pset.sversion, false,
+ sverbuf, sizeof(sverbuf)));
+ return true;
+ }
+
+ initPQExpBuffer(&buf);
+
+ printfPQExpBuffer(&buf, "/* %s */\n", _("Get matching subscriptions"));
+ appendPQExpBuffer(&buf,
+ "SELECT oid, subname AS \"%s\"\n"
+ ", (SELECT nspname FROM pg_namespace WHERE oid = " CppAsString2(PG_CONFLICT_NAMESPACE) ") AS \"%s\"\n"
+ ", pg_catalog.pg_get_userbyid(subowner) AS \"%s\"\n"
+ ", subenabled AS \"%s\"\n"
+ ", subpublications AS \"%s\"\n",
+ gettext_noop("Name"),
+ gettext_noop("Conflict_schema"),
+ gettext_noop("Owner"),
+ gettext_noop("Enabled"),
+ gettext_noop("Publication"));
+
+ /*
+ * oid, subname and conflict_schema columns are internal and not displayed,
+ * so only 3 visible columns.
+ */
+ ncols = 3;
+
+ /* Binary mode and streaming are only supported in v14 and higher */
+ if (pset.sversion >= 140000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subbinary AS \"%s\"\n",
+ gettext_noop("Binary"));
+ ncols++;
+
+ if (pset.sversion >= 160000)
appendPQExpBuffer(&buf,
- ", subfailover AS \"%s\"\n",
- gettext_noop("Failover"));
- if (pset.sversion >= 190000)
- {
+ ", (CASE substream\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_OFF) " THEN 'off'\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_ON) " THEN 'on'\n"
+ " WHEN " CppAsString2(LOGICALREP_STREAM_PARALLEL) " THEN 'parallel'\n"
+ " END) AS \"%s\"\n",
+ gettext_noop("Streaming"));
+ else
appendPQExpBuffer(&buf,
- ", (select srvname from pg_catalog.pg_foreign_server where oid=subserver) AS \"%s\"\n",
- gettext_noop("Server"));
+ ", substream AS \"%s\"\n",
+ gettext_noop("Streaming"));
- appendPQExpBuffer(&buf,
- ", subretaindeadtuples AS \"%s\"\n",
- gettext_noop("Retain dead tuples"));
+ ncols++;
+ }
- appendPQExpBuffer(&buf,
- ", submaxretention AS \"%s\"\n",
- gettext_noop("Max retention duration"));
+ /* Two_phase and disable_on_error are only supported in v15 and higher */
+ if (pset.sversion >= 150000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subtwophasestate AS \"%s\"\n"
+ ", subdisableonerr AS \"%s\"\n",
+ gettext_noop("Two-phase commit"),
+ gettext_noop("Disable on error"));
+ ncols += 2;
+ }
- appendPQExpBuffer(&buf,
- ", subretentionactive AS \"%s\"\n",
- gettext_noop("Retention active"));
- }
+ if (pset.sversion >= 160000)
+ {
+ appendPQExpBuffer(&buf,
+ ", suborigin AS \"%s\"\n"
+ ", subpasswordrequired AS \"%s\"\n"
+ ", subrunasowner AS \"%s\"\n",
+ gettext_noop("Origin"),
+ gettext_noop("Password required"),
+ gettext_noop("Run as owner?"));
+ ncols += 3;
+ }
+ if (pset.sversion >= 170000)
+ {
appendPQExpBuffer(&buf,
- ", subsynccommit AS \"%s\"\n"
- ", subconninfo AS \"%s\"\n",
- gettext_noop("Synchronous commit"),
- gettext_noop("Conninfo"));
+ ", subfailover AS \"%s\"\n",
+ gettext_noop("Failover"));
+ ncols++;
+ }
- if (pset.sversion >= 190000)
- appendPQExpBuffer(&buf,
- ", subwalrcvtimeout AS \"%s\"\n",
- gettext_noop("Receiver timeout"));
+ if (pset.sversion >= 190000)
+ {
+ appendPQExpBuffer(&buf,
+ ", (select srvname from pg_catalog.pg_foreign_server where oid=subserver) AS \"%s\"\n"
+ ", subretaindeadtuples AS \"%s\"\n"
+ ", submaxretention AS \"%s\"\n"
+ ", subretentionactive AS \"%s\"\n",
+ gettext_noop("Server"),
+ gettext_noop("Retain dead tuples"),
+ gettext_noop("Max retention duration"),
+ gettext_noop("Retention active"));
+ ncols += 4;
+ }
- /* Skip LSN is only supported in v15 and higher */
- if (pset.sversion >= 150000)
- appendPQExpBuffer(&buf,
- ", subskiplsn AS \"%s\"\n",
- gettext_noop("Skip LSN"));
+ appendPQExpBuffer(&buf,
+ ", subsynccommit AS \"%s\"\n"
+ ", subconninfo AS \"%s\"\n",
+ gettext_noop("Synchronous commit"),
+ gettext_noop("Conninfo"));
+ ncols += 2;
+ if (pset.sversion >= 190000)
+ {
appendPQExpBuffer(&buf,
- ", pg_catalog.obj_description(oid, 'pg_subscription') AS \"%s\"\n",
- gettext_noop("Description"));
+ ", subwalrcvtimeout AS \"%s\"\n",
+ gettext_noop("Receiver timeout"));
+ ncols++;
+ }
+
+ /* Skip LSN is only supported in v15 and higher */
+ if (pset.sversion >= 150000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subskiplsn AS \"%s\"\n",
+ gettext_noop("Skip LSN"));
+ ncols++;
+ }
+
+ /* Conflict log destination is supported in v19 and higher */
+ if (pset.sversion >= 190000)
+ {
+ appendPQExpBuffer(&buf,
+ ", subconflictlogdest AS \"%s\"\n",
+ gettext_noop("Conflict log destination"));
+ ncols++;
}
+ appendPQExpBuffer(&buf,
+ ", pg_catalog.obj_description(oid, 'pg_subscription') AS \"%s\"\n",
+ gettext_noop("Description"));
+ ncols++;
+
/* Only display subscriptions in current database. */
appendPQExpBufferStr(&buf,
"FROM pg_catalog.pg_subscription\n"
@@ -7219,20 +7352,143 @@ describeSubscriptions(const char *pattern, bool verbose)
return false;
}
- appendPQExpBufferStr(&buf, "ORDER BY 1;");
+ appendPQExpBufferStr(&buf, "ORDER BY subname;");
res = PSQLexec(buf.data);
termPQExpBuffer(&buf);
if (!res)
return false;
- myopt.title = _("List of subscriptions");
- myopt.translate_header = true;
- myopt.translate_columns = translate_columns;
- myopt.n_translate_columns = lengthof(translate_columns);
+ if (PQntuples(res) == 0)
+ {
+ if (!pset.quiet)
+ {
+ if (pattern)
+ pg_log_error("Did not find any subscription named \"%s\".",
+ pattern);
+ else
+ pg_log_error("Did not find any subscriptions.");
+ }
- printQuery(res, &myopt, pset.queryFout, false, pset.logfile);
+ termPQExpBuffer(&buf);
+ PQclear(res);
+ return false;
+ }
+
+ for (i = 0; i < PQntuples(res); i++)
+ {
+ const char align = 'l';
+ Oid subid = atooid(PQgetvalue(res, i, 0));
+ char *subname = PQgetvalue(res, i, 1);
+ char *conflict_schema = PQgetvalue(res, i, 2);
+ int current_col = 3;
+ char *logdest = "log";
+ printTableOpt myopt = pset.popt.topt;
+
+ initPQExpBuffer(&title);
+ printfPQExpBuffer(&title, _("Subscription %s"), subname);
+ printTableInit(&cont, &myopt, title.data, ncols, nrows);
+
+ printTableAddHeader(&cont, gettext_noop("Owner"), true, align);
+ printTableAddHeader(&cont, gettext_noop("Enabled"), true, align);
+ printTableAddHeader(&cont, gettext_noop("Publication"), true, align);
+
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ if (pset.sversion >= 140000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Binary"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Streaming"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 150000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Two-phase commit"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Disable on error"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 160000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Origin"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Password required"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Run as owner?"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 170000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Failover"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Server"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Retain dead tuples"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Max retention duration"),
+ true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ printTableAddHeader(&cont, gettext_noop("Retention active"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ printTableAddHeader(&cont, gettext_noop("Synchronous commit"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ printTableAddHeader(&cont, gettext_noop("Conninfo"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Receiver timeout"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 150000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Skip LSN"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+ }
+
+ if (pset.sversion >= 190000)
+ {
+ printTableAddHeader(&cont, gettext_noop("Conflict log destination"),
+ true, align);
+
+ logdest = PQgetvalue(res, i, current_col++);
+
+ printTableAddCell(&cont, logdest, false, false);
+ }
+
+ printTableAddHeader(&cont, gettext_noop("Description"), true, align);
+ printTableAddCell(&cont, PQgetvalue(res, i, current_col++), false, false);
+
+ addFooterToSubscriptionDesc(logdest, conflict_schema, subid, &cont);
+
+ printTable(&cont, pset.queryFout, false, pset.logfile);
+ printTableCleanup(&cont);
+
+ termPQExpBuffer(&title);
+ }
+
+ termPQExpBuffer(&buf);
PQclear(res);
return true;
}
diff --git a/src/bin/psql/describe.h b/src/bin/psql/describe.h
index 47fae5ceafb..15c6c685323 100644
--- a/src/bin/psql/describe.h
+++ b/src/bin/psql/describe.h
@@ -126,7 +126,10 @@ bool listPublications(const char *pattern);
bool describePublications(const char *pattern);
/* \dRs */
-bool describeSubscriptions(const char *pattern, bool verbose);
+bool listSubscriptions(const char *pattern);
+
+/* \dRs+ */
+bool describeSubscriptions(const char *pattern);
/* \dAc */
extern bool listOperatorClasses(const char *access_method_pattern,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index d201ad764f0..56d6aa94a4c 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -139,18 +139,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub4
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | none | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub4
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -237,10 +237,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log | test subscription
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -249,10 +249,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | f | t | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | log | test subscription
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -268,10 +268,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00012345 | log | test subscription
(1 row)
-- ok - with lsn = NONE
@@ -280,10 +280,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | test subscription
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist2 | -1 | 0/00000000 | log | test subscription
(1 row)
BEGIN;
@@ -319,10 +319,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = '80s');
ALTER SUBSCRIPTION regress_testsub_foo SET (wal_receiver_timeout = 'foobar');
ERROR: invalid value for parameter "wal_receiver_timeout": "foobar"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+-------------------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | test subscription
+ Subscription regress_testsub_foo
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+------------------------------+------------------+------------+--------------------------+-------------------
+ regress_subscription_user | f | {testpub2,testpub3} | f | parallel | d | f | any | t | f | f | | f | 0 | f | local | dbname=regress_doesnotexist2 | 80s | 0/00000000 | log | test subscription
(1 row)
-- rename back to keep the rest simple
@@ -351,19 +351,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | t | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -375,27 +375,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- fail - publication already exists
@@ -410,10 +410,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- fail - publication used more than once
@@ -428,10 +428,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -467,19 +467,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- we can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -489,10 +489,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -505,18 +505,18 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | t | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -529,10 +529,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -549,10 +549,10 @@ NOTICE: max_retention_duration is ineffective when retain_dead_tuples is disabl
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 1000 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
-- fail - max_retention_duration must be non-negative
@@ -561,10 +561,10 @@ ERROR: max_retention_duration cannot be negative
-- ok
ALTER SUBSCRIPTION regress_testsub SET (max_retention_duration = 0);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Description
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+-------------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 |
+ Subscription regress_testsub
+ Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Server | Retain dead tuples | Max retention duration | Retention active | Synchronous commit | Conninfo | Receiver timeout | Skip LSN | Conflict log destination | Description
+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------+--------------------+------------------------+------------------+--------------------+-----------------------------+------------------+------------+--------------------------+-------------
+ regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | f | | f | 0 | f | off | dbname=regress_doesnotexist | -1 | 0/00000000 | log |
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
--
2.49.0
[application/octet-stream] v63-0002-Preserve-conflict-log-destination-and-subscripti.patch (23.4K, ../CAFiTN-s45_nOdvE8mBquZO=vyKbVY1ww_q3pejUz7DC29NcSxQ@mail.gmail.com/4-v63-0002-Preserve-conflict-log-destination-and-subscripti.patch)
download | inline diff:
From c6018aa2e45f05dac00e1489b6cf128bf857bc11 Mon Sep 17 00:00:00 2001
From: Dilip Kumar <[email protected]>
Date: Thu, 2 Jul 2026 13:20:58 +0530
Subject: [PATCH v63 2/3] Preserve conflict log destination and subscription
OID for subscriptions
Support pg_dump to dump and restore the conflict_log_destination setting for
subscriptions.
During a normal CREATE SUBSCRIPTION, a conflict log table is created
automatically when required. However, during binary upgrade, the conflict
log table will already exist and must be reused rather than recreated, and
the subscription must retain its original OID to correctly re-establish
catalog relationships.
To ensure correct behavior, pg_dump now emits an ALTER SUBSCRIPTION command
after subscription creation to restore the conflict_log_destination setting.
TODO: Ensure all 190000 series references are updated to 200000
when the version number is bumped.
---
src/backend/catalog/heap.c | 2 +
src/backend/commands/subscriptioncmds.c | 56 +++++++-
src/backend/replication/logical/conflict.c | 2 +-
src/backend/utils/adt/pg_upgrade_support.c | 10 ++
src/bin/pg_dump/pg_dump.c | 125 +++++++++++++++++-
src/bin/pg_dump/pg_dump.h | 2 +
src/bin/pg_dump/pg_dump_sort.c | 31 +++++
src/bin/pg_dump/t/002_pg_dump.pl | 5 +-
src/bin/pg_upgrade/pg_upgrade.c | 4 +
src/bin/pg_upgrade/t/004_subscription.pl | 14 +-
src/include/catalog/binary_upgrade.h | 1 +
src/include/catalog/pg_proc.dat | 4 +
src/include/catalog/pg_subscription.h | 3 +
.../expected/spgist_name_ops.out | 6 +-
14 files changed, 249 insertions(+), 16 deletions(-)
diff --git a/src/backend/catalog/heap.c b/src/backend/catalog/heap.c
index 88087654de9..08d15de4083 100644
--- a/src/backend/catalog/heap.c
+++ b/src/backend/catalog/heap.c
@@ -312,6 +312,8 @@ heap_create(const char *relname,
* But allow creating indexes on relations in pg_catalog even if
* allow_system_table_mods = off, upper layers already guarantee it's on a
* user defined relation, not a system one.
+ *
+ * Allow creation of the conflict log table in binary-upgrade mode.
*/
if (!allow_system_table_mods &&
((IsCatalogNamespace(relnamespace) && relkind != RELKIND_INDEX) ||
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 4292e7fb8f4..0bd73876349 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -19,6 +19,7 @@
#include "access/table.h"
#include "access/twophase.h"
#include "access/xact.h"
+#include "catalog/binary_upgrade.h"
#include "catalog/catalog.h"
#include "catalog/dependency.h"
#include "catalog/indexing.h"
@@ -86,6 +87,12 @@
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
+/*
+ * This will be set by the pg_upgrade_support function --
+ * binary_upgrade_set_next_pg_subscription_oid().
+ */
+Oid binary_upgrade_next_pg_subscription_oid = InvalidOid;
+
/*
* Structure to hold a bitmap representing the user-provided CREATE/ALTER
* SUBSCRIPTION command options and the parsed/default values of each of them.
@@ -828,8 +835,21 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
memset(values, 0, sizeof(values));
memset(nulls, false, sizeof(nulls));
- subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
- Anum_pg_subscription_oid);
+ /* Use binary-upgrade override for pg_subscription.oid? */
+ if (IsBinaryUpgrade)
+ {
+ if (!OidIsValid(binary_upgrade_next_pg_subscription_oid))
+ ereport(ERROR,
+ (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+ errmsg("pg_subscription OID value not set when in binary upgrade mode")));
+
+ subid = binary_upgrade_next_pg_subscription_oid;
+ binary_upgrade_next_pg_subscription_oid = InvalidOid;
+ }
+ else
+ subid = GetNewOidWithIndex(rel, SubscriptionObjectIndexId,
+ Anum_pg_subscription_oid);
+
values[Anum_pg_subscription_oid - 1] = ObjectIdGetDatum(subid);
values[Anum_pg_subscription_subdbid - 1] = ObjectIdGetDatum(MyDatabaseId);
values[Anum_pg_subscription_subskiplsn - 1] = LSNGetDatum(InvalidXLogRecPtr);
@@ -1542,18 +1562,40 @@ alter_sub_conflict_log_dest(Subscription *sub, ConflictLogDest oldlogdest,
{
ObjectAddress cltaddr;
ObjectAddress subobj;
+ char relname[NAMEDATALEN];
- relid = create_conflict_log_table(sub->oid, sub->name, sub->owner);
- update_relid = true;
+ snprintf(relname, NAMEDATALEN, CONFLICT_LOG_RELATION_NAME_FMT, sub->oid);
/*
- * Establish an internal dependency between the conflict log table
- * and the subscription. For details refer comments in
- * CreateSubscription function.
+ * In upgrade scenarios, old_dest reflects the default behavior of
+ * a newly created subscription (i.e., no conflict logging to
+ * table) However, the conflict log table will already exist from
+ * the upgraded cluster. In such cases, we need to detect the
+ * pre-existing table and update the catalog state to associate it
+ * with the subscription instead of creating a new one.
+ */
+ relid = get_relname_relid(relname, PG_CONFLICT_NAMESPACE);
+ if (OidIsValid(relid))
+ {
+ /* Existing table from upgrade */
+ Assert(IsBinaryUpgrade);
+ }
+ else
+ {
+ relid = create_conflict_log_table(sub->oid, sub->name,
+ sub->owner);
+ }
+
+ /*
+ * Establish an internal dependency between the conflict log
+ * table and the subscription. Refer to comments in the
+ * CreateSubscription function for details.
*/
ObjectAddressSet(cltaddr, RelationRelationId, relid);
ObjectAddressSet(subobj, SubscriptionRelationId, sub->oid);
recordDependencyOn(&cltaddr, &subobj, DEPENDENCY_INTERNAL);
+
+ update_relid = true;
}
}
diff --git a/src/backend/replication/logical/conflict.c b/src/backend/replication/logical/conflict.c
index 5551d87d1bc..a3c06e22ac1 100644
--- a/src/backend/replication/logical/conflict.c
+++ b/src/backend/replication/logical/conflict.c
@@ -207,7 +207,7 @@ create_conflict_log_table(Oid subid, char *subname, Oid subowner)
Oid relid;
char relname[NAMEDATALEN];
- snprintf(relname, NAMEDATALEN, "pg_conflict_log_%u", subid);
+ snprintf(relname, NAMEDATALEN, CONFLICT_LOG_RELATION_NAME_FMT, subid);
/* Build the tuple descriptor for the new table. */
tupdesc = create_conflict_log_table_tupdesc();
diff --git a/src/backend/utils/adt/pg_upgrade_support.c b/src/backend/utils/adt/pg_upgrade_support.c
index b505a6b4fee..59c3e7f0146 100644
--- a/src/backend/utils/adt/pg_upgrade_support.c
+++ b/src/backend/utils/adt/pg_upgrade_support.c
@@ -181,6 +181,16 @@ binary_upgrade_set_next_pg_authid_oid(PG_FUNCTION_ARGS)
PG_RETURN_VOID();
}
+Datum
+binary_upgrade_set_next_pg_subscription_oid(PG_FUNCTION_ARGS)
+{
+ Oid subid = PG_GETARG_OID(0);
+
+ CHECK_IS_BINARY_UPGRADE;
+ binary_upgrade_next_pg_subscription_oid = subid;
+ PG_RETURN_VOID();
+}
+
Datum
binary_upgrade_create_empty_extension(PG_FUNCTION_ARGS)
{
diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c
index 25f66b56d40..9d60d12dfb5 100644
--- a/src/bin/pg_dump/pg_dump.c
+++ b/src/bin/pg_dump/pg_dump.c
@@ -1981,6 +1981,8 @@ checkExtensionMembership(DumpableObject *dobj, Archive *fout)
static void
selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
{
+ DumpOptions *dopt = fout->dopt;
+
/*
* DUMP_COMPONENT_DEFINITION typically implies a CREATE SCHEMA statement
* and (for --clean) a DROP SCHEMA statement. (In the absence of
@@ -2010,6 +2012,32 @@ selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
*/
nsinfo->dobj.dump_contains = nsinfo->dobj.dump = DUMP_COMPONENT_ACL;
}
+ else if (strcmp(nsinfo->dobj.name, "pg_conflict") == 0)
+ {
+ if (dopt->binary_upgrade)
+ {
+ /*
+ * The pg_conflict schema is a strange beast that sits in a sort
+ * of no-mans-land between being a system object and a user
+ * object. CREATE SCHEMA would fail, so its
+ * DUMP_COMPONENT_DEFINITION is just a comment.
+ */
+ nsinfo->create = false;
+ nsinfo->dobj.dump = DUMP_COMPONENT_ALL;
+ nsinfo->dobj.dump &= ~DUMP_COMPONENT_DEFINITION;
+ nsinfo->dobj.dump_contains = DUMP_COMPONENT_ALL;
+
+ /*
+ * Also, make like it has a comment even if it doesn't; this is so
+ * that we'll emit a command to drop the comment, if appropriate.
+ * (Without this, we'd not call dumpCommentExtended for it.)
+ */
+ nsinfo->dobj.components |= DUMP_COMPONENT_COMMENT;
+ }
+ else
+ nsinfo->dobj.dump_contains = nsinfo->dobj.dump =
+ DUMP_COMPONENT_NONE;
+ }
else if (strncmp(nsinfo->dobj.name, "pg_", 3) == 0 ||
strcmp(nsinfo->dobj.name, "information_schema") == 0)
{
@@ -2067,9 +2095,38 @@ selectDumpableNamespace(NamespaceInfo *nsinfo, Archive *fout)
static void
selectDumpableTable(TableInfo *tbinfo, Archive *fout)
{
+ DumpOptions *dopt = fout->dopt;
+
if (checkExtensionMembership(&tbinfo->dobj, fout))
return; /* extension membership overrides all else */
+ if (strcmp(tbinfo->dobj.namespace->dobj.name, "pg_conflict") == 0)
+ {
+ /*
+ * Dump pg_conflict tables only during binary upgrade. The schema is
+ * assumed to already exist.
+ */
+ if (dopt->binary_upgrade)
+ {
+ tbinfo->dobj.dump = DUMP_COMPONENT_DEFINITION;
+
+ /*
+ * Suppress the "ALTER TABLE ... OWNER TO ..." command for this
+ * table. This prevents pg_dump from outputting the owner change.
+ */
+ tbinfo->rolname = NULL;
+ }
+ else
+ tbinfo->dobj.dump = DUMP_COMPONENT_NONE;
+
+ /*
+ * Do not apply table selection or exclusion rules to pg_conflict
+ * tables. During binary upgrade, these tables are dumped to preserve
+ * catalog state; otherwise, they are not dumped at all.
+ */
+ return;
+ }
+
/*
* If specific tables are being dumped, dump just those tables; else, dump
* according to the parent namespace's dump flag.
@@ -5184,6 +5241,8 @@ getSubscriptions(Archive *fout)
int i_subfailover;
int i_subretaindeadtuples;
int i_submaxretention;
+ int i_subconflictlogrelid;
+ int i_sublogdestination;
int i,
ntups;
@@ -5282,9 +5341,16 @@ getSubscriptions(Archive *fout)
" '-1' AS subwalrcvtimeout,\n");
if (fout->remoteVersion >= 190000)
- appendPQExpBufferStr(query, " fs.srvname AS subservername\n");
+ appendPQExpBufferStr(query, " fs.srvname AS subservername,\n");
+ else
+ appendPQExpBufferStr(query, " NULL AS subservername,\n");
+
+ if (fout->remoteVersion >= 190000)
+ appendPQExpBufferStr(query,
+ " s.subconflictlogrelid, s.subconflictlogdest\n");
else
- appendPQExpBufferStr(query, " NULL AS subservername\n");
+ appendPQExpBufferStr(query,
+ " NULL AS subconflictlogrelid, NULL AS subconflictlogdest\n");
appendPQExpBufferStr(query,
"FROM pg_subscription s\n");
@@ -5333,6 +5399,8 @@ getSubscriptions(Archive *fout)
i_subpublications = PQfnumber(res, "subpublications");
i_suborigin = PQfnumber(res, "suborigin");
i_suboriginremotelsn = PQfnumber(res, "suboriginremotelsn");
+ i_subconflictlogrelid = PQfnumber(res, "subconflictlogrelid");
+ i_sublogdestination = PQfnumber(res, "subconflictlogdest");
subinfo = pg_malloc_array(SubscriptionInfo, ntups);
@@ -5391,6 +5459,32 @@ getSubscriptions(Archive *fout)
subinfo[i].suboriginremotelsn =
pg_strdup(PQgetvalue(res, i, i_suboriginremotelsn));
+ if (PQgetisnull(res, i, i_subconflictlogrelid))
+ subinfo[i].subconflictlogrelid = InvalidOid;
+ else
+ {
+ subinfo[i].subconflictlogrelid =
+ atooid(PQgetvalue(res, i, i_subconflictlogrelid));
+
+ if (subinfo[i].subconflictlogrelid)
+ {
+ TableInfo *tableInfo;
+
+ tableInfo = findTableByOid(subinfo[i].subconflictlogrelid);
+ if (!tableInfo)
+ pg_fatal("could not find conflict log table with OID %u",
+ subinfo[i].subconflictlogrelid);
+
+ addObjectDependency(&subinfo[i].dobj, tableInfo->dobj.dumpId);
+ }
+ }
+
+ if (PQgetisnull(res, i, i_sublogdestination))
+ subinfo[i].subconflictlogdest = NULL;
+ else
+ subinfo[i].subconflictlogdest =
+ pg_strdup(PQgetvalue(res, i, i_sublogdestination));
+
/* Decide whether we want to dump it */
selectDumpableObject(&(subinfo[i].dobj), fout);
}
@@ -5583,6 +5677,14 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBuffer(delq, "DROP SUBSCRIPTION %s;\n",
qsubname);
+ if (dopt->binary_upgrade)
+ {
+ appendPQExpBufferStr(query, "\n-- For binary upgrade, must preserve pg_subscription.oid\n");
+ appendPQExpBuffer(query,
+ "SELECT pg_catalog.binary_upgrade_set_next_pg_subscription_oid('%u'::pg_catalog.oid);\n\n",
+ subinfo->dobj.catId.oid);
+ }
+
appendPQExpBuffer(query, "CREATE SUBSCRIPTION %s ",
qsubname);
if (subinfo->subservername)
@@ -5656,6 +5758,25 @@ dumpSubscription(Archive *fout, const SubscriptionInfo *subinfo)
appendPQExpBufferStr(query, ");\n");
+ /*
+ * Restore the conflict log destination separately from CREATE SUBSCRIPTION.
+ * Setting conflict_log_destination to 'table' may create and associate
+ * a conflict log table whose name is derived from the subscription OID.
+ * Since the subscription must already exist before that OID is known,
+ * emit this as a separate ALTER SUBSCRIPTION so that any required conflict
+ * log table can be created and linked after the subscription itself has
+ * been restored.
+ *
+ * We skip the default value ('log') to match the handling of other
+ * subscription options.
+ */
+ if (subinfo->subconflictlogdest &&
+ (pg_strcasecmp(subinfo->subconflictlogdest, "log") != 0))
+ appendPQExpBuffer(query,
+ "\n\nALTER SUBSCRIPTION %s SET (conflict_log_destination = %s);\n",
+ qsubname,
+ subinfo->subconflictlogdest);
+
/*
* In binary-upgrade mode, we allow the replication to continue after the
* upgrade.
diff --git a/src/bin/pg_dump/pg_dump.h b/src/bin/pg_dump/pg_dump.h
index 5a6726d8b12..a43a3049343 100644
--- a/src/bin/pg_dump/pg_dump.h
+++ b/src/bin/pg_dump/pg_dump.h
@@ -722,6 +722,7 @@ typedef struct _SubscriptionInfo
bool subfailover;
bool subretaindeadtuples;
int submaxretention;
+ Oid subconflictlogrelid;
char *subservername;
char *subconninfo;
char *subslotname;
@@ -730,6 +731,7 @@ typedef struct _SubscriptionInfo
char *subpublications;
char *suborigin;
char *suboriginremotelsn;
+ char *subconflictlogdest;
} SubscriptionInfo;
/*
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 5b3ea0f2a18..18b7b78adda 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -1131,6 +1131,19 @@ repairTableAttrDefMultiLoop(DumpableObject *tableobj,
addObjectDependency(attrdefobj, tableobj->dumpId);
}
+/*
+ * Because we make subscriptions depend on their conflict log tables, while
+ * there is an automatic dependency in the other direction, we need to break
+ * the loop. Remove the automatic dependency, allowing the table to be created
+ * first.
+ */
+static void
+repairSubscriptionTableLoop(DumpableObject *subobj, DumpableObject *tableobj)
+{
+ /* Remove table's dependency on subscription */
+ removeObjectDependency(tableobj, subobj->dumpId);
+}
+
/*
* CHECK, NOT NULL constraints on domains work just like those on tables ...
*/
@@ -1361,6 +1374,24 @@ repairDependencyLoop(DumpableObject **loop,
return;
}
+ /*
+ * Subscription and its Conflict Log Table
+ */
+ if (nLoop == 2 &&
+ loop[0]->objType == DO_TABLE &&
+ loop[1]->objType == DO_SUBSCRIPTION)
+ {
+ repairSubscriptionTableLoop(loop[1], loop[0]);
+ return;
+ }
+ if (nLoop == 2 &&
+ loop[0]->objType == DO_SUBSCRIPTION &&
+ loop[1]->objType == DO_TABLE)
+ {
+ repairSubscriptionTableLoop(loop[0], loop[1]);
+ return;
+ }
+
/* index on partitioned table and corresponding index on partition */
if (nLoop == 2 &&
loop[0]->objType == DO_INDEX &&
diff --git a/src/bin/pg_dump/t/002_pg_dump.pl b/src/bin/pg_dump/t/002_pg_dump.pl
index 9258948b583..8c2653e852e 100644
--- a/src/bin/pg_dump/t/002_pg_dump.pl
+++ b/src/bin/pg_dump/t/002_pg_dump.pl
@@ -3317,9 +3317,10 @@ my %tests = (
create_order => 50,
create_sql => 'CREATE SUBSCRIPTION sub3
CONNECTION \'dbname=doesnotexist\' PUBLICATION pub1
- WITH (connect = false, origin = any, streaming = on);',
+ WITH (connect = false, origin = any, streaming = on, conflict_log_destination= table);',
regexp => qr/^
- \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E
+ \QCREATE SUBSCRIPTION sub3 CONNECTION 'dbname=doesnotexist' PUBLICATION pub1 WITH (connect = false, slot_name = 'sub3', streaming = on);\E\n\n\n
+ \QALTER SUBSCRIPTION sub3 SET (conflict_log_destination = table);\E
/xm,
like => { %full_runs, section_post_data => 1, },
unlike => {
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 2127d297bfe..135ef658c2c 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -35,6 +35,10 @@
*
* We control all assignments of pg_database.oid because we want the directory
* names to match between the old and new cluster.
+ *
+ * We control assignment of pg_subscription.oid because we want the oid to
+ * match between the old and new cluster to make use of subscription's
+ * conflict log table which is named using the subscription oid.
*/
diff --git a/src/bin/pg_upgrade/t/004_subscription.pl b/src/bin/pg_upgrade/t/004_subscription.pl
index 36f059faa2e..27731686d87 100644
--- a/src/bin/pg_upgrade/t/004_subscription.pl
+++ b/src/bin/pg_upgrade/t/004_subscription.pl
@@ -290,7 +290,7 @@ $publisher->safe_psql(
$old_sub->safe_psql(
'postgres', qq[
CREATE TABLE tab_upgraded2(id int);
- CREATE SUBSCRIPTION regress_sub5 CONNECTION '$connstr' PUBLICATION regress_pub5;
+ CREATE SUBSCRIPTION regress_sub5 CONNECTION '$connstr' PUBLICATION regress_pub5 with (conflict_log_destination = 'table');
]);
# The table tab_upgraded2 will be in the init state as the subscriber's
@@ -312,7 +312,10 @@ my $tab_upgraded1_oid = $old_sub->safe_psql('postgres',
"SELECT oid FROM pg_class WHERE relname = 'tab_upgraded1'");
my $tab_upgraded2_oid = $old_sub->safe_psql('postgres',
"SELECT oid FROM pg_class WHERE relname = 'tab_upgraded2'");
-
+my $sub5_oid = $old_sub->safe_psql('postgres',
+ "SELECT oid FROM pg_subscription where subname = 'regress_sub5'");
+my $sub_clt_relid = $old_sub->safe_psql('postgres',
+ "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'regress_sub5'");
$old_sub->stop;
# Change configuration so that initial table sync does not get started
@@ -394,6 +397,13 @@ $result = $new_sub->safe_psql('postgres',
);
is($result, qq(t), "conflict detection slot exists");
+# The subscription oid and the subscription conflict log table relid should be preserved
+$result = $new_sub->safe_psql('postgres', "SELECT oid FROM pg_subscription WHERE subname = 'regress_sub5'");
+is($result, qq($sub5_oid), "subscription oid should have been preserved");
+
+$result = $new_sub->safe_psql('postgres', "SELECT subconflictlogrelid FROM pg_subscription WHERE subname = 'regress_sub5'");
+is($result, qq($sub_clt_relid), "subscription conflict log table relid should have been preserved");
+
# Resume the initial sync and wait until all tables of subscription
# 'regress_sub5' are synchronized
$new_sub->append_conf('postgresql.conf',
diff --git a/src/include/catalog/binary_upgrade.h b/src/include/catalog/binary_upgrade.h
index 7bf7ae44385..b15b18e7dc9 100644
--- a/src/include/catalog/binary_upgrade.h
+++ b/src/include/catalog/binary_upgrade.h
@@ -32,6 +32,7 @@ extern PGDLLIMPORT RelFileNumber binary_upgrade_next_toast_pg_class_relfilenumbe
extern PGDLLIMPORT Oid binary_upgrade_next_pg_enum_oid;
extern PGDLLIMPORT Oid binary_upgrade_next_pg_authid_oid;
+extern PGDLLIMPORT Oid binary_upgrade_next_pg_subscription_oid;
extern PGDLLIMPORT bool binary_upgrade_record_init_privs;
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index 3cb84359adf..80aba5e8fd8 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11994,6 +11994,10 @@
proisstrict => 'f', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => '',
prosrc => 'binary_upgrade_create_conflict_detection_slot' },
+{ oid => '8407', descr => 'for use by pg_upgrade',
+ proname => 'binary_upgrade_set_next_pg_subscription_oid', provolatile => 'v',
+ proparallel => 'r', prorettype => 'void', proargtypes => 'oid',
+ prosrc => 'binary_upgrade_set_next_pg_subscription_oid' },
# conversion functions
{ oid => '4310', descr => 'internal conversion function for KOI8R to WIN1251',
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index 65ce8e145fb..f7e1f48aa8f 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -220,6 +220,9 @@ typedef struct Subscription
*/
#define LOGICALREP_STREAM_PARALLEL 'p'
+/* Conflict log table name format. %u is replaced with the subscription OID. */
+#define CONFLICT_LOG_RELATION_NAME_FMT "pg_conflict_log_%u"
+
#endif /* EXPOSE_TO_CLIENT_CODE */
extern Subscription *GetSubscription(Oid subid, bool missing_ok,
diff --git a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
index 1ee65ede243..39d43368c42 100644
--- a/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
+++ b/src/test/modules/spgist_name_ops/expected/spgist_name_ops.out
@@ -59,11 +59,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
-- Verify clean failure when INCLUDE'd columns result in overlength tuple
-- The error message details are platform-dependent, so show only SQLSTATE
@@ -108,11 +109,12 @@ select * from t
binary_upgrade_set_next_multirange_pg_type_oid | 1 | binary_upgrade_set_next_multirange_pg_type_oid
binary_upgrade_set_next_pg_authid_oid | | binary_upgrade_set_next_pg_authid_oid
binary_upgrade_set_next_pg_enum_oid | | binary_upgrade_set_next_pg_enum_oid
+ binary_upgrade_set_next_pg_subscription_oid | | binary_upgrade_set_next_pg_subscription_oid
binary_upgrade_set_next_pg_tablespace_oid | | binary_upgrade_set_next_pg_tablespace_oid
binary_upgrade_set_next_pg_type_oid | | binary_upgrade_set_next_pg_type_oid
binary_upgrade_set_next_toast_pg_class_oid | 1 | binary_upgrade_set_next_toast_pg_class_oid
binary_upgrade_set_next_toast_relfilenode | | binary_upgrade_set_next_toast_relfilenode
-(13 rows)
+(14 rows)
\set VERBOSITY sqlstate
insert into t values(repeat('xyzzy', 12), 42, repeat('xyzzy', 4000));
--
2.49.0
view thread (227+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: Re: Proposal: Conflict log history table for Logical Replication
In-Reply-To: <CAFiTN-s45_nOdvE8mBquZO=vyKbVY1ww_q3pejUz7DC29NcSxQ@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