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