PostgreSQL Source Code  git master
hashinsert.c
Go to the documentation of this file.
1 /*-------------------------------------------------------------------------
2  *
3  * hashinsert.c
4  * Item insertion in hash tables for Postgres.
5  *
6  * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
7  * Portions Copyright (c) 1994, Regents of the University of California
8  *
9  *
10  * IDENTIFICATION
11  * src/backend/access/hash/hashinsert.c
12  *
13  *-------------------------------------------------------------------------
14  */
15 
16 #include "postgres.h"
17 
18 #include "access/hash.h"
19 #include "access/hash_xlog.h"
20 #include "access/xloginsert.h"
21 #include "miscadmin.h"
22 #include "storage/buf_internals.h"
23 #include "storage/lwlock.h"
24 #include "storage/predicate.h"
25 #include "utils/rel.h"
26 
27 static void _hash_vacuum_one_page(Relation rel, Relation hrel,
28  Buffer metabuf, Buffer buf);
29 
30 /*
31  * _hash_doinsert() -- Handle insertion of a single index tuple.
32  *
33  * This routine is called by the public interface routines, hashbuild
34  * and hashinsert. By here, itup is completely filled in.
35  *
36  * 'sorted' must only be passed as 'true' when inserts are done in hashkey
37  * order.
38  */
39 void
40 _hash_doinsert(Relation rel, IndexTuple itup, Relation heapRel, bool sorted)
41 {
43  Buffer bucket_buf;
44  Buffer metabuf;
45  HashMetaPage metap;
46  HashMetaPage usedmetap = NULL;
47  Page metapage;
48  Page page;
49  HashPageOpaque pageopaque;
50  Size itemsz;
51  bool do_expand;
52  uint32 hashkey;
53  Bucket bucket;
54  OffsetNumber itup_off;
55 
56  /*
57  * Get the hash key for the item (it's stored in the index tuple itself).
58  */
59  hashkey = _hash_get_indextuple_hashkey(itup);
60 
61  /* compute item size too */
62  itemsz = IndexTupleSize(itup);
63  itemsz = MAXALIGN(itemsz); /* be safe, PageAddItem will do this but we
64  * need to be consistent */
65 
66 restart_insert:
67 
68  /*
69  * Read the metapage. We don't lock it yet; HashMaxItemSize() will
70  * examine pd_pagesize_version, but that can't change so we can examine it
71  * without a lock.
72  */
74  metapage = BufferGetPage(metabuf);
75 
76  /*
77  * Check whether the item can fit on a hash page at all. (Eventually, we
78  * ought to try to apply TOAST methods if not.) Note that at this point,
79  * itemsz doesn't include the ItemId.
80  *
81  * XXX this is useless code if we are only storing hash keys.
82  */
83  if (itemsz > HashMaxItemSize(metapage))
84  ereport(ERROR,
85  (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
86  errmsg("index row size %zu exceeds hash maximum %zu",
87  itemsz, HashMaxItemSize(metapage)),
88  errhint("Values larger than a buffer page cannot be indexed.")));
89 
90  /* Lock the primary bucket page for the target bucket. */
92  &usedmetap);
93  Assert(usedmetap != NULL);
94 
96 
97  /* remember the primary bucket buffer to release the pin on it at end. */
98  bucket_buf = buf;
99 
100  page = BufferGetPage(buf);
101  pageopaque = HashPageGetOpaque(page);
102  bucket = pageopaque->hasho_bucket;
103 
104  /*
105  * If this bucket is in the process of being split, try to finish the
106  * split before inserting, because that might create room for the
107  * insertion to proceed without allocating an additional overflow page.
108  * It's only interesting to finish the split if we're trying to insert
109  * into the bucket from which we're removing tuples (the "old" bucket),
110  * not if we're trying to insert into the bucket into which tuples are
111  * being moved (the "new" bucket).
112  */
113  if (H_BUCKET_BEING_SPLIT(pageopaque) && IsBufferCleanupOK(buf))
114  {
115  /* release the lock on bucket buffer, before completing the split. */
117 
118  _hash_finish_split(rel, metabuf, buf, bucket,
119  usedmetap->hashm_maxbucket,
120  usedmetap->hashm_highmask,
121  usedmetap->hashm_lowmask);
122 
123  /* release the pin on old and meta buffer. retry for insert. */
124  _hash_dropbuf(rel, buf);
125  _hash_dropbuf(rel, metabuf);
126  goto restart_insert;
127  }
128 
129  /* Do the insertion */
130  while (PageGetFreeSpace(page) < itemsz)
131  {
132  BlockNumber nextblkno;
133 
134  /*
135  * Check if current page has any DEAD tuples. If yes, delete these
136  * tuples and see if we can get a space for the new item to be
137  * inserted before moving to the next page in the bucket chain.
138  */
139  if (H_HAS_DEAD_TUPLES(pageopaque))
140  {
141 
142  if (IsBufferCleanupOK(buf))
143  {
144  _hash_vacuum_one_page(rel, heapRel, metabuf, buf);
145 
146  if (PageGetFreeSpace(page) >= itemsz)
147  break; /* OK, now we have enough space */
148  }
149  }
150 
151  /*
152  * no space on this page; check for an overflow page
153  */
154  nextblkno = pageopaque->hasho_nextblkno;
155 
156  if (BlockNumberIsValid(nextblkno))
157  {
158  /*
159  * ovfl page exists; go get it. if it doesn't have room, we'll
160  * find out next pass through the loop test above. we always
161  * release both the lock and pin if this is an overflow page, but
162  * only the lock if this is the primary bucket page, since the pin
163  * on the primary bucket must be retained throughout the scan.
164  */
165  if (buf != bucket_buf)
166  _hash_relbuf(rel, buf);
167  else
169  buf = _hash_getbuf(rel, nextblkno, HASH_WRITE, LH_OVERFLOW_PAGE);
170  page = BufferGetPage(buf);
171  }
172  else
173  {
174  /*
175  * we're at the end of the bucket chain and we haven't found a
176  * page with enough room. allocate a new overflow page.
177  */
178 
179  /* release our write lock without modifying buffer */
181 
182  /* chain to a new overflow page */
183  buf = _hash_addovflpage(rel, metabuf, buf, (buf == bucket_buf));
184  page = BufferGetPage(buf);
185 
186  /* should fit now, given test above */
187  Assert(PageGetFreeSpace(page) >= itemsz);
188  }
189  pageopaque = HashPageGetOpaque(page);
190  Assert((pageopaque->hasho_flag & LH_PAGE_TYPE) == LH_OVERFLOW_PAGE);
191  Assert(pageopaque->hasho_bucket == bucket);
192  }
193 
194  /*
195  * Write-lock the metapage so we can increment the tuple count. After
196  * incrementing it, check to see if it's time for a split.
197  */
199 
200  /* Do the update. No ereport(ERROR) until changes are logged */
202 
203  /* found page with enough space, so add the item here */
204  itup_off = _hash_pgaddtup(rel, buf, itemsz, itup, sorted);
206 
207  /* metapage operations */
208  metap = HashPageGetMeta(metapage);
209  metap->hashm_ntuples += 1;
210 
211  /* Make sure this stays in sync with _hash_expandtable() */
212  do_expand = metap->hashm_ntuples >
213  (double) metap->hashm_ffactor * (metap->hashm_maxbucket + 1);
214 
215  MarkBufferDirty(metabuf);
216 
217  /* XLOG stuff */
218  if (RelationNeedsWAL(rel))
219  {
220  xl_hash_insert xlrec;
221  XLogRecPtr recptr;
222 
223  xlrec.offnum = itup_off;
224 
225  XLogBeginInsert();
226  XLogRegisterData((char *) &xlrec, SizeOfHashInsert);
227 
228  XLogRegisterBuffer(1, metabuf, REGBUF_STANDARD);
229 
231  XLogRegisterBufData(0, (char *) itup, IndexTupleSize(itup));
232 
233  recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_INSERT);
234 
235  PageSetLSN(BufferGetPage(buf), recptr);
236  PageSetLSN(BufferGetPage(metabuf), recptr);
237  }
238 
240 
241  /* drop lock on metapage, but keep pin */
242  LockBuffer(metabuf, BUFFER_LOCK_UNLOCK);
243 
244  /*
245  * Release the modified page and ensure to release the pin on primary
246  * page.
247  */
248  _hash_relbuf(rel, buf);
249  if (buf != bucket_buf)
250  _hash_dropbuf(rel, bucket_buf);
251 
252  /* Attempt to split if a split is needed */
253  if (do_expand)
254  _hash_expandtable(rel, metabuf);
255 
256  /* Finally drop our pin on the metapage */
257  _hash_dropbuf(rel, metabuf);
258 }
259 
260 /*
261  * _hash_pgaddtup() -- add a tuple to a particular page in the index.
262  *
263  * This routine adds the tuple to the page as requested; it does not write out
264  * the page. It is an error to call this function without pin and write lock
265  * on the target buffer.
266  *
267  * Returns the offset number at which the tuple was inserted. This function
268  * is responsible for preserving the condition that tuples in a hash index
269  * page are sorted by hashkey value, however, if the caller is certain that
270  * the hashkey for the tuple being added is >= the hashkeys of all existing
271  * tuples on the page, then the 'appendtup' flag may be passed as true. This
272  * saves from having to binary search for the correct location to insert the
273  * tuple.
274  */
277  bool appendtup)
278 {
279  OffsetNumber itup_off;
280  Page page;
281 
283  page = BufferGetPage(buf);
284 
285  /*
286  * Find where to insert the tuple (preserving page's hashkey ordering). If
287  * 'appendtup' is true then we just insert it at the end.
288  */
289  if (appendtup)
290  {
291  itup_off = PageGetMaxOffsetNumber(page) + 1;
292 
293 #ifdef USE_ASSERT_CHECKING
294  /* ensure this tuple's hashkey is >= the final existing tuple */
295  if (PageGetMaxOffsetNumber(page) > 0)
296  {
297  IndexTuple lasttup;
298  ItemId itemid;
299 
300  itemid = PageGetItemId(page, PageGetMaxOffsetNumber(page));
301  lasttup = (IndexTuple) PageGetItem(page, itemid);
302 
305  }
306 #endif
307  }
308  else
309  {
310  uint32 hashkey = _hash_get_indextuple_hashkey(itup);
311 
312  itup_off = _hash_binsearch(page, hashkey);
313  }
314 
315  if (PageAddItem(page, (Item) itup, itemsize, itup_off, false, false)
317  elog(ERROR, "failed to add index item to \"%s\"",
319 
320  return itup_off;
321 }
322 
323 /*
324  * _hash_pgaddmultitup() -- add a tuple vector to a particular page in the
325  * index.
326  *
327  * This routine has same requirements for locking and tuple ordering as
328  * _hash_pgaddtup().
329  *
330  * Returns the offset number array at which the tuples were inserted.
331  */
332 void
334  OffsetNumber *itup_offsets, uint16 nitups)
335 {
336  OffsetNumber itup_off;
337  Page page;
338  uint32 hashkey;
339  int i;
340 
342  page = BufferGetPage(buf);
343 
344  for (i = 0; i < nitups; i++)
345  {
346  Size itemsize;
347 
348  itemsize = IndexTupleSize(itups[i]);
349  itemsize = MAXALIGN(itemsize);
350 
351  /* Find where to insert the tuple (preserving page's hashkey ordering) */
352  hashkey = _hash_get_indextuple_hashkey(itups[i]);
353  itup_off = _hash_binsearch(page, hashkey);
354 
355  itup_offsets[i] = itup_off;
356 
357  if (PageAddItem(page, (Item) itups[i], itemsize, itup_off, false, false)
359  elog(ERROR, "failed to add index item to \"%s\"",
361  }
362 }
363 
364 /*
365  * _hash_vacuum_one_page - vacuum just one index page.
366  *
367  * Try to remove LP_DEAD items from the given page. We must acquire cleanup
368  * lock on the page being modified before calling this function.
369  */
370 
371 static void
373 {
374  OffsetNumber deletable[MaxOffsetNumber];
375  int ndeletable = 0;
376  OffsetNumber offnum,
377  maxoff;
378  Page page = BufferGetPage(buf);
379  HashPageOpaque pageopaque;
380  HashMetaPage metap;
381 
382  /* Scan each tuple in page to see if it is marked as LP_DEAD */
383  maxoff = PageGetMaxOffsetNumber(page);
384  for (offnum = FirstOffsetNumber;
385  offnum <= maxoff;
386  offnum = OffsetNumberNext(offnum))
387  {
388  ItemId itemId = PageGetItemId(page, offnum);
389 
390  if (ItemIdIsDead(itemId))
391  deletable[ndeletable++] = offnum;
392  }
393 
394  if (ndeletable > 0)
395  {
396  TransactionId snapshotConflictHorizon;
397 
398  snapshotConflictHorizon =
400  deletable, ndeletable);
401 
402  /*
403  * Write-lock the meta page so that we can decrement tuple count.
404  */
406 
407  /* No ereport(ERROR) until changes are logged */
409 
410  PageIndexMultiDelete(page, deletable, ndeletable);
411 
412  /*
413  * Mark the page as not containing any LP_DEAD items. This is not
414  * certainly true (there might be some that have recently been marked,
415  * but weren't included in our target-item list), but it will almost
416  * always be true and it doesn't seem worth an additional page scan to
417  * check it. Remember that LH_PAGE_HAS_DEAD_TUPLES is only a hint
418  * anyway.
419  */
420  pageopaque = HashPageGetOpaque(page);
421  pageopaque->hasho_flag &= ~LH_PAGE_HAS_DEAD_TUPLES;
422 
423  metap = HashPageGetMeta(BufferGetPage(metabuf));
424  metap->hashm_ntuples -= ndeletable;
425 
427  MarkBufferDirty(metabuf);
428 
429  /* XLOG stuff */
430  if (RelationNeedsWAL(rel))
431  {
433  XLogRecPtr recptr;
434 
435  xlrec.snapshotConflictHorizon = snapshotConflictHorizon;
436  xlrec.ntuples = ndeletable;
437 
438  XLogBeginInsert();
440  XLogRegisterData((char *) &xlrec, SizeOfHashVacuumOnePage);
441 
442  /*
443  * We need the target-offsets array whether or not we store the
444  * whole buffer, to allow us to find the snapshotConflictHorizon
445  * on a standby server.
446  */
447  XLogRegisterData((char *) deletable,
448  ndeletable * sizeof(OffsetNumber));
449 
450  XLogRegisterBuffer(1, metabuf, REGBUF_STANDARD);
451 
452  recptr = XLogInsert(RM_HASH_ID, XLOG_HASH_VACUUM_ONE_PAGE);
453 
454  PageSetLSN(BufferGetPage(buf), recptr);
455  PageSetLSN(BufferGetPage(metabuf), recptr);
456  }
457 
459 
460  /*
461  * Releasing write lock on meta page as we have updated the tuple
462  * count.
463  */
464  LockBuffer(metabuf, BUFFER_LOCK_UNLOCK);
465  }
466 }
uint32 BlockNumber
Definition: block.h:31
static bool BlockNumberIsValid(BlockNumber blockNumber)
Definition: block.h:71
int Buffer
Definition: buf.h:23
#define InvalidBuffer
Definition: buf.h:25
BlockNumber BufferGetBlockNumber(Buffer buffer)
Definition: bufmgr.c:2763
bool IsBufferCleanupOK(Buffer buffer)
Definition: bufmgr.c:4462
void MarkBufferDirty(Buffer buffer)
Definition: bufmgr.c:1583
void LockBuffer(Buffer buffer, int mode)
Definition: bufmgr.c:4172
#define BUFFER_LOCK_UNLOCK
Definition: bufmgr.h:105
static Page BufferGetPage(Buffer buffer)
Definition: bufmgr.h:280
#define BUFFER_LOCK_EXCLUSIVE
Definition: bufmgr.h:107
void PageIndexMultiDelete(Page page, OffsetNumber *itemnos, int nitems)
Definition: bufpage.c:1161
Size PageGetFreeSpace(Page page)
Definition: bufpage.c:907
Pointer Page
Definition: bufpage.h:78
static Item PageGetItem(Page page, ItemId itemId)
Definition: bufpage.h:351
static ItemId PageGetItemId(Page page, OffsetNumber offsetNumber)
Definition: bufpage.h:240
static void PageSetLSN(Page page, XLogRecPtr lsn)
Definition: bufpage.h:388
static OffsetNumber PageGetMaxOffsetNumber(Page page)
Definition: bufpage.h:369
#define PageAddItem(page, item, size, offsetNumber, overwrite, is_heap)
Definition: bufpage.h:468
unsigned short uint16
Definition: c.h:441
unsigned int uint32
Definition: c.h:442
#define MAXALIGN(LEN)
Definition: c.h:747
uint32 TransactionId
Definition: c.h:588
size_t Size
Definition: c.h:541
int errhint(const char *fmt,...)
Definition: elog.c:1153
int errcode(int sqlerrcode)
Definition: elog.c:695
int errmsg(const char *fmt,...)
Definition: elog.c:906
#define ERROR
Definition: elog.h:35
#define ereport(elevel,...)
Definition: elog.h:145
TransactionId index_compute_xid_horizon_for_tuples(Relation irel, Relation hrel, Buffer ibuf, OffsetNumber *itemnos, int nitems)
Definition: genam.c:294
#define HASH_NOLOCK
Definition: hash.h:341
#define HashPageGetOpaque(page)
Definition: hash.h:88
#define LH_BUCKET_PAGE
Definition: hash.h:55
#define HASH_WRITE
Definition: hash.h:340
#define H_BUCKET_BEING_SPLIT(opaque)
Definition: hash.h:91
#define LH_META_PAGE
Definition: hash.h:57
#define HashPageGetMeta(page)
Definition: hash.h:323
#define HASH_METAPAGE
Definition: hash.h:198
#define H_HAS_DEAD_TUPLES(opaque)
Definition: hash.h:93
#define LH_PAGE_TYPE
Definition: hash.h:63
uint32 Bucket
Definition: hash.h:35
#define HashMaxItemSize(page)
Definition: hash.h:287
#define LH_PAGE_HAS_DEAD_TUPLES
Definition: hash.h:61
#define LH_OVERFLOW_PAGE
Definition: hash.h:54
#define SizeOfHashInsert
Definition: hash_xlog.h:61
#define XLOG_HASH_INSERT
Definition: hash_xlog.h:29
#define XLOG_HASH_VACUUM_ONE_PAGE
Definition: hash_xlog.h:40
#define SizeOfHashVacuumOnePage
Definition: hash_xlog.h:253
void _hash_doinsert(Relation rel, IndexTuple itup, Relation heapRel, bool sorted)
Definition: hashinsert.c:40
OffsetNumber _hash_pgaddtup(Relation rel, Buffer buf, Size itemsize, IndexTuple itup, bool appendtup)
Definition: hashinsert.c:276
void _hash_pgaddmultitup(Relation rel, Buffer buf, IndexTuple *itups, OffsetNumber *itup_offsets, uint16 nitups)
Definition: hashinsert.c:333
static void _hash_vacuum_one_page(Relation rel, Relation hrel, Buffer metabuf, Buffer buf)
Definition: hashinsert.c:372
Buffer _hash_addovflpage(Relation rel, Buffer metabuf, Buffer buf, bool retain_pin)
Definition: hashovfl.c:112
void _hash_relbuf(Relation rel, Buffer buf)
Definition: hashpage.c:266
void _hash_dropbuf(Relation rel, Buffer buf)
Definition: hashpage.c:277
Buffer _hash_getbuf(Relation rel, BlockNumber blkno, int access, int flags)
Definition: hashpage.c:70
void _hash_finish_split(Relation rel, Buffer metabuf, Buffer obuf, Bucket obucket, uint32 maxbucket, uint32 highmask, uint32 lowmask)
Definition: hashpage.c:1356
Buffer _hash_getbucketbuf_from_hashkey(Relation rel, uint32 hashkey, int access, HashMetaPage *cachedmetap)
Definition: hashpage.c:1559
void _hash_expandtable(Relation rel, Buffer metabuf)
Definition: hashpage.c:614
uint32 _hash_get_indextuple_hashkey(IndexTuple itup)
Definition: hashutil.c:292
OffsetNumber _hash_binsearch(Page page, uint32 hash_value)
Definition: hashutil.c:351
void _hash_checkpage(Relation rel, Buffer buf, int flags)
Definition: hashutil.c:211
int i
Definition: isn.c:73
Pointer Item
Definition: item.h:17
#define ItemIdIsDead(itemId)
Definition: itemid.h:113
IndexTupleData * IndexTuple
Definition: itup.h:53
#define IndexTupleSize(itup)
Definition: itup.h:70
Assert(fmt[strlen(fmt) - 1] !='\n')
#define START_CRIT_SECTION()
Definition: miscadmin.h:148
#define END_CRIT_SECTION()
Definition: miscadmin.h:150
#define InvalidOffsetNumber
Definition: off.h:26
#define OffsetNumberNext(offsetNumber)
Definition: off.h:52
uint16 OffsetNumber
Definition: off.h:24
#define FirstOffsetNumber
Definition: off.h:27
#define MaxOffsetNumber
Definition: off.h:28
static char * buf
Definition: pg_test_fsync.c:67
void CheckForSerializableConflictIn(Relation relation, ItemPointer tid, BlockNumber blkno)
Definition: predicate.c:4441
#define RelationGetRelationName(relation)
Definition: rel.h:535
#define RelationNeedsWAL(relation)
Definition: rel.h:626
uint32 hashm_lowmask
Definition: hash.h:256
uint32 hashm_maxbucket
Definition: hash.h:254
double hashm_ntuples
Definition: hash.h:248
uint32 hashm_highmask
Definition: hash.h:255
uint16 hashm_ffactor
Definition: hash.h:249
BlockNumber hasho_nextblkno
Definition: hash.h:80
uint16 hasho_flag
Definition: hash.h:82
Bucket hasho_bucket
Definition: hash.h:81
OffsetNumber offnum
Definition: hash_xlog.h:58
TransactionId snapshotConflictHorizon
Definition: hash_xlog.h:247
uint64 XLogRecPtr
Definition: xlogdefs.h:21
void XLogRegisterData(char *data, uint32 len)
Definition: xloginsert.c:351
XLogRecPtr XLogInsert(RmgrId rmid, uint8 info)
Definition: xloginsert.c:451
void XLogRegisterBufData(uint8 block_id, char *data, uint32 len)
Definition: xloginsert.c:389
void XLogRegisterBuffer(uint8 block_id, Buffer buffer, uint8 flags)
Definition: xloginsert.c:243
void XLogBeginInsert(void)
Definition: xloginsert.c:150
#define REGBUF_STANDARD
Definition: xloginsert.h:34