1616#include "access/bufmask.h"
1717#include "access/gist_private.h"
1818#include "access/gistxlog.h"
19+ #include "access/heapam_xlog.h"
20+ #include "access/transam.h"
1921#include "access/xloginsert.h"
2022#include "access/xlogutils.h"
23+ #include "miscadmin.h"
24+ #include "storage/procarray.h"
2125#include "utils/memutils.h"
2226
2327static MemoryContext opCtx ; /* working memory for operations */
@@ -60,6 +64,155 @@ gistRedoClearFollowRight(XLogReaderState *record, uint8 block_id)
6064 UnlockReleaseBuffer (buffer );
6165}
6266
67+ /*
68+ * Get the latestRemovedXid from the heap pages pointed at by the index
69+ * tuples being deleted. See also btree_xlog_delete_get_latestRemovedXid,
70+ * on which this function is based.
71+ */
72+ static TransactionId
73+ gistRedoPageUpdateRecordGetLatestRemovedXid (XLogReaderState * record )
74+ {
75+ gistxlogPageUpdate * xlrec = (gistxlogPageUpdate * ) XLogRecGetData (record );
76+ OffsetNumber * todelete ;
77+ Buffer ibuffer ,
78+ hbuffer ;
79+ Page ipage ,
80+ hpage ;
81+ RelFileNode rnode ,
82+ * hnode ;
83+ BlockNumber blkno ;
84+ ItemId iitemid ,
85+ hitemid ;
86+ IndexTuple itup ;
87+ HeapTupleHeader htuphdr ;
88+ BlockNumber hblkno ;
89+ OffsetNumber hoffnum ;
90+ TransactionId latestRemovedXid = InvalidTransactionId ;
91+ int i ;
92+
93+ /*
94+ * If there's nothing running on the standby we don't need to derive a
95+ * full latestRemovedXid value, so use a fast path out of here. This
96+ * returns InvalidTransactionId, and so will conflict with all HS
97+ * transactions; but since we just worked out that that's zero people,
98+ * it's OK.
99+ *
100+ * XXX There is a race condition here, which is that a new backend might
101+ * start just after we look. If so, it cannot need to conflict, but this
102+ * coding will result in throwing a conflict anyway.
103+ */
104+ if (CountDBBackends (InvalidOid ) == 0 )
105+ return latestRemovedXid ;
106+
107+ /*
108+ * In what follows, we have to examine the previous state of the index
109+ * page, as well as the heap page(s) it points to. This is only valid if
110+ * WAL replay has reached a consistent database state; which means that
111+ * the preceding check is not just an optimization, but is *necessary*. We
112+ * won't have let in any user sessions before we reach consistency.
113+ */
114+ if (!reachedConsistency )
115+ elog (PANIC , "gistRedoDeleteRecordGetLatestRemovedXid: cannot operate with inconsistent data" );
116+
117+ /*
118+ * Get index page. If the DB is consistent, this should not fail, nor
119+ * should any of the heap page fetches below. If one does, we return
120+ * InvalidTransactionId to cancel all HS transactions. That's probably
121+ * overkill, but it's safe, and certainly better than panicking here.
122+ */
123+ XLogRecGetBlockTag (record , 0 , & rnode , NULL , & blkno );
124+ ibuffer = XLogReadBufferExtended (rnode , MAIN_FORKNUM , blkno , RBM_NORMAL );
125+ if (!BufferIsValid (ibuffer ))
126+ return InvalidTransactionId ;
127+ LockBuffer (ibuffer , BUFFER_LOCK_EXCLUSIVE );
128+ ipage = (Page ) BufferGetPage (ibuffer );
129+
130+ /*
131+ * Loop through the deleted index items to obtain the TransactionId from
132+ * the heap items they point to.
133+ */
134+ hnode = (RelFileNode * ) ((char * ) xlrec + sizeof (gistxlogPageUpdate ));
135+ todelete = (OffsetNumber * ) ((char * ) hnode + sizeof (RelFileNode ));
136+
137+ for (i = 0 ; i < xlrec -> ntodelete ; i ++ )
138+ {
139+ /*
140+ * Identify the index tuple about to be deleted
141+ */
142+ iitemid = PageGetItemId (ipage , todelete [i ]);
143+ itup = (IndexTuple ) PageGetItem (ipage , iitemid );
144+
145+ /*
146+ * Locate the heap page that the index tuple points at
147+ */
148+ hblkno = ItemPointerGetBlockNumber (& (itup -> t_tid ));
149+ hbuffer = XLogReadBufferExtended (* hnode , MAIN_FORKNUM , hblkno , RBM_NORMAL );
150+ if (!BufferIsValid (hbuffer ))
151+ {
152+ UnlockReleaseBuffer (ibuffer );
153+ return InvalidTransactionId ;
154+ }
155+ LockBuffer (hbuffer , BUFFER_LOCK_SHARE );
156+ hpage = (Page ) BufferGetPage (hbuffer );
157+
158+ /*
159+ * Look up the heap tuple header that the index tuple points at by
160+ * using the heap node supplied with the xlrec. We can't use
161+ * heap_fetch, since it uses ReadBuffer rather than XLogReadBuffer.
162+ * Note that we are not looking at tuple data here, just headers.
163+ */
164+ hoffnum = ItemPointerGetOffsetNumber (& (itup -> t_tid ));
165+ hitemid = PageGetItemId (hpage , hoffnum );
166+
167+ /*
168+ * Follow any redirections until we find something useful.
169+ */
170+ while (ItemIdIsRedirected (hitemid ))
171+ {
172+ hoffnum = ItemIdGetRedirect (hitemid );
173+ hitemid = PageGetItemId (hpage , hoffnum );
174+ CHECK_FOR_INTERRUPTS ();
175+ }
176+
177+ /*
178+ * If the heap item has storage, then read the header and use that to
179+ * set latestRemovedXid.
180+ *
181+ * Some LP_DEAD items may not be accessible, so we ignore them.
182+ */
183+ if (ItemIdHasStorage (hitemid ))
184+ {
185+ htuphdr = (HeapTupleHeader ) PageGetItem (hpage , hitemid );
186+
187+ HeapTupleHeaderAdvanceLatestRemovedXid (htuphdr , & latestRemovedXid );
188+ }
189+ else if (ItemIdIsDead (hitemid ))
190+ {
191+ /*
192+ * Conjecture: if hitemid is dead then it had xids before the xids
193+ * marked on LP_NORMAL items. So we just ignore this item and move
194+ * onto the next, for the purposes of calculating
195+ * latestRemovedxids.
196+ */
197+ }
198+ else
199+ Assert (!ItemIdIsUsed (hitemid ));
200+
201+ UnlockReleaseBuffer (hbuffer );
202+ }
203+
204+ UnlockReleaseBuffer (ibuffer );
205+
206+ /*
207+ * If all heap tuples were LP_DEAD then we will be returning
208+ * InvalidTransactionId here, which avoids conflicts. This matches
209+ * existing logic which assumes that LP_DEAD tuples must already be older
210+ * than the latestRemovedXid on the cleanup record that set them as
211+ * LP_DEAD, hence must already have generated a conflict.
212+ */
213+ return latestRemovedXid ;
214+ }
215+
63216/*
64217 * redo any page update (except page split)
65218 */
@@ -71,6 +224,34 @@ gistRedoPageUpdateRecord(XLogReaderState *record)
71224 Buffer buffer ;
72225 Page page ;
73226
227+ /*
228+ * If we have any conflict processing to do, it must happen before we
229+ * update the page.
230+ *
231+ * Support for conflict processing in GiST has been backpatched. This is
232+ * why we have to use tricky way of saving WAL-compatibility between minor
233+ * versions. Information required for conflict processing is just
234+ * appended to data of XLOG_GIST_PAGE_UPDATE record. So, PostgreSQL
235+ * version, which doesn't know about conflict processing, will just ignore
236+ * that.
237+ *
238+ * GiST delete records can conflict with standby queries. You might think
239+ * that vacuum records would conflict as well, but we've handled that
240+ * already. XLOG_HEAP2_CLEANUP_INFO records provide the highest xid
241+ * cleaned by the vacuum of the heap and so we can resolve any conflicts
242+ * just once when that arrives. After that we know that no conflicts
243+ * exist from individual gist vacuum records on that index.
244+ */
245+ if (InHotStandby && XLogRecGetDataLen (record ) > sizeof (gistxlogPageUpdate ))
246+ {
247+ TransactionId latestRemovedXid = gistRedoPageUpdateRecordGetLatestRemovedXid (record );
248+ RelFileNode rnode ;
249+
250+ XLogRecGetBlockTag (record , 0 , & rnode , NULL , NULL );
251+
252+ ResolveRecoveryConflictWithSnapshot (latestRemovedXid , rnode );
253+ }
254+
74255 if (XLogReadBufferForRedo (record , 0 , & buffer ) == BLK_NEEDS_REDO )
75256 {
76257 char * begin ;
@@ -457,7 +638,7 @@ XLogRecPtr
457638gistXLogUpdate (Buffer buffer ,
458639 OffsetNumber * todelete , int ntodelete ,
459640 IndexTuple * itup , int ituplen ,
460- Buffer leftchildbuf )
641+ Buffer leftchildbuf , RelFileNode * hnode )
461642{
462643 gistxlogPageUpdate xlrec ;
463644 int i ;
@@ -469,6 +650,16 @@ gistXLogUpdate(Buffer buffer,
469650 XLogBeginInsert ();
470651 XLogRegisterData ((char * ) & xlrec , sizeof (gistxlogPageUpdate ));
471652
653+ /*
654+ * Append the information required for standby conflict processing if it
655+ * is provided by caller.
656+ */
657+ if (hnode )
658+ {
659+ XLogRegisterData ((char * ) hnode , sizeof (RelFileNode ));
660+ XLogRegisterData ((char * ) todelete , sizeof (OffsetNumber ) * ntodelete );
661+ }
662+
472663 XLogRegisterBuffer (0 , buffer , REGBUF_STANDARD );
473664 XLogRegisterBufData (0 , (char * ) todelete , sizeof (OffsetNumber ) * ntodelete );
474665
0 commit comments