Thread: vacuum.c refactoring
. rename variables
. cur_buffer -> dst_buffer
. ToPage -> dst_page
. cur_page -> dst_vacpage
. move variable declarations into block where variable is used
. various Asserts instead of elog(ERROR, ...)
. extract functionality from repair_frag() into new routines
. move_chain_tuple()
. move_plain_tuple()
. update_hint_bits()
. create type ExecContext
. add comments
This patch does not intend to change any behaviour. It passes make
check, make installcheck and some manual tests. It might be hard to
review, because some lines are affected by more than one change. If
it's too much to swallow at once, I can provide it in smaller chunks ...
Servus
Manfred
diff -Ncr ../base/src/backend/commands/vacuum.c src/backend/commands/vacuum.c
*** ../base/src/backend/commands/vacuum.c Mon May 31 21:24:05 2004
--- src/backend/commands/vacuum.c Wed Jun 2 21:46:59 2004
***************
*** 99,104 ****
--- 99,162 ----
VTupleLink vtlinks;
} VRelStats;
+ /*----------------------------------------------------------------------
+ * ExecContext:
+ *
+ * As these variables always appear together, we put them into one struct
+ * and pull initialization and cleanup into separate routines.
+ * ExecContext is used by repair_frag() and move_xxx_tuple(). More
+ * accurately: It is *used* only in move_xxx_tuple(), but because this
+ * routine is called many times, we initialize the struct just once in
+ * repair_frag() and pass it on to move_xxx_tuple().
+ */
+ typedef struct ExecContextData
+ {
+ ResultRelInfo *resultRelInfo;
+ EState *estate;
+ TupleTable tupleTable;
+ TupleTableSlot *slot;
+ } ExecContextData;
+ typedef ExecContextData *ExecContext;
+
+ static void
+ ExecContext_Init(ExecContext ec, Relation rel)
+ {
+ TupleDesc tupdesc = RelationGetDescr(rel);
+
+ /*
+ * We need a ResultRelInfo and an EState so we can use the regular
+ * executor's index-entry-making machinery.
+ */
+ ec->estate = CreateExecutorState();
+
+ ec->resultRelInfo = makeNode(ResultRelInfo);
+ ec->resultRelInfo->ri_RangeTableIndex = 1; /* dummy */
+ ec->resultRelInfo->ri_RelationDesc = rel;
+ ec->resultRelInfo->ri_TrigDesc = NULL; /* we don't fire triggers */
+
+ ExecOpenIndices(ec->resultRelInfo);
+
+ ec->estate->es_result_relations = ec->resultRelInfo;
+ ec->estate->es_num_result_relations = 1;
+ ec->estate->es_result_relation_info = ec->resultRelInfo;
+
+ /* Set up a dummy tuple table too */
+ ec->tupleTable = ExecCreateTupleTable(1);
+ ec->slot = ExecAllocTableSlot(ec->tupleTable);
+ ExecSetSlotDescriptor(ec->slot, tupdesc, false);
+ }
+
+ static void
+ ExecContext_Finish(ExecContext ec)
+ {
+ ExecDropTupleTable(ec->tupleTable, true);
+ ExecCloseIndices(ec->resultRelInfo);
+ FreeExecutorState(ec->estate);
+ }
+ /*
+ * End of ExecContext Implementation
+ *----------------------------------------------------------------------
+ */
static MemoryContext vac_context = NULL;
***************
*** 122,127 ****
--- 180,196 ----
static void repair_frag(VRelStats *vacrelstats, Relation onerel,
VacPageList vacuum_pages, VacPageList fraged_pages,
int nindexes, Relation *Irel);
+ static void move_chain_tuple(Relation rel,
+ Buffer old_buf, Page old_page, HeapTuple old_tup,
+ Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
+ ExecContext ec, ItemPointer ctid, bool cleanVpd);
+ static void move_plain_tuple(Relation rel,
+ Buffer old_buf, Page old_page, HeapTuple old_tup,
+ Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
+ ExecContext ec);
+ static void update_hint_bits(Relation rel, VacPageList fraged_pages,
+ int num_fraged_pages, BlockNumber last_move_dest_block,
+ int num_moved);
static void vacuum_heap(VRelStats *vacrelstats, Relation onerel,
VacPageList vacpagelist);
static void vacuum_page(Relation onerel, Buffer buffer, VacPage vacpage);
***************
*** 675,681 ****
static void
vac_truncate_clog(TransactionId vacuumXID, TransactionId frozenXID)
{
! TransactionId myXID;
Relation relation;
HeapScanDesc scan;
HeapTuple tuple;
--- 744,750 ----
static void
vac_truncate_clog(TransactionId vacuumXID, TransactionId frozenXID)
{
! TransactionId myXID = GetCurrentTransactionId();
Relation relation;
HeapScanDesc scan;
HeapTuple tuple;
***************
*** 683,689 ****
bool vacuumAlreadyWrapped = false;
bool frozenAlreadyWrapped = false;
- myXID = GetCurrentTransactionId();
relation = heap_openr(DatabaseRelationName, AccessShareLock);
--- 752,757 ----
***************
*** 1059,1075 ****
{
BlockNumber nblocks,
blkno;
- ItemId itemid;
- Buffer buf;
HeapTupleData tuple;
- OffsetNumber offnum,
- maxoff;
- bool pgchanged,
- tupgone,
- notup;
char *relname;
! VacPage vacpage,
! vacpagecopy;
BlockNumber empty_pages,
empty_end_pages;
double num_tuples,
--- 1127,1135 ----
{
BlockNumber nblocks,
blkno;
HeapTupleData tuple;
char *relname;
! VacPage vacpage;
BlockNumber empty_pages,
empty_end_pages;
double num_tuples,
***************
*** 1080,1086 ****
usable_free_space;
Size min_tlen = MaxTupleSize;
Size max_tlen = 0;
- int i;
bool do_shrinking = true;
VTupleLink vtlinks = (VTupleLink) palloc(100 * sizeof(VTupleLinkData));
int num_vtlinks = 0;
--- 1140,1145 ----
***************
*** 1113,1118 ****
--- 1172,1182 ----
tempPage = NULL;
bool do_reap,
do_frag;
+ Buffer buf;
+ OffsetNumber offnum,
+ maxoff;
+ bool pgchanged,
+ notup;
vacuum_delay_point();
***************
*** 1125,1130 ****
--- 1189,1196 ----
if (PageIsNew(page))
{
+ VacPage vacpagecopy;
+
ereport(WARNING,
(errmsg("relation \"%s\" page %u is uninitialized --- fixing",
relname, blkno)));
***************
*** 1142,1147 ****
--- 1208,1215 ----
if (PageIsEmpty(page))
{
+ VacPage vacpagecopy;
+
vacpage->free = ((PageHeader) page)->pd_upper - ((PageHeader) page)->pd_lower;
free_space += vacpage->free;
empty_pages++;
***************
*** 1161,1168 ****
offnum = OffsetNumberNext(offnum))
{
uint16 sv_infomask;
!
! itemid = PageGetItemId(page, offnum);
/*
* Collect un-used items too - it's possible to have indexes
--- 1229,1236 ----
offnum = OffsetNumberNext(offnum))
{
uint16 sv_infomask;
! ItemId itemid = PageGetItemId(page, offnum);
! bool tupgone = false;
/*
* Collect un-used items too - it's possible to have indexes
***************
*** 1180,1186 ****
tuple.t_len = ItemIdGetLength(itemid);
ItemPointerSet(&(tuple.t_self), blkno, offnum);
- tupgone = false;
sv_infomask = tuple.t_data->t_infomask;
switch (HeapTupleSatisfiesVacuum(tuple.t_data, OldestXmin))
--- 1248,1253 ----
***************
*** 1269,1275 ****
do_shrinking = false;
break;
default:
! elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
break;
}
--- 1336,1343 ----
do_shrinking = false;
break;
default:
! /* unexpected HeapTupleSatisfiesVacuum result */
! Assert(false);
break;
}
***************
*** 1344,1350 ****
if (do_reap || do_frag)
{
! vacpagecopy = copy_vac_page(vacpage);
if (do_reap)
vpage_insert(vacuum_pages, vacpagecopy);
if (do_frag)
--- 1412,1418 ----
if (do_reap || do_frag)
{
! VacPage vacpagecopy = copy_vac_page(vacpage);
if (do_reap)
vpage_insert(vacuum_pages, vacpagecopy);
if (do_frag)
***************
*** 1390,1395 ****
--- 1458,1465 ----
*/
if (do_shrinking)
{
+ int i;
+
Assert((BlockNumber) fraged_pages->num_pages >= empty_end_pages);
fraged_pages->num_pages -= empty_end_pages;
usable_free_space = 0;
***************
*** 1453,1528 ****
VacPageList vacuum_pages, VacPageList fraged_pages,
int nindexes, Relation *Irel)
{
! TransactionId myXID;
! CommandId myCID;
! Buffer buf,
! cur_buffer;
BlockNumber nblocks,
blkno;
BlockNumber last_move_dest_block = 0,
last_vacuum_block;
! Page page,
! ToPage = NULL;
! OffsetNumber offnum,
! maxoff,
! newoff,
! max_offset;
! ItemId itemid,
! newitemid;
! HeapTupleData tuple,
! newtup;
! TupleDesc tupdesc;
! ResultRelInfo *resultRelInfo;
! EState *estate;
! TupleTable tupleTable;
! TupleTableSlot *slot;
VacPageListData Nvacpagelist;
! VacPage cur_page = NULL,
last_vacuum_page,
vacpage,
*curpage;
- int cur_item = 0;
int i;
! Size tuple_len;
! int num_moved,
num_fraged_pages,
vacuumed_pages;
! int checked_moved,
! num_tuples,
! keep_tuples = 0;
! bool isempty,
! dowrite,
! chain_tuple_moved;
VacRUsage ru0;
vac_init_rusage(&ru0);
! myXID = GetCurrentTransactionId();
! myCID = GetCurrentCommandId();
!
! tupdesc = RelationGetDescr(onerel);
!
! /*
! * We need a ResultRelInfo and an EState so we can use the regular
! * executor's index-entry-making machinery.
! */
! estate = CreateExecutorState();
!
! resultRelInfo = makeNode(ResultRelInfo);
! resultRelInfo->ri_RangeTableIndex = 1; /* dummy */
! resultRelInfo->ri_RelationDesc = onerel;
! resultRelInfo->ri_TrigDesc = NULL; /* we don't fire triggers */
!
! ExecOpenIndices(resultRelInfo);
!
! estate->es_result_relations = resultRelInfo;
! estate->es_num_result_relations = 1;
! estate->es_result_relation_info = resultRelInfo;
!
! /* Set up a dummy tuple table too */
! tupleTable = ExecCreateTupleTable(1);
! slot = ExecAllocTableSlot(tupleTable);
! ExecSetSlotDescriptor(slot, tupdesc, false);
Nvacpagelist.num_pages = 0;
num_fraged_pages = fraged_pages->num_pages;
--- 1523,1551 ----
VacPageList vacuum_pages, VacPageList fraged_pages,
int nindexes, Relation *Irel)
{
! TransactionId myXID = GetCurrentTransactionId();
! Buffer dst_buffer = InvalidBuffer;
BlockNumber nblocks,
blkno;
BlockNumber last_move_dest_block = 0,
last_vacuum_block;
! Page dst_page = NULL;
! ExecContextData ec;
VacPageListData Nvacpagelist;
! VacPage dst_vacpage = NULL,
last_vacuum_page,
vacpage,
*curpage;
int i;
! int num_moved = 0,
num_fraged_pages,
vacuumed_pages;
! int keep_tuples = 0;
VacRUsage ru0;
vac_init_rusage(&ru0);
! ExecContext_Init(&ec, onerel);
Nvacpagelist.num_pages = 0;
num_fraged_pages = fraged_pages->num_pages;
***************
*** 1539,1546 ****
last_vacuum_page = NULL;
last_vacuum_block = InvalidBlockNumber;
}
- cur_buffer = InvalidBuffer;
- num_moved = 0;
vacpage = (VacPage) palloc(sizeof(VacPageData) + MaxOffsetNumber * sizeof(OffsetNumber));
vacpage->offsets_used = vacpage->offsets_free = 0;
--- 1562,1567 ----
***************
*** 1560,1565 ****
--- 1581,1594 ----
blkno > last_move_dest_block;
blkno--)
{
+ Buffer buf;
+ Page page;
+ OffsetNumber offnum,
+ maxoff;
+ bool isempty,
+ dowrite,
+ chain_tuple_moved;
+
vacuum_delay_point();
/*
***************
*** 1635,1641 ****
offnum <= maxoff;
offnum = OffsetNumberNext(offnum))
{
! itemid = PageGetItemId(page, offnum);
if (!ItemIdIsUsed(itemid))
continue;
--- 1664,1672 ----
offnum <= maxoff;
offnum = OffsetNumberNext(offnum))
{
! Size tuple_len;
! HeapTupleData tuple;
! ItemId itemid = PageGetItemId(page, offnum);
if (!ItemIdIsUsed(itemid))
continue;
***************
*** 1645,1689 ****
tuple_len = tuple.t_len = ItemIdGetLength(itemid);
ItemPointerSet(&(tuple.t_self), blkno, offnum);
if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
{
! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
! elog(ERROR, "HEAP_MOVED_IN was not expected");
/*
* If this (chain) tuple is moved by me already then I
* have to check is it in vacpage or not - i.e. is it
* moved while cleaning this page or some previous one.
*/
! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
! {
! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
! elog(ERROR, "invalid XVAC in tuple header");
! if (keep_tuples == 0)
! continue;
! if (chain_tuple_moved) /* some chains was moved
! * while */
! { /* cleaning this page */
! Assert(vacpage->offsets_free > 0);
! for (i = 0; i < vacpage->offsets_free; i++)
! {
! if (vacpage->offsets[i] == offnum)
! break;
! }
! if (i >= vacpage->offsets_free) /* not found */
! {
! vacpage->offsets[vacpage->offsets_free++] = offnum;
! keep_tuples--;
! }
}
! else
{
vacpage->offsets[vacpage->offsets_free++] = offnum;
keep_tuples--;
}
- continue;
}
! elog(ERROR, "HEAP_MOVED_OFF was expected");
}
/*
--- 1676,1746 ----
tuple_len = tuple.t_len = ItemIdGetLength(itemid);
ItemPointerSet(&(tuple.t_self), blkno, offnum);
+ /*
+ * VACUUM FULL has an exclusive lock on the relation. So
+ * normally no other transaction can have pending INSERTs or
+ * DELETEs in this relation. A tuple is either
+ * (a) a tuple in a system catalog, inserted or deleted by
+ * a not yet committed transaction or
+ * (b) dead (XMIN_INVALID or XMAX_COMMITTED) or
+ * (c) inserted by a committed xact (XMIN_COMMITTED) or
+ * (d) moved by the currently running VACUUM.
+ * In case (a) we wouldn't be in repair_frag() at all.
+ * In case (b) we cannot be here, because scan_heap() has
+ * already marked the item as unused, see continue above.
+ * Case (c) is what normally is to be expected.
+ * Case (d) is only possible, if a whole tuple chain has been
+ * moved while processing this or a higher numbered block.
+ */
if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
{
! /*
! * There cannot be another concurrently running VACUUM. If
! * the tuple had been moved in by a previous VACUUM, the
! * visibility check would have set XMIN_COMMITTED. If the
! * tuple had been moved in by the currently running VACUUM,
! * the loop would have been terminated. We had
! * elog(ERROR, ...) here, but as we are testing for a
! * can't-happen condition, Assert() seems more appropriate.
! */
! Assert(!(tuple.t_data->t_infomask & HEAP_MOVED_IN));
/*
* If this (chain) tuple is moved by me already then I
* have to check is it in vacpage or not - i.e. is it
* moved while cleaning this page or some previous one.
*/
! Assert(tuple.t_data->t_infomask & HEAP_MOVED_OFF);
! /*
! * MOVED_OFF by another VACUUM would have caused the
! * visibility check to set XMIN_COMMITTED or XMIN_INVALID.
! */
! Assert(HeapTupleHeaderGetXvac(tuple.t_data) == myXID);
!
! /* Can't we Assert(keep_tuples > 0) here? */
! if (keep_tuples == 0)
! continue;
! if (chain_tuple_moved) /* some chains was moved
! * while */
! { /* cleaning this page */
! Assert(vacpage->offsets_free > 0);
! for (i = 0; i < vacpage->offsets_free; i++)
! {
! if (vacpage->offsets[i] == offnum)
! break;
}
! if (i >= vacpage->offsets_free) /* not found */
{
vacpage->offsets[vacpage->offsets_free++] = offnum;
keep_tuples--;
}
}
! else
! {
! vacpage->offsets[vacpage->offsets_free++] = offnum;
! keep_tuples--;
! }
! continue;
}
/*
***************
*** 1716,1723 ****
Buffer Cbuf = buf;
bool freeCbuf = false;
bool chain_move_failed = false;
- Page Cpage;
- ItemId Citemid;
ItemPointerData Ctid;
HeapTupleData tp = tuple;
Size tlen = tuple_len;
--- 1773,1778 ----
***************
*** 1728,1737 ****
int to_item = 0;
int ti;
! if (cur_buffer != InvalidBuffer)
{
! WriteBuffer(cur_buffer);
! cur_buffer = InvalidBuffer;
}
/* Quick exit if we have no vtlinks to search in */
--- 1783,1792 ----
int to_item = 0;
int ti;
! if (dst_buffer != InvalidBuffer)
{
! WriteBuffer(dst_buffer);
! dst_buffer = InvalidBuffer;
}
/* Quick exit if we have no vtlinks to search in */
***************
*** 1754,1759 ****
--- 1809,1818 ----
!(ItemPointerEquals(&(tp.t_self),
&(tp.t_data->t_ctid))))
{
+ Page Cpage;
+ ItemId Citemid;
+ ItemPointerData Ctid;
+
Ctid = tp.t_data->t_ctid;
if (freeCbuf)
ReleaseBuffer(Cbuf);
***************
*** 1929,1940 ****
}
/*
! * Okay, move the whle tuple chain
*/
ItemPointerSetInvalid(&Ctid);
for (ti = 0; ti < num_vtmove; ti++)
{
VacPage destvacpage = vtmove[ti].vacpage;
/* Get page to move from */
tuple.t_self = vtmove[ti].tid;
--- 1988,2001 ----
}
/*
! * Okay, move the whole tuple chain
*/
ItemPointerSetInvalid(&Ctid);
for (ti = 0; ti < num_vtmove; ti++)
{
VacPage destvacpage = vtmove[ti].vacpage;
+ Page Cpage;
+ ItemId Citemid;
/* Get page to move from */
tuple.t_self = vtmove[ti].tid;
***************
*** 1942,1954 ****
ItemPointerGetBlockNumber(&(tuple.t_self)));
/* Get page to move to */
! cur_buffer = ReadBuffer(onerel, destvacpage->blkno);
! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
! if (cur_buffer != Cbuf)
LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
! ToPage = BufferGetPage(cur_buffer);
Cpage = BufferGetPage(Cbuf);
Citemid = PageGetItemId(Cpage,
--- 2003,2015 ----
ItemPointerGetBlockNumber(&(tuple.t_self)));
/* Get page to move to */
! dst_buffer = ReadBuffer(onerel, destvacpage->blkno);
! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
! if (dst_buffer != Cbuf)
LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
! dst_page = BufferGetPage(dst_buffer);
Cpage = BufferGetPage(Cbuf);
Citemid = PageGetItemId(Cpage,
***************
*** 1961,2081 ****
* make a copy of the source tuple, and then mark the
* source tuple MOVED_OFF.
*/
! heap_copytuple_with_tuple(&tuple, &newtup);
!
! /*
! * register invalidation of source tuple in catcaches.
! */
! CacheInvalidateHeapTuple(onerel, &tuple);
!
! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
! START_CRIT_SECTION();
!
! tuple.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_IN);
! tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
! HeapTupleHeaderSetXvac(tuple.t_data, myXID);
!
! /*
! * If this page was not used before - clean it.
! *
! * NOTE: a nasty bug used to lurk here. It is possible
! * for the source and destination pages to be the same
! * (since this tuple-chain member can be on a page
! * lower than the one we're currently processing in
! * the outer loop). If that's true, then after
! * vacuum_page() the source tuple will have been
! * moved, and tuple.t_data will be pointing at
! * garbage. Therefore we must do everything that uses
! * tuple.t_data BEFORE this step!!
! *
! * This path is different from the other callers of
! * vacuum_page, because we have already incremented
! * the vacpage's offsets_used field to account for the
! * tuple(s) we expect to move onto the page. Therefore
! * vacuum_page's check for offsets_used == 0 is wrong.
! * But since that's a good debugging check for all
! * other callers, we work around it here rather than
! * remove it.
! */
! if (!PageIsEmpty(ToPage) && vtmove[ti].cleanVpd)
! {
! int sv_offsets_used = destvacpage->offsets_used;
!
! destvacpage->offsets_used = 0;
! vacuum_page(onerel, cur_buffer, destvacpage);
! destvacpage->offsets_used = sv_offsets_used;
! }
!
! /*
! * Update the state of the copied tuple, and store it
! * on the destination page.
! */
! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_OFF);
! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
! newoff = PageAddItem(ToPage,
! (Item) newtup.t_data,
! tuple_len,
! InvalidOffsetNumber,
! LP_USED);
! if (newoff == InvalidOffsetNumber)
! {
! elog(PANIC, "failed to add item with len = %lu to page %u while moving tuple chain",
! (unsigned long) tuple_len, destvacpage->blkno);
! }
! newitemid = PageGetItemId(ToPage, newoff);
! pfree(newtup.t_data);
! newtup.t_datamcxt = NULL;
! newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
! ItemPointerSet(&(newtup.t_self), destvacpage->blkno, newoff);
!
! /* XLOG stuff */
! if (!onerel->rd_istemp)
! {
! XLogRecPtr recptr =
! log_heap_move(onerel, Cbuf, tuple.t_self,
! cur_buffer, &newtup);
!
! if (Cbuf != cur_buffer)
! {
! PageSetLSN(Cpage, recptr);
! PageSetSUI(Cpage, ThisStartUpID);
! }
! PageSetLSN(ToPage, recptr);
! PageSetSUI(ToPage, ThisStartUpID);
! }
! else
! {
! /*
! * No XLOG record, but still need to flag that XID
! * exists on disk
! */
! MyXactMadeTempRelUpdate = true;
! }
!
! END_CRIT_SECTION();
if (destvacpage->blkno > last_move_dest_block)
last_move_dest_block = destvacpage->blkno;
/*
- * Set new tuple's t_ctid pointing to itself for last
- * tuple in chain, and to next tuple in chain
- * otherwise.
- */
- if (!ItemPointerIsValid(&Ctid))
- newtup.t_data->t_ctid = newtup.t_self;
- else
- newtup.t_data->t_ctid = Ctid;
- Ctid = newtup.t_self;
-
- num_moved++;
-
- /*
* Remember that we moved tuple from the current page
* (corresponding index tuple will be cleaned).
*/
--- 2022,2036 ----
* make a copy of the source tuple, and then mark the
* source tuple MOVED_OFF.
*/
! move_chain_tuple(onerel, Cbuf, Cpage, &tuple,
! dst_buffer, dst_page, destvacpage,
! &ec, &Ctid, vtmove[ti].cleanVpd);
+ num_moved++;
if (destvacpage->blkno > last_move_dest_block)
last_move_dest_block = destvacpage->blkno;
/*
* Remember that we moved tuple from the current page
* (corresponding index tuple will be cleaned).
*/
***************
*** 2085,2107 ****
else
keep_tuples++;
! LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
! if (cur_buffer != Cbuf)
! LockBuffer(Cbuf, BUFFER_LOCK_UNLOCK);
!
! /* Create index entries for the moved tuple */
! if (resultRelInfo->ri_NumIndices > 0)
! {
! ExecStoreTuple(&newtup, slot, InvalidBuffer, false);
! ExecInsertIndexTuples(slot, &(newtup.t_self),
! estate, true);
! }
!
! WriteBuffer(cur_buffer);
WriteBuffer(Cbuf);
} /* end of move-the-tuple-chain loop */
! cur_buffer = InvalidBuffer;
pfree(vtmove);
chain_tuple_moved = true;
--- 2040,2050 ----
else
keep_tuples++;
! WriteBuffer(dst_buffer);
WriteBuffer(Cbuf);
} /* end of move-the-tuple-chain loop */
! dst_buffer = InvalidBuffer;
pfree(vtmove);
chain_tuple_moved = true;
***************
*** 2110,2122 ****
} /* end of is-tuple-in-chain test */
/* try to find new page for this tuple */
! if (cur_buffer == InvalidBuffer ||
! !enough_space(cur_page, tuple_len))
{
! if (cur_buffer != InvalidBuffer)
{
! WriteBuffer(cur_buffer);
! cur_buffer = InvalidBuffer;
}
for (i = 0; i < num_fraged_pages; i++)
{
--- 2053,2065 ----
} /* end of is-tuple-in-chain test */
/* try to find new page for this tuple */
! if (dst_buffer == InvalidBuffer ||
! !enough_space(dst_vacpage, tuple_len))
{
! if (dst_buffer != InvalidBuffer)
{
! WriteBuffer(dst_buffer);
! dst_buffer = InvalidBuffer;
}
for (i = 0; i < num_fraged_pages; i++)
{
***************
*** 2125,2234 ****
}
if (i == num_fraged_pages)
break; /* can't move item anywhere */
! cur_item = i;
! cur_page = fraged_pages->pagedesc[cur_item];
! cur_buffer = ReadBuffer(onerel, cur_page->blkno);
! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
! ToPage = BufferGetPage(cur_buffer);
/* if this page was not used before - clean it */
! if (!PageIsEmpty(ToPage) && cur_page->offsets_used == 0)
! vacuum_page(onerel, cur_buffer, cur_page);
}
else
! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
! /* copy tuple */
! heap_copytuple_with_tuple(&tuple, &newtup);
!
! /*
! * register invalidation of source tuple in catcaches.
! *
! * (Note: we do not need to register the copied tuple, because we
! * are not changing the tuple contents and so there cannot be
! * any need to flush negative catcache entries.)
! */
! CacheInvalidateHeapTuple(onerel, &tuple);
- /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
- START_CRIT_SECTION();
! /*
! * Mark new tuple as MOVED_IN by me.
! */
! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_OFF);
! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
!
! /* add tuple to the page */
! newoff = PageAddItem(ToPage, (Item) newtup.t_data, tuple_len,
! InvalidOffsetNumber, LP_USED);
! if (newoff == InvalidOffsetNumber)
! {
! elog(PANIC, "failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)",
! (unsigned long) tuple_len,
! cur_page->blkno, (unsigned long) cur_page->free,
! cur_page->offsets_used, cur_page->offsets_free);
! }
! newitemid = PageGetItemId(ToPage, newoff);
! pfree(newtup.t_data);
! newtup.t_datamcxt = NULL;
! newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
! ItemPointerSet(&(newtup.t_data->t_ctid), cur_page->blkno, newoff);
! newtup.t_self = newtup.t_data->t_ctid;
/*
! * Mark old tuple as MOVED_OFF by me.
*/
- tuple.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
- HEAP_XMIN_INVALID |
- HEAP_MOVED_IN);
- tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
- HeapTupleHeaderSetXvac(tuple.t_data, myXID);
-
- /* XLOG stuff */
- if (!onerel->rd_istemp)
- {
- XLogRecPtr recptr =
- log_heap_move(onerel, buf, tuple.t_self,
- cur_buffer, &newtup);
-
- PageSetLSN(page, recptr);
- PageSetSUI(page, ThisStartUpID);
- PageSetLSN(ToPage, recptr);
- PageSetSUI(ToPage, ThisStartUpID);
- }
- else
- {
- /*
- * No XLOG record, but still need to flag that XID exists
- * on disk
- */
- MyXactMadeTempRelUpdate = true;
- }
-
- END_CRIT_SECTION();
-
- cur_page->offsets_used++;
- num_moved++;
- cur_page->free = ((PageHeader) ToPage)->pd_upper - ((PageHeader) ToPage)->pd_lower;
- if (cur_page->blkno > last_move_dest_block)
- last_move_dest_block = cur_page->blkno;
-
vacpage->offsets[vacpage->offsets_free++] = offnum;
-
- LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
- LockBuffer(buf, BUFFER_LOCK_UNLOCK);
-
- /* insert index' tuples if needed */
- if (resultRelInfo->ri_NumIndices > 0)
- {
- ExecStoreTuple(&newtup, slot, InvalidBuffer, false);
- ExecInsertIndexTuples(slot, &(newtup.t_self), estate, true);
- }
} /* walk along page */
/*
--- 2068,2099 ----
}
if (i == num_fraged_pages)
break; /* can't move item anywhere */
! dst_vacpage = fraged_pages->pagedesc[i];
! dst_buffer = ReadBuffer(onerel, dst_vacpage->blkno);
! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
! dst_page = BufferGetPage(dst_buffer);
/* if this page was not used before - clean it */
! if (!PageIsEmpty(dst_page) && dst_vacpage->offsets_used == 0)
! vacuum_page(onerel, dst_buffer, dst_vacpage);
}
else
! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
! move_plain_tuple(onerel, buf, page, &tuple,
! dst_buffer, dst_page, dst_vacpage, &ec);
! num_moved++;
! if (dst_vacpage->blkno > last_move_dest_block)
! last_move_dest_block = dst_vacpage->blkno;
/*
! * Remember that we moved tuple from the current page
! * (corresponding index tuple will be cleaned).
*/
vacpage->offsets[vacpage->offsets_free++] = offnum;
} /* walk along page */
/*
***************
*** 2249,2284 ****
off <= maxoff;
off = OffsetNumberNext(off))
{
! itemid = PageGetItemId(page, off);
if (!ItemIdIsUsed(itemid))
continue;
! tuple.t_datamcxt = NULL;
! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
! if (tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED)
continue;
! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
! elog(ERROR, "HEAP_MOVED_IN was not expected");
! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
{
! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
! elog(ERROR, "invalid XVAC in tuple header");
! /* some chains was moved while */
! if (chain_tuple_moved)
! { /* cleaning this page */
! Assert(vacpage->offsets_free > 0);
! for (i = 0; i < vacpage->offsets_free; i++)
! {
! if (vacpage->offsets[i] == off)
! break;
! }
! if (i >= vacpage->offsets_free) /* not found */
! {
! vacpage->offsets[vacpage->offsets_free++] = off;
! Assert(keep_tuples > 0);
! keep_tuples--;
! }
}
! else
{
vacpage->offsets[vacpage->offsets_free++] = off;
Assert(keep_tuples > 0);
--- 2114,2144 ----
off <= maxoff;
off = OffsetNumberNext(off))
{
! ItemId itemid = PageGetItemId(page, off);
! HeapTupleHeader htup;
!
if (!ItemIdIsUsed(itemid))
continue;
! htup = (HeapTupleHeader) PageGetItem(page, itemid);
! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
continue;
! /*
! ** See comments in the walk-along-page loop above, why we
! ** have Asserts here instead of if (...) elog(ERROR).
! */
! Assert(!(htup->t_infomask & HEAP_MOVED_IN));
! Assert(htup->t_infomask & HEAP_MOVED_OFF);
! Assert(HeapTupleHeaderGetXvac(htup) == myXID);
! if (chain_tuple_moved)
{
! /* some chains was moved while cleaning this page */
! Assert(vacpage->offsets_free > 0);
! for (i = 0; i < vacpage->offsets_free; i++)
! {
! if (vacpage->offsets[i] == off)
! break;
}
! if (i >= vacpage->offsets_free) /* not found */
{
vacpage->offsets[vacpage->offsets_free++] = off;
Assert(keep_tuples > 0);
***************
*** 2286,2292 ****
}
}
else
! elog(ERROR, "HEAP_MOVED_OFF was expected");
}
}
--- 2146,2156 ----
}
}
else
! {
! vacpage->offsets[vacpage->offsets_free++] = off;
! Assert(keep_tuples > 0);
! keep_tuples--;
! }
}
}
***************
*** 2312,2321 ****
blkno++; /* new number of blocks */
! if (cur_buffer != InvalidBuffer)
{
Assert(num_moved > 0);
! WriteBuffer(cur_buffer);
}
if (num_moved > 0)
--- 2176,2185 ----
blkno++; /* new number of blocks */
! if (dst_buffer != InvalidBuffer)
{
Assert(num_moved > 0);
! WriteBuffer(dst_buffer);
}
if (num_moved > 0)
***************
*** 2348,2353 ****
--- 2212,2220 ----
Assert((*curpage)->blkno < blkno);
if ((*curpage)->offsets_used == 0)
{
+ Buffer buf;
+ Page page;
+
/* this page was not used as a move target, so must clean it */
buf = ReadBuffer(onerel, (*curpage)->blkno);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
***************
*** 2363,2424 ****
* Now scan all the pages that we moved tuples onto and update tuple
* status bits. This is not really necessary, but will save time for
* future transactions examining these tuples.
- *
- * XXX NOTICE that this code fails to clear HEAP_MOVED_OFF tuples from
- * pages that were move source pages but not move dest pages. One
- * also wonders whether it wouldn't be better to skip this step and
- * let the tuple status updates happen someplace that's not holding an
- * exclusive lock on the relation.
*/
! checked_moved = 0;
! for (i = 0, curpage = fraged_pages->pagedesc;
! i < num_fraged_pages;
! i++, curpage++)
! {
! vacuum_delay_point();
!
! Assert((*curpage)->blkno < blkno);
! if ((*curpage)->blkno > last_move_dest_block)
! break; /* no need to scan any further */
! if ((*curpage)->offsets_used == 0)
! continue; /* this page was never used as a move dest */
! buf = ReadBuffer(onerel, (*curpage)->blkno);
! LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
! page = BufferGetPage(buf);
! num_tuples = 0;
! max_offset = PageGetMaxOffsetNumber(page);
! for (newoff = FirstOffsetNumber;
! newoff <= max_offset;
! newoff = OffsetNumberNext(newoff))
! {
! itemid = PageGetItemId(page, newoff);
! if (!ItemIdIsUsed(itemid))
! continue;
! tuple.t_datamcxt = NULL;
! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
! if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
! {
! if (!(tuple.t_data->t_infomask & HEAP_MOVED))
! elog(ERROR, "HEAP_MOVED_OFF/HEAP_MOVED_IN was expected");
! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
! elog(ERROR, "invalid XVAC in tuple header");
! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
! {
! tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
! tuple.t_data->t_infomask &= ~HEAP_MOVED;
! num_tuples++;
! }
! else
! tuple.t_data->t_infomask |= HEAP_XMIN_INVALID;
! }
! }
! LockBuffer(buf, BUFFER_LOCK_UNLOCK);
! WriteBuffer(buf);
! Assert((*curpage)->offsets_used == num_tuples);
! checked_moved += num_tuples;
! }
! Assert(num_moved == checked_moved);
!
/*
* It'd be cleaner to make this report at the bottom of this routine,
* but then the rusage would double-count the second pass of index
--- 2230,2239 ----
* Now scan all the pages that we moved tuples onto and update tuple
* status bits. This is not really necessary, but will save time for
* future transactions examining these tuples.
*/
! update_hint_bits(onerel, fraged_pages, num_fraged_pages,
! last_move_dest_block, num_moved);
!
/*
* It'd be cleaner to make this report at the bottom of this routine,
* but then the rusage would double-count the second pass of index
***************
*** 2455,2460 ****
--- 2270,2282 ----
*vpleft = *vpright;
*vpright = vpsave;
}
+ /*
+ * keep_tuples is the number of tuples that have been moved
+ * off a page during chain moves but not been scanned over
+ * subsequently. The tuple ids of these tuples are not
+ * recorded as free offsets for any VacPage, so they will not
+ * be cleared from the indexes.
+ */
Assert(keep_tuples >= 0);
for (i = 0; i < nindexes; i++)
vacuum_index(&Nvacpagelist, Irel[i],
***************
*** 2465,2500 ****
if (vacpage->blkno == (blkno - 1) &&
vacpage->offsets_free > 0)
{
! OffsetNumber unused[BLCKSZ / sizeof(OffsetNumber)];
! int uncnt;
buf = ReadBuffer(onerel, vacpage->blkno);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
page = BufferGetPage(buf);
- num_tuples = 0;
maxoff = PageGetMaxOffsetNumber(page);
for (offnum = FirstOffsetNumber;
offnum <= maxoff;
offnum = OffsetNumberNext(offnum))
{
! itemid = PageGetItemId(page, offnum);
if (!ItemIdIsUsed(itemid))
continue;
! tuple.t_datamcxt = NULL;
! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
! if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
! {
! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
! {
! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
! elog(ERROR, "invalid XVAC in tuple header");
! itemid->lp_flags &= ~LP_USED;
! num_tuples++;
! }
! else
! elog(ERROR, "HEAP_MOVED_OFF was expected");
! }
}
Assert(vacpage->offsets_free == num_tuples);
--- 2287,2327 ----
if (vacpage->blkno == (blkno - 1) &&
vacpage->offsets_free > 0)
{
! Buffer buf;
! Page page;
! OffsetNumber unused[BLCKSZ / sizeof(OffsetNumber)];
! OffsetNumber offnum,
! maxoff;
! int uncnt;
! int num_tuples = 0;
buf = ReadBuffer(onerel, vacpage->blkno);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
page = BufferGetPage(buf);
maxoff = PageGetMaxOffsetNumber(page);
for (offnum = FirstOffsetNumber;
offnum <= maxoff;
offnum = OffsetNumberNext(offnum))
{
! ItemId itemid = PageGetItemId(page, offnum);
! HeapTupleHeader htup;
!
if (!ItemIdIsUsed(itemid))
continue;
! htup = (HeapTupleHeader) PageGetItem(page, itemid);
! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
! continue;
! /*
! ** See comments in the walk-along-page loop above, why we
! ** have Asserts here instead of if (...) elog(ERROR).
! */
! Assert(!(htup->t_infomask & HEAP_MOVED_IN));
! Assert(htup->t_infomask & HEAP_MOVED_OFF);
! Assert(HeapTupleHeaderGetXvac(htup) == myXID);
!
! itemid->lp_flags &= ~LP_USED;
! num_tuples++;
}
Assert(vacpage->offsets_free == num_tuples);
***************
*** 2554,2564 ****
if (vacrelstats->vtlinks != NULL)
pfree(vacrelstats->vtlinks);
! ExecDropTupleTable(tupleTable, true);
! ExecCloseIndices(resultRelInfo);
! FreeExecutorState(estate);
}
/*
--- 2381,2717 ----
if (vacrelstats->vtlinks != NULL)
pfree(vacrelstats->vtlinks);
! ExecContext_Finish(&ec);
! }
!
! /*
! * move_chain_tuple() -- move one tuple that is part of a tuple chain
! *
! * This routine moves old_tup from old_page to dst_page.
! * old_page and dst_page might be the same page.
! * On entry old_buf and dst_buf are locked exclusively, both locks (or
! * the single lock, if this is a intra-page-move) are released before
! * exit.
! *
! * Yes, a routine with ten parameters is ugly, but it's still better
! * than having these 120 lines of code in repair_frag() which is
! * already too long and almost unreadable.
! */
! static void
! move_chain_tuple(Relation rel,
! Buffer old_buf, Page old_page, HeapTuple old_tup,
! Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
! ExecContext ec, ItemPointer ctid, bool cleanVpd)
! {
! TransactionId myXID = GetCurrentTransactionId();
! HeapTupleData newtup;
! OffsetNumber newoff;
! ItemId newitemid;
! Size tuple_len = old_tup->t_len;
!
! heap_copytuple_with_tuple(old_tup, &newtup);
!
! /*
! * register invalidation of source tuple in catcaches.
! */
! CacheInvalidateHeapTuple(rel, old_tup);
!
! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
! START_CRIT_SECTION();
!
! old_tup->t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_IN);
! old_tup->t_data->t_infomask |= HEAP_MOVED_OFF;
! HeapTupleHeaderSetXvac(old_tup->t_data, myXID);
!
! /*
! * If this page was not used before - clean it.
! *
! * NOTE: a nasty bug used to lurk here. It is possible
! * for the source and destination pages to be the same
! * (since this tuple-chain member can be on a page
! * lower than the one we're currently processing in
! * the outer loop). If that's true, then after
! * vacuum_page() the source tuple will have been
! * moved, and tuple.t_data will be pointing at
! * garbage. Therefore we must do everything that uses
! * old_tup->t_data BEFORE this step!!
! *
! * This path is different from the other callers of
! * vacuum_page, because we have already incremented
! * the vacpage's offsets_used field to account for the
! * tuple(s) we expect to move onto the page. Therefore
! * vacuum_page's check for offsets_used == 0 is wrong.
! * But since that's a good debugging check for all
! * other callers, we work around it here rather than
! * remove it.
! */
! if (!PageIsEmpty(dst_page) && cleanVpd)
! {
! int sv_offsets_used = dst_vacpage->offsets_used;
!
! dst_vacpage->offsets_used = 0;
! vacuum_page(rel, dst_buf, dst_vacpage);
! dst_vacpage->offsets_used = sv_offsets_used;
! }
! /*
! * Update the state of the copied tuple, and store it
! * on the destination page.
! */
! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_OFF);
! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
! newoff = PageAddItem(dst_page, (Item) newtup.t_data, tuple_len,
! InvalidOffsetNumber, LP_USED);
! if (newoff == InvalidOffsetNumber)
! {
! elog(PANIC, "failed to add item with len = %lu to page %u while moving tuple chain",
! (unsigned long) tuple_len, dst_vacpage->blkno);
! }
! newitemid = PageGetItemId(dst_page, newoff);
! pfree(newtup.t_data);
! newtup.t_datamcxt = NULL;
! newtup.t_data = (HeapTupleHeader) PageGetItem(dst_page, newitemid);
! ItemPointerSet(&(newtup.t_self), dst_vacpage->blkno, newoff);
!
! /* XLOG stuff */
! if (!rel->rd_istemp)
! {
! XLogRecPtr recptr = log_heap_move(rel, old_buf, old_tup->t_self,
! dst_buf, &newtup);
!
! if (old_buf != dst_buf)
! {
! PageSetLSN(old_page, recptr);
! PageSetSUI(old_page, ThisStartUpID);
! }
! PageSetLSN(dst_page, recptr);
! PageSetSUI(dst_page, ThisStartUpID);
! }
! else
! {
! /*
! * No XLOG record, but still need to flag that XID
! * exists on disk
! */
! MyXactMadeTempRelUpdate = true;
! }
!
! END_CRIT_SECTION();
!
! /*
! * Set new tuple's t_ctid pointing to itself for last
! * tuple in chain, and to next tuple in chain
! * otherwise.
! */
! /* Is this ok after log_heap_move() and END_CRIT_SECTION()? */
! if (!ItemPointerIsValid(ctid))
! newtup.t_data->t_ctid = newtup.t_self;
! else
! newtup.t_data->t_ctid = *ctid;
! *ctid = newtup.t_self;
! LockBuffer(dst_buf, BUFFER_LOCK_UNLOCK);
! if (dst_buf != old_buf)
! LockBuffer(old_buf, BUFFER_LOCK_UNLOCK);
!
! /* Create index entries for the moved tuple */
! if (ec->resultRelInfo->ri_NumIndices > 0)
! {
! ExecStoreTuple(&newtup, ec->slot, InvalidBuffer, false);
! ExecInsertIndexTuples(ec->slot, &(newtup.t_self), ec->estate, true);
! }
! }
!
! /*
! * move_plain_tuple() -- move one tuple that is not part of a chain
! *
! * This routine moves old_tup from old_page to dst_page.
! * On entry old_buf and dst_buf are locked exclusively, both locks are
! * released before exit.
! *
! * Yes, a routine with eight parameters is ugly, but it's still better
! * than having these 90 lines of code in repair_frag() which is already
! * too long and almost unreadable.
! */
! static void
! move_plain_tuple(Relation rel,
! Buffer old_buf, Page old_page, HeapTuple old_tup,
! Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
! ExecContext ec)
! {
! TransactionId myXID = GetCurrentTransactionId();
! HeapTupleData newtup;
! OffsetNumber newoff;
! ItemId newitemid;
! Size tuple_len = old_tup->t_len;
!
! /* copy tuple */
! heap_copytuple_with_tuple(old_tup, &newtup);
!
! /*
! * register invalidation of source tuple in catcaches.
! *
! * (Note: we do not need to register the copied tuple, because we
! * are not changing the tuple contents and so there cannot be
! * any need to flush negative catcache entries.)
! */
! CacheInvalidateHeapTuple(rel, old_tup);
!
! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
! START_CRIT_SECTION();
!
! /*
! * Mark new tuple as MOVED_IN by me.
! */
! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_OFF);
! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
!
! /* add tuple to the page */
! newoff = PageAddItem(dst_page, (Item) newtup.t_data, tuple_len,
! InvalidOffsetNumber, LP_USED);
! if (newoff == InvalidOffsetNumber)
! {
! elog(PANIC, "failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)",
! (unsigned long) tuple_len,
! dst_vacpage->blkno, (unsigned long) dst_vacpage->free,
! dst_vacpage->offsets_used, dst_vacpage->offsets_free);
! }
! newitemid = PageGetItemId(dst_page, newoff);
! pfree(newtup.t_data);
! newtup.t_datamcxt = NULL;
! newtup.t_data = (HeapTupleHeader) PageGetItem(dst_page, newitemid);
! ItemPointerSet(&(newtup.t_data->t_ctid), dst_vacpage->blkno, newoff);
! newtup.t_self = newtup.t_data->t_ctid;
!
! /*
! * Mark old tuple as MOVED_OFF by me.
! */
! old_tup->t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
! HEAP_XMIN_INVALID |
! HEAP_MOVED_IN);
! old_tup->t_data->t_infomask |= HEAP_MOVED_OFF;
! HeapTupleHeaderSetXvac(old_tup->t_data, myXID);
!
! /* XLOG stuff */
! if (!rel->rd_istemp)
! {
! XLogRecPtr recptr = log_heap_move(rel, old_buf, old_tup->t_self,
! dst_buf, &newtup);
!
! PageSetLSN(old_page, recptr);
! PageSetSUI(old_page, ThisStartUpID);
! PageSetLSN(dst_page, recptr);
! PageSetSUI(dst_page, ThisStartUpID);
! }
! else
! {
! /*
! * No XLOG record, but still need to flag that XID exists
! * on disk
! */
! MyXactMadeTempRelUpdate = true;
! }
!
! END_CRIT_SECTION();
!
! dst_vacpage->free = ((PageHeader) dst_page)->pd_upper -
! ((PageHeader) dst_page)->pd_lower;
! LockBuffer(dst_buf, BUFFER_LOCK_UNLOCK);
! LockBuffer(old_buf, BUFFER_LOCK_UNLOCK);
!
! dst_vacpage->offsets_used++;
!
! /* insert index' tuples if needed */
! if (ec->resultRelInfo->ri_NumIndices > 0)
! {
! ExecStoreTuple(&newtup, ec->slot, InvalidBuffer, false);
! ExecInsertIndexTuples(ec->slot, &(newtup.t_self), ec->estate, true);
! }
! }
!
! /*
! * update_hint_bits() -- update hint bits in destination pages
! *
! * Scan all the pages that we moved tuples onto and update tuple
! * status bits. This is not really necessary, but will save time for
! * future transactions examining these tuples.
! *
! * XXX NOTICE that this code fails to clear HEAP_MOVED_OFF tuples from
! * pages that were move source pages but not move dest pages. One
! * also wonders whether it wouldn't be better to skip this step and
! * let the tuple status updates happen someplace that's not holding an
! * exclusive lock on the relation.
! */
! static void
! update_hint_bits(Relation rel, VacPageList fraged_pages, int num_fraged_pages,
! BlockNumber last_move_dest_block, int num_moved)
! {
! int checked_moved = 0;
! int i;
! VacPage *curpage;
!
! for (i = 0, curpage = fraged_pages->pagedesc;
! i < num_fraged_pages;
! i++, curpage++)
! {
! Buffer buf;
! Page page;
! OffsetNumber max_offset;
! OffsetNumber off;
! int num_tuples = 0;
!
! vacuum_delay_point();
!
! if ((*curpage)->blkno > last_move_dest_block)
! break; /* no need to scan any further */
! if ((*curpage)->offsets_used == 0)
! continue; /* this page was never used as a move dest */
! buf = ReadBuffer(rel, (*curpage)->blkno);
! LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
! page = BufferGetPage(buf);
! max_offset = PageGetMaxOffsetNumber(page);
! for (off = FirstOffsetNumber;
! off <= max_offset;
! off = OffsetNumberNext(off))
! {
! ItemId itemid = PageGetItemId(page, off);
! HeapTupleHeader htup;
!
! if (!ItemIdIsUsed(itemid))
! continue;
! htup = (HeapTupleHeader) PageGetItem(page, itemid);
! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
! continue;
! /*
! * See comments in the walk-along-page loop above, why we
! * have Asserts here instead of if (...) elog(ERROR). The
! * difference here is that we may see MOVED_IN.
! */
! Assert(htup->t_infomask & HEAP_MOVED);
! Assert(HeapTupleHeaderGetXvac(htup) == GetCurrentTransactionId());
! if (htup->t_infomask & HEAP_MOVED_IN)
! {
! htup->t_infomask |= HEAP_XMIN_COMMITTED;
! htup->t_infomask &= ~HEAP_MOVED;
! num_tuples++;
! }
! else
! htup->t_infomask |= HEAP_XMIN_INVALID;
! }
! LockBuffer(buf, BUFFER_LOCK_UNLOCK);
! WriteBuffer(buf);
! Assert((*curpage)->offsets_used == num_tuples);
! checked_moved += num_tuples;
! }
! Assert(num_moved == checked_moved);
}
/*
Patch applied. Thanks.
---------------------------------------------------------------------------
Manfred Koizar wrote:
> . rename variables
> . cur_buffer -> dst_buffer
> . ToPage -> dst_page
> . cur_page -> dst_vacpage
> . move variable declarations into block where variable is used
> . various Asserts instead of elog(ERROR, ...)
> . extract functionality from repair_frag() into new routines
> . move_chain_tuple()
> . move_plain_tuple()
> . update_hint_bits()
> . create type ExecContext
> . add comments
>
> This patch does not intend to change any behaviour. It passes make
> check, make installcheck and some manual tests. It might be hard to
> review, because some lines are affected by more than one change. If
> it's too much to swallow at once, I can provide it in smaller chunks ...
>
> Servus
> Manfred
> diff -Ncr ../base/src/backend/commands/vacuum.c src/backend/commands/vacuum.c
> *** ../base/src/backend/commands/vacuum.c Mon May 31 21:24:05 2004
> --- src/backend/commands/vacuum.c Wed Jun 2 21:46:59 2004
> ***************
> *** 99,104 ****
> --- 99,162 ----
> VTupleLink vtlinks;
> } VRelStats;
>
> + /*----------------------------------------------------------------------
> + * ExecContext:
> + *
> + * As these variables always appear together, we put them into one struct
> + * and pull initialization and cleanup into separate routines.
> + * ExecContext is used by repair_frag() and move_xxx_tuple(). More
> + * accurately: It is *used* only in move_xxx_tuple(), but because this
> + * routine is called many times, we initialize the struct just once in
> + * repair_frag() and pass it on to move_xxx_tuple().
> + */
> + typedef struct ExecContextData
> + {
> + ResultRelInfo *resultRelInfo;
> + EState *estate;
> + TupleTable tupleTable;
> + TupleTableSlot *slot;
> + } ExecContextData;
> + typedef ExecContextData *ExecContext;
> +
> + static void
> + ExecContext_Init(ExecContext ec, Relation rel)
> + {
> + TupleDesc tupdesc = RelationGetDescr(rel);
> +
> + /*
> + * We need a ResultRelInfo and an EState so we can use the regular
> + * executor's index-entry-making machinery.
> + */
> + ec->estate = CreateExecutorState();
> +
> + ec->resultRelInfo = makeNode(ResultRelInfo);
> + ec->resultRelInfo->ri_RangeTableIndex = 1; /* dummy */
> + ec->resultRelInfo->ri_RelationDesc = rel;
> + ec->resultRelInfo->ri_TrigDesc = NULL; /* we don't fire triggers */
> +
> + ExecOpenIndices(ec->resultRelInfo);
> +
> + ec->estate->es_result_relations = ec->resultRelInfo;
> + ec->estate->es_num_result_relations = 1;
> + ec->estate->es_result_relation_info = ec->resultRelInfo;
> +
> + /* Set up a dummy tuple table too */
> + ec->tupleTable = ExecCreateTupleTable(1);
> + ec->slot = ExecAllocTableSlot(ec->tupleTable);
> + ExecSetSlotDescriptor(ec->slot, tupdesc, false);
> + }
> +
> + static void
> + ExecContext_Finish(ExecContext ec)
> + {
> + ExecDropTupleTable(ec->tupleTable, true);
> + ExecCloseIndices(ec->resultRelInfo);
> + FreeExecutorState(ec->estate);
> + }
> + /*
> + * End of ExecContext Implementation
> + *----------------------------------------------------------------------
> + */
>
> static MemoryContext vac_context = NULL;
>
> ***************
> *** 122,127 ****
> --- 180,196 ----
> static void repair_frag(VRelStats *vacrelstats, Relation onerel,
> VacPageList vacuum_pages, VacPageList fraged_pages,
> int nindexes, Relation *Irel);
> + static void move_chain_tuple(Relation rel,
> + Buffer old_buf, Page old_page, HeapTuple old_tup,
> + Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
> + ExecContext ec, ItemPointer ctid, bool cleanVpd);
> + static void move_plain_tuple(Relation rel,
> + Buffer old_buf, Page old_page, HeapTuple old_tup,
> + Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
> + ExecContext ec);
> + static void update_hint_bits(Relation rel, VacPageList fraged_pages,
> + int num_fraged_pages, BlockNumber last_move_dest_block,
> + int num_moved);
> static void vacuum_heap(VRelStats *vacrelstats, Relation onerel,
> VacPageList vacpagelist);
> static void vacuum_page(Relation onerel, Buffer buffer, VacPage vacpage);
> ***************
> *** 675,681 ****
> static void
> vac_truncate_clog(TransactionId vacuumXID, TransactionId frozenXID)
> {
> ! TransactionId myXID;
> Relation relation;
> HeapScanDesc scan;
> HeapTuple tuple;
> --- 744,750 ----
> static void
> vac_truncate_clog(TransactionId vacuumXID, TransactionId frozenXID)
> {
> ! TransactionId myXID = GetCurrentTransactionId();
> Relation relation;
> HeapScanDesc scan;
> HeapTuple tuple;
> ***************
> *** 683,689 ****
> bool vacuumAlreadyWrapped = false;
> bool frozenAlreadyWrapped = false;
>
> - myXID = GetCurrentTransactionId();
>
> relation = heap_openr(DatabaseRelationName, AccessShareLock);
>
> --- 752,757 ----
> ***************
> *** 1059,1075 ****
> {
> BlockNumber nblocks,
> blkno;
> - ItemId itemid;
> - Buffer buf;
> HeapTupleData tuple;
> - OffsetNumber offnum,
> - maxoff;
> - bool pgchanged,
> - tupgone,
> - notup;
> char *relname;
> ! VacPage vacpage,
> ! vacpagecopy;
> BlockNumber empty_pages,
> empty_end_pages;
> double num_tuples,
> --- 1127,1135 ----
> {
> BlockNumber nblocks,
> blkno;
> HeapTupleData tuple;
> char *relname;
> ! VacPage vacpage;
> BlockNumber empty_pages,
> empty_end_pages;
> double num_tuples,
> ***************
> *** 1080,1086 ****
> usable_free_space;
> Size min_tlen = MaxTupleSize;
> Size max_tlen = 0;
> - int i;
> bool do_shrinking = true;
> VTupleLink vtlinks = (VTupleLink) palloc(100 * sizeof(VTupleLinkData));
> int num_vtlinks = 0;
> --- 1140,1145 ----
> ***************
> *** 1113,1118 ****
> --- 1172,1182 ----
> tempPage = NULL;
> bool do_reap,
> do_frag;
> + Buffer buf;
> + OffsetNumber offnum,
> + maxoff;
> + bool pgchanged,
> + notup;
>
> vacuum_delay_point();
>
> ***************
> *** 1125,1130 ****
> --- 1189,1196 ----
>
> if (PageIsNew(page))
> {
> + VacPage vacpagecopy;
> +
> ereport(WARNING,
> (errmsg("relation \"%s\" page %u is uninitialized --- fixing",
> relname, blkno)));
> ***************
> *** 1142,1147 ****
> --- 1208,1215 ----
>
> if (PageIsEmpty(page))
> {
> + VacPage vacpagecopy;
> +
> vacpage->free = ((PageHeader) page)->pd_upper - ((PageHeader) page)->pd_lower;
> free_space += vacpage->free;
> empty_pages++;
> ***************
> *** 1161,1168 ****
> offnum = OffsetNumberNext(offnum))
> {
> uint16 sv_infomask;
> !
> ! itemid = PageGetItemId(page, offnum);
>
> /*
> * Collect un-used items too - it's possible to have indexes
> --- 1229,1236 ----
> offnum = OffsetNumberNext(offnum))
> {
> uint16 sv_infomask;
> ! ItemId itemid = PageGetItemId(page, offnum);
> ! bool tupgone = false;
>
> /*
> * Collect un-used items too - it's possible to have indexes
> ***************
> *** 1180,1186 ****
> tuple.t_len = ItemIdGetLength(itemid);
> ItemPointerSet(&(tuple.t_self), blkno, offnum);
>
> - tupgone = false;
> sv_infomask = tuple.t_data->t_infomask;
>
> switch (HeapTupleSatisfiesVacuum(tuple.t_data, OldestXmin))
> --- 1248,1253 ----
> ***************
> *** 1269,1275 ****
> do_shrinking = false;
> break;
> default:
> ! elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
> break;
> }
>
> --- 1336,1343 ----
> do_shrinking = false;
> break;
> default:
> ! /* unexpected HeapTupleSatisfiesVacuum result */
> ! Assert(false);
> break;
> }
>
> ***************
> *** 1344,1350 ****
>
> if (do_reap || do_frag)
> {
> ! vacpagecopy = copy_vac_page(vacpage);
> if (do_reap)
> vpage_insert(vacuum_pages, vacpagecopy);
> if (do_frag)
> --- 1412,1418 ----
>
> if (do_reap || do_frag)
> {
> ! VacPage vacpagecopy = copy_vac_page(vacpage);
> if (do_reap)
> vpage_insert(vacuum_pages, vacpagecopy);
> if (do_frag)
> ***************
> *** 1390,1395 ****
> --- 1458,1465 ----
> */
> if (do_shrinking)
> {
> + int i;
> +
> Assert((BlockNumber) fraged_pages->num_pages >= empty_end_pages);
> fraged_pages->num_pages -= empty_end_pages;
> usable_free_space = 0;
> ***************
> *** 1453,1528 ****
> VacPageList vacuum_pages, VacPageList fraged_pages,
> int nindexes, Relation *Irel)
> {
> ! TransactionId myXID;
> ! CommandId myCID;
> ! Buffer buf,
> ! cur_buffer;
> BlockNumber nblocks,
> blkno;
> BlockNumber last_move_dest_block = 0,
> last_vacuum_block;
> ! Page page,
> ! ToPage = NULL;
> ! OffsetNumber offnum,
> ! maxoff,
> ! newoff,
> ! max_offset;
> ! ItemId itemid,
> ! newitemid;
> ! HeapTupleData tuple,
> ! newtup;
> ! TupleDesc tupdesc;
> ! ResultRelInfo *resultRelInfo;
> ! EState *estate;
> ! TupleTable tupleTable;
> ! TupleTableSlot *slot;
> VacPageListData Nvacpagelist;
> ! VacPage cur_page = NULL,
> last_vacuum_page,
> vacpage,
> *curpage;
> - int cur_item = 0;
> int i;
> ! Size tuple_len;
> ! int num_moved,
> num_fraged_pages,
> vacuumed_pages;
> ! int checked_moved,
> ! num_tuples,
> ! keep_tuples = 0;
> ! bool isempty,
> ! dowrite,
> ! chain_tuple_moved;
> VacRUsage ru0;
>
> vac_init_rusage(&ru0);
>
> ! myXID = GetCurrentTransactionId();
> ! myCID = GetCurrentCommandId();
> !
> ! tupdesc = RelationGetDescr(onerel);
> !
> ! /*
> ! * We need a ResultRelInfo and an EState so we can use the regular
> ! * executor's index-entry-making machinery.
> ! */
> ! estate = CreateExecutorState();
> !
> ! resultRelInfo = makeNode(ResultRelInfo);
> ! resultRelInfo->ri_RangeTableIndex = 1; /* dummy */
> ! resultRelInfo->ri_RelationDesc = onerel;
> ! resultRelInfo->ri_TrigDesc = NULL; /* we don't fire triggers */
> !
> ! ExecOpenIndices(resultRelInfo);
> !
> ! estate->es_result_relations = resultRelInfo;
> ! estate->es_num_result_relations = 1;
> ! estate->es_result_relation_info = resultRelInfo;
> !
> ! /* Set up a dummy tuple table too */
> ! tupleTable = ExecCreateTupleTable(1);
> ! slot = ExecAllocTableSlot(tupleTable);
> ! ExecSetSlotDescriptor(slot, tupdesc, false);
>
> Nvacpagelist.num_pages = 0;
> num_fraged_pages = fraged_pages->num_pages;
> --- 1523,1551 ----
> VacPageList vacuum_pages, VacPageList fraged_pages,
> int nindexes, Relation *Irel)
> {
> ! TransactionId myXID = GetCurrentTransactionId();
> ! Buffer dst_buffer = InvalidBuffer;
> BlockNumber nblocks,
> blkno;
> BlockNumber last_move_dest_block = 0,
> last_vacuum_block;
> ! Page dst_page = NULL;
> ! ExecContextData ec;
> VacPageListData Nvacpagelist;
> ! VacPage dst_vacpage = NULL,
> last_vacuum_page,
> vacpage,
> *curpage;
> int i;
> ! int num_moved = 0,
> num_fraged_pages,
> vacuumed_pages;
> ! int keep_tuples = 0;
> VacRUsage ru0;
>
> vac_init_rusage(&ru0);
>
> ! ExecContext_Init(&ec, onerel);
>
> Nvacpagelist.num_pages = 0;
> num_fraged_pages = fraged_pages->num_pages;
> ***************
> *** 1539,1546 ****
> last_vacuum_page = NULL;
> last_vacuum_block = InvalidBlockNumber;
> }
> - cur_buffer = InvalidBuffer;
> - num_moved = 0;
>
> vacpage = (VacPage) palloc(sizeof(VacPageData) + MaxOffsetNumber * sizeof(OffsetNumber));
> vacpage->offsets_used = vacpage->offsets_free = 0;
> --- 1562,1567 ----
> ***************
> *** 1560,1565 ****
> --- 1581,1594 ----
> blkno > last_move_dest_block;
> blkno--)
> {
> + Buffer buf;
> + Page page;
> + OffsetNumber offnum,
> + maxoff;
> + bool isempty,
> + dowrite,
> + chain_tuple_moved;
> +
> vacuum_delay_point();
>
> /*
> ***************
> *** 1635,1641 ****
> offnum <= maxoff;
> offnum = OffsetNumberNext(offnum))
> {
> ! itemid = PageGetItemId(page, offnum);
>
> if (!ItemIdIsUsed(itemid))
> continue;
> --- 1664,1672 ----
> offnum <= maxoff;
> offnum = OffsetNumberNext(offnum))
> {
> ! Size tuple_len;
> ! HeapTupleData tuple;
> ! ItemId itemid = PageGetItemId(page, offnum);
>
> if (!ItemIdIsUsed(itemid))
> continue;
> ***************
> *** 1645,1689 ****
> tuple_len = tuple.t_len = ItemIdGetLength(itemid);
> ItemPointerSet(&(tuple.t_self), blkno, offnum);
>
> if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
> {
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
> ! elog(ERROR, "HEAP_MOVED_IN was not expected");
>
> /*
> * If this (chain) tuple is moved by me already then I
> * have to check is it in vacpage or not - i.e. is it
> * moved while cleaning this page or some previous one.
> */
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
> ! {
> ! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
> ! elog(ERROR, "invalid XVAC in tuple header");
> ! if (keep_tuples == 0)
> ! continue;
> ! if (chain_tuple_moved) /* some chains was moved
> ! * while */
> ! { /* cleaning this page */
> ! Assert(vacpage->offsets_free > 0);
> ! for (i = 0; i < vacpage->offsets_free; i++)
> ! {
> ! if (vacpage->offsets[i] == offnum)
> ! break;
> ! }
> ! if (i >= vacpage->offsets_free) /* not found */
> ! {
> ! vacpage->offsets[vacpage->offsets_free++] = offnum;
> ! keep_tuples--;
> ! }
> }
> ! else
> {
> vacpage->offsets[vacpage->offsets_free++] = offnum;
> keep_tuples--;
> }
> - continue;
> }
> ! elog(ERROR, "HEAP_MOVED_OFF was expected");
> }
>
> /*
> --- 1676,1746 ----
> tuple_len = tuple.t_len = ItemIdGetLength(itemid);
> ItemPointerSet(&(tuple.t_self), blkno, offnum);
>
> + /*
> + * VACUUM FULL has an exclusive lock on the relation. So
> + * normally no other transaction can have pending INSERTs or
> + * DELETEs in this relation. A tuple is either
> + * (a) a tuple in a system catalog, inserted or deleted by
> + * a not yet committed transaction or
> + * (b) dead (XMIN_INVALID or XMAX_COMMITTED) or
> + * (c) inserted by a committed xact (XMIN_COMMITTED) or
> + * (d) moved by the currently running VACUUM.
> + * In case (a) we wouldn't be in repair_frag() at all.
> + * In case (b) we cannot be here, because scan_heap() has
> + * already marked the item as unused, see continue above.
> + * Case (c) is what normally is to be expected.
> + * Case (d) is only possible, if a whole tuple chain has been
> + * moved while processing this or a higher numbered block.
> + */
> if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
> {
> ! /*
> ! * There cannot be another concurrently running VACUUM. If
> ! * the tuple had been moved in by a previous VACUUM, the
> ! * visibility check would have set XMIN_COMMITTED. If the
> ! * tuple had been moved in by the currently running VACUUM,
> ! * the loop would have been terminated. We had
> ! * elog(ERROR, ...) here, but as we are testing for a
> ! * can't-happen condition, Assert() seems more appropriate.
> ! */
> ! Assert(!(tuple.t_data->t_infomask & HEAP_MOVED_IN));
>
> /*
> * If this (chain) tuple is moved by me already then I
> * have to check is it in vacpage or not - i.e. is it
> * moved while cleaning this page or some previous one.
> */
> ! Assert(tuple.t_data->t_infomask & HEAP_MOVED_OFF);
> ! /*
> ! * MOVED_OFF by another VACUUM would have caused the
> ! * visibility check to set XMIN_COMMITTED or XMIN_INVALID.
> ! */
> ! Assert(HeapTupleHeaderGetXvac(tuple.t_data) == myXID);
> !
> ! /* Can't we Assert(keep_tuples > 0) here? */
> ! if (keep_tuples == 0)
> ! continue;
> ! if (chain_tuple_moved) /* some chains was moved
> ! * while */
> ! { /* cleaning this page */
> ! Assert(vacpage->offsets_free > 0);
> ! for (i = 0; i < vacpage->offsets_free; i++)
> ! {
> ! if (vacpage->offsets[i] == offnum)
> ! break;
> }
> ! if (i >= vacpage->offsets_free) /* not found */
> {
> vacpage->offsets[vacpage->offsets_free++] = offnum;
> keep_tuples--;
> }
> }
> ! else
> ! {
> ! vacpage->offsets[vacpage->offsets_free++] = offnum;
> ! keep_tuples--;
> ! }
> ! continue;
> }
>
> /*
> ***************
> *** 1716,1723 ****
> Buffer Cbuf = buf;
> bool freeCbuf = false;
> bool chain_move_failed = false;
> - Page Cpage;
> - ItemId Citemid;
> ItemPointerData Ctid;
> HeapTupleData tp = tuple;
> Size tlen = tuple_len;
> --- 1773,1778 ----
> ***************
> *** 1728,1737 ****
> int to_item = 0;
> int ti;
>
> ! if (cur_buffer != InvalidBuffer)
> {
> ! WriteBuffer(cur_buffer);
> ! cur_buffer = InvalidBuffer;
> }
>
> /* Quick exit if we have no vtlinks to search in */
> --- 1783,1792 ----
> int to_item = 0;
> int ti;
>
> ! if (dst_buffer != InvalidBuffer)
> {
> ! WriteBuffer(dst_buffer);
> ! dst_buffer = InvalidBuffer;
> }
>
> /* Quick exit if we have no vtlinks to search in */
> ***************
> *** 1754,1759 ****
> --- 1809,1818 ----
> !(ItemPointerEquals(&(tp.t_self),
> &(tp.t_data->t_ctid))))
> {
> + Page Cpage;
> + ItemId Citemid;
> + ItemPointerData Ctid;
> +
> Ctid = tp.t_data->t_ctid;
> if (freeCbuf)
> ReleaseBuffer(Cbuf);
> ***************
> *** 1929,1940 ****
> }
>
> /*
> ! * Okay, move the whle tuple chain
> */
> ItemPointerSetInvalid(&Ctid);
> for (ti = 0; ti < num_vtmove; ti++)
> {
> VacPage destvacpage = vtmove[ti].vacpage;
>
> /* Get page to move from */
> tuple.t_self = vtmove[ti].tid;
> --- 1988,2001 ----
> }
>
> /*
> ! * Okay, move the whole tuple chain
> */
> ItemPointerSetInvalid(&Ctid);
> for (ti = 0; ti < num_vtmove; ti++)
> {
> VacPage destvacpage = vtmove[ti].vacpage;
> + Page Cpage;
> + ItemId Citemid;
>
> /* Get page to move from */
> tuple.t_self = vtmove[ti].tid;
> ***************
> *** 1942,1954 ****
> ItemPointerGetBlockNumber(&(tuple.t_self)));
>
> /* Get page to move to */
> ! cur_buffer = ReadBuffer(onerel, destvacpage->blkno);
>
> ! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
> ! if (cur_buffer != Cbuf)
> LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
>
> ! ToPage = BufferGetPage(cur_buffer);
> Cpage = BufferGetPage(Cbuf);
>
> Citemid = PageGetItemId(Cpage,
> --- 2003,2015 ----
> ItemPointerGetBlockNumber(&(tuple.t_self)));
>
> /* Get page to move to */
> ! dst_buffer = ReadBuffer(onerel, destvacpage->blkno);
>
> ! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
> ! if (dst_buffer != Cbuf)
> LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
>
> ! dst_page = BufferGetPage(dst_buffer);
> Cpage = BufferGetPage(Cbuf);
>
> Citemid = PageGetItemId(Cpage,
> ***************
> *** 1961,2081 ****
> * make a copy of the source tuple, and then mark the
> * source tuple MOVED_OFF.
> */
> ! heap_copytuple_with_tuple(&tuple, &newtup);
> !
> ! /*
> ! * register invalidation of source tuple in catcaches.
> ! */
> ! CacheInvalidateHeapTuple(onerel, &tuple);
> !
> ! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
> ! START_CRIT_SECTION();
> !
> ! tuple.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_IN);
> ! tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
> ! HeapTupleHeaderSetXvac(tuple.t_data, myXID);
> !
> ! /*
> ! * If this page was not used before - clean it.
> ! *
> ! * NOTE: a nasty bug used to lurk here. It is possible
> ! * for the source and destination pages to be the same
> ! * (since this tuple-chain member can be on a page
> ! * lower than the one we're currently processing in
> ! * the outer loop). If that's true, then after
> ! * vacuum_page() the source tuple will have been
> ! * moved, and tuple.t_data will be pointing at
> ! * garbage. Therefore we must do everything that uses
> ! * tuple.t_data BEFORE this step!!
> ! *
> ! * This path is different from the other callers of
> ! * vacuum_page, because we have already incremented
> ! * the vacpage's offsets_used field to account for the
> ! * tuple(s) we expect to move onto the page. Therefore
> ! * vacuum_page's check for offsets_used == 0 is wrong.
> ! * But since that's a good debugging check for all
> ! * other callers, we work around it here rather than
> ! * remove it.
> ! */
> ! if (!PageIsEmpty(ToPage) && vtmove[ti].cleanVpd)
> ! {
> ! int sv_offsets_used = destvacpage->offsets_used;
> !
> ! destvacpage->offsets_used = 0;
> ! vacuum_page(onerel, cur_buffer, destvacpage);
> ! destvacpage->offsets_used = sv_offsets_used;
> ! }
> !
> ! /*
> ! * Update the state of the copied tuple, and store it
> ! * on the destination page.
> ! */
> ! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_OFF);
> ! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
> ! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
> ! newoff = PageAddItem(ToPage,
> ! (Item) newtup.t_data,
> ! tuple_len,
> ! InvalidOffsetNumber,
> ! LP_USED);
> ! if (newoff == InvalidOffsetNumber)
> ! {
> ! elog(PANIC, "failed to add item with len = %lu to page %u while moving tuple chain",
> ! (unsigned long) tuple_len, destvacpage->blkno);
> ! }
> ! newitemid = PageGetItemId(ToPage, newoff);
> ! pfree(newtup.t_data);
> ! newtup.t_datamcxt = NULL;
> ! newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
> ! ItemPointerSet(&(newtup.t_self), destvacpage->blkno, newoff);
> !
> ! /* XLOG stuff */
> ! if (!onerel->rd_istemp)
> ! {
> ! XLogRecPtr recptr =
> ! log_heap_move(onerel, Cbuf, tuple.t_self,
> ! cur_buffer, &newtup);
> !
> ! if (Cbuf != cur_buffer)
> ! {
> ! PageSetLSN(Cpage, recptr);
> ! PageSetSUI(Cpage, ThisStartUpID);
> ! }
> ! PageSetLSN(ToPage, recptr);
> ! PageSetSUI(ToPage, ThisStartUpID);
> ! }
> ! else
> ! {
> ! /*
> ! * No XLOG record, but still need to flag that XID
> ! * exists on disk
> ! */
> ! MyXactMadeTempRelUpdate = true;
> ! }
> !
> ! END_CRIT_SECTION();
>
> if (destvacpage->blkno > last_move_dest_block)
> last_move_dest_block = destvacpage->blkno;
>
> /*
> - * Set new tuple's t_ctid pointing to itself for last
> - * tuple in chain, and to next tuple in chain
> - * otherwise.
> - */
> - if (!ItemPointerIsValid(&Ctid))
> - newtup.t_data->t_ctid = newtup.t_self;
> - else
> - newtup.t_data->t_ctid = Ctid;
> - Ctid = newtup.t_self;
> -
> - num_moved++;
> -
> - /*
> * Remember that we moved tuple from the current page
> * (corresponding index tuple will be cleaned).
> */
> --- 2022,2036 ----
> * make a copy of the source tuple, and then mark the
> * source tuple MOVED_OFF.
> */
> ! move_chain_tuple(onerel, Cbuf, Cpage, &tuple,
> ! dst_buffer, dst_page, destvacpage,
> ! &ec, &Ctid, vtmove[ti].cleanVpd);
>
> + num_moved++;
> if (destvacpage->blkno > last_move_dest_block)
> last_move_dest_block = destvacpage->blkno;
>
> /*
> * Remember that we moved tuple from the current page
> * (corresponding index tuple will be cleaned).
> */
> ***************
> *** 2085,2107 ****
> else
> keep_tuples++;
>
> ! LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
> ! if (cur_buffer != Cbuf)
> ! LockBuffer(Cbuf, BUFFER_LOCK_UNLOCK);
> !
> ! /* Create index entries for the moved tuple */
> ! if (resultRelInfo->ri_NumIndices > 0)
> ! {
> ! ExecStoreTuple(&newtup, slot, InvalidBuffer, false);
> ! ExecInsertIndexTuples(slot, &(newtup.t_self),
> ! estate, true);
> ! }
> !
> ! WriteBuffer(cur_buffer);
> WriteBuffer(Cbuf);
> } /* end of move-the-tuple-chain loop */
>
> ! cur_buffer = InvalidBuffer;
> pfree(vtmove);
> chain_tuple_moved = true;
>
> --- 2040,2050 ----
> else
> keep_tuples++;
>
> ! WriteBuffer(dst_buffer);
> WriteBuffer(Cbuf);
> } /* end of move-the-tuple-chain loop */
>
> ! dst_buffer = InvalidBuffer;
> pfree(vtmove);
> chain_tuple_moved = true;
>
> ***************
> *** 2110,2122 ****
> } /* end of is-tuple-in-chain test */
>
> /* try to find new page for this tuple */
> ! if (cur_buffer == InvalidBuffer ||
> ! !enough_space(cur_page, tuple_len))
> {
> ! if (cur_buffer != InvalidBuffer)
> {
> ! WriteBuffer(cur_buffer);
> ! cur_buffer = InvalidBuffer;
> }
> for (i = 0; i < num_fraged_pages; i++)
> {
> --- 2053,2065 ----
> } /* end of is-tuple-in-chain test */
>
> /* try to find new page for this tuple */
> ! if (dst_buffer == InvalidBuffer ||
> ! !enough_space(dst_vacpage, tuple_len))
> {
> ! if (dst_buffer != InvalidBuffer)
> {
> ! WriteBuffer(dst_buffer);
> ! dst_buffer = InvalidBuffer;
> }
> for (i = 0; i < num_fraged_pages; i++)
> {
> ***************
> *** 2125,2234 ****
> }
> if (i == num_fraged_pages)
> break; /* can't move item anywhere */
> ! cur_item = i;
> ! cur_page = fraged_pages->pagedesc[cur_item];
> ! cur_buffer = ReadBuffer(onerel, cur_page->blkno);
> ! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
> ! ToPage = BufferGetPage(cur_buffer);
> /* if this page was not used before - clean it */
> ! if (!PageIsEmpty(ToPage) && cur_page->offsets_used == 0)
> ! vacuum_page(onerel, cur_buffer, cur_page);
> }
> else
> ! LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
>
> LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
>
> ! /* copy tuple */
> ! heap_copytuple_with_tuple(&tuple, &newtup);
> !
> ! /*
> ! * register invalidation of source tuple in catcaches.
> ! *
> ! * (Note: we do not need to register the copied tuple, because we
> ! * are not changing the tuple contents and so there cannot be
> ! * any need to flush negative catcache entries.)
> ! */
> ! CacheInvalidateHeapTuple(onerel, &tuple);
>
> - /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
> - START_CRIT_SECTION();
>
> ! /*
> ! * Mark new tuple as MOVED_IN by me.
> ! */
> ! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_OFF);
> ! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
> ! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
> !
> ! /* add tuple to the page */
> ! newoff = PageAddItem(ToPage, (Item) newtup.t_data, tuple_len,
> ! InvalidOffsetNumber, LP_USED);
> ! if (newoff == InvalidOffsetNumber)
> ! {
> ! elog(PANIC, "failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)",
> ! (unsigned long) tuple_len,
> ! cur_page->blkno, (unsigned long) cur_page->free,
> ! cur_page->offsets_used, cur_page->offsets_free);
> ! }
> ! newitemid = PageGetItemId(ToPage, newoff);
> ! pfree(newtup.t_data);
> ! newtup.t_datamcxt = NULL;
> ! newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
> ! ItemPointerSet(&(newtup.t_data->t_ctid), cur_page->blkno, newoff);
> ! newtup.t_self = newtup.t_data->t_ctid;
>
> /*
> ! * Mark old tuple as MOVED_OFF by me.
> */
> - tuple.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> - HEAP_XMIN_INVALID |
> - HEAP_MOVED_IN);
> - tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
> - HeapTupleHeaderSetXvac(tuple.t_data, myXID);
> -
> - /* XLOG stuff */
> - if (!onerel->rd_istemp)
> - {
> - XLogRecPtr recptr =
> - log_heap_move(onerel, buf, tuple.t_self,
> - cur_buffer, &newtup);
> -
> - PageSetLSN(page, recptr);
> - PageSetSUI(page, ThisStartUpID);
> - PageSetLSN(ToPage, recptr);
> - PageSetSUI(ToPage, ThisStartUpID);
> - }
> - else
> - {
> - /*
> - * No XLOG record, but still need to flag that XID exists
> - * on disk
> - */
> - MyXactMadeTempRelUpdate = true;
> - }
> -
> - END_CRIT_SECTION();
> -
> - cur_page->offsets_used++;
> - num_moved++;
> - cur_page->free = ((PageHeader) ToPage)->pd_upper - ((PageHeader) ToPage)->pd_lower;
> - if (cur_page->blkno > last_move_dest_block)
> - last_move_dest_block = cur_page->blkno;
> -
> vacpage->offsets[vacpage->offsets_free++] = offnum;
> -
> - LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
> - LockBuffer(buf, BUFFER_LOCK_UNLOCK);
> -
> - /* insert index' tuples if needed */
> - if (resultRelInfo->ri_NumIndices > 0)
> - {
> - ExecStoreTuple(&newtup, slot, InvalidBuffer, false);
> - ExecInsertIndexTuples(slot, &(newtup.t_self), estate, true);
> - }
> } /* walk along page */
>
> /*
> --- 2068,2099 ----
> }
> if (i == num_fraged_pages)
> break; /* can't move item anywhere */
> ! dst_vacpage = fraged_pages->pagedesc[i];
> ! dst_buffer = ReadBuffer(onerel, dst_vacpage->blkno);
> ! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
> ! dst_page = BufferGetPage(dst_buffer);
> /* if this page was not used before - clean it */
> ! if (!PageIsEmpty(dst_page) && dst_vacpage->offsets_used == 0)
> ! vacuum_page(onerel, dst_buffer, dst_vacpage);
> }
> else
> ! LockBuffer(dst_buffer, BUFFER_LOCK_EXCLUSIVE);
>
> LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
>
> ! move_plain_tuple(onerel, buf, page, &tuple,
> ! dst_buffer, dst_page, dst_vacpage, &ec);
>
>
> ! num_moved++;
> ! if (dst_vacpage->blkno > last_move_dest_block)
> ! last_move_dest_block = dst_vacpage->blkno;
>
> /*
> ! * Remember that we moved tuple from the current page
> ! * (corresponding index tuple will be cleaned).
> */
> vacpage->offsets[vacpage->offsets_free++] = offnum;
> } /* walk along page */
>
> /*
> ***************
> *** 2249,2284 ****
> off <= maxoff;
> off = OffsetNumberNext(off))
> {
> ! itemid = PageGetItemId(page, off);
> if (!ItemIdIsUsed(itemid))
> continue;
> ! tuple.t_datamcxt = NULL;
> ! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
> ! if (tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED)
> continue;
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
> ! elog(ERROR, "HEAP_MOVED_IN was not expected");
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
> {
> ! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
> ! elog(ERROR, "invalid XVAC in tuple header");
> ! /* some chains was moved while */
> ! if (chain_tuple_moved)
> ! { /* cleaning this page */
> ! Assert(vacpage->offsets_free > 0);
> ! for (i = 0; i < vacpage->offsets_free; i++)
> ! {
> ! if (vacpage->offsets[i] == off)
> ! break;
> ! }
> ! if (i >= vacpage->offsets_free) /* not found */
> ! {
> ! vacpage->offsets[vacpage->offsets_free++] = off;
> ! Assert(keep_tuples > 0);
> ! keep_tuples--;
> ! }
> }
> ! else
> {
> vacpage->offsets[vacpage->offsets_free++] = off;
> Assert(keep_tuples > 0);
> --- 2114,2144 ----
> off <= maxoff;
> off = OffsetNumberNext(off))
> {
> ! ItemId itemid = PageGetItemId(page, off);
> ! HeapTupleHeader htup;
> !
> if (!ItemIdIsUsed(itemid))
> continue;
> ! htup = (HeapTupleHeader) PageGetItem(page, itemid);
> ! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
> continue;
> ! /*
> ! ** See comments in the walk-along-page loop above, why we
> ! ** have Asserts here instead of if (...) elog(ERROR).
> ! */
> ! Assert(!(htup->t_infomask & HEAP_MOVED_IN));
> ! Assert(htup->t_infomask & HEAP_MOVED_OFF);
> ! Assert(HeapTupleHeaderGetXvac(htup) == myXID);
> ! if (chain_tuple_moved)
> {
> ! /* some chains was moved while cleaning this page */
> ! Assert(vacpage->offsets_free > 0);
> ! for (i = 0; i < vacpage->offsets_free; i++)
> ! {
> ! if (vacpage->offsets[i] == off)
> ! break;
> }
> ! if (i >= vacpage->offsets_free) /* not found */
> {
> vacpage->offsets[vacpage->offsets_free++] = off;
> Assert(keep_tuples > 0);
> ***************
> *** 2286,2292 ****
> }
> }
> else
> ! elog(ERROR, "HEAP_MOVED_OFF was expected");
> }
> }
>
> --- 2146,2156 ----
> }
> }
> else
> ! {
> ! vacpage->offsets[vacpage->offsets_free++] = off;
> ! Assert(keep_tuples > 0);
> ! keep_tuples--;
> ! }
> }
> }
>
> ***************
> *** 2312,2321 ****
>
> blkno++; /* new number of blocks */
>
> ! if (cur_buffer != InvalidBuffer)
> {
> Assert(num_moved > 0);
> ! WriteBuffer(cur_buffer);
> }
>
> if (num_moved > 0)
> --- 2176,2185 ----
>
> blkno++; /* new number of blocks */
>
> ! if (dst_buffer != InvalidBuffer)
> {
> Assert(num_moved > 0);
> ! WriteBuffer(dst_buffer);
> }
>
> if (num_moved > 0)
> ***************
> *** 2348,2353 ****
> --- 2212,2220 ----
> Assert((*curpage)->blkno < blkno);
> if ((*curpage)->offsets_used == 0)
> {
> + Buffer buf;
> + Page page;
> +
> /* this page was not used as a move target, so must clean it */
> buf = ReadBuffer(onerel, (*curpage)->blkno);
> LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
> ***************
> *** 2363,2424 ****
> * Now scan all the pages that we moved tuples onto and update tuple
> * status bits. This is not really necessary, but will save time for
> * future transactions examining these tuples.
> - *
> - * XXX NOTICE that this code fails to clear HEAP_MOVED_OFF tuples from
> - * pages that were move source pages but not move dest pages. One
> - * also wonders whether it wouldn't be better to skip this step and
> - * let the tuple status updates happen someplace that's not holding an
> - * exclusive lock on the relation.
> */
> ! checked_moved = 0;
> ! for (i = 0, curpage = fraged_pages->pagedesc;
> ! i < num_fraged_pages;
> ! i++, curpage++)
> ! {
> ! vacuum_delay_point();
> !
> ! Assert((*curpage)->blkno < blkno);
> ! if ((*curpage)->blkno > last_move_dest_block)
> ! break; /* no need to scan any further */
> ! if ((*curpage)->offsets_used == 0)
> ! continue; /* this page was never used as a move dest */
> ! buf = ReadBuffer(onerel, (*curpage)->blkno);
> ! LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
> ! page = BufferGetPage(buf);
> ! num_tuples = 0;
> ! max_offset = PageGetMaxOffsetNumber(page);
> ! for (newoff = FirstOffsetNumber;
> ! newoff <= max_offset;
> ! newoff = OffsetNumberNext(newoff))
> ! {
> ! itemid = PageGetItemId(page, newoff);
> ! if (!ItemIdIsUsed(itemid))
> ! continue;
> ! tuple.t_datamcxt = NULL;
> ! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
> ! if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
> ! {
> ! if (!(tuple.t_data->t_infomask & HEAP_MOVED))
> ! elog(ERROR, "HEAP_MOVED_OFF/HEAP_MOVED_IN was expected");
> ! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
> ! elog(ERROR, "invalid XVAC in tuple header");
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
> ! {
> ! tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
> ! tuple.t_data->t_infomask &= ~HEAP_MOVED;
> ! num_tuples++;
> ! }
> ! else
> ! tuple.t_data->t_infomask |= HEAP_XMIN_INVALID;
> ! }
> ! }
> ! LockBuffer(buf, BUFFER_LOCK_UNLOCK);
> ! WriteBuffer(buf);
> ! Assert((*curpage)->offsets_used == num_tuples);
> ! checked_moved += num_tuples;
> ! }
> ! Assert(num_moved == checked_moved);
> !
> /*
> * It'd be cleaner to make this report at the bottom of this routine,
> * but then the rusage would double-count the second pass of index
> --- 2230,2239 ----
> * Now scan all the pages that we moved tuples onto and update tuple
> * status bits. This is not really necessary, but will save time for
> * future transactions examining these tuples.
> */
> ! update_hint_bits(onerel, fraged_pages, num_fraged_pages,
> ! last_move_dest_block, num_moved);
> !
> /*
> * It'd be cleaner to make this report at the bottom of this routine,
> * but then the rusage would double-count the second pass of index
> ***************
> *** 2455,2460 ****
> --- 2270,2282 ----
> *vpleft = *vpright;
> *vpright = vpsave;
> }
> + /*
> + * keep_tuples is the number of tuples that have been moved
> + * off a page during chain moves but not been scanned over
> + * subsequently. The tuple ids of these tuples are not
> + * recorded as free offsets for any VacPage, so they will not
> + * be cleared from the indexes.
> + */
> Assert(keep_tuples >= 0);
> for (i = 0; i < nindexes; i++)
> vacuum_index(&Nvacpagelist, Irel[i],
> ***************
> *** 2465,2500 ****
> if (vacpage->blkno == (blkno - 1) &&
> vacpage->offsets_free > 0)
> {
> ! OffsetNumber unused[BLCKSZ / sizeof(OffsetNumber)];
> ! int uncnt;
>
> buf = ReadBuffer(onerel, vacpage->blkno);
> LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
> page = BufferGetPage(buf);
> - num_tuples = 0;
> maxoff = PageGetMaxOffsetNumber(page);
> for (offnum = FirstOffsetNumber;
> offnum <= maxoff;
> offnum = OffsetNumberNext(offnum))
> {
> ! itemid = PageGetItemId(page, offnum);
> if (!ItemIdIsUsed(itemid))
> continue;
> ! tuple.t_datamcxt = NULL;
> ! tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
>
> ! if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
> ! {
> ! if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
> ! {
> ! if (HeapTupleHeaderGetXvac(tuple.t_data) != myXID)
> ! elog(ERROR, "invalid XVAC in tuple header");
> ! itemid->lp_flags &= ~LP_USED;
> ! num_tuples++;
> ! }
> ! else
> ! elog(ERROR, "HEAP_MOVED_OFF was expected");
> ! }
>
> }
> Assert(vacpage->offsets_free == num_tuples);
> --- 2287,2327 ----
> if (vacpage->blkno == (blkno - 1) &&
> vacpage->offsets_free > 0)
> {
> ! Buffer buf;
> ! Page page;
> ! OffsetNumber unused[BLCKSZ / sizeof(OffsetNumber)];
> ! OffsetNumber offnum,
> ! maxoff;
> ! int uncnt;
> ! int num_tuples = 0;
>
> buf = ReadBuffer(onerel, vacpage->blkno);
> LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
> page = BufferGetPage(buf);
> maxoff = PageGetMaxOffsetNumber(page);
> for (offnum = FirstOffsetNumber;
> offnum <= maxoff;
> offnum = OffsetNumberNext(offnum))
> {
> ! ItemId itemid = PageGetItemId(page, offnum);
> ! HeapTupleHeader htup;
> !
> if (!ItemIdIsUsed(itemid))
> continue;
> ! htup = (HeapTupleHeader) PageGetItem(page, itemid);
> ! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
> ! continue;
>
> ! /*
> ! ** See comments in the walk-along-page loop above, why we
> ! ** have Asserts here instead of if (...) elog(ERROR).
> ! */
> ! Assert(!(htup->t_infomask & HEAP_MOVED_IN));
> ! Assert(htup->t_infomask & HEAP_MOVED_OFF);
> ! Assert(HeapTupleHeaderGetXvac(htup) == myXID);
> !
> ! itemid->lp_flags &= ~LP_USED;
> ! num_tuples++;
>
> }
> Assert(vacpage->offsets_free == num_tuples);
> ***************
> *** 2554,2564 ****
> if (vacrelstats->vtlinks != NULL)
> pfree(vacrelstats->vtlinks);
>
> ! ExecDropTupleTable(tupleTable, true);
>
> ! ExecCloseIndices(resultRelInfo);
>
> ! FreeExecutorState(estate);
> }
>
> /*
> --- 2381,2717 ----
> if (vacrelstats->vtlinks != NULL)
> pfree(vacrelstats->vtlinks);
>
> ! ExecContext_Finish(&ec);
> ! }
> !
> ! /*
> ! * move_chain_tuple() -- move one tuple that is part of a tuple chain
> ! *
> ! * This routine moves old_tup from old_page to dst_page.
> ! * old_page and dst_page might be the same page.
> ! * On entry old_buf and dst_buf are locked exclusively, both locks (or
> ! * the single lock, if this is a intra-page-move) are released before
> ! * exit.
> ! *
> ! * Yes, a routine with ten parameters is ugly, but it's still better
> ! * than having these 120 lines of code in repair_frag() which is
> ! * already too long and almost unreadable.
> ! */
> ! static void
> ! move_chain_tuple(Relation rel,
> ! Buffer old_buf, Page old_page, HeapTuple old_tup,
> ! Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
> ! ExecContext ec, ItemPointer ctid, bool cleanVpd)
> ! {
> ! TransactionId myXID = GetCurrentTransactionId();
> ! HeapTupleData newtup;
> ! OffsetNumber newoff;
> ! ItemId newitemid;
> ! Size tuple_len = old_tup->t_len;
> !
> ! heap_copytuple_with_tuple(old_tup, &newtup);
> !
> ! /*
> ! * register invalidation of source tuple in catcaches.
> ! */
> ! CacheInvalidateHeapTuple(rel, old_tup);
> !
> ! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
> ! START_CRIT_SECTION();
> !
> ! old_tup->t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_IN);
> ! old_tup->t_data->t_infomask |= HEAP_MOVED_OFF;
> ! HeapTupleHeaderSetXvac(old_tup->t_data, myXID);
> !
> ! /*
> ! * If this page was not used before - clean it.
> ! *
> ! * NOTE: a nasty bug used to lurk here. It is possible
> ! * for the source and destination pages to be the same
> ! * (since this tuple-chain member can be on a page
> ! * lower than the one we're currently processing in
> ! * the outer loop). If that's true, then after
> ! * vacuum_page() the source tuple will have been
> ! * moved, and tuple.t_data will be pointing at
> ! * garbage. Therefore we must do everything that uses
> ! * old_tup->t_data BEFORE this step!!
> ! *
> ! * This path is different from the other callers of
> ! * vacuum_page, because we have already incremented
> ! * the vacpage's offsets_used field to account for the
> ! * tuple(s) we expect to move onto the page. Therefore
> ! * vacuum_page's check for offsets_used == 0 is wrong.
> ! * But since that's a good debugging check for all
> ! * other callers, we work around it here rather than
> ! * remove it.
> ! */
> ! if (!PageIsEmpty(dst_page) && cleanVpd)
> ! {
> ! int sv_offsets_used = dst_vacpage->offsets_used;
> !
> ! dst_vacpage->offsets_used = 0;
> ! vacuum_page(rel, dst_buf, dst_vacpage);
> ! dst_vacpage->offsets_used = sv_offsets_used;
> ! }
>
> ! /*
> ! * Update the state of the copied tuple, and store it
> ! * on the destination page.
> ! */
> ! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_OFF);
> ! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
> ! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
> ! newoff = PageAddItem(dst_page, (Item) newtup.t_data, tuple_len,
> ! InvalidOffsetNumber, LP_USED);
> ! if (newoff == InvalidOffsetNumber)
> ! {
> ! elog(PANIC, "failed to add item with len = %lu to page %u while moving tuple chain",
> ! (unsigned long) tuple_len, dst_vacpage->blkno);
> ! }
> ! newitemid = PageGetItemId(dst_page, newoff);
> ! pfree(newtup.t_data);
> ! newtup.t_datamcxt = NULL;
> ! newtup.t_data = (HeapTupleHeader) PageGetItem(dst_page, newitemid);
> ! ItemPointerSet(&(newtup.t_self), dst_vacpage->blkno, newoff);
> !
> ! /* XLOG stuff */
> ! if (!rel->rd_istemp)
> ! {
> ! XLogRecPtr recptr = log_heap_move(rel, old_buf, old_tup->t_self,
> ! dst_buf, &newtup);
> !
> ! if (old_buf != dst_buf)
> ! {
> ! PageSetLSN(old_page, recptr);
> ! PageSetSUI(old_page, ThisStartUpID);
> ! }
> ! PageSetLSN(dst_page, recptr);
> ! PageSetSUI(dst_page, ThisStartUpID);
> ! }
> ! else
> ! {
> ! /*
> ! * No XLOG record, but still need to flag that XID
> ! * exists on disk
> ! */
> ! MyXactMadeTempRelUpdate = true;
> ! }
> !
> ! END_CRIT_SECTION();
> !
> ! /*
> ! * Set new tuple's t_ctid pointing to itself for last
> ! * tuple in chain, and to next tuple in chain
> ! * otherwise.
> ! */
> ! /* Is this ok after log_heap_move() and END_CRIT_SECTION()? */
> ! if (!ItemPointerIsValid(ctid))
> ! newtup.t_data->t_ctid = newtup.t_self;
> ! else
> ! newtup.t_data->t_ctid = *ctid;
> ! *ctid = newtup.t_self;
>
> ! LockBuffer(dst_buf, BUFFER_LOCK_UNLOCK);
> ! if (dst_buf != old_buf)
> ! LockBuffer(old_buf, BUFFER_LOCK_UNLOCK);
> !
> ! /* Create index entries for the moved tuple */
> ! if (ec->resultRelInfo->ri_NumIndices > 0)
> ! {
> ! ExecStoreTuple(&newtup, ec->slot, InvalidBuffer, false);
> ! ExecInsertIndexTuples(ec->slot, &(newtup.t_self), ec->estate, true);
> ! }
> ! }
> !
> ! /*
> ! * move_plain_tuple() -- move one tuple that is not part of a chain
> ! *
> ! * This routine moves old_tup from old_page to dst_page.
> ! * On entry old_buf and dst_buf are locked exclusively, both locks are
> ! * released before exit.
> ! *
> ! * Yes, a routine with eight parameters is ugly, but it's still better
> ! * than having these 90 lines of code in repair_frag() which is already
> ! * too long and almost unreadable.
> ! */
> ! static void
> ! move_plain_tuple(Relation rel,
> ! Buffer old_buf, Page old_page, HeapTuple old_tup,
> ! Buffer dst_buf, Page dst_page, VacPage dst_vacpage,
> ! ExecContext ec)
> ! {
> ! TransactionId myXID = GetCurrentTransactionId();
> ! HeapTupleData newtup;
> ! OffsetNumber newoff;
> ! ItemId newitemid;
> ! Size tuple_len = old_tup->t_len;
> !
> ! /* copy tuple */
> ! heap_copytuple_with_tuple(old_tup, &newtup);
> !
> ! /*
> ! * register invalidation of source tuple in catcaches.
> ! *
> ! * (Note: we do not need to register the copied tuple, because we
> ! * are not changing the tuple contents and so there cannot be
> ! * any need to flush negative catcache entries.)
> ! */
> ! CacheInvalidateHeapTuple(rel, old_tup);
> !
> ! /* NO EREPORT(ERROR) TILL CHANGES ARE LOGGED */
> ! START_CRIT_SECTION();
> !
> ! /*
> ! * Mark new tuple as MOVED_IN by me.
> ! */
> ! newtup.t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_OFF);
> ! newtup.t_data->t_infomask |= HEAP_MOVED_IN;
> ! HeapTupleHeaderSetXvac(newtup.t_data, myXID);
> !
> ! /* add tuple to the page */
> ! newoff = PageAddItem(dst_page, (Item) newtup.t_data, tuple_len,
> ! InvalidOffsetNumber, LP_USED);
> ! if (newoff == InvalidOffsetNumber)
> ! {
> ! elog(PANIC, "failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)",
> ! (unsigned long) tuple_len,
> ! dst_vacpage->blkno, (unsigned long) dst_vacpage->free,
> ! dst_vacpage->offsets_used, dst_vacpage->offsets_free);
> ! }
> ! newitemid = PageGetItemId(dst_page, newoff);
> ! pfree(newtup.t_data);
> ! newtup.t_datamcxt = NULL;
> ! newtup.t_data = (HeapTupleHeader) PageGetItem(dst_page, newitemid);
> ! ItemPointerSet(&(newtup.t_data->t_ctid), dst_vacpage->blkno, newoff);
> ! newtup.t_self = newtup.t_data->t_ctid;
> !
> ! /*
> ! * Mark old tuple as MOVED_OFF by me.
> ! */
> ! old_tup->t_data->t_infomask &= ~(HEAP_XMIN_COMMITTED |
> ! HEAP_XMIN_INVALID |
> ! HEAP_MOVED_IN);
> ! old_tup->t_data->t_infomask |= HEAP_MOVED_OFF;
> ! HeapTupleHeaderSetXvac(old_tup->t_data, myXID);
> !
> ! /* XLOG stuff */
> ! if (!rel->rd_istemp)
> ! {
> ! XLogRecPtr recptr = log_heap_move(rel, old_buf, old_tup->t_self,
> ! dst_buf, &newtup);
> !
> ! PageSetLSN(old_page, recptr);
> ! PageSetSUI(old_page, ThisStartUpID);
> ! PageSetLSN(dst_page, recptr);
> ! PageSetSUI(dst_page, ThisStartUpID);
> ! }
> ! else
> ! {
> ! /*
> ! * No XLOG record, but still need to flag that XID exists
> ! * on disk
> ! */
> ! MyXactMadeTempRelUpdate = true;
> ! }
> !
> ! END_CRIT_SECTION();
> !
> ! dst_vacpage->free = ((PageHeader) dst_page)->pd_upper -
> ! ((PageHeader) dst_page)->pd_lower;
> ! LockBuffer(dst_buf, BUFFER_LOCK_UNLOCK);
> ! LockBuffer(old_buf, BUFFER_LOCK_UNLOCK);
> !
> ! dst_vacpage->offsets_used++;
> !
> ! /* insert index' tuples if needed */
> ! if (ec->resultRelInfo->ri_NumIndices > 0)
> ! {
> ! ExecStoreTuple(&newtup, ec->slot, InvalidBuffer, false);
> ! ExecInsertIndexTuples(ec->slot, &(newtup.t_self), ec->estate, true);
> ! }
> ! }
> !
> ! /*
> ! * update_hint_bits() -- update hint bits in destination pages
> ! *
> ! * Scan all the pages that we moved tuples onto and update tuple
> ! * status bits. This is not really necessary, but will save time for
> ! * future transactions examining these tuples.
> ! *
> ! * XXX NOTICE that this code fails to clear HEAP_MOVED_OFF tuples from
> ! * pages that were move source pages but not move dest pages. One
> ! * also wonders whether it wouldn't be better to skip this step and
> ! * let the tuple status updates happen someplace that's not holding an
> ! * exclusive lock on the relation.
> ! */
> ! static void
> ! update_hint_bits(Relation rel, VacPageList fraged_pages, int num_fraged_pages,
> ! BlockNumber last_move_dest_block, int num_moved)
> ! {
> ! int checked_moved = 0;
> ! int i;
> ! VacPage *curpage;
> !
> ! for (i = 0, curpage = fraged_pages->pagedesc;
> ! i < num_fraged_pages;
> ! i++, curpage++)
> ! {
> ! Buffer buf;
> ! Page page;
> ! OffsetNumber max_offset;
> ! OffsetNumber off;
> ! int num_tuples = 0;
> !
> ! vacuum_delay_point();
> !
> ! if ((*curpage)->blkno > last_move_dest_block)
> ! break; /* no need to scan any further */
> ! if ((*curpage)->offsets_used == 0)
> ! continue; /* this page was never used as a move dest */
> ! buf = ReadBuffer(rel, (*curpage)->blkno);
> ! LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
> ! page = BufferGetPage(buf);
> ! max_offset = PageGetMaxOffsetNumber(page);
> ! for (off = FirstOffsetNumber;
> ! off <= max_offset;
> ! off = OffsetNumberNext(off))
> ! {
> ! ItemId itemid = PageGetItemId(page, off);
> ! HeapTupleHeader htup;
> !
> ! if (!ItemIdIsUsed(itemid))
> ! continue;
> ! htup = (HeapTupleHeader) PageGetItem(page, itemid);
> ! if (htup->t_infomask & HEAP_XMIN_COMMITTED)
> ! continue;
> ! /*
> ! * See comments in the walk-along-page loop above, why we
> ! * have Asserts here instead of if (...) elog(ERROR). The
> ! * difference here is that we may see MOVED_IN.
> ! */
> ! Assert(htup->t_infomask & HEAP_MOVED);
> ! Assert(HeapTupleHeaderGetXvac(htup) == GetCurrentTransactionId());
> ! if (htup->t_infomask & HEAP_MOVED_IN)
> ! {
> ! htup->t_infomask |= HEAP_XMIN_COMMITTED;
> ! htup->t_infomask &= ~HEAP_MOVED;
> ! num_tuples++;
> ! }
> ! else
> ! htup->t_infomask |= HEAP_XMIN_INVALID;
> ! }
> ! LockBuffer(buf, BUFFER_LOCK_UNLOCK);
> ! WriteBuffer(buf);
> ! Assert((*curpage)->offsets_used == num_tuples);
> ! checked_moved += num_tuples;
> ! }
> ! Assert(num_moved == checked_moved);
> }
>
> /*
>
> ---------------------------(end of broadcast)---------------------------
> TIP 6: Have you searched our list archives?
>
> http://archives.postgresql.org
--
Bruce Momjian | http://candle.pha.pa.us
pgman@candle.pha.pa.us | (610) 359-1001
+ If your life is a hard drive, | 13 Roberts Road
+ Christ can be your backup. | Newtown Square, Pennsylvania 19073