Re: Conflict detection for update_deleted in logical replication - Mailing list pgsql-hackers

On Tue, 7 Jan 2025 at 18:04, Zhijie Hou (Fujitsu)
<houzj.fnst@fujitsu.com> wrote:
>
> On Friday, January 3, 2025 1:53 PM Amit Kapila <amit.kapila16@gmail.com> wrote:
> >
> > On Wed, Dec 25, 2024 at 8:13 AM Zhijie Hou (Fujitsu)
> > <houzj.fnst@fujitsu.com> wrote:
> > >
> > > Attach the new version patch set which addressed all other comments.
> > >
> >
> > Some more miscellaneous comments:
>
> Thanks for the comments!
>
> > =============================
> > 1.
> > @@ -1431,9 +1431,9 @@ RecordTransactionCommit(void)
> >   * modifying it.  This makes checkpoint's determination of which xacts
> >   * are delaying the checkpoint a bit fuzzy, but it doesn't matter.
> >   */
> > - Assert((MyProc->delayChkptFlags & DELAY_CHKPT_START) == 0);
> > + Assert((MyProc->delayChkptFlags & DELAY_CHKPT_IN_COMMIT) == 0);
> >   START_CRIT_SECTION();
> > - MyProc->delayChkptFlags |= DELAY_CHKPT_START;
> > + MyProc->delayChkptFlags |= DELAY_CHKPT_IN_COMMIT;
> >
> >   /*
> >   * Insert the commit XLOG record.
> > @@ -1536,7 +1536,7 @@ RecordTransactionCommit(void)
> >   */
> >   if (markXidCommitted)
> >   {
> > - MyProc->delayChkptFlags &= ~DELAY_CHKPT_START;
> > + MyProc->delayChkptFlags &= ~DELAY_CHKPT_IN_COMMIT;
> >   END_CRIT_SECTION();
> >
> > The comments related to this change should be updated in EndPrepare()
> > and RecordTransactionCommitPrepared(). They still refer to the
> > DELAY_CHKPT_START flag. We should update the comments explaining why
> > a
> > similar change is not required for prepare or commit_prepare, if there
> > is one.
>
> After considering more, I think we need to use the new flag in
> RecordTransactionCommitPrepared() as well, because it is assigned a commit
> timestamp and would be replicated as normal transaction if sub's two_phase is
> not enabled.
>
> > 3.
> > +FindMostRecentlyDeletedTupleInfo(Relation rel, TupleTableSlot *searchslot,
> > + TransactionId *delete_xid,
> > + RepOriginId *delete_origin,
> > + TimestampTz *delete_time)
> > ...
> > ...
> > + /* Try to find the tuple */
> > + while (table_scan_getnextslot(scan, ForwardScanDirection, scanslot))
> > + {
> > + bool dead = false;
> > + TransactionId xmax;
> > + TimestampTz localts;
> > + RepOriginId localorigin;
> > +
> > + if (!tuples_equal(scanslot, searchslot, eq, indexbitmap))
> > + continue;
> > +
> > + tuple = ExecFetchSlotHeapTuple(scanslot, false, NULL);
> > + buf = hslot->buffer;
> > +
> > + LockBuffer(buf, BUFFER_LOCK_SHARE);
> > +
> > + if (HeapTupleSatisfiesVacuum(tuple, oldestXmin, buf) ==
> > HEAPTUPLE_RECENTLY_DEAD)
> > + dead = true;
> > +
> > + LockBuffer(buf, BUFFER_LOCK_UNLOCK);
> > +
> > + if (!dead)
> > + continue;
> >
> > Why do we need to check only for HEAPTUPLE_RECENTLY_DEAD and not
> > HEAPTUPLE_DEAD? IIUC, we came here because we couldn't find the live
> > tuple, now whether the tuple is DEAD or RECENTLY_DEAD, why should it
> > matter to detect update_delete conflict?
>
> The HEAPTUPLE_DEAD could indicate tuples whose inserting transaction was
> aborted, in which case we could not get the commit timestamp or origin for the
> transaction. Or it could indicate tuples deleted by a transaction older than
> oldestXmin(we would take the new replication slot's xmin into account when
> computing this value), which means any subsequent transaction would have commit
> timestamp later than that old delete transaction, so I think it's OK to ignore
> this dead tuple and even detect update_missing because the resolution is to
> apply the subsequent UPDATEs anyway (assuming we are using last update win
> strategy). I added some comments along these lines in the patch.
>
> >
> > 5.
> > +
> > +      <varlistentry
> > id="sql-createsubscription-params-with-detect-update-deleted">
> > +        <term><literal>detect_update_deleted</literal>
> > (<type>boolean</type>)</term>
> > +        <listitem>
> > +         <para>
> > +          Specifies whether the detection of <xref
> > linkend="conflict-update-deleted"/>
> > +          is enabled. The default is <literal>false</literal>. If set to
> > +          true, the dead tuples on the subscriber that are still useful for
> > +          detecting <xref linkend="conflict-update-deleted"/>
> > +          are retained,
> >
> > One of the purposes of retaining dead tuples is to detect
> > update_delete conflict. But, I also see the following in 0001's commit
> > message: "Since the mechanism relies on a single replication slot, it
> > not only assists in retaining dead tuples but also preserves commit
> > timestamps and origin data. These information will be displayed in the
> > additional logs generated for logical replication conflicts.
> > Furthermore, the preserved commit timestamps and origin data are
> > essential for consistently detecting update_origin_differs conflicts."
> > which indicates there are other cases where retaining dead tuples can
> > help. So, I was thinking about whether to name this new option as
> > retain_dead_tuples or something along those lines?
>
> I used the retain_conflict_info in this version as it looks more general and we
> are already using similar name in patch(RetainConflictInfoData), but we can
> change it later if people have better ideas.
>
> Attached the V19 patch which addressed comments in [1][2][3][4][5][6][7].

Few comments:
1) All other options are ordered, we can mention retain_conflict_info
after password_required to keep it consistent, I think it got
misplaced because of the name change from detect_update_deleted to
retain_conflict_info:
diff --git a/src/bin/psql/tab-complete.in.c b/src/bin/psql/tab-complete.in.c
index bbd08770c3..9d07fbf07a 100644
--- a/src/bin/psql/tab-complete.in.c
+++ b/src/bin/psql/tab-complete.in.c
@@ -2278,9 +2278,10 @@ match_previous_words(int pattern_id,
                COMPLETE_WITH("(", "PUBLICATION");
        /* ALTER SUBSCRIPTION <name> SET ( */
        else if (Matches("ALTER", "SUBSCRIPTION", MatchAny, MatchAnyN,
"SET", "("))
-               COMPLETE_WITH("binary", "disable_on_error",
"failover", "origin",
-                                         "password_required",
"run_as_owner", "slot_name",
-                                         "streaming",
"synchronous_commit", "two_phase");
+               COMPLETE_WITH("binary", "retain_conflict_info",
"disable_on_error",
+                                         "failover", "origin",
"password_required",
+                                         "run_as_owner", "slot_name",
"streaming",
+                                         "synchronous_commit", "two_phase");

2) Similarly here too:
        /* Complete "CREATE SUBSCRIPTION <name> ...  WITH ( <opt>" */
        else if (Matches("CREATE", "SUBSCRIPTION", MatchAnyN, "WITH", "("))
                COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
-                                         "disable_on_error",
"enabled", "failover", "origin",
-                                         "password_required",
"run_as_owner", "slot_name",
-                                         "streaming",
"synchronous_commit", "two_phase");
+                                         "retain_conflict_info",
"disable_on_error", "enabled",

