granicus.if.org Git - postgresql/blob - src/backend/commands/vacuum.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * vacuum.c
   4  *        the postgres vacuum cleaner
   5  *
   6  * Portions Copyright (c) 1996-2000, PostgreSQL, Inc
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/commands/vacuum.c,v 1.172 2000/11/16 05:50:59 momjian Exp $
  12  *
  13
  14  *-------------------------------------------------------------------------
  15  */
  16 #include <sys/types.h>
  17 #include <sys/file.h>
  18 #include <sys/stat.h>
  19 #include <fcntl.h>
  20 #include <unistd.h>
  21
  22 #include "postgres.h"
  23
  24 #include "access/genam.h"
  25 #include "access/heapam.h"
  26 #include "catalog/catalog.h"
  27 #include "catalog/catname.h"
  28 #include "catalog/index.h"
  29 #include "commands/vacuum.h"
  30 #include "miscadmin.h"
  31 #include "nodes/execnodes.h"
  32 #include "storage/sinval.h"
  33 #include "storage/smgr.h"
  34 #include "tcop/tcopprot.h"
  35 #include "utils/acl.h"
  36 #include "utils/builtins.h"
  37 #include "utils/fmgroids.h"
  38 #include "utils/inval.h"
  39 #include "utils/relcache.h"
  40 #include "utils/syscache.h"
  41 #include "utils/temprel.h"
  42
  43 #ifndef HAVE_GETRUSAGE
  44 #include "rusagestub.h"
  45 #else
  46 #include <sys/time.h>
  47 #include <sys/resource.h>
  48 #endif
  49
  50 #ifdef XLOG
  51 #include "access/xlog.h"
  52 XLogRecPtr      log_heap_move(Relation reln,
  53                                 ItemPointerData from, HeapTuple newtup);
  54 #endif
  55
  56 static MemoryContext vac_context = NULL;
  57
  58 static int      MESSAGE_LEVEL;          /* message level */
  59
  60 static TransactionId XmaxRecent;
  61
  62 /* non-export function prototypes */
  63 static void vacuum_init(void);
  64 static void vacuum_shutdown(void);
  65 static void vac_vacuum(NameData *VacRelP, bool analyze, List *anal_cols2);
  66 static VRelList getrels(NameData *VacRelP);
  67 static void vacuum_rel(Oid relid, bool analyze, bool is_toastrel);
  68 static void scan_heap(VRelStats *vacrelstats, Relation onerel, VacPageList vacuum_pages, VacPageList fraged_pages);
  69 static void repair_frag(VRelStats *vacrelstats, Relation onerel, VacPageList vacuum_pages, VacPageList fraged_pages, int nindices, Relation *Irel);
  70 static void vacuum_heap(VRelStats *vacrelstats, Relation onerel, VacPageList vacpagelist);
  71 static void vacuum_page(Page page, VacPage vacpage);
  72 static void vacuum_index(VacPageList vacpagelist, Relation indrel, int num_tuples, int keep_tuples);
  73 static void scan_index(Relation indrel, int num_tuples);
  74 static void update_relstats(Oid relid, int num_pages, int num_tuples, bool hasindex, VRelStats *vacrelstats);
  75 static VacPage tid_reaped(ItemPointer itemptr, VacPageList vacpagelist);
  76 static void reap_page(VacPageList vacpagelist, VacPage vacpage);
  77 static void vpage_insert(VacPageList vacpagelist, VacPage vpnew);
  78 static void get_indices(Relation relation, int *nindices, Relation **Irel);
  79 static void close_indices(int nindices, Relation *Irel);
  80 static IndexInfo **get_index_desc(Relation onerel, int nindices,
  81                                                                   Relation *Irel);
  82 static void *vac_find_eq(void *bot, int nelem, int size, void *elm,
  83                          int (*compar) (const void *, const void *));
  84 static int      vac_cmp_blk(const void *left, const void *right);
  85 static int      vac_cmp_offno(const void *left, const void *right);
  86 static int      vac_cmp_vtlinks(const void *left, const void *right);
  87 static bool enough_space(VacPage vacpage, Size len);
  88 static char *show_rusage(struct rusage * ru0);
  89
  90
  91 void
  92 vacuum(char *vacrel, bool verbose, bool analyze, List *anal_cols)
  93 {
  94         NameData        VacRel;
  95         Name            VacRelName;
  96         MemoryContext old;
  97         List       *le;
  98         List       *anal_cols2 = NIL;
  99
 100         if (anal_cols != NIL && !analyze)
 101                 elog(ERROR, "Can't vacuum columns, only tables.  You can 'vacuum analyze' columns.");
 102
 103         /*
 104          * We cannot run VACUUM inside a user transaction block; if we were
 105          * inside a transaction, then our commit- and
 106          * start-transaction-command calls would not have the intended effect!
 107          * Furthermore, the forced commit that occurs before truncating the
 108          * relation's file would have the effect of committing the rest of the
 109          * user's transaction too, which would certainly not be the desired
 110          * behavior.
 111          */
 112         if (IsTransactionBlock())
 113                 elog(ERROR, "VACUUM cannot run inside a BEGIN/END block");
 114
 115         if (verbose)
 116                 MESSAGE_LEVEL = NOTICE;
 117         else
 118                 MESSAGE_LEVEL = DEBUG;
 119
 120         /*
 121          * Create special memory context for cross-transaction storage.
 122          *
 123          * Since it is a child of QueryContext, it will go away eventually
 124          * even if we suffer an error; there's no need for special abort
 125          * cleanup logic.
 126          */
 127         vac_context = AllocSetContextCreate(QueryContext,
 128                                                                                 "Vacuum",
 129                                                                                 ALLOCSET_DEFAULT_MINSIZE,
 130                                                                                 ALLOCSET_DEFAULT_INITSIZE,
 131                                                                                 ALLOCSET_DEFAULT_MAXSIZE);
 132
 133         /* vacrel gets de-allocated on xact commit, so copy it to safe storage */
 134         if (vacrel)
 135         {
 136                 namestrcpy(&VacRel, vacrel);
 137                 VacRelName = &VacRel;
 138         }
 139         else
 140                 VacRelName = NULL;
 141
 142         /* must also copy the column list, if any, to safe storage */
 143         old = MemoryContextSwitchTo(vac_context);
 144         foreach(le, anal_cols)
 145         {
 146                 char       *col = (char *) lfirst(le);
 147
 148                 anal_cols2 = lappend(anal_cols2, pstrdup(col));
 149         }
 150         MemoryContextSwitchTo(old);
 151
 152         /*
 153          * Start up the vacuum cleaner.
 154          *
 155          * NOTE: since this commits the current transaction, the memory holding
 156          * any passed-in parameters gets freed here.  We must have already
 157          * copied pass-by-reference parameters to safe storage.  Don't make me
 158          * fix this again!
 159          */
 160         vacuum_init();
 161
 162         /* vacuum the database */
 163         vac_vacuum(VacRelName, analyze, anal_cols2);
 164
 165         /* clean up */
 166         vacuum_shutdown();
 167 }
 168
 169 /*
 170  *      vacuum_init(), vacuum_shutdown() -- start up and shut down the vacuum cleaner.
 171  *
 172  *              Formerly, there was code here to prevent more than one VACUUM from
 173  *              executing concurrently in the same database.  However, there's no
 174  *              good reason to prevent that, and manually removing lockfiles after
 175  *              a vacuum crash was a pain for dbadmins.  So, forget about lockfiles,
 176  *              and just rely on the exclusive lock we grab on each target table
 177  *              to ensure that there aren't two VACUUMs running on the same table
 178  *              at the same time.
 179  *
 180  *              The strangeness with committing and starting transactions in the
 181  *              init and shutdown routines is due to the fact that the vacuum cleaner
 182  *              is invoked via an SQL command, and so is already executing inside
 183  *              a transaction.  We need to leave ourselves in a predictable state
 184  *              on entry and exit to the vacuum cleaner.  We commit the transaction
 185  *              started in PostgresMain() inside vacuum_init(), and start one in
 186  *              vacuum_shutdown() to match the commit waiting for us back in
 187  *              PostgresMain().
 188  */
 189 static void
 190 vacuum_init()
 191 {
 192         /* matches the StartTransaction in PostgresMain() */
 193         CommitTransactionCommand();
 194 }
 195
 196 static void
 197 vacuum_shutdown()
 198 {
 199         /* on entry, we are not in a transaction */
 200
 201         /*
 202          * Flush the init file that relcache.c uses to save startup time. The
 203          * next backend startup will rebuild the init file with up-to-date
 204          * information from pg_class.  This lets the optimizer see the stats
 205          * that we've collected for certain critical system indexes.  See
 206          * relcache.c for more details.
 207          *
 208          * Ignore any failure to unlink the file, since it might not be there if
 209          * no backend has been started since the last vacuum...
 210          */
 211         unlink(RELCACHE_INIT_FILENAME);
 212
 213         /* matches the CommitTransaction in PostgresMain() */
 214         StartTransactionCommand();
 215
 216         /*
 217          * Clean up working storage --- note we must do this after
 218          * StartTransactionCommand, else we might be trying to delete
 219          * the active context!
 220          */
 221         MemoryContextDelete(vac_context);
 222         vac_context = NULL;
 223 }
 224
 225 /*
 226  *      vac_vacuum() -- vacuum the database.
 227  *
 228  *              This routine builds a list of relations to vacuum, and then calls
 229  *              code that vacuums them one at a time.  We are careful to vacuum each
 230  *              relation in a separate transaction in order to avoid holding too many
 231  *              locks at one time.
 232  */
 233 static void
 234 vac_vacuum(NameData *VacRelP, bool analyze, List *anal_cols2)
 235 {
 236         VRelList        vrl,
 237                                 cur;
 238
 239         /* get list of relations */
 240         vrl = getrels(VacRelP);
 241
 242         /* vacuum each heap relation */
 243         for (cur = vrl; cur != (VRelList) NULL; cur = cur->vrl_next)
 244         {
 245                 vacuum_rel(cur->vrl_relid, analyze, false);
 246                 /* analyze separately so locking is minimized */
 247                 if (analyze)
 248                         analyze_rel(cur->vrl_relid, anal_cols2, MESSAGE_LEVEL);
 249         }
 250 }
 251
 252 static VRelList
 253 getrels(NameData *VacRelP)
 254 {
 255         Relation        rel;
 256         TupleDesc       tupdesc;
 257         HeapScanDesc scan;
 258         HeapTuple       tuple;
 259         VRelList        vrl,
 260                                 cur;
 261         Datum           d;
 262         char       *rname;
 263         char            rkind;
 264         bool            n;
 265         bool            found = false;
 266         ScanKeyData key;
 267
 268         StartTransactionCommand();
 269
 270         if (NameStr(*VacRelP))
 271         {
 272
 273                 /*
 274                  * we could use the cache here, but it is clearer to use scankeys
 275                  * for both vacuum cases, bjm 2000/01/19
 276                  */
 277                 char       *nontemp_relname;
 278
 279                 /* We must re-map temp table names bjm 2000-04-06 */
 280                 if ((nontemp_relname =
 281                          get_temp_rel_by_username(NameStr(*VacRelP))) == NULL)
 282                         nontemp_relname = NameStr(*VacRelP);
 283
 284                 ScanKeyEntryInitialize(&key, 0x0, Anum_pg_class_relname,
 285                                                            F_NAMEEQ,
 286                                                            PointerGetDatum(nontemp_relname));
 287         }
 288         else
 289         {
 290                 ScanKeyEntryInitialize(&key, 0x0, Anum_pg_class_relkind,
 291                                                            F_CHAREQ, CharGetDatum('r'));
 292         }
 293
 294         vrl = cur = (VRelList) NULL;
 295
 296         rel = heap_openr(RelationRelationName, AccessShareLock);
 297         tupdesc = RelationGetDescr(rel);
 298
 299         scan = heap_beginscan(rel, false, SnapshotNow, 1, &key);
 300
 301         while (HeapTupleIsValid(tuple = heap_getnext(scan, 0)))
 302         {
 303                 found = true;
 304
 305                 d = heap_getattr(tuple, Anum_pg_class_relname, tupdesc, &n);
 306                 rname = (char *) d;
 307
 308                 d = heap_getattr(tuple, Anum_pg_class_relkind, tupdesc, &n);
 309
 310                 rkind = DatumGetChar(d);
 311
 312                 if (rkind != RELKIND_RELATION)
 313                 {
 314                         elog(NOTICE, "Vacuum: can not process indecies, views and certain system tables");
 315                         continue;
 316                 }
 317
 318                 /* get a relation list entry for this guy */
 319                 if (vrl == (VRelList) NULL)
 320                         vrl = cur = (VRelList)
 321                                 MemoryContextAlloc(vac_context, sizeof(VRelListData));
 322                 else
 323                 {
 324                         cur->vrl_next = (VRelList)
 325                                 MemoryContextAlloc(vac_context, sizeof(VRelListData));
 326                         cur = cur->vrl_next;
 327                 }
 328
 329                 cur->vrl_relid = tuple->t_data->t_oid;
 330                 cur->vrl_next = (VRelList) NULL;
 331         }
 332
 333         heap_endscan(scan);
 334         heap_close(rel, AccessShareLock);
 335
 336         if (!found)
 337                 elog(NOTICE, "Vacuum: table not found");
 338
 339         CommitTransactionCommand();
 340
 341         return vrl;
 342 }
 343
 344 /*
 345  *      vacuum_rel() -- vacuum one heap relation
 346  *
 347  *              This routine vacuums a single heap, cleans out its indices, and
 348  *              updates its statistics num_pages and num_tuples statistics.
 349  *
 350  *              Doing one heap at a time incurs extra overhead, since we need to
 351  *              check that the heap exists again just before we vacuum it.      The
 352  *              reason that we do this is so that vacuuming can be spread across
 353  *              many small transactions.  Otherwise, two-phase locking would require
 354  *              us to lock the entire database during one pass of the vacuum cleaner.
 355  */
 356 static void
 357 vacuum_rel(Oid relid, bool analyze, bool is_toastrel)
 358 {
 359         HeapTuple       tuple;
 360         Relation        onerel;
 361         VacPageListData vacuum_pages; /* List of pages to vacuum and/or clean
 362                                                                  * indices */
 363         VacPageListData fraged_pages; /* List of pages with space enough for
 364                                                                  * re-using */
 365         VacPage    *vacpage;
 366         Relation   *Irel;
 367         int32           nindices,
 368                                 i;
 369         VRelStats  *vacrelstats;
 370         bool            reindex = false;
 371         Oid                     toast_relid;
 372
 373         if (!is_toastrel)
 374                 StartTransactionCommand();
 375
 376         /*
 377          * Check for user-requested abort.      Note we want this to be inside a
 378          * transaction, so xact.c doesn't issue useless NOTICE.
 379          */
 380         if (QueryCancel)
 381                 CancelQuery();
 382
 383         /*
 384          * Race condition -- if the pg_class tuple has gone away since the
 385          * last time we saw it, we don't need to vacuum it.
 386          */
 387         tuple = SearchSysCacheTuple(RELOID,
 388                                                                 ObjectIdGetDatum(relid),
 389                                                                 0, 0, 0);
 390         if (!HeapTupleIsValid(tuple))
 391         {
 392                 if (!is_toastrel)
 393                         CommitTransactionCommand();
 394                 return;
 395         }
 396
 397         /*
 398          * Open the class, get an exclusive lock on it, and check permissions.
 399          *
 400          * Note we choose to treat permissions failure as a NOTICE and keep
 401          * trying to vacuum the rest of the DB --- is this appropriate?
 402          */
 403         onerel = heap_open(relid, AccessExclusiveLock);
 404
 405         if (!pg_ownercheck(GetUserId(), RelationGetRelationName(onerel),
 406                                            RELNAME))
 407         {
 408                 elog(NOTICE, "Skipping \"%s\" --- only table owner can VACUUM it",
 409                          RelationGetRelationName(onerel));
 410                 heap_close(onerel, AccessExclusiveLock);
 411                 if (!is_toastrel)
 412                         CommitTransactionCommand();
 413                 return;
 414         }
 415
 416         /*
 417          * Remember the relation'ss TOAST relation for later
 418          */
 419         toast_relid = onerel->rd_rel->reltoastrelid;
 420
 421         /*
 422          * Set up statistics-gathering machinery.
 423          */
 424         vacrelstats = (VRelStats *) palloc(sizeof(VRelStats));
 425         vacrelstats->relid = relid;
 426         vacrelstats->num_pages = vacrelstats->num_tuples = 0;
 427         vacrelstats->hasindex = false;
 428
 429         GetXmaxRecent(&XmaxRecent);
 430
 431         /* scan it */
 432         reindex = false;
 433         vacuum_pages.num_pages = fraged_pages.num_pages = 0;
 434         scan_heap(vacrelstats, onerel, &vacuum_pages, &fraged_pages);
 435         if (IsIgnoringSystemIndexes() &&
 436                 IsSystemRelationName(RelationGetRelationName(onerel)))
 437                 reindex = true;
 438
 439         /* Now open indices */
 440         nindices = 0;
 441         Irel = (Relation *) NULL;
 442         get_indices(onerel, &nindices, &Irel);
 443         if (!Irel)
 444                 reindex = false;
 445         else if (!RelationGetForm(onerel)->relhasindex)
 446                 reindex = true;
 447         if (nindices > 0)
 448                 vacrelstats->hasindex = true;
 449         else
 450                 vacrelstats->hasindex = false;
 451         if (reindex)
 452         {
 453                 for (i = 0; i < nindices; i++)
 454                         index_close(Irel[i]);
 455                 Irel = (Relation *) NULL;
 456                 activate_indexes_of_a_table(relid, false);
 457         }
 458
 459         /* Clean/scan index relation(s) */
 460         if (Irel != (Relation *) NULL)
 461         {
 462                 if (vacuum_pages.num_pages > 0)
 463                 {
 464                         for (i = 0; i < nindices; i++)
 465                                 vacuum_index(&vacuum_pages, Irel[i],
 466                                                          vacrelstats->num_tuples, 0);
 467                 }
 468                 else
 469                 {
 470                         /* just scan indices to update statistic */
 471                         for (i = 0; i < nindices; i++)
 472                                 scan_index(Irel[i], vacrelstats->num_tuples);
 473                 }
 474         }
 475
 476         if (fraged_pages.num_pages > 0)
 477         {
 478                 /* Try to shrink heap */
 479                 repair_frag(vacrelstats, onerel, &vacuum_pages, &fraged_pages,
 480                                         nindices, Irel);
 481         }
 482         else
 483         {
 484                 if (Irel != (Relation *) NULL)
 485                         close_indices(nindices, Irel);
 486                 if (vacuum_pages.num_pages > 0)
 487                 {
 488                         /* Clean pages from vacuum_pages list */
 489                         vacuum_heap(vacrelstats, onerel, &vacuum_pages);
 490                 }
 491                 else
 492                 {
 493                         /*
 494                          * Flush dirty pages out to disk.  We must do this even if we
 495                          * didn't do anything else, because we want to ensure that all
 496                          * tuples have correct on-row commit status on disk (see
 497                          * bufmgr.c's comments for FlushRelationBuffers()).
 498                          */
 499                         i = FlushRelationBuffers(onerel, vacrelstats->num_pages);
 500                         if (i < 0)
 501                                 elog(ERROR, "VACUUM (vacuum_rel): FlushRelationBuffers returned %d",
 502                                          i);
 503                 }
 504         }
 505         if (reindex)
 506                 activate_indexes_of_a_table(relid, true);
 507
 508         /*
 509          * ok - free vacuum_pages list of reaped pages
 510          *
 511          * Isn't this a waste of code?  Upcoming commit should free memory, no?
 512          */
 513         if (vacuum_pages.num_pages > 0)
 514         {
 515                 vacpage = vacuum_pages.pagedesc;
 516                 for (i = 0; i < vacuum_pages.num_pages; i++, vacpage++)
 517                         pfree(*vacpage);
 518                 pfree(vacuum_pages.pagedesc);
 519                 if (fraged_pages.num_pages > 0)
 520                         pfree(fraged_pages.pagedesc);
 521         }
 522
 523         /* all done with this class, but hold lock until commit */
 524         heap_close(onerel, NoLock);
 525
 526         /* update statistics in pg_class */
 527         update_relstats(vacrelstats->relid, vacrelstats->num_pages,
 528                                         vacrelstats->num_tuples, vacrelstats->hasindex,
 529                                         vacrelstats);
 530
 531         /*
 532          * If the relation has a secondary toast one, vacuum that too
 533          * while we still hold the lock on the master table. We don't
 534          * need to propagate "analyze" to it, because the toaster
 535          * always uses hardcoded index access and statistics are
 536          * totally unimportant for toast relations
 537          */
 538         if (toast_relid != InvalidOid)
 539                 vacuum_rel(toast_relid, false, true);
 540
 541         /* next command frees attribute stats */
 542         if (!is_toastrel)
 543                 CommitTransactionCommand();
 544 }
 545
 546 /*
 547  *      scan_heap() -- scan an open heap relation
 548  *
 549  *              This routine sets commit times, constructs vacuum_pages list of
 550  *              empty/uninitialized pages and pages with dead tuples and
 551  *              ~LP_USED line pointers, constructs fraged_pages list of pages
 552  *              appropriate for purposes of shrinking and maintains statistics
 553  *              on the number of live tuples in a heap.
 554  */
 555 static void
 556 scan_heap(VRelStats *vacrelstats, Relation onerel,
 557                         VacPageList vacuum_pages, VacPageList fraged_pages)
 558 {
 559         BlockNumber nblocks,
 560                                 blkno;
 561         ItemId          itemid;
 562         Buffer          buf;
 563         HeapTupleData tuple;
 564         Page            page,
 565                                 tempPage = NULL;
 566         OffsetNumber offnum,
 567                                 maxoff;
 568         bool            pgchanged,
 569                                 tupgone,
 570                                 dobufrel,
 571                                 notup;
 572         char       *relname;
 573         VacPage         vacpage,
 574                                 vp;
 575         uint32          tups_vacuumed,
 576                                 num_tuples,
 577                                 nkeep,
 578                                 nunused,
 579                                 ncrash,
 580                                 empty_pages,
 581                                 new_pages,
 582                                 changed_pages,
 583                                 empty_end_pages;
 584         Size            free_size,
 585                                 usable_free_size;
 586         Size            min_tlen = MaxTupleSize;
 587         Size            max_tlen = 0;
 588         int32           i;
 589         bool            do_shrinking = true;
 590         VTupleLink      vtlinks = (VTupleLink) palloc(100 * sizeof(VTupleLinkData));
 591         int                     num_vtlinks = 0;
 592         int                     free_vtlinks = 100;
 593         struct rusage ru0;
 594
 595         getrusage(RUSAGE_SELF, &ru0);
 596
 597         relname = RelationGetRelationName(onerel);
 598         elog(MESSAGE_LEVEL, "--Relation %s--", relname);
 599
 600         tups_vacuumed = num_tuples = nkeep = nunused = ncrash = empty_pages =
 601                 new_pages = changed_pages = empty_end_pages = 0;
 602         free_size = usable_free_size = 0;
 603
 604         nblocks = RelationGetNumberOfBlocks(onerel);
 605
 606         vacpage = (VacPage) palloc(sizeof(VacPageData) + MaxOffsetNumber * sizeof(OffsetNumber));
 607         vacpage->offsets_used = 0;
 608
 609         for (blkno = 0; blkno < nblocks; blkno++)
 610         {
 611                 buf = ReadBuffer(onerel, blkno);
 612                 page = BufferGetPage(buf);
 613                 vacpage->blkno = blkno;
 614                 vacpage->offsets_free = 0;
 615
 616                 if (PageIsNew(page))
 617                 {
 618                         elog(NOTICE, "Rel %s: Uninitialized page %u - fixing",
 619                                  relname, blkno);
 620                         PageInit(page, BufferGetPageSize(buf), 0);
 621                         vacpage->free = ((PageHeader) page)->pd_upper - ((PageHeader) page)->pd_lower;
 622                         free_size += (vacpage->free - sizeof(ItemIdData));
 623                         new_pages++;
 624                         empty_end_pages++;
 625                         reap_page(vacuum_pages, vacpage);
 626                         WriteBuffer(buf);
 627                         continue;
 628                 }
 629
 630                 if (PageIsEmpty(page))
 631                 {
 632                         vacpage->free = ((PageHeader) page)->pd_upper - ((PageHeader) page)->pd_lower;
 633                         free_size += (vacpage->free - sizeof(ItemIdData));
 634                         empty_pages++;
 635                         empty_end_pages++;
 636                         reap_page(vacuum_pages, vacpage);
 637                         ReleaseBuffer(buf);
 638                         continue;
 639                 }
 640
 641                 pgchanged = false;
 642                 notup = true;
 643                 maxoff = PageGetMaxOffsetNumber(page);
 644                 for (offnum = FirstOffsetNumber;
 645                          offnum <= maxoff;
 646                          offnum = OffsetNumberNext(offnum))
 647                 {
 648                         itemid = PageGetItemId(page, offnum);
 649
 650                         /*
 651                          * Collect un-used items too - it's possible to have indices
 652                          * pointing here after crash.
 653                          */
 654                         if (!ItemIdIsUsed(itemid))
 655                         {
 656                                 vacpage->offsets[vacpage->offsets_free++] = offnum;
 657                                 nunused++;
 658                                 continue;
 659                         }
 660
 661                         tuple.t_datamcxt = NULL;
 662                         tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
 663                         tuple.t_len = ItemIdGetLength(itemid);
 664                         ItemPointerSet(&(tuple.t_self), blkno, offnum);
 665                         tupgone = false;
 666
 667                         if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
 668                         {
 669                                 if (tuple.t_data->t_infomask & HEAP_XMIN_INVALID)
 670                                         tupgone = true;
 671                                 else if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
 672                                 {
 673                                         if (TransactionIdDidCommit((TransactionId)
 674                                                                                            tuple.t_data->t_cmin))
 675                                         {
 676                                                 tuple.t_data->t_infomask |= HEAP_XMIN_INVALID;
 677                                                 pgchanged = true;
 678                                                 tupgone = true;
 679                                         }
 680                                         else
 681                                         {
 682                                                 tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
 683                                                 pgchanged = true;
 684                                         }
 685                                 }
 686                                 else if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
 687                                 {
 688                                         if (!TransactionIdDidCommit((TransactionId)
 689                                                                                                 tuple.t_data->t_cmin))
 690                                         {
 691                                                 tuple.t_data->t_infomask |= HEAP_XMIN_INVALID;
 692                                                 pgchanged = true;
 693                                                 tupgone = true;
 694                                         }
 695                                         else
 696                                         {
 697                                                 tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
 698                                                 pgchanged = true;
 699                                         }
 700                                 }
 701                                 else
 702                                 {
 703                                         if (TransactionIdDidAbort(tuple.t_data->t_xmin))
 704                                                 tupgone = true;
 705                                         else if (TransactionIdDidCommit(tuple.t_data->t_xmin))
 706                                         {
 707                                                 tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
 708                                                 pgchanged = true;
 709                                         }
 710                                         else if (!TransactionIdIsInProgress(tuple.t_data->t_xmin))
 711                                         {
 712
 713                                                 /*
 714                                                  * Not Aborted, Not Committed, Not in Progress -
 715                                                  * so it's from crashed process. - vadim 11/26/96
 716                                                  */
 717                                                 ncrash++;
 718                                                 tupgone = true;
 719                                         }
 720                                         else
 721                                         {
 722                                                 elog(NOTICE, "Rel %s: TID %u/%u: InsertTransactionInProgress %u - can't shrink relation",
 723                                                    relname, blkno, offnum, tuple.t_data->t_xmin);
 724                                                 do_shrinking = false;
 725                                         }
 726                                 }
 727                         }
 728
 729                         /*
 730                          * here we are concerned about tuples with xmin committed and
 731                          * xmax unknown or committed
 732                          */
 733                         if (tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED &&
 734                                 !(tuple.t_data->t_infomask & HEAP_XMAX_INVALID))
 735                         {
 736                                 if (tuple.t_data->t_infomask & HEAP_XMAX_COMMITTED)
 737                                 {
 738                                         if (tuple.t_data->t_infomask & HEAP_MARKED_FOR_UPDATE)
 739                                         {
 740                                                 tuple.t_data->t_infomask |= HEAP_XMAX_INVALID;
 741                                                 tuple.t_data->t_infomask &=
 742                                                         ~(HEAP_XMAX_COMMITTED | HEAP_MARKED_FOR_UPDATE);
 743                                                 pgchanged = true;
 744                                         }
 745                                         else
 746                                                 tupgone = true;
 747                                 }
 748                                 else if (TransactionIdDidAbort(tuple.t_data->t_xmax))
 749                                 {
 750                                         tuple.t_data->t_infomask |= HEAP_XMAX_INVALID;
 751                                         pgchanged = true;
 752                                 }
 753                                 else if (TransactionIdDidCommit(tuple.t_data->t_xmax))
 754                                 {
 755                                         if (tuple.t_data->t_infomask & HEAP_MARKED_FOR_UPDATE)
 756                                         {
 757                                                 tuple.t_data->t_infomask |= HEAP_XMAX_INVALID;
 758                                                 tuple.t_data->t_infomask &=
 759                                                         ~(HEAP_XMAX_COMMITTED | HEAP_MARKED_FOR_UPDATE);
 760                                                 pgchanged = true;
 761                                         }
 762                                         else
 763                                                 tupgone = true;
 764                                 }
 765                                 else if (!TransactionIdIsInProgress(tuple.t_data->t_xmax))
 766                                 {
 767
 768                                         /*
 769                                          * Not Aborted, Not Committed, Not in Progress - so it
 770                                          * from crashed process. - vadim 06/02/97
 771                                          */
 772                                         tuple.t_data->t_infomask |= HEAP_XMAX_INVALID;
 773                                         tuple.t_data->t_infomask &=
 774                                                 ~(HEAP_XMAX_COMMITTED | HEAP_MARKED_FOR_UPDATE);
 775                                         pgchanged = true;
 776                                 }
 777                                 else
 778                                 {
 779                                         elog(NOTICE, "Rel %s: TID %u/%u: DeleteTransactionInProgress %u - can't shrink relation",
 780                                                  relname, blkno, offnum, tuple.t_data->t_xmax);
 781                                         do_shrinking = false;
 782                                 }
 783
 784                                 /*
 785                                  * If tuple is recently deleted then we must not remove it
 786                                  * from relation.
 787                                  */
 788                                 if (tupgone && (tuple.t_data->t_infomask & HEAP_XMIN_INVALID) == 0 && tuple.t_data->t_xmax >= XmaxRecent)
 789                                 {
 790                                         tupgone = false;
 791                                         nkeep++;
 792                                         if (!(tuple.t_data->t_infomask & HEAP_XMAX_COMMITTED))
 793                                         {
 794                                                 tuple.t_data->t_infomask |= HEAP_XMAX_COMMITTED;
 795                                                 pgchanged = true;
 796                                         }
 797
 798                                         /*
 799                                          * If we do shrinking and this tuple is updated one
 800                                          * then remember it to construct updated tuple
 801                                          * dependencies.
 802                                          */
 803                                         if (do_shrinking && !(ItemPointerEquals(&(tuple.t_self),
 804                                                                                            &(tuple.t_data->t_ctid))))
 805                                         {
 806                                                 if (free_vtlinks == 0)
 807                                                 {
 808                                                         free_vtlinks = 1000;
 809                                                         vtlinks = (VTupleLink) repalloc(vtlinks,
 810                                                                                    (free_vtlinks + num_vtlinks) *
 811                                                                                                  sizeof(VTupleLinkData));
 812                                                 }
 813                                                 vtlinks[num_vtlinks].new_tid = tuple.t_data->t_ctid;
 814                                                 vtlinks[num_vtlinks].this_tid = tuple.t_self;
 815                                                 free_vtlinks--;
 816                                                 num_vtlinks++;
 817                                         }
 818                                 }
 819                         }
 820
 821                         /*
 822                          * Other checks...
 823                          */
 824                         if (!OidIsValid(tuple.t_data->t_oid))
 825                         {
 826                                 elog(NOTICE, "Rel %s: TID %u/%u: OID IS INVALID. TUPGONE %d.",
 827                                          relname, blkno, offnum, tupgone);
 828                         }
 829
 830                         if (tupgone)
 831                         {
 832                                 ItemId          lpp;
 833
 834                                 /*
 835                                  * Here we are building a temporary copy of the page with
 836                                  * dead tuples removed.  Below we will apply
 837                                  * PageRepairFragmentation to the copy, so that we can
 838                                  * determine how much space will be available after
 839                                  * removal of dead tuples.  But note we are NOT changing
 840                                  * the real page yet...
 841                                  */
 842                                 if (tempPage == (Page) NULL)
 843                                 {
 844                                         Size            pageSize;
 845
 846                                         pageSize = PageGetPageSize(page);
 847                                         tempPage = (Page) palloc(pageSize);
 848                                         memmove(tempPage, page, pageSize);
 849                                 }
 850
 851                                 /* mark it unused on the temp page */
 852                                 lpp = &(((PageHeader) tempPage)->pd_linp[offnum - 1]);
 853                                 lpp->lp_flags &= ~LP_USED;
 854
 855                                 vacpage->offsets[vacpage->offsets_free++] = offnum;
 856                                 tups_vacuumed++;
 857                         }
 858                         else
 859                         {
 860                                 num_tuples++;
 861                                 notup = false;
 862                                 if (tuple.t_len < min_tlen)
 863                                         min_tlen = tuple.t_len;
 864                                 if (tuple.t_len > max_tlen)
 865                                         max_tlen = tuple.t_len;
 866                         }
 867                 }
 868
 869                 if (pgchanged)
 870                 {
 871                         WriteBuffer(buf);
 872                         dobufrel = false;
 873                         changed_pages++;
 874                 }
 875                 else
 876                         dobufrel = true;
 877
 878                 if (tempPage != (Page) NULL)
 879                 {                                               /* Some tuples are gone */
 880                         PageRepairFragmentation(tempPage);
 881                         vacpage->free = ((PageHeader) tempPage)->pd_upper - ((PageHeader) tempPage)->pd_lower;
 882                         free_size += vacpage->free;
 883                         reap_page(vacuum_pages, vacpage);
 884                         pfree(tempPage);
 885                         tempPage = (Page) NULL;
 886                 }
 887                 else if (vacpage->offsets_free > 0)
 888                 {                                               /* there are only ~LP_USED line pointers */
 889                         vacpage->free = ((PageHeader) page)->pd_upper - ((PageHeader) page)->pd_lower;
 890                         free_size += vacpage->free;
 891                         reap_page(vacuum_pages, vacpage);
 892                 }
 893                 if (dobufrel)
 894                         ReleaseBuffer(buf);
 895                 if (notup)
 896                         empty_end_pages++;
 897                 else
 898                         empty_end_pages = 0;
 899         }
 900
 901         pfree(vacpage);
 902
 903         /* save stats in the rel list for use later */
 904         vacrelstats->num_tuples = num_tuples;
 905         vacrelstats->num_pages = nblocks;
 906 /*        vacrelstats->natts = attr_cnt;*/
 907         if (num_tuples == 0)
 908                 min_tlen = max_tlen = 0;
 909         vacrelstats->min_tlen = min_tlen;
 910         vacrelstats->max_tlen = max_tlen;
 911
 912         vacuum_pages->empty_end_pages = empty_end_pages;
 913         fraged_pages->empty_end_pages = empty_end_pages;
 914
 915         /*
 916          * Try to make fraged_pages keeping in mind that we can't use free
 917          * space of "empty" end-pages and last page if it reaped.
 918          */
 919         if (do_shrinking && vacuum_pages->num_pages - empty_end_pages > 0)
 920         {
 921                 int                     nusf;           /* blocks usefull for re-using */
 922
 923                 nusf = vacuum_pages->num_pages - empty_end_pages;
 924                 if ((vacuum_pages->pagedesc[nusf - 1])->blkno == nblocks - empty_end_pages - 1)
 925                         nusf--;
 926
 927                 for (i = 0; i < nusf; i++)
 928                 {
 929                         vp = vacuum_pages->pagedesc[i];
 930                         if (enough_space(vp, min_tlen))
 931                         {
 932                                 vpage_insert(fraged_pages, vp);
 933                                 usable_free_size += vp->free;
 934                         }
 935                 }
 936         }
 937
 938         if (usable_free_size > 0 && num_vtlinks > 0)
 939         {
 940                 qsort((char *) vtlinks, num_vtlinks, sizeof(VTupleLinkData),
 941                           vac_cmp_vtlinks);
 942                 vacrelstats->vtlinks = vtlinks;
 943                 vacrelstats->num_vtlinks = num_vtlinks;
 944         }
 945         else
 946         {
 947                 vacrelstats->vtlinks = NULL;
 948                 vacrelstats->num_vtlinks = 0;
 949                 pfree(vtlinks);
 950         }
 951
 952         elog(MESSAGE_LEVEL, "Pages %u: Changed %u, reaped %u, Empty %u, New %u; \
 953 Tup %u: Vac %u, Keep/VTL %u/%u, Crash %u, UnUsed %u, MinLen %lu, MaxLen %lu; \
 954 Re-using: Free/Avail. Space %lu/%lu; EndEmpty/Avail. Pages %u/%u. %s",
 955                  nblocks, changed_pages, vacuum_pages->num_pages, empty_pages,
 956                  new_pages, num_tuples, tups_vacuumed,
 957                  nkeep, vacrelstats->num_vtlinks, ncrash,
 958                  nunused, (unsigned long)min_tlen, (unsigned long)max_tlen,
 959                  (unsigned long)free_size, (unsigned long)usable_free_size,
 960                  empty_end_pages, fraged_pages->num_pages,
 961                  show_rusage(&ru0));
 962
 963 }
 964
 965
 966 /*
 967  *      repair_frag() -- try to repair relation's fragmentation
 968  *
 969  *              This routine marks dead tuples as unused and tries re-use dead space
 970  *              by moving tuples (and inserting indices if needed). It constructs
 971  *              Nvacpagelist list of free-ed pages (moved tuples) and clean indices
 972  *              for them after committing (in hack-manner - without losing locks
 973  *              and freeing memory!) current transaction. It truncates relation
 974  *              if some end-blocks are gone away.
 975  */
 976 static void
 977 repair_frag(VRelStats *vacrelstats, Relation onerel,
 978                            VacPageList vacuum_pages, VacPageList fraged_pages,
 979                            int nindices, Relation *Irel)
 980 {
 981         TransactionId myXID;
 982         CommandId       myCID;
 983         Buffer          buf,
 984                                 cur_buffer;
 985         int                     nblocks,
 986                                 blkno;
 987         Page            page,
 988                                 ToPage = NULL;
 989         OffsetNumber offnum = 0,
 990                                 maxoff = 0,
 991                                 newoff,
 992                                 max_offset;
 993         ItemId          itemid,
 994                                 newitemid;
 995         HeapTupleData tuple,
 996                                 newtup;
 997         TupleDesc       tupdesc;
 998         IndexInfo **indexInfo = NULL;
 999         Datum           idatum[INDEX_MAX_KEYS];
1000         char            inulls[INDEX_MAX_KEYS];
1001         InsertIndexResult iresult;
1002         VacPageListData Nvacpagelist;
1003         VacPage         cur_page = NULL,
1004                                 last_vacuum_page,
1005                                 vacpage,
1006                            *curpage;
1007         int                     cur_item = 0;
1008         int                     last_move_dest_block = -1,
1009                                 last_vacuum_block,
1010                                 i = 0;
1011         Size            tuple_len;
1012         int                     num_moved,
1013                                 num_fraged_pages,
1014                                 vacuumed_pages;
1015         int                     checked_moved,
1016                                 num_tuples,
1017                                 keep_tuples = 0;
1018         bool            isempty,
1019                                 dowrite,
1020                                 chain_tuple_moved;
1021         struct rusage ru0;
1022
1023         getrusage(RUSAGE_SELF, &ru0);
1024
1025         myXID = GetCurrentTransactionId();
1026         myCID = GetCurrentCommandId();
1027
1028         tupdesc = RelationGetDescr(onerel);
1029
1030         if (Irel != (Relation *) NULL)          /* preparation for index' inserts */
1031                 indexInfo = get_index_desc(onerel, nindices, Irel);
1032
1033         Nvacpagelist.num_pages = 0;
1034         num_fraged_pages = fraged_pages->num_pages;
1035         Assert(vacuum_pages->num_pages > vacuum_pages->empty_end_pages);
1036         vacuumed_pages = vacuum_pages->num_pages - vacuum_pages->empty_end_pages;
1037         last_vacuum_page = vacuum_pages->pagedesc[vacuumed_pages - 1];
1038         last_vacuum_block = last_vacuum_page->blkno;
1039         cur_buffer = InvalidBuffer;
1040         num_moved = 0;
1041
1042         vacpage = (VacPage) palloc(sizeof(VacPageData) + MaxOffsetNumber * sizeof(OffsetNumber));
1043         vacpage->offsets_used = vacpage->offsets_free = 0;
1044
1045         /*
1046          * Scan pages backwards from the last nonempty page, trying to move
1047          * tuples down to lower pages.  Quit when we reach a page that we have
1048          * moved any tuples onto.  Note that if a page is still in the
1049          * fraged_pages list (list of candidate move-target pages) when we
1050          * reach it, we will remove it from the list.  This ensures we never
1051          * move a tuple up to a higher page number.
1052          *
1053          * NB: this code depends on the vacuum_pages and fraged_pages lists being
1054          * in order, and on fraged_pages being a subset of vacuum_pages.
1055          */
1056         nblocks = vacrelstats->num_pages;
1057         for (blkno = nblocks - vacuum_pages->empty_end_pages - 1;
1058                  blkno > last_move_dest_block;
1059                  blkno--)
1060         {
1061                 buf = ReadBuffer(onerel, blkno);
1062                 page = BufferGetPage(buf);
1063
1064                 vacpage->offsets_free = 0;
1065
1066                 isempty = PageIsEmpty(page);
1067
1068                 dowrite = false;
1069                 if (blkno == last_vacuum_block) /* it's reaped page */
1070                 {
1071                         if (last_vacuum_page->offsets_free > 0) /* there are dead tuples */
1072                         {                                       /* on this page - clean */
1073                                 Assert(!isempty);
1074                                 vacuum_page(page, last_vacuum_page);
1075                                 dowrite = true;
1076                         }
1077                         else
1078                                 Assert(isempty);
1079                         --vacuumed_pages;
1080                         if (vacuumed_pages > 0)
1081                         {
1082                                 /* get prev reaped page from vacuum_pages */
1083                                 last_vacuum_page = vacuum_pages->pagedesc[vacuumed_pages - 1];
1084                                 last_vacuum_block = last_vacuum_page->blkno;
1085                         }
1086                         else
1087                         {
1088                                 last_vacuum_page = NULL;
1089                                 last_vacuum_block = -1;
1090                         }
1091                         if (num_fraged_pages > 0 &&
1092                         fraged_pages->pagedesc[num_fraged_pages - 1]->blkno ==
1093                                 (BlockNumber) blkno)
1094                         {
1095                                 /* page is in fraged_pages too; remove it */
1096                                 --num_fraged_pages;
1097                         }
1098                         if (isempty)
1099                         {
1100                                 ReleaseBuffer(buf);
1101                                 continue;
1102                         }
1103                 }
1104                 else
1105                         Assert(!isempty);
1106
1107                 chain_tuple_moved = false;              /* no one chain-tuple was moved
1108                                                                                  * off this page, yet */
1109                 vacpage->blkno = blkno;
1110                 maxoff = PageGetMaxOffsetNumber(page);
1111                 for (offnum = FirstOffsetNumber;
1112                          offnum <= maxoff;
1113                          offnum = OffsetNumberNext(offnum))
1114                 {
1115                         itemid = PageGetItemId(page, offnum);
1116
1117                         if (!ItemIdIsUsed(itemid))
1118                                 continue;
1119
1120                         tuple.t_datamcxt = NULL;
1121                         tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
1122                         tuple_len = tuple.t_len = ItemIdGetLength(itemid);
1123                         ItemPointerSet(&(tuple.t_self), blkno, offnum);
1124
1125                         if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
1126                         {
1127                                 if ((TransactionId) tuple.t_data->t_cmin != myXID)
1128                                         elog(ERROR, "Invalid XID in t_cmin");
1129                                 if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
1130                                         elog(ERROR, "HEAP_MOVED_IN was not expected");
1131
1132                                 /*
1133                                  * If this (chain) tuple is moved by me already then I
1134                                  * have to check is it in vacpage or not - i.e. is it moved
1135                                  * while cleaning this page or some previous one.
1136                                  */
1137                                 if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
1138                                 {
1139                                         if (keep_tuples == 0)
1140                                                 continue;
1141                                         if (chain_tuple_moved)          /* some chains was moved
1142                                                                                                  * while */
1143                                         {                       /* cleaning this page */
1144                                                 Assert(vacpage->offsets_free > 0);
1145                                                 for (i = 0; i < vacpage->offsets_free; i++)
1146                                                 {
1147                                                         if (vacpage->offsets[i] == offnum)
1148                                                                 break;
1149                                                 }
1150                                                 if (i >= vacpage->offsets_free) /* not found */
1151                                                 {
1152                                                         vacpage->offsets[vacpage->offsets_free++] = offnum;
1153                                                         keep_tuples--;
1154                                                 }
1155                                         }
1156                                         else
1157                                         {
1158                                                 vacpage->offsets[vacpage->offsets_free++] = offnum;
1159                                                 keep_tuples--;
1160                                         }
1161                                         continue;
1162                                 }
1163                                 elog(ERROR, "HEAP_MOVED_OFF was expected");
1164                         }
1165
1166                         /*
1167                          * If this tuple is in the chain of tuples created in updates
1168                          * by "recent" transactions then we have to move all chain of
1169                          * tuples to another places.
1170                          */
1171                         if ((tuple.t_data->t_infomask & HEAP_UPDATED &&
1172                                  tuple.t_data->t_xmin >= XmaxRecent) ||
1173                                 (!(tuple.t_data->t_infomask & HEAP_XMAX_INVALID) &&
1174                                  !(ItemPointerEquals(&(tuple.t_self), &(tuple.t_data->t_ctid)))))
1175                         {
1176                                 Buffer          Cbuf = buf;
1177                                 Page            Cpage;
1178                                 ItemId          Citemid;
1179                                 ItemPointerData Ctid;
1180                                 HeapTupleData tp = tuple;
1181                                 Size            tlen = tuple_len;
1182                                 VTupleMove      vtmove = (VTupleMove)
1183                                 palloc(100 * sizeof(VTupleMoveData));
1184                                 int                     num_vtmove = 0;
1185                                 int                     free_vtmove = 100;
1186                                 VacPage         to_vacpage = NULL;
1187                                 int                     to_item = 0;
1188                                 bool            freeCbuf = false;
1189                                 int                     ti;
1190
1191                                 if (vacrelstats->vtlinks == NULL)
1192                                         elog(ERROR, "No one parent tuple was found");
1193                                 if (cur_buffer != InvalidBuffer)
1194                                 {
1195                                         WriteBuffer(cur_buffer);
1196                                         cur_buffer = InvalidBuffer;
1197                                 }
1198
1199                                 /*
1200                                  * If this tuple is in the begin/middle of the chain then
1201                                  * we have to move to the end of chain.
1202                                  */
1203                                 while (!(tp.t_data->t_infomask & HEAP_XMAX_INVALID) &&
1204                                 !(ItemPointerEquals(&(tp.t_self), &(tp.t_data->t_ctid))))
1205                                 {
1206                                         Ctid = tp.t_data->t_ctid;
1207                                         if (freeCbuf)
1208                                                 ReleaseBuffer(Cbuf);
1209                                         freeCbuf = true;
1210                                         Cbuf = ReadBuffer(onerel,
1211                                                                           ItemPointerGetBlockNumber(&Ctid));
1212                                         Cpage = BufferGetPage(Cbuf);
1213                                         Citemid = PageGetItemId(Cpage,
1214                                                                           ItemPointerGetOffsetNumber(&Ctid));
1215                                         if (!ItemIdIsUsed(Citemid))
1216                                         {
1217
1218                                                 /*
1219                                                  * This means that in the middle of chain there
1220                                                  * was tuple updated by older (than XmaxRecent)
1221                                                  * xaction and this tuple is already deleted by
1222                                                  * me. Actually, upper part of chain should be
1223                                                  * removed and seems that this should be handled
1224                                                  * in scan_heap(), but it's not implemented at
1225                                                  * the moment and so we just stop shrinking here.
1226                                                  */
1227                                                 ReleaseBuffer(Cbuf);
1228                                                 pfree(vtmove);
1229                                                 vtmove = NULL;
1230                                                 elog(NOTICE, "Child itemid in update-chain marked as unused - can't continue repair_frag");
1231                                                 break;
1232                                         }
1233                                         tp.t_datamcxt = NULL;
1234                                         tp.t_data = (HeapTupleHeader) PageGetItem(Cpage, Citemid);
1235                                         tp.t_self = Ctid;
1236                                         tlen = tp.t_len = ItemIdGetLength(Citemid);
1237                                 }
1238                                 if (vtmove == NULL)
1239                                         break;
1240                                 /* first, can chain be moved ? */
1241                                 for (;;)
1242                                 {
1243                                         if (to_vacpage == NULL ||
1244                                                 !enough_space(to_vacpage, tlen))
1245                                         {
1246
1247                                                 /*
1248                                                  * if to_vacpage no longer has enough free space to be
1249                                                  * useful, remove it from fraged_pages list
1250                                                  */
1251                                                 if (to_vacpage != NULL &&
1252                                                  !enough_space(to_vacpage, vacrelstats->min_tlen))
1253                                                 {
1254                                                         Assert(num_fraged_pages > to_item);
1255                                                         memmove(fraged_pages->pagedesc + to_item,
1256                                                                 fraged_pages->pagedesc + to_item + 1,
1257                                                                         sizeof(VacPage) * (num_fraged_pages - to_item - 1));
1258                                                         num_fraged_pages--;
1259                                                 }
1260                                                 for (i = 0; i < num_fraged_pages; i++)
1261                                                 {
1262                                                         if (enough_space(fraged_pages->pagedesc[i], tlen))
1263                                                                 break;
1264                                                 }
1265
1266                                                 /* can't move item anywhere */
1267                                                 if (i == num_fraged_pages)
1268                                                 {
1269                                                         for (i = 0; i < num_vtmove; i++)
1270                                                         {
1271                                                                 Assert(vtmove[i].vacpage->offsets_used > 0);
1272                                                                 (vtmove[i].vacpage->offsets_used)--;
1273                                                         }
1274                                                         num_vtmove = 0;
1275                                                         break;
1276                                                 }
1277                                                 to_item = i;
1278                                                 to_vacpage = fraged_pages->pagedesc[to_item];
1279                                         }
1280                                         to_vacpage->free -= MAXALIGN(tlen);
1281                                         if (to_vacpage->offsets_used >= to_vacpage->offsets_free)
1282                                                 to_vacpage->free -= MAXALIGN(sizeof(ItemIdData));
1283                                         (to_vacpage->offsets_used)++;
1284                                         if (free_vtmove == 0)
1285                                         {
1286                                                 free_vtmove = 1000;
1287                                                 vtmove = (VTupleMove) repalloc(vtmove,
1288                                                                                          (free_vtmove + num_vtmove) *
1289                                                                                                  sizeof(VTupleMoveData));
1290                                         }
1291                                         vtmove[num_vtmove].tid = tp.t_self;
1292                                         vtmove[num_vtmove].vacpage = to_vacpage;
1293                                         if (to_vacpage->offsets_used == 1)
1294                                                 vtmove[num_vtmove].cleanVpd = true;
1295                                         else
1296                                                 vtmove[num_vtmove].cleanVpd = false;
1297                                         free_vtmove--;
1298                                         num_vtmove++;
1299
1300                                         /* All done ? */
1301                                         if (!(tp.t_data->t_infomask & HEAP_UPDATED) ||
1302                                                 tp.t_data->t_xmin < XmaxRecent)
1303                                                 break;
1304
1305                                         /* Well, try to find tuple with old row version */
1306                                         for (;;)
1307                                         {
1308                                                 Buffer          Pbuf;
1309                                                 Page            Ppage;
1310                                                 ItemId          Pitemid;
1311                                                 HeapTupleData Ptp;
1312                                                 VTupleLinkData vtld,
1313                                                                    *vtlp;
1314
1315                                                 vtld.new_tid = tp.t_self;
1316                                                 vtlp = (VTupleLink)
1317                                                         vac_find_eq((void *) (vacrelstats->vtlinks),
1318                                                                            vacrelstats->num_vtlinks,
1319                                                                            sizeof(VTupleLinkData),
1320                                                                            (void *) &vtld,
1321                                                                            vac_cmp_vtlinks);
1322                                                 if (vtlp == NULL)
1323                                                         elog(ERROR, "Parent tuple was not found");
1324                                                 tp.t_self = vtlp->this_tid;
1325                                                 Pbuf = ReadBuffer(onerel,
1326                                                                 ItemPointerGetBlockNumber(&(tp.t_self)));
1327                                                 Ppage = BufferGetPage(Pbuf);
1328                                                 Pitemid = PageGetItemId(Ppage,
1329                                                            ItemPointerGetOffsetNumber(&(tp.t_self)));
1330                                                 if (!ItemIdIsUsed(Pitemid))
1331                                                         elog(ERROR, "Parent itemid marked as unused");
1332                                                 Ptp.t_datamcxt = NULL;
1333                                                 Ptp.t_data = (HeapTupleHeader) PageGetItem(Ppage, Pitemid);
1334                                                 Assert(ItemPointerEquals(&(vtld.new_tid),
1335                                                                                                  &(Ptp.t_data->t_ctid)));
1336
1337                                                 /*
1338                                                  * Read above about cases when
1339                                                  * !ItemIdIsUsed(Citemid) (child item is
1340                                                  * removed)... Due to the fact that at the moment
1341                                                  * we don't remove unuseful part of update-chain,
1342                                                  * it's possible to get too old parent row here.
1343                                                  * Like as in the case which caused this problem,
1344                                                  * we stop shrinking here. I could try to find
1345                                                  * real parent row but want not to do it because
1346                                                  * of real solution will be implemented anyway,
1347                                                  * latter, and we are too close to 6.5 release. -
1348                                                  * vadim 06/11/99
1349                                                  */
1350                                                 if (Ptp.t_data->t_xmax != tp.t_data->t_xmin)
1351                                                 {
1352                                                         if (freeCbuf)
1353                                                                 ReleaseBuffer(Cbuf);
1354                                                         freeCbuf = false;
1355                                                         ReleaseBuffer(Pbuf);
1356                                                         for (i = 0; i < num_vtmove; i++)
1357                                                         {
1358                                                                 Assert(vtmove[i].vacpage->offsets_used > 0);
1359                                                                 (vtmove[i].vacpage->offsets_used)--;
1360                                                         }
1361                                                         num_vtmove = 0;
1362                                                         elog(NOTICE, "Too old parent tuple found - can't continue repair_frag");
1363                                                         break;
1364                                                 }
1365 #ifdef NOT_USED                                 /* I'm not sure that this will wotk
1366                                                                  * properly... */
1367
1368                                                 /*
1369                                                  * If this tuple is updated version of row and it
1370                                                  * was created by the same transaction then no one
1371                                                  * is interested in this tuple - mark it as
1372                                                  * removed.
1373                                                  */
1374                                                 if (Ptp.t_data->t_infomask & HEAP_UPDATED &&
1375                                                         Ptp.t_data->t_xmin == Ptp.t_data->t_xmax)
1376                                                 {
1377                                                         TransactionIdStore(myXID,
1378                                                                 (TransactionId *) &(Ptp.t_data->t_cmin));
1379                                                         Ptp.t_data->t_infomask &=
1380                                                                 ~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
1381                                                         Ptp.t_data->t_infomask |= HEAP_MOVED_OFF;
1382                                                         WriteBuffer(Pbuf);
1383                                                         continue;
1384                                                 }
1385 #endif
1386                                                 tp.t_datamcxt = Ptp.t_datamcxt;
1387                                                 tp.t_data = Ptp.t_data;
1388                                                 tlen = tp.t_len = ItemIdGetLength(Pitemid);
1389                                                 if (freeCbuf)
1390                                                         ReleaseBuffer(Cbuf);
1391                                                 Cbuf = Pbuf;
1392                                                 freeCbuf = true;
1393                                                 break;
1394                                         }
1395                                         if (num_vtmove == 0)
1396                                                 break;
1397                                 }
1398                                 if (freeCbuf)
1399                                         ReleaseBuffer(Cbuf);
1400                                 if (num_vtmove == 0)    /* chain can't be moved */
1401                                 {
1402                                         pfree(vtmove);
1403                                         break;
1404                                 }
1405                                 ItemPointerSetInvalid(&Ctid);
1406                                 for (ti = 0; ti < num_vtmove; ti++)
1407                                 {
1408                                         VacPage destvacpage = vtmove[ti].vacpage;
1409
1410                                         /* Get page to move from */
1411                                         tuple.t_self = vtmove[ti].tid;
1412                                         Cbuf = ReadBuffer(onerel,
1413                                                          ItemPointerGetBlockNumber(&(tuple.t_self)));
1414
1415                                         /* Get page to move to */
1416                                         cur_buffer = ReadBuffer(onerel, destvacpage->blkno);
1417
1418                                         LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
1419                                         if (cur_buffer != Cbuf)
1420                                                 LockBuffer(Cbuf, BUFFER_LOCK_EXCLUSIVE);
1421
1422                                         ToPage = BufferGetPage(cur_buffer);
1423                                         Cpage = BufferGetPage(Cbuf);
1424
1425                                         /* NO ELOG(ERROR) TILL CHANGES ARE LOGGED */
1426
1427                                         Citemid = PageGetItemId(Cpage,
1428                                                         ItemPointerGetOffsetNumber(&(tuple.t_self)));
1429                                         tuple.t_datamcxt = NULL;
1430                                         tuple.t_data = (HeapTupleHeader) PageGetItem(Cpage, Citemid);
1431                                         tuple_len = tuple.t_len = ItemIdGetLength(Citemid);
1432
1433                                         /*
1434                                          * make a copy of the source tuple, and then mark the
1435                                          * source tuple MOVED_OFF.
1436                                          */
1437                                         heap_copytuple_with_tuple(&tuple, &newtup);
1438
1439                                         RelationInvalidateHeapTuple(onerel, &tuple);
1440
1441                                         TransactionIdStore(myXID, (TransactionId *) &(tuple.t_data->t_cmin));
1442                                         tuple.t_data->t_infomask &=
1443                                                 ~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
1444                                         tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
1445
1446                                         /*
1447                                          * If this page was not used before - clean it.
1448                                          *
1449                                          * NOTE: a nasty bug used to lurk here.  It is possible
1450                                          * for the source and destination pages to be the same
1451                                          * (since this tuple-chain member can be on a page lower
1452                                          * than the one we're currently processing in the outer
1453                                          * loop).  If that's true, then after vacuum_page() the
1454                                          * source tuple will have been moved, and tuple.t_data
1455                                          * will be pointing at garbage.  Therefore we must do
1456                                          * everything that uses tuple.t_data BEFORE this step!!
1457                                          *
1458                                          * This path is different from the other callers of
1459                                          * vacuum_page, because we have already incremented the
1460                                          * vacpage's offsets_used field to account for the
1461                                          * tuple(s) we expect to move onto the page. Therefore
1462                                          * vacuum_page's check for offsets_used == 0 is
1463                                          * wrong. But since that's a good debugging check for
1464                                          * all other callers, we work around it here rather
1465                                          * than remove it.
1466                                          */
1467                                         if (!PageIsEmpty(ToPage) && vtmove[ti].cleanVpd)
1468                                         {
1469                                                 int                     sv_offsets_used = destvacpage->offsets_used;
1470
1471                                                 destvacpage->offsets_used = 0;
1472                                                 vacuum_page(ToPage, destvacpage);
1473                                                 destvacpage->offsets_used = sv_offsets_used;
1474                                         }
1475
1476                                         /*
1477                                          * Update the state of the copied tuple, and store it
1478                                          * on the destination page.
1479                                          */
1480                                         TransactionIdStore(myXID, (TransactionId *) &(newtup.t_data->t_cmin));
1481                                         newtup.t_data->t_infomask &=
1482                                                 ~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_OFF);
1483                                         newtup.t_data->t_infomask |= HEAP_MOVED_IN;
1484                                         newoff = PageAddItem(ToPage, (Item) newtup.t_data, tuple_len,
1485                                                                                  InvalidOffsetNumber, LP_USED);
1486                                         if (newoff == InvalidOffsetNumber)
1487                                         {
1488                                                 elog(STOP, "moving chain: failed to add item with len = %lu to page %u",
1489                                                          (unsigned long)tuple_len, destvacpage->blkno);
1490                                         }
1491                                         newitemid = PageGetItemId(ToPage, newoff);
1492                                         pfree(newtup.t_data);
1493                                         newtup.t_datamcxt = NULL;
1494                                         newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
1495                                         ItemPointerSet(&(newtup.t_self), destvacpage->blkno, newoff);
1496
1497 #ifdef XLOG
1498                                         {
1499                                                 XLogRecPtr      recptr =
1500                                                         log_heap_move(onerel, tuple.t_self, &newtup);
1501
1502                                                 if (Cbuf != cur_buffer)
1503                                                 {
1504                                                         PageSetLSN(Cpage, recptr);
1505                                                         PageSetSUI(Cpage, ThisStartUpID);
1506                                                 }
1507                                                 PageSetLSN(ToPage, recptr);
1508                                                 PageSetSUI(ToPage, ThisStartUpID);
1509                                         }
1510 #endif
1511
1512                                         if (((int) destvacpage->blkno) > last_move_dest_block)
1513                                                 last_move_dest_block = destvacpage->blkno;
1514
1515                                         /*
1516                                          * Set new tuple's t_ctid pointing to itself for last
1517                                          * tuple in chain, and to next tuple in chain otherwise.
1518                                          */
1519                                         if (!ItemPointerIsValid(&Ctid))
1520                                                 newtup.t_data->t_ctid = newtup.t_self;
1521                                         else
1522                                                 newtup.t_data->t_ctid = Ctid;
1523                                         Ctid = newtup.t_self;
1524
1525                                         num_moved++;
1526
1527                                         /*
1528                                          * Remember that we moved tuple from the current page
1529                                          * (corresponding index tuple will be cleaned).
1530                                          */
1531                                         if (Cbuf == buf)
1532                                                 vacpage->offsets[vacpage->offsets_free++] =
1533                                                         ItemPointerGetOffsetNumber(&(tuple.t_self));
1534                                         else
1535                                                 keep_tuples++;
1536
1537                                         LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
1538                                         if (cur_buffer != Cbuf)
1539                                                 LockBuffer(Cbuf, BUFFER_LOCK_UNLOCK);
1540
1541                                         if (Irel != (Relation *) NULL)
1542                                         {
1543                                                 /*
1544                                                  * XXX using CurrentMemoryContext here means
1545                                                  * intra-vacuum memory leak for functional indexes.
1546                                                  * Should fix someday.
1547                                                  *
1548                                                  * XXX This code fails to handle partial indexes!
1549                                                  * Probably should change it to use ExecOpenIndices.
1550                                                  */
1551                                                 for (i = 0; i < nindices; i++)
1552                                                 {
1553                                                         FormIndexDatum(indexInfo[i],
1554                                                                                    &newtup,
1555                                                                                    tupdesc,
1556                                                                                    CurrentMemoryContext,
1557                                                                                    idatum,
1558                                                                                    inulls);
1559                                                         iresult = index_insert(Irel[i],
1560                                                                                                    idatum,
1561                                                                                                    inulls,
1562                                                                                                    &newtup.t_self,
1563                                                                                                    onerel);
1564                                                         if (iresult)
1565                                                                 pfree(iresult);
1566                                                 }
1567                                         }
1568                                         WriteBuffer(cur_buffer);
1569                                         WriteBuffer(Cbuf);
1570                                 }
1571                                 cur_buffer = InvalidBuffer;
1572                                 pfree(vtmove);
1573                                 chain_tuple_moved = true;
1574                                 continue;
1575                         }
1576
1577                         /* try to find new page for this tuple */
1578                         if (cur_buffer == InvalidBuffer ||
1579                                 !enough_space(cur_page, tuple_len))
1580                         {
1581                                 if (cur_buffer != InvalidBuffer)
1582                                 {
1583                                         WriteBuffer(cur_buffer);
1584                                         cur_buffer = InvalidBuffer;
1585
1586                                         /*
1587                                          * If previous target page is now too full to add *any*
1588                                          * tuple to it, remove it from fraged_pages.
1589                                          */
1590                                         if (!enough_space(cur_page, vacrelstats->min_tlen))
1591                                         {
1592                                                 Assert(num_fraged_pages > cur_item);
1593                                                 memmove(fraged_pages->pagedesc + cur_item,
1594                                                                 fraged_pages->pagedesc + cur_item + 1,
1595                                                                 sizeof(VacPage) * (num_fraged_pages - cur_item - 1));
1596                                                 num_fraged_pages--;
1597                                         }
1598                                 }
1599                                 for (i = 0; i < num_fraged_pages; i++)
1600                                 {
1601                                         if (enough_space(fraged_pages->pagedesc[i], tuple_len))
1602                                                 break;
1603                                 }
1604                                 if (i == num_fraged_pages)
1605                                         break;          /* can't move item anywhere */
1606                                 cur_item = i;
1607                                 cur_page = fraged_pages->pagedesc[cur_item];
1608                                 cur_buffer = ReadBuffer(onerel, cur_page->blkno);
1609                                 LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
1610                                 ToPage = BufferGetPage(cur_buffer);
1611                                 /* if this page was not used before - clean it */
1612                                 if (!PageIsEmpty(ToPage) && cur_page->offsets_used == 0)
1613                                         vacuum_page(ToPage, cur_page);
1614                         }
1615                         else
1616                                 LockBuffer(cur_buffer, BUFFER_LOCK_EXCLUSIVE);
1617
1618                         LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
1619
1620                         /* copy tuple */
1621                         heap_copytuple_with_tuple(&tuple, &newtup);
1622
1623                         RelationInvalidateHeapTuple(onerel, &tuple);
1624
1625                         /*
1626                          * Mark new tuple as moved_in by vacuum and store vacuum XID
1627                          * in t_cmin !!!
1628                          */
1629                         TransactionIdStore(myXID, (TransactionId *) &(newtup.t_data->t_cmin));
1630                         newtup.t_data->t_infomask &=
1631                                 ~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_OFF);
1632                         newtup.t_data->t_infomask |= HEAP_MOVED_IN;
1633
1634                         /* add tuple to the page */
1635                         newoff = PageAddItem(ToPage, (Item) newtup.t_data, tuple_len,
1636                                                                  InvalidOffsetNumber, LP_USED);
1637                         if (newoff == InvalidOffsetNumber)
1638                         {
1639                                 elog(ERROR, "\
1640 failed to add item with len = %lu to page %u (free space %lu, nusd %u, noff %u)",
1641                                          (unsigned long)tuple_len, cur_page->blkno, (unsigned long)cur_page->free,
1642                                  cur_page->offsets_used, cur_page->offsets_free);
1643                         }
1644                         newitemid = PageGetItemId(ToPage, newoff);
1645                         pfree(newtup.t_data);
1646                         newtup.t_datamcxt = NULL;
1647                         newtup.t_data = (HeapTupleHeader) PageGetItem(ToPage, newitemid);
1648                         ItemPointerSet(&(newtup.t_data->t_ctid), cur_page->blkno, newoff);
1649                         newtup.t_self = newtup.t_data->t_ctid;
1650
1651                         /*
1652                          * Mark old tuple as moved_off by vacuum and store vacuum XID
1653                          * in t_cmin !!!
1654                          */
1655                         TransactionIdStore(myXID, (TransactionId *) &(tuple.t_data->t_cmin));
1656                         tuple.t_data->t_infomask &=
1657                                 ~(HEAP_XMIN_COMMITTED | HEAP_XMIN_INVALID | HEAP_MOVED_IN);
1658                         tuple.t_data->t_infomask |= HEAP_MOVED_OFF;
1659
1660 #ifdef XLOG
1661                         {
1662                                 XLogRecPtr      recptr =
1663                                         log_heap_move(onerel, tuple.t_self, &newtup);
1664
1665                                 PageSetLSN(page, recptr);
1666                                 PageSetSUI(page, ThisStartUpID);
1667                                 PageSetLSN(ToPage, recptr);
1668                                 PageSetSUI(ToPage, ThisStartUpID);
1669                         }
1670 #endif
1671
1672                         cur_page->offsets_used++;
1673                         num_moved++;
1674                         cur_page->free = ((PageHeader) ToPage)->pd_upper - ((PageHeader) ToPage)->pd_lower;
1675                         if (((int) cur_page->blkno) > last_move_dest_block)
1676                                 last_move_dest_block = cur_page->blkno;
1677
1678                         vacpage->offsets[vacpage->offsets_free++] = offnum;
1679
1680                         LockBuffer(cur_buffer, BUFFER_LOCK_UNLOCK);
1681                         LockBuffer(buf, BUFFER_LOCK_UNLOCK);
1682
1683                         /* insert index' tuples if needed */
1684                         if (Irel != (Relation *) NULL)
1685                         {
1686                                 /*
1687                                  * XXX using CurrentMemoryContext here means
1688                                  * intra-vacuum memory leak for functional indexes.
1689                                  * Should fix someday.
1690                                  *
1691                                  * XXX This code fails to handle partial indexes!
1692                                  * Probably should change it to use ExecOpenIndices.
1693                                  */
1694                                 for (i = 0; i < nindices; i++)
1695                                 {
1696                                         FormIndexDatum(indexInfo[i],
1697                                                                    &newtup,
1698                                                                    tupdesc,
1699                                                                    CurrentMemoryContext,
1700                                                                    idatum,
1701                                                                    inulls);
1702                                         iresult = index_insert(Irel[i],
1703                                                                                    idatum,
1704                                                                                    inulls,
1705                                                                                    &newtup.t_self,
1706                                                                                    onerel);
1707                                         if (iresult)
1708                                                 pfree(iresult);
1709                                 }
1710                         }
1711
1712                 }                                               /* walk along page */
1713
1714                 if (offnum < maxoff && keep_tuples > 0)
1715                 {
1716                         OffsetNumber off;
1717
1718                         for (off = OffsetNumberNext(offnum);
1719                                  off <= maxoff;
1720                                  off = OffsetNumberNext(off))
1721                         {
1722                                 itemid = PageGetItemId(page, off);
1723                                 if (!ItemIdIsUsed(itemid))
1724                                         continue;
1725                                 tuple.t_datamcxt = NULL;
1726                                 tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
1727                                 if (tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED)
1728                                         continue;
1729                                 if ((TransactionId) tuple.t_data->t_cmin != myXID)
1730                                         elog(ERROR, "Invalid XID in t_cmin (4)");
1731                                 if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
1732                                         elog(ERROR, "HEAP_MOVED_IN was not expected (2)");
1733                                 if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
1734                                 {
1735                                         /* some chains was moved while */
1736                                         if (chain_tuple_moved)
1737                                         {                       /* cleaning this page */
1738                                                 Assert(vacpage->offsets_free > 0);
1739                                                 for (i = 0; i < vacpage->offsets_free; i++)
1740                                                 {
1741                                                         if (vacpage->offsets[i] == off)
1742                                                                 break;
1743                                                 }
1744                                                 if (i >= vacpage->offsets_free) /* not found */
1745                                                 {
1746                                                         vacpage->offsets[vacpage->offsets_free++] = off;
1747                                                         Assert(keep_tuples > 0);
1748                                                         keep_tuples--;
1749                                                 }
1750                                         }
1751                                         else
1752                                         {
1753                                                 vacpage->offsets[vacpage->offsets_free++] = off;
1754                                                 Assert(keep_tuples > 0);
1755                                                 keep_tuples--;
1756                                         }
1757                                 }
1758                         }
1759                 }
1760
1761                 if (vacpage->offsets_free > 0)  /* some tuples were moved */
1762                 {
1763                         if (chain_tuple_moved)          /* else - they are ordered */
1764                         {
1765                                 qsort((char *) (vacpage->offsets), vacpage->offsets_free,
1766                                           sizeof(OffsetNumber), vac_cmp_offno);
1767                         }
1768                         reap_page(&Nvacpagelist, vacpage);
1769                         WriteBuffer(buf);
1770                 }
1771                 else if (dowrite)
1772                         WriteBuffer(buf);
1773                 else
1774                         ReleaseBuffer(buf);
1775
1776                 if (offnum <= maxoff)
1777                         break;                          /* some item(s) left */
1778
1779         }                                                       /* walk along relation */
1780
1781         blkno++;                                        /* new number of blocks */
1782
1783         if (cur_buffer != InvalidBuffer)
1784         {
1785                 Assert(num_moved > 0);
1786                 WriteBuffer(cur_buffer);
1787         }
1788
1789         if (num_moved > 0)
1790         {
1791 #ifdef XLOG
1792                 RecordTransactionCommit();
1793 #else
1794                 /*
1795                  * We have to commit our tuple' movings before we'll truncate
1796                  * relation, but we shouldn't lose our locks. And so - quick hack:
1797                  * flush buffers and record status of current transaction as
1798                  * committed, and continue. - vadim 11/13/96
1799                  */
1800                 FlushBufferPool();
1801                 TransactionIdCommit(myXID);
1802                 FlushBufferPool();
1803 #endif
1804         }
1805
1806         /*
1807          * Clean uncleaned reaped pages from vacuum_pages list list and set
1808          * xmin committed for inserted tuples
1809          */
1810         checked_moved = 0;
1811         for (i = 0, curpage = vacuum_pages->pagedesc; i < vacuumed_pages; i++, curpage++)
1812         {
1813                 Assert((*curpage)->blkno < (BlockNumber) blkno);
1814                 buf = ReadBuffer(onerel, (*curpage)->blkno);
1815                 page = BufferGetPage(buf);
1816                 if ((*curpage)->offsets_used == 0)              /* this page was not used */
1817                 {
1818                         if (!PageIsEmpty(page))
1819                                 vacuum_page(page, *curpage);
1820                 }
1821                 else
1822 /* this page was used */
1823                 {
1824                         num_tuples = 0;
1825                         max_offset = PageGetMaxOffsetNumber(page);
1826                         for (newoff = FirstOffsetNumber;
1827                                  newoff <= max_offset;
1828                                  newoff = OffsetNumberNext(newoff))
1829                         {
1830                                 itemid = PageGetItemId(page, newoff);
1831                                 if (!ItemIdIsUsed(itemid))
1832                                         continue;
1833                                 tuple.t_datamcxt = NULL;
1834                                 tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
1835                                 if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
1836                                 {
1837                                         if ((TransactionId) tuple.t_data->t_cmin != myXID)
1838                                                 elog(ERROR, "Invalid XID in t_cmin (2)");
1839                                         if (tuple.t_data->t_infomask & HEAP_MOVED_IN)
1840                                         {
1841                                                 tuple.t_data->t_infomask |= HEAP_XMIN_COMMITTED;
1842                                                 num_tuples++;
1843                                         }
1844                                         else if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
1845                                                 tuple.t_data->t_infomask |= HEAP_XMIN_INVALID;
1846                                         else
1847                                                 elog(ERROR, "HEAP_MOVED_OFF/HEAP_MOVED_IN was expected");
1848                                 }
1849                         }
1850                         Assert((*curpage)->offsets_used == num_tuples);
1851                         checked_moved += num_tuples;
1852                 }
1853                 WriteBuffer(buf);
1854         }
1855         Assert(num_moved == checked_moved);
1856
1857         elog(MESSAGE_LEVEL, "Rel %s: Pages: %u --> %u; Tuple(s) moved: %u. %s",
1858                  RelationGetRelationName(onerel),
1859                  nblocks, blkno, num_moved,
1860                  show_rusage(&ru0));
1861
1862         if (Nvacpagelist.num_pages > 0)
1863         {
1864                 /* vacuum indices again if needed */
1865                 if (Irel != (Relation *) NULL)
1866                 {
1867                         VacPage    *vpleft,
1868                                            *vpright,
1869                                                 vpsave;
1870
1871                         /* re-sort Nvacpagelist.pagedesc */
1872                         for (vpleft = Nvacpagelist.pagedesc,
1873                                  vpright = Nvacpagelist.pagedesc + Nvacpagelist.num_pages - 1;
1874                                  vpleft < vpright; vpleft++, vpright--)
1875                         {
1876                                 vpsave = *vpleft;
1877                                 *vpleft = *vpright;
1878                                 *vpright = vpsave;
1879                         }
1880                         Assert(keep_tuples >= 0);
1881                         for (i = 0; i < nindices; i++)
1882                                 vacuum_index(&Nvacpagelist, Irel[i],
1883                                                          vacrelstats->num_tuples, keep_tuples);
1884                 }
1885
1886                 /* clean moved tuples from last page in Nvacpagelist list */
1887                 if (vacpage->blkno == (BlockNumber) (blkno - 1) &&
1888                         vacpage->offsets_free > 0)
1889                 {
1890                         buf = ReadBuffer(onerel, vacpage->blkno);
1891                         page = BufferGetPage(buf);
1892                         num_tuples = 0;
1893                         for (offnum = FirstOffsetNumber;
1894                                  offnum <= maxoff;
1895                                  offnum = OffsetNumberNext(offnum))
1896                         {
1897                                 itemid = PageGetItemId(page, offnum);
1898                                 if (!ItemIdIsUsed(itemid))
1899                                         continue;
1900                                 tuple.t_datamcxt = NULL;
1901                                 tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
1902
1903                                 if (!(tuple.t_data->t_infomask & HEAP_XMIN_COMMITTED))
1904                                 {
1905                                         if ((TransactionId) tuple.t_data->t_cmin != myXID)
1906                                                 elog(ERROR, "Invalid XID in t_cmin (3)");
1907                                         if (tuple.t_data->t_infomask & HEAP_MOVED_OFF)
1908                                         {
1909                                                 itemid->lp_flags &= ~LP_USED;
1910                                                 num_tuples++;
1911                                         }
1912                                         else
1913                                                 elog(ERROR, "HEAP_MOVED_OFF was expected (2)");
1914                                 }
1915
1916                         }
1917                         Assert(vacpage->offsets_free == num_tuples);
1918                         PageRepairFragmentation(page);
1919                         WriteBuffer(buf);
1920                 }
1921
1922                 /* now - free new list of reaped pages */
1923                 curpage = Nvacpagelist.pagedesc;
1924                 for (i = 0; i < Nvacpagelist.num_pages; i++, curpage++)
1925                         pfree(*curpage);
1926                 pfree(Nvacpagelist.pagedesc);
1927         }
1928
1929         /*
1930          * Flush dirty pages out to disk.  We do this unconditionally, even if
1931          * we don't need to truncate, because we want to ensure that all tuples
1932          * have correct on-row commit status on disk (see bufmgr.c's comments
1933          * for FlushRelationBuffers()).
1934          */
1935         i = FlushRelationBuffers(onerel, blkno);
1936         if (i < 0)
1937                 elog(ERROR, "VACUUM (repair_frag): FlushRelationBuffers returned %d",
1938                          i);
1939
1940         /* truncate relation, if needed */
1941         if (blkno < nblocks)
1942         {
1943                 blkno = smgrtruncate(DEFAULT_SMGR, onerel, blkno);
1944                 Assert(blkno >= 0);
1945                 vacrelstats->num_pages = blkno; /* set new number of blocks */
1946         }
1947
1948         if (Irel != (Relation *) NULL)          /* pfree index' allocations */
1949         {
1950                 close_indices(nindices, Irel);
1951                 pfree(indexInfo);
1952         }
1953
1954         pfree(vacpage);
1955         if (vacrelstats->vtlinks != NULL)
1956                 pfree(vacrelstats->vtlinks);
1957 }
1958
1959 /*
1960  *      vacuum_heap() -- free dead tuples
1961  *
1962  *              This routine marks dead tuples as unused and truncates relation
1963  *              if there are "empty" end-blocks.
1964  */
1965 static void
1966 vacuum_heap(VRelStats *vacrelstats, Relation onerel, VacPageList vacuum_pages)
1967 {
1968         Buffer          buf;
1969         Page            page;
1970         VacPage    *vacpage;
1971         int                     nblocks;
1972         int                     i;
1973
1974         nblocks = vacuum_pages->num_pages;
1975         nblocks -= vacuum_pages->empty_end_pages;               /* nothing to do with
1976                                                                                                          * them */
1977
1978         for (i = 0, vacpage = vacuum_pages->pagedesc; i < nblocks; i++, vacpage++)
1979         {
1980                 if ((*vacpage)->offsets_free > 0)
1981                 {
1982                         buf = ReadBuffer(onerel, (*vacpage)->blkno);
1983                         page = BufferGetPage(buf);
1984                         vacuum_page(page, *vacpage);
1985                         WriteBuffer(buf);
1986                 }
1987         }
1988
1989         /*
1990          * Flush dirty pages out to disk.  We do this unconditionally, even if
1991          * we don't need to truncate, because we want to ensure that all tuples
1992          * have correct on-row commit status on disk (see bufmgr.c's comments
1993          * for FlushRelationBuffers()).
1994          */
1995         Assert(vacrelstats->num_pages >= vacuum_pages->empty_end_pages);
1996         nblocks = vacrelstats->num_pages - vacuum_pages->empty_end_pages;
1997
1998         i = FlushRelationBuffers(onerel, nblocks);
1999         if (i < 0)
2000                 elog(ERROR, "VACUUM (vacuum_heap): FlushRelationBuffers returned %d",
2001                          i);
2002
2003         /* truncate relation if there are some empty end-pages */
2004         if (vacuum_pages->empty_end_pages > 0)
2005         {
2006                 elog(MESSAGE_LEVEL, "Rel %s: Pages: %u --> %u.",
2007                          RelationGetRelationName(onerel),
2008                          vacrelstats->num_pages, nblocks);
2009                 nblocks = smgrtruncate(DEFAULT_SMGR, onerel, nblocks);
2010                 Assert(nblocks >= 0);
2011                 vacrelstats->num_pages = nblocks; /* set new number of blocks */
2012         }
2013 }
2014
2015 /*
2016  *      vacuum_page() -- free dead tuples on a page
2017  *                                       and repair its fragmentation.
2018  */
2019 static void
2020 vacuum_page(Page page, VacPage vacpage)
2021 {
2022         ItemId          itemid;
2023         int                     i;
2024
2025         /* There shouldn't be any tuples moved onto the page yet! */
2026         Assert(vacpage->offsets_used == 0);
2027
2028         for (i = 0; i < vacpage->offsets_free; i++)
2029         {
2030                 itemid = &(((PageHeader) page)->pd_linp[vacpage->offsets[i] - 1]);
2031                 itemid->lp_flags &= ~LP_USED;
2032         }
2033         PageRepairFragmentation(page);
2034
2035 }
2036
2037 /*
2038  *      _scan_index() -- scan one index relation to update statistic.
2039  *
2040  */
2041 static void
2042 scan_index(Relation indrel, int num_tuples)
2043 {
2044         RetrieveIndexResult res;
2045         IndexScanDesc iscan;
2046         int                     nitups;
2047         int                     nipages;
2048         struct rusage ru0;
2049
2050         getrusage(RUSAGE_SELF, &ru0);
2051
2052         /* walk through the entire index */
2053         iscan = index_beginscan(indrel, false, 0, (ScanKey) NULL);
2054         nitups = 0;
2055
2056         while ((res = index_getnext(iscan, ForwardScanDirection))
2057                    != (RetrieveIndexResult) NULL)
2058         {
2059                 nitups++;
2060                 pfree(res);
2061         }
2062
2063         index_endscan(iscan);
2064
2065         /* now update statistics in pg_class */
2066         nipages = RelationGetNumberOfBlocks(indrel);
2067         update_relstats(RelationGetRelid(indrel), nipages, nitups, false, NULL);
2068
2069         elog(MESSAGE_LEVEL, "Index %s: Pages %u; Tuples %u. %s",
2070                  RelationGetRelationName(indrel), nipages, nitups,
2071                  show_rusage(&ru0));
2072
2073         if (nitups != num_tuples)
2074                 elog(NOTICE, "Index %s: NUMBER OF INDEX' TUPLES (%u) IS NOT THE SAME AS HEAP' (%u).\
2075 \n\tRecreate the index.",
2076                          RelationGetRelationName(indrel), nitups, num_tuples);
2077
2078 }
2079
2080 /*
2081  *      vacuum_index() -- vacuum one index relation.
2082  *
2083  *              Vpl is the VacPageList of the heap we're currently vacuuming.
2084  *              It's locked. Indrel is an index relation on the vacuumed heap.
2085  *              We don't set locks on the index relation here, since the indexed
2086  *              access methods support locking at different granularities.
2087  *              We let them handle it.
2088  *
2089  *              Finally, we arrange to update the index relation's statistics in
2090  *              pg_class.
2091  */
2092 static void
2093 vacuum_index(VacPageList vacpagelist, Relation indrel, int num_tuples, int keep_tuples)
2094 {
2095         RetrieveIndexResult res;
2096         IndexScanDesc iscan;
2097         ItemPointer heapptr;
2098         int                     tups_vacuumed;
2099         int                     num_index_tuples;
2100         int                     num_pages;
2101         VacPage         vp;
2102         struct rusage ru0;
2103
2104         getrusage(RUSAGE_SELF, &ru0);
2105
2106         /* walk through the entire index */
2107         iscan = index_beginscan(indrel, false, 0, (ScanKey) NULL);
2108         tups_vacuumed = 0;
2109         num_index_tuples = 0;
2110
2111         while ((res = index_getnext(iscan, ForwardScanDirection))
2112                    != (RetrieveIndexResult) NULL)
2113         {
2114                 heapptr = &res->heap_iptr;
2115
2116                 if ((vp = tid_reaped(heapptr, vacpagelist)) != (VacPage) NULL)
2117                 {
2118 #ifdef NOT_USED
2119                         elog(DEBUG, "<%x,%x> -> <%x,%x>",
2120                                  ItemPointerGetBlockNumber(&(res->index_iptr)),
2121                                  ItemPointerGetOffsetNumber(&(res->index_iptr)),
2122                                  ItemPointerGetBlockNumber(&(res->heap_iptr)),
2123                                  ItemPointerGetOffsetNumber(&(res->heap_iptr)));
2124 #endif
2125                         if (vp->offsets_free == 0)
2126                         {
2127                                 elog(NOTICE, "Index %s: pointer to EmptyPage (blk %u off %u) - fixing",
2128                                          RelationGetRelationName(indrel),
2129                                          vp->blkno, ItemPointerGetOffsetNumber(heapptr));
2130                         }
2131                         ++tups_vacuumed;
2132                         index_delete(indrel, &res->index_iptr);
2133                 }
2134                 else
2135                         num_index_tuples++;
2136
2137                 pfree(res);
2138         }
2139
2140         index_endscan(iscan);
2141
2142         /* now update statistics in pg_class */
2143         num_pages = RelationGetNumberOfBlocks(indrel);
2144         update_relstats(RelationGetRelid(indrel), num_pages, num_index_tuples, false, NULL);
2145
2146         elog(MESSAGE_LEVEL, "Index %s: Pages %u; Tuples %u: Deleted %u. %s",
2147                  RelationGetRelationName(indrel), num_pages,
2148                  num_index_tuples - keep_tuples, tups_vacuumed,
2149                  show_rusage(&ru0));
2150
2151         if (num_index_tuples != num_tuples + keep_tuples)
2152                 elog(NOTICE, "Index %s: NUMBER OF INDEX' TUPLES (%u) IS NOT THE SAME AS HEAP' (%u).\
2153 \n\tRecreate the index.",
2154                   RelationGetRelationName(indrel), num_index_tuples, num_tuples);
2155
2156 }
2157
2158 /*
2159  *      tid_reaped() -- is a particular tid reaped?
2160  *
2161  *              vacpagelist->VacPage_array is sorted in right order.
2162  */
2163 static VacPage
2164 tid_reaped(ItemPointer itemptr, VacPageList vacpagelist)
2165 {
2166         OffsetNumber ioffno;
2167         OffsetNumber *voff;
2168         VacPage         vp,
2169                            *vpp;
2170         VacPageData vacpage;
2171
2172         vacpage.blkno = ItemPointerGetBlockNumber(itemptr);
2173         ioffno = ItemPointerGetOffsetNumber(itemptr);
2174
2175         vp = &vacpage;
2176         vpp = (VacPage *) vac_find_eq((void *) (vacpagelist->pagedesc),
2177                                         vacpagelist->num_pages, sizeof(VacPage), (void *) &vp,
2178                                                                         vac_cmp_blk);
2179
2180         if (vpp == (VacPage *) NULL)
2181                 return (VacPage) NULL;
2182         vp = *vpp;
2183
2184         /* ok - we are on true page */
2185
2186         if (vp->offsets_free == 0)
2187         {                                                       /* this is EmptyPage !!! */
2188                 return vp;
2189         }
2190
2191         voff = (OffsetNumber *) vac_find_eq((void *) (vp->offsets),
2192                         vp->offsets_free, sizeof(OffsetNumber), (void *) &ioffno,
2193                                                                            vac_cmp_offno);
2194
2195         if (voff == (OffsetNumber *) NULL)
2196                 return (VacPage) NULL;
2197
2198         return vp;
2199
2200 }
2201
2202 /*
2203  *      update_relstats() -- update statistics for one relation
2204  *
2205  *              Statistics are stored in several places: the pg_class row for the
2206  *              relation has stats about the whole relation, the pg_attribute rows
2207  *              for each attribute store "dispersion", and there is a pg_statistic
2208  *              row for each (non-system) attribute.  (Dispersion probably ought to
2209  *              be moved to pg_statistic, but it's not worth doing unless there's
2210  *              another reason to have to change pg_attribute.)  Dispersion and
2211  *              pg_statistic values are only updated by VACUUM ANALYZE, but we
2212  *              always update the stats in pg_class.
2213  *
2214  *              This routine works for both index and heap relation entries in
2215  *              pg_class.  We violate no-overwrite semantics here by storing new
2216  *              values for the statistics columns directly into the pg_class
2217  *              tuple that's already on the page.  The reason for this is that if
2218  *              we updated these tuples in the usual way, vacuuming pg_class itself
2219  *              wouldn't work very well --- by the time we got done with a vacuum
2220  *              cycle, most of the tuples in pg_class would've been obsoleted.
2221  *              Updating pg_class's own statistics would be especially tricky.
2222  *              Of course, this only works for fixed-size never-null columns, but
2223  *              these are.
2224  */
2225 static void
2226 update_relstats(Oid relid, int num_pages, int num_tuples, bool hasindex,
2227                         VRelStats *vacrelstats)
2228 {
2229         Relation        rd;
2230         HeapTupleData rtup;
2231         HeapTuple       ctup;
2232         Form_pg_class pgcform;
2233         Buffer          buffer;
2234
2235         /*
2236          * update number of tuples and number of pages in pg_class
2237          */
2238         rd = heap_openr(RelationRelationName, RowExclusiveLock);
2239
2240         ctup = SearchSysCacheTupleCopy(RELOID,
2241                                                                    ObjectIdGetDatum(relid),
2242                                                                    0, 0, 0);
2243         if (!HeapTupleIsValid(ctup))
2244                 elog(ERROR, "pg_class entry for relid %u vanished during vacuuming",
2245                          relid);
2246
2247         /* get the buffer cache tuple */
2248         rtup.t_self = ctup->t_self;
2249         heap_fetch(rd, SnapshotNow, &rtup, &buffer);
2250         heap_freetuple(ctup);
2251
2252         /* overwrite the existing statistics in the tuple */
2253         pgcform = (Form_pg_class) GETSTRUCT(&rtup);
2254         pgcform->reltuples = num_tuples;
2255         pgcform->relpages = num_pages;
2256         pgcform->relhasindex = hasindex;
2257
2258         /* invalidate the tuple in the cache and write the buffer */
2259         RelationInvalidateHeapTuple(rd, &rtup);
2260         WriteBuffer(buffer);
2261
2262         heap_close(rd, RowExclusiveLock);
2263 }
2264
2265 /*
2266  *      reap_page() -- save a page on the array of reaped pages.
2267  *
2268  *              As a side effect of the way that the vacuuming loop for a given
2269  *              relation works, higher pages come after lower pages in the array
2270  *              (and highest tid on a page is last).
2271  */
2272 static void
2273 reap_page(VacPageList vacpagelist, VacPage vacpage)
2274 {
2275         VacPage newvacpage;
2276
2277         /* allocate a VacPageData entry */
2278         newvacpage = (VacPage) palloc(sizeof(VacPageData) + vacpage->offsets_free * sizeof(OffsetNumber));
2279
2280         /* fill it in */
2281         if (vacpage->offsets_free > 0)
2282                 memmove(newvacpage->offsets, vacpage->offsets, vacpage->offsets_free * sizeof(OffsetNumber));
2283         newvacpage->blkno = vacpage->blkno;
2284         newvacpage->free = vacpage->free;
2285         newvacpage->offsets_used = vacpage->offsets_used;
2286         newvacpage->offsets_free = vacpage->offsets_free;
2287
2288         /* insert this page into vacpagelist list */
2289         vpage_insert(vacpagelist, newvacpage);
2290
2291 }
2292
2293 static void
2294 vpage_insert(VacPageList vacpagelist, VacPage vpnew)
2295 {
2296 #define PG_NPAGEDESC 1024
2297
2298         /* allocate a VacPage entry if needed */
2299         if (vacpagelist->num_pages == 0)
2300         {
2301                 vacpagelist->pagedesc = (VacPage *) palloc(PG_NPAGEDESC * sizeof(VacPage));
2302                 vacpagelist->num_allocated_pages = PG_NPAGEDESC;
2303         }
2304         else if (vacpagelist->num_pages >= vacpagelist->num_allocated_pages)
2305         {
2306                 vacpagelist->num_allocated_pages *= 2;
2307                 vacpagelist->pagedesc = (VacPage *) repalloc(vacpagelist->pagedesc, vacpagelist->num_allocated_pages * sizeof(VacPage));
2308         }
2309         vacpagelist->pagedesc[vacpagelist->num_pages] = vpnew;
2310         (vacpagelist->num_pages)++;
2311
2312 }
2313
2314 static void *
2315 vac_find_eq(void *bot, int nelem, int size, void *elm,
2316                    int (*compar) (const void *, const void *))
2317 {
2318         int                     res;
2319         int                     last = nelem - 1;
2320         int                     celm = nelem / 2;
2321         bool            last_move,
2322                                 first_move;
2323
2324         last_move = first_move = true;
2325         for (;;)
2326         {
2327                 if (first_move == true)
2328                 {
2329                         res = compar(bot, elm);
2330                         if (res > 0)
2331                                 return NULL;
2332                         if (res == 0)
2333                                 return bot;
2334                         first_move = false;
2335                 }
2336                 if (last_move == true)
2337                 {
2338                         res = compar(elm, (void *) ((char *) bot + last * size));
2339                         if (res > 0)
2340                                 return NULL;
2341                         if (res == 0)
2342                                 return (void *) ((char *) bot + last * size);
2343                         last_move = false;
2344                 }
2345                 res = compar(elm, (void *) ((char *) bot + celm * size));
2346                 if (res == 0)
2347                         return (void *) ((char *) bot + celm * size);
2348                 if (res < 0)
2349                 {
2350                         if (celm == 0)
2351                                 return NULL;
2352                         last = celm - 1;
2353                         celm = celm / 2;
2354                         last_move = true;
2355                         continue;
2356                 }
2357
2358                 if (celm == last)
2359                         return NULL;
2360
2361                 last = last - celm - 1;
2362                 bot = (void *) ((char *) bot + (celm + 1) * size);
2363                 celm = (last + 1) / 2;
2364                 first_move = true;
2365         }
2366
2367 }
2368
2369 static int
2370 vac_cmp_blk(const void *left, const void *right)
2371 {
2372         BlockNumber lblk,
2373                                 rblk;
2374
2375         lblk = (*((VacPage *) left))->blkno;
2376         rblk = (*((VacPage *) right))->blkno;
2377
2378         if (lblk < rblk)
2379                 return -1;
2380         if (lblk == rblk)
2381                 return 0;
2382         return 1;
2383
2384 }
2385
2386 static int
2387 vac_cmp_offno(const void *left, const void *right)
2388 {
2389
2390         if (*(OffsetNumber *) left < *(OffsetNumber *) right)
2391                 return -1;
2392         if (*(OffsetNumber *) left == *(OffsetNumber *) right)
2393                 return 0;
2394         return 1;
2395
2396 }
2397
2398 static int
2399 vac_cmp_vtlinks(const void *left, const void *right)
2400 {
2401
2402         if (((VTupleLink) left)->new_tid.ip_blkid.bi_hi <
2403                 ((VTupleLink) right)->new_tid.ip_blkid.bi_hi)
2404                 return -1;
2405         if (((VTupleLink) left)->new_tid.ip_blkid.bi_hi >
2406                 ((VTupleLink) right)->new_tid.ip_blkid.bi_hi)
2407                 return 1;
2408         /* bi_hi-es are equal */
2409         if (((VTupleLink) left)->new_tid.ip_blkid.bi_lo <
2410                 ((VTupleLink) right)->new_tid.ip_blkid.bi_lo)
2411                 return -1;
2412         if (((VTupleLink) left)->new_tid.ip_blkid.bi_lo >
2413                 ((VTupleLink) right)->new_tid.ip_blkid.bi_lo)
2414                 return 1;
2415         /* bi_lo-es are equal */
2416         if (((VTupleLink) left)->new_tid.ip_posid <
2417                 ((VTupleLink) right)->new_tid.ip_posid)
2418                 return -1;
2419         if (((VTupleLink) left)->new_tid.ip_posid >
2420                 ((VTupleLink) right)->new_tid.ip_posid)
2421                 return 1;
2422         return 0;
2423
2424 }
2425
2426
2427 static void
2428 get_indices(Relation relation, int *nindices, Relation **Irel)
2429 {
2430         List       *indexoidlist,
2431                            *indexoidscan;
2432         int                     i;
2433
2434         indexoidlist = RelationGetIndexList(relation);
2435
2436         *nindices = length(indexoidlist);
2437
2438         if (*nindices > 0)
2439                 *Irel = (Relation *) palloc(*nindices * sizeof(Relation));
2440         else
2441                 *Irel = NULL;
2442
2443         i = 0;
2444         foreach(indexoidscan, indexoidlist)
2445         {
2446                 Oid                     indexoid = lfirsti(indexoidscan);
2447
2448                 (*Irel)[i] = index_open(indexoid);
2449                 i++;
2450         }
2451
2452         freeList(indexoidlist);
2453 }
2454
2455
2456 static void
2457 close_indices(int nindices, Relation *Irel)
2458 {
2459
2460         if (Irel == (Relation *) NULL)
2461                 return;
2462
2463         while (nindices--)
2464                 index_close(Irel[nindices]);
2465         pfree(Irel);
2466
2467 }
2468
2469
2470 /*
2471  * Obtain IndexInfo data for each index on the rel
2472  */
2473 static IndexInfo **
2474 get_index_desc(Relation onerel, int nindices, Relation *Irel)
2475 {
2476         IndexInfo **indexInfo;
2477         int                     i;
2478         HeapTuple       cachetuple;
2479
2480         indexInfo = (IndexInfo **) palloc(nindices * sizeof(IndexInfo *));
2481
2482         for (i = 0; i < nindices; i++)
2483         {
2484                 cachetuple = SearchSysCacheTuple(INDEXRELID,
2485                                                          ObjectIdGetDatum(RelationGetRelid(Irel[i])),
2486                                                                                  0, 0, 0);
2487                 if (!HeapTupleIsValid(cachetuple))
2488                         elog(ERROR, "get_index_desc: index %u not found",
2489                                  RelationGetRelid(Irel[i]));
2490                 indexInfo[i] = BuildIndexInfo(cachetuple);
2491         }
2492
2493         return indexInfo;
2494 }
2495
2496
2497 static bool
2498 enough_space(VacPage vacpage, Size len)
2499 {
2500
2501         len = MAXALIGN(len);
2502
2503         if (len > vacpage->free)
2504                 return false;
2505
2506         if (vacpage->offsets_used < vacpage->offsets_free)      /* there are free
2507                                                                                                                  * itemid(s) */
2508                 return true;                    /* and len <= free_space */
2509
2510         /* ok. noff_usd >= noff_free and so we'll have to allocate new itemid */
2511         if (len + MAXALIGN(sizeof(ItemIdData)) <= vacpage->free)
2512                 return true;
2513
2514         return false;
2515
2516 }
2517
2518
2519 /*
2520  * Compute elapsed time since ru0 usage snapshot, and format into
2521  * a displayable string.  Result is in a static string, which is
2522  * tacky, but no one ever claimed that the Postgres backend is
2523  * threadable...
2524  */
2525 static char *
2526 show_rusage(struct rusage * ru0)
2527 {
2528         static char result[64];
2529         struct rusage ru1;
2530
2531         getrusage(RUSAGE_SELF, &ru1);
2532
2533         if (ru1.ru_stime.tv_usec < ru0->ru_stime.tv_usec)
2534         {
2535                 ru1.ru_stime.tv_sec--;
2536                 ru1.ru_stime.tv_usec += 1000000;
2537         }
2538         if (ru1.ru_utime.tv_usec < ru0->ru_utime.tv_usec)
2539         {
2540                 ru1.ru_utime.tv_sec--;
2541                 ru1.ru_utime.tv_usec += 1000000;
2542         }
2543
2544         snprintf(result, sizeof(result),
2545                          "CPU %d.%02ds/%d.%02du sec.",
2546                          (int) (ru1.ru_stime.tv_sec - ru0->ru_stime.tv_sec),
2547                          (int) (ru1.ru_stime.tv_usec - ru0->ru_stime.tv_usec) / 10000,
2548                          (int) (ru1.ru_utime.tv_sec - ru0->ru_utime.tv_sec),
2549                    (int) (ru1.ru_utime.tv_usec - ru0->ru_utime.tv_usec) / 10000);
2550
2551         return result;
2552 }