granicus.if.org Git - postgresql/blob - src/backend/access/heap/hio.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * hio.c--
   4  *    POSTGRES heap access method input/output code.
   5  *
   6  * Copyright (c) 1994, Regents of the University of California
   7  *
   8  *
   9  * IDENTIFICATION
  10  *    $Id: hio.c,v 1.3 1996/10/20 06:56:02 scrappy Exp $
  11  *
  12  *-------------------------------------------------------------------------
  13  */
  14
  15 #include "postgres.h"
  16 #include "utils/rel.h"
  17 #include "access/htup.h"
  18 #include "storage/buf.h"
  19 #include "storage/bufpage.h"
  20 #include "access/relscan.h"
  21 #include "access/heapam.h"
  22 #include "storage/bufmgr.h"
  23
  24 /*
  25  * amputunique  - place tuple at tid
  26  *   Currently on errors, calls elog.  Perhaps should return -1?
  27  *   Possible errors include the addition of a tuple to the page
  28  *   between the time the linep is chosen and the page is L_UP'd.
  29  *
  30  *   This should be coordinated with the B-tree code.
  31  *   Probably needs to have an amdelunique to allow for
  32  *   internal index records to be deleted and reordered as needed.
  33  *   For the heap AM, this should never be needed.
  34  */
  35 void
  36 RelationPutHeapTuple(Relation relation,
  37                      BlockNumber blockIndex,
  38                      HeapTuple tuple)
  39 {
  40     Buffer              buffer;
  41     Page                pageHeader;
  42     BlockNumber         numberOfBlocks;
  43     OffsetNumber        offnum;
  44     unsigned int        len;
  45     ItemId              itemId;
  46     Item                item;
  47
  48     /* ----------------
  49      *  increment access statistics
  50      * ----------------
  51      */
  52     IncrHeapAccessStat(local_RelationPutHeapTuple);
  53     IncrHeapAccessStat(global_RelationPutHeapTuple);
  54
  55     Assert(RelationIsValid(relation));
  56     Assert(HeapTupleIsValid(tuple));
  57
  58     numberOfBlocks = RelationGetNumberOfBlocks(relation);
  59     Assert(blockIndex < numberOfBlocks);
  60
  61     buffer = ReadBuffer(relation, blockIndex);
  62 #ifndef NO_BUFFERISVALID
  63     if (!BufferIsValid(buffer)) {
  64         elog(WARN, "RelationPutHeapTuple: no buffer for %ld in %s",
  65              blockIndex, &relation->rd_rel->relname);
  66     }
  67 #endif
  68
  69     pageHeader = (Page)BufferGetPage(buffer);
  70     len = (unsigned)DOUBLEALIGN(tuple->t_len);  /* be conservative */
  71     Assert((int)len <= PageGetFreeSpace(pageHeader));
  72
  73     offnum = PageAddItem((Page)pageHeader, (Item)tuple,
  74                          tuple->t_len, InvalidOffsetNumber, LP_USED);
  75
  76     itemId = PageGetItemId((Page)pageHeader, offnum);
  77     item = PageGetItem((Page)pageHeader, itemId);
  78
  79     ItemPointerSet(&((HeapTuple)item)->t_ctid, blockIndex, offnum);
  80
  81     WriteBuffer(buffer);
  82     /* return an accurate tuple */
  83     ItemPointerSet(&tuple->t_ctid, blockIndex, offnum);
  84 }
  85
  86 /*
  87  * This routine is another in the series of attempts to reduce the number
  88  * of I/O's and system calls executed in the various benchmarks.  In
  89  * particular, this routine is used to append data to the end of a relation
  90  * file without excessive lseeks.  This code should do no more than 2 semops
  91  * in the ideal case.
  92  *
  93  * Eventually, we should cache the number of blocks in a relation somewhere.
  94  * Until that time, this code will have to do an lseek to determine the number
  95  * of blocks in a relation.
  96  *
  97  * This code should ideally do at most 4 semops, 1 lseek, and possibly 1 write
  98  * to do an append; it's possible to eliminate 2 of the semops if we do direct
  99  * buffer stuff (!); the lseek and the write can go if we get
 100  * RelationGetNumberOfBlocks to be useful.
 101  *
 102  * NOTE: This code presumes that we have a write lock on the relation.
 103  *
 104  * Also note that this routine probably shouldn't have to exist, and does
 105  * screw up the call graph rather badly, but we are wasting so much time and
 106  * system resources being massively general that we are losing badly in our
 107  * performance benchmarks.
 108  */
 109 void
 110 RelationPutHeapTupleAtEnd(Relation relation, HeapTuple tuple)
 111 {
 112     Buffer              buffer;
 113     Page                pageHeader;
 114     BlockNumber         lastblock;
 115     OffsetNumber        offnum;
 116     unsigned int        len;
 117     ItemId              itemId;
 118     Item                item;
 119
 120     Assert(RelationIsValid(relation));
 121     Assert(HeapTupleIsValid(tuple));
 122
 123     /*
 124      * XXX This does an lseek - VERY expensive - but at the moment it
 125      * is the only way to accurately determine how many blocks are in
 126      * a relation.  A good optimization would be to get this to actually
 127      * work properly.
 128      */
 129
 130     lastblock = RelationGetNumberOfBlocks(relation);
 131
 132     if (lastblock == 0)
 133         {
 134             buffer = ReadBuffer(relation, lastblock);
 135             pageHeader = (Page)BufferGetPage(buffer);
 136             if (PageIsNew((PageHeader) pageHeader))
 137                 {
 138                     buffer = ReleaseAndReadBuffer(buffer, relation, P_NEW);
 139                     pageHeader = (Page)BufferGetPage(buffer);
 140                     PageInit(pageHeader, BufferGetPageSize(buffer), 0);
 141                 }
 142         }
 143     else
 144         buffer = ReadBuffer(relation, lastblock - 1);
 145
 146     pageHeader = (Page)BufferGetPage(buffer);
 147     len = (unsigned)DOUBLEALIGN(tuple->t_len);  /* be conservative */
 148
 149     /*
 150      * Note that this is true if the above returned a bogus page, which
 151      * it will do for a completely empty relation.
 152      */
 153
 154     if (len > PageGetFreeSpace(pageHeader))
 155         {
 156             buffer = ReleaseAndReadBuffer(buffer, relation, P_NEW);
 157             pageHeader = (Page)BufferGetPage(buffer);
 158             PageInit(pageHeader, BufferGetPageSize(buffer), 0);
 159
 160             if (len > PageGetFreeSpace(pageHeader))
 161                 elog(WARN, "Tuple is too big: size %d", len);
 162         }
 163
 164     offnum = PageAddItem((Page)pageHeader, (Item)tuple,
 165                          tuple->t_len, InvalidOffsetNumber, LP_USED);
 166
 167     itemId = PageGetItemId((Page)pageHeader, offnum);
 168     item = PageGetItem((Page)pageHeader, itemId);
 169
 170     lastblock = BufferGetBlockNumber(buffer);
 171
 172     ItemPointerSet(&((HeapTuple)item)->t_ctid, lastblock, offnum);
 173
 174     /* return an accurate tuple */
 175     ItemPointerSet(&tuple->t_ctid, lastblock, offnum);
 176
 177     WriteBuffer(buffer);
 178 }