3) Now that the option detect_update_deleted is changed to
retain_conflict_info, we can change this to "Retain conflict info":
+               if (pset.sversion >= 180000)
+                       appendPQExpBuffer(&buf,
+                                                         ",
subretainconflictinfo AS \"%s\"\n",
+
gettext_noop("Detect update deleted"));

4) The corresponding test changes also should be updated:
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,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 refresh the
subscription.
 \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 | Synchronous commit |
      Conninfo           | Skip LSN

-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f       | {testpub}
| f      | parallel  | d                | f                | none   |
t                 | f             | f        | off                |
dbname=regress_doesnotexist | 0/0
+
                                                       List of
subscriptions
+       Name       |           Owner           | Enabled | Publication
| Binary | Streaming | Two-phase commit | Disable on error | Origin |
Password required | Run as owner? | Failover | Detect update deleted |
Synchronous commit |          Conninfo           | Skip LSN

+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+-----------------------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f       | {testpub}
| f      | parallel  | d                | f                | none   |
t                 | f             | f        | f                     |
off                | dbname=regress_doesnotexist | 0/0

5) This part of code is not very easy to understand that it is done
for handling wrap around, could we add some comments here:
+       if (!TimestampDifferenceExceeds(data->candidate_xid_time, now,
+
 data->xid_advance_interval))
+               return;
+
+       data->candidate_xid_time = now;
+
+       oldest_running_xid = GetOldestActiveTransactionId();
+       next_full_xid = ReadNextFullTransactionId();
+       epoch = EpochFromFullTransactionId(next_full_xid);
+
+       /* Compute the epoch of the oldest_running_xid */
+       if (oldest_running_xid > XidFromFullTransactionId(next_full_xid))
+               epoch--;

Regards,
Vignesh