vacuumlazy.c File Reference

#include "postgres.h"
#include <math.h>
#include "access/amapi.h"
#include "access/genam.h"
#include "access/heapam.h"
#include "access/heapam_xlog.h"
#include "access/htup_details.h"
#include "access/multixact.h"
#include "access/parallel.h"
#include "access/transam.h"
#include "access/visibilitymap.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "catalog/index.h"
#include "catalog/storage.h"
#include "commands/dbcommands.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "executor/instrument.h"
#include "miscadmin.h"
#include "optimizer/paths.h"
#include "pgstat.h"
#include "portability/instr_time.h"
#include "postmaster/autovacuum.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
#include "storage/lmgr.h"
#include "tcop/tcopprot.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/pg_rusage.h"
#include "utils/timestamp.h"

Include dependency graph for vacuumlazy.c:

Go to the source code of this file.

Data Structures
struct	LVDeadTuples
struct	LVShared
struct	LVSharedIndStats
struct	LVParallelState
struct	LVRelState
struct	LVPagePruneState
struct	LVSavedErrInfo

Macros
#define	REL_TRUNCATE_MINIMUM   1000
#define	REL_TRUNCATE_FRACTION   16
#define	VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL   20 /* ms */
#define	VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL   50 /* ms */
#define	VACUUM_TRUNCATE_LOCK_TIMEOUT   5000 /* ms */
#define	BYPASS_THRESHOLD_PAGES   0.02 /* i.e. 2% of rel_pages */
#define	FAILSAFE_EVERY_PAGES   ((BlockNumber) (((uint64) 4 * 1024 * 1024 * 1024) / BLCKSZ))
#define	VACUUM_FSM_EVERY_PAGES   ((BlockNumber) (((uint64) 8 * 1024 * 1024 * 1024) / BLCKSZ))
#define	LAZY_ALLOC_TUPLES   MaxHeapTuplesPerPage
#define	SKIP_PAGES_THRESHOLD   ((BlockNumber) 32)
#define	PREFETCH_SIZE   ((BlockNumber) 32)
#define	PARALLEL_VACUUM_KEY_SHARED   1
#define	PARALLEL_VACUUM_KEY_DEAD_TUPLES   2
#define	PARALLEL_VACUUM_KEY_QUERY_TEXT   3
#define	PARALLEL_VACUUM_KEY_BUFFER_USAGE   4
#define	PARALLEL_VACUUM_KEY_WAL_USAGE   5
#define	ParallelVacuumIsActive(vacrel)   ((vacrel)->lps != NULL)
#define	SizeOfDeadTuples(cnt)
#define	MAXDEADTUPLES(max_size)   (((max_size) - offsetof(LVDeadTuples, itemptrs)) / sizeof(ItemPointerData))
#define	SizeOfLVShared   (offsetof(LVShared, bitmap) + sizeof(bits8))
#define	GetSharedIndStats(s)   ((LVSharedIndStats *)((char *)(s) + ((LVShared *)(s))->offset))
#define	IndStatsIsNull(s, i)   (!(((LVShared *)(s))->bitmap[(i) >> 3] & (1 << ((i) & 0x07))))
#define	FORCE_CHECK_PAGE()   (blkno == nblocks - 1 && should_attempt_truncation(vacrel))

Typedefs
typedef struct LVDeadTuples	LVDeadTuples
typedef struct LVShared	LVShared
typedef struct LVSharedIndStats	LVSharedIndStats
typedef struct LVParallelState	LVParallelState
typedef struct LVRelState	LVRelState
typedef struct LVPagePruneState	LVPagePruneState
typedef struct LVSavedErrInfo	LVSavedErrInfo

Enumerations
enum	VacErrPhase {   VACUUM_ERRCB_PHASE_UNKNOWN, VACUUM_ERRCB_PHASE_SCAN_HEAP, VACUUM_ERRCB_PHASE_VACUUM_INDEX, VACUUM_ERRCB_PHASE_VACUUM_HEAP,   VACUUM_ERRCB_PHASE_INDEX_CLEANUP, VACUUM_ERRCB_PHASE_TRUNCATE }

Functions
static void	lazy_scan_heap (LVRelState *vacrel, VacuumParams *params, bool aggressive)
static void	lazy_scan_prune (LVRelState *vacrel, Buffer buf, BlockNumber blkno, Page page, GlobalVisState *vistest, LVPagePruneState *prunestate)
static void	lazy_vacuum (LVRelState *vacrel)
static bool	lazy_vacuum_all_indexes (LVRelState *vacrel)
static void	lazy_vacuum_heap_rel (LVRelState *vacrel)
static int	lazy_vacuum_heap_page (LVRelState *vacrel, BlockNumber blkno, Buffer buffer, int tupindex, Buffer *vmbuffer)
static bool	lazy_check_needs_freeze (Buffer buf, bool *hastup, LVRelState *vacrel)
static bool	lazy_check_wraparound_failsafe (LVRelState *vacrel)
static void	do_parallel_lazy_vacuum_all_indexes (LVRelState *vacrel)
static void	do_parallel_lazy_cleanup_all_indexes (LVRelState *vacrel)
static void	do_parallel_vacuum_or_cleanup (LVRelState *vacrel, int nworkers)
static void	do_parallel_processing (LVRelState *vacrel, LVShared *lvshared)
static void	do_serial_processing_for_unsafe_indexes (LVRelState *vacrel, LVShared *lvshared)
static IndexBulkDeleteResult *	parallel_process_one_index (Relation indrel, IndexBulkDeleteResult *istat, LVShared *lvshared, LVSharedIndStats *shared_indstats, LVRelState *vacrel)
static void	lazy_cleanup_all_indexes (LVRelState *vacrel)
static IndexBulkDeleteResult *	lazy_vacuum_one_index (Relation indrel, IndexBulkDeleteResult *istat, double reltuples, LVRelState *vacrel)
static IndexBulkDeleteResult *	lazy_cleanup_one_index (Relation indrel, IndexBulkDeleteResult *istat, double reltuples, bool estimated_count, LVRelState *vacrel)
static bool	should_attempt_truncation (LVRelState *vacrel)
static void	lazy_truncate_heap (LVRelState *vacrel)
static BlockNumber	count_nondeletable_pages (LVRelState *vacrel, bool *lock_waiter_detected)
static long	compute_max_dead_tuples (BlockNumber relblocks, bool hasindex)
static void	lazy_space_alloc (LVRelState *vacrel, int nworkers, BlockNumber relblocks)
static void	lazy_space_free (LVRelState *vacrel)
static bool	lazy_tid_reaped (ItemPointer itemptr, void *state)
static int	vac_cmp_itemptr (const void *left, const void *right)
static bool	heap_page_is_all_visible (LVRelState *vacrel, Buffer buf, TransactionId *visibility_cutoff_xid, bool *all_frozen)
static int	compute_parallel_vacuum_workers (LVRelState *vacrel, int nrequested, bool *can_parallel_vacuum)
static void	update_index_statistics (LVRelState *vacrel)
static LVParallelState *	begin_parallel_vacuum (LVRelState *vacrel, BlockNumber nblocks, int nrequested)
static void	end_parallel_vacuum (LVRelState *vacrel)
static LVSharedIndStats *	parallel_stats_for_idx (LVShared *lvshared, int getidx)
static bool	parallel_processing_is_safe (Relation indrel, LVShared *lvshared)
static void	vacuum_error_callback (void *arg)
static void	update_vacuum_error_info (LVRelState *vacrel, LVSavedErrInfo *saved_vacrel, int phase, BlockNumber blkno, OffsetNumber offnum)
static void	restore_vacuum_error_info (LVRelState *vacrel, const LVSavedErrInfo *saved_vacrel)
void	heap_vacuum_rel (Relation rel, VacuumParams *params, BufferAccessStrategy bstrategy)
void	parallel_vacuum_main (dsm_segment *seg, shm_toc *toc)

Variables
static int	elevel = -1

Macro Definition Documentation

BYPASS_THRESHOLD_PAGES

#define BYPASS_THRESHOLD_PAGES   0.02 /* i.e. 2% of rel_pages */

Definition at line 110 of file vacuumlazy.c.

Referenced by lazy_vacuum().

FAILSAFE_EVERY_PAGES

#define FAILSAFE_EVERY_PAGES   ((BlockNumber) (((uint64) 4 * 1024 * 1024 * 1024) / BLCKSZ))

Definition at line 115 of file vacuumlazy.c.

Referenced by lazy_scan_heap().

FORCE_CHECK_PAGE

#define FORCE_CHECK_PAGE

(

)

(blkno == nblocks - 1 && should_attempt_truncation(vacrel))

Referenced by lazy_scan_heap().

GetSharedIndStats

#define GetSharedIndStats

(

s

)

((LVSharedIndStats *)((char *)(s) + ((LVShared *)(s))->offset))

Definition at line 267 of file vacuumlazy.c.

Referenced by parallel_stats_for_idx().

IndStatsIsNull

#define IndStatsIsNull	(		s,
			i
	)		(!(((LVShared *)(s))->bitmap[(i) >> 3] & (1 << ((i) & 0x07))))

Definition at line 269 of file vacuumlazy.c.

Referenced by parallel_stats_for_idx().

LAZY_ALLOC_TUPLES

#define LAZY_ALLOC_TUPLES   MaxHeapTuplesPerPage

Definition at line 132 of file vacuumlazy.c.

Referenced by compute_max_dead_tuples().

MAXDEADTUPLES

#define MAXDEADTUPLES

(

max_size

)

(((max_size) - offsetof(LVDeadTuples, itemptrs)) / sizeof(ItemPointerData))

Definition at line 193 of file vacuumlazy.c.

Referenced by compute_max_dead_tuples(), and lazy_vacuum().

PARALLEL_VACUUM_KEY_BUFFER_USAGE

#define PARALLEL_VACUUM_KEY_BUFFER_USAGE   4

Definition at line 154 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and parallel_vacuum_main().

PARALLEL_VACUUM_KEY_DEAD_TUPLES

#define PARALLEL_VACUUM_KEY_DEAD_TUPLES   2

Definition at line 152 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and parallel_vacuum_main().

PARALLEL_VACUUM_KEY_QUERY_TEXT

#define PARALLEL_VACUUM_KEY_QUERY_TEXT   3

Definition at line 153 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and parallel_vacuum_main().

PARALLEL_VACUUM_KEY_SHARED

#define PARALLEL_VACUUM_KEY_SHARED   1

Definition at line 151 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and parallel_vacuum_main().

PARALLEL_VACUUM_KEY_WAL_USAGE

#define PARALLEL_VACUUM_KEY_WAL_USAGE   5

Definition at line 155 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and parallel_vacuum_main().

ParallelVacuumIsActive

#define ParallelVacuumIsActive

(

vacrel

)

((vacrel)->lps != NULL)

Definition at line 161 of file vacuumlazy.c.

Referenced by do_parallel_vacuum_or_cleanup(), lazy_cleanup_all_indexes(), lazy_space_alloc(), lazy_space_free(), and lazy_vacuum_all_indexes().

PREFETCH_SIZE

#define PREFETCH_SIZE   ((BlockNumber) 32)

Definition at line 144 of file vacuumlazy.c.

Referenced by count_nondeletable_pages().

REL_TRUNCATE_FRACTION

#define REL_TRUNCATE_FRACTION   16

Definition at line 93 of file vacuumlazy.c.

Referenced by should_attempt_truncation().

REL_TRUNCATE_MINIMUM

#define REL_TRUNCATE_MINIMUM   1000

Definition at line 92 of file vacuumlazy.c.

Referenced by should_attempt_truncation().

SizeOfDeadTuples

#define SizeOfDeadTuples

(

cnt

)

Value:

add_size(offsetof(LVDeadTuples, itemptrs), \
             mul_size(sizeof(ItemPointerData), cnt))

Definition at line 190 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum(), and lazy_space_alloc().

SizeOfLVShared

#define SizeOfLVShared   (offsetof(LVShared, bitmap) + sizeof(bits8))

Definition at line 266 of file vacuumlazy.c.

Referenced by begin_parallel_vacuum().

SKIP_PAGES_THRESHOLD

#define SKIP_PAGES_THRESHOLD   ((BlockNumber) 32)

Definition at line 138 of file vacuumlazy.c.

Referenced by lazy_scan_heap().

VACUUM_FSM_EVERY_PAGES

#define VACUUM_FSM_EVERY_PAGES   ((BlockNumber) (((uint64) 8 * 1024 * 1024 * 1024) / BLCKSZ))

Definition at line 124 of file vacuumlazy.c.

Referenced by lazy_scan_heap().

VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL

#define VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL   20 /* ms */

Definition at line 102 of file vacuumlazy.c.

Referenced by count_nondeletable_pages().

VACUUM_TRUNCATE_LOCK_TIMEOUT

#define VACUUM_TRUNCATE_LOCK_TIMEOUT   5000 /* ms */

Definition at line 104 of file vacuumlazy.c.

Referenced by lazy_truncate_heap().

VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL

#define VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL   50 /* ms */

Definition at line 103 of file vacuumlazy.c.

Referenced by lazy_truncate_heap().

Typedef Documentation

LVDeadTuples

typedef struct LVDeadTuples LVDeadTuples

LVPagePruneState

typedef struct LVPagePruneState LVPagePruneState

LVParallelState

typedef struct LVParallelState LVParallelState

LVRelState

typedef struct LVRelState LVRelState

LVSavedErrInfo

typedef struct LVSavedErrInfo LVSavedErrInfo

LVShared

typedef struct LVShared LVShared

LVSharedIndStats

typedef struct LVSharedIndStats LVSharedIndStats

Enumeration Type Documentation

VacErrPhase

enum VacErrPhase

Enumerator
VACUUM_ERRCB_PHASE_UNKNOWN
VACUUM_ERRCB_PHASE_SCAN_HEAP
VACUUM_ERRCB_PHASE_VACUUM_INDEX
VACUUM_ERRCB_PHASE_VACUUM_HEAP
VACUUM_ERRCB_PHASE_INDEX_CLEANUP
VACUUM_ERRCB_PHASE_TRUNCATE

Definition at line 164 of file vacuumlazy.c.

{
    VACUUM_ERRCB_PHASE_UNKNOWN,
    VACUUM_ERRCB_PHASE_SCAN_HEAP,
    VACUUM_ERRCB_PHASE_VACUUM_INDEX,
    VACUUM_ERRCB_PHASE_VACUUM_HEAP,
    VACUUM_ERRCB_PHASE_INDEX_CLEANUP,
    VACUUM_ERRCB_PHASE_TRUNCATE
} VacErrPhase;

Function Documentation

begin_parallel_vacuum()

static LVParallelState * begin_parallel_vacuum ( LVRelState *  vacrel, BlockNumber  nblocks, int  nrequested  )

static

Definition at line 3845 of file vacuumlazy.c.

References LVShared::active_nworkers, add_size(), IndexAmRoutine::amparallelvacuumoptions, IndexAmRoutine::amusemaintenanceworkmem, Assert, LVShared::bitmap, BITMAPLEN, LVParallelState::buffer_usage, compute_max_dead_tuples(), compute_parallel_vacuum_workers(), LVShared::cost_balance, CreateParallelContext(), LVRelState::dead_tuples, debug_query_string, LVShared::elevel, elevel, EnterParallelMode(), ParallelContext::estimator, LVShared::idx, idx(), LVRelState::indrels, InitializeParallelDSM(), LVDeadTuples::itemptrs, LVParallelState::lvshared, maintenance_work_mem, LVShared::maintenance_work_mem_worker, LVDeadTuples::max_tuples, MAXALIGN, MemSet, Min, mul_size(), LVRelState::nindexes, LVParallelState::nindexes_parallel_bulkdel, LVParallelState::nindexes_parallel_cleanup, LVParallelState::nindexes_parallel_condcleanup, LVDeadTuples::num_tuples, ParallelContext::nworkers, LVShared::offset, palloc0(), PARALLEL_VACUUM_KEY_BUFFER_USAGE, PARALLEL_VACUUM_KEY_DEAD_TUPLES, PARALLEL_VACUUM_KEY_QUERY_TEXT, PARALLEL_VACUUM_KEY_SHARED, PARALLEL_VACUUM_KEY_WAL_USAGE, LVParallelState::pcxt, pfree(), pg_atomic_init_u32(), RelationData::rd_indam, LVRelState::rel, RelationGetRelid, LVShared::relid, shm_toc_allocate(), shm_toc_estimate_chunk, shm_toc_estimate_keys, shm_toc_insert(), SizeOfDeadTuples, SizeOfLVShared, ParallelContext::toc, VACUUM_OPTION_MAX_VALID_VALUE, VACUUM_OPTION_PARALLEL_BULKDEL, VACUUM_OPTION_PARALLEL_CLEANUP, VACUUM_OPTION_PARALLEL_COND_CLEANUP, and LVParallelState::wal_usage.

Referenced by lazy_space_alloc().

{
    LVParallelState *lps = NULL;
    Relation   *indrels = vacrel->indrels;
    int         nindexes = vacrel->nindexes;
    ParallelContext *pcxt;
    LVShared   *shared;
    LVDeadTuples *dead_tuples;
    BufferUsage *buffer_usage;
    WalUsage   *wal_usage;
    bool       *can_parallel_vacuum;
    long        maxtuples;
    Size        est_shared;
    Size        est_deadtuples;
    int         nindexes_mwm = 0;
    int         parallel_workers = 0;
    int         querylen;

    /*
     * A parallel vacuum must be requested and there must be indexes on the
     * relation
     */
    Assert(nrequested >= 0);
    Assert(nindexes > 0);

    /*
     * Compute the number of parallel vacuum workers to launch
     */
    can_parallel_vacuum = (bool *) palloc0(sizeof(bool) * nindexes);
    parallel_workers = compute_parallel_vacuum_workers(vacrel,
                                                       nrequested,
                                                       can_parallel_vacuum);

    /* Can't perform vacuum in parallel */
    if (parallel_workers <= 0)
    {
        pfree(can_parallel_vacuum);
        return lps;
    }

    lps = (LVParallelState *) palloc0(sizeof(LVParallelState));

    EnterParallelMode();
    pcxt = CreateParallelContext("postgres", "parallel_vacuum_main",
                                 parallel_workers);
    Assert(pcxt->nworkers > 0);
    lps->pcxt = pcxt;

    /* Estimate size for shared information -- PARALLEL_VACUUM_KEY_SHARED */
    est_shared = MAXALIGN(add_size(SizeOfLVShared, BITMAPLEN(nindexes)));
    for (int idx = 0; idx < nindexes; idx++)
    {
        Relation    indrel = indrels[idx];
        uint8       vacoptions = indrel->rd_indam->amparallelvacuumoptions;

        /*
         * Cleanup option should be either disabled, always performing in
         * parallel or conditionally performing in parallel.
         */
        Assert(((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) == 0) ||
               ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) == 0));
        Assert(vacoptions <= VACUUM_OPTION_MAX_VALID_VALUE);

        /* Skip indexes that don't participate in parallel vacuum */
        if (!can_parallel_vacuum[idx])
            continue;

        if (indrel->rd_indam->amusemaintenanceworkmem)
            nindexes_mwm++;

        est_shared = add_size(est_shared, sizeof(LVSharedIndStats));

        /*
         * Remember the number of indexes that support parallel operation for
         * each phase.
         */
        if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
            lps->nindexes_parallel_bulkdel++;
        if ((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0)
            lps->nindexes_parallel_cleanup++;
        if ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0)
            lps->nindexes_parallel_condcleanup++;
    }
    shm_toc_estimate_chunk(&pcxt->estimator, est_shared);
    shm_toc_estimate_keys(&pcxt->estimator, 1);

    /* Estimate size for dead tuples -- PARALLEL_VACUUM_KEY_DEAD_TUPLES */
    maxtuples = compute_max_dead_tuples(nblocks, true);
    est_deadtuples = MAXALIGN(SizeOfDeadTuples(maxtuples));
    shm_toc_estimate_chunk(&pcxt->estimator, est_deadtuples);
    shm_toc_estimate_keys(&pcxt->estimator, 1);

    /*
     * Estimate space for BufferUsage and WalUsage --
     * PARALLEL_VACUUM_KEY_BUFFER_USAGE and PARALLEL_VACUUM_KEY_WAL_USAGE.
     *
     * If there are no extensions loaded that care, we could skip this.  We
     * have no way of knowing whether anyone's looking at pgBufferUsage or
     * pgWalUsage, so do it unconditionally.
     */
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);
    shm_toc_estimate_chunk(&pcxt->estimator,
                           mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_estimate_keys(&pcxt->estimator, 1);

    /* Finally, estimate PARALLEL_VACUUM_KEY_QUERY_TEXT space */
    if (debug_query_string)
    {
        querylen = strlen(debug_query_string);
        shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
        shm_toc_estimate_keys(&pcxt->estimator, 1);
    }
    else
        querylen = 0;           /* keep compiler quiet */

    InitializeParallelDSM(pcxt);

    /* Prepare shared information */
    shared = (LVShared *) shm_toc_allocate(pcxt->toc, est_shared);
    MemSet(shared, 0, est_shared);
    shared->relid = RelationGetRelid(vacrel->rel);
    shared->elevel = elevel;
    shared->maintenance_work_mem_worker =
        (nindexes_mwm > 0) ?
        maintenance_work_mem / Min(parallel_workers, nindexes_mwm) :
        maintenance_work_mem;

    pg_atomic_init_u32(&(shared->cost_balance), 0);
    pg_atomic_init_u32(&(shared->active_nworkers), 0);
    pg_atomic_init_u32(&(shared->idx), 0);
    shared->offset = MAXALIGN(add_size(SizeOfLVShared, BITMAPLEN(nindexes)));

    /*
     * Initialize variables for shared index statistics, set NULL bitmap and
     * the size of stats for each index.
     */
    memset(shared->bitmap, 0x00, BITMAPLEN(nindexes));
    for (int idx = 0; idx < nindexes; idx++)
    {
        if (!can_parallel_vacuum[idx])
            continue;

        /* Set NOT NULL as this index does support parallelism */
        shared->bitmap[idx >> 3] |= 1 << (idx & 0x07);
    }

    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_SHARED, shared);
    lps->lvshared = shared;

    /* Prepare the dead tuple space */
    dead_tuples = (LVDeadTuples *) shm_toc_allocate(pcxt->toc, est_deadtuples);
    dead_tuples->max_tuples = maxtuples;
    dead_tuples->num_tuples = 0;
    MemSet(dead_tuples->itemptrs, 0, sizeof(ItemPointerData) * maxtuples);
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_DEAD_TUPLES, dead_tuples);
    vacrel->dead_tuples = dead_tuples;

    /*
     * Allocate space for each worker's BufferUsage and WalUsage; no need to
     * initialize
     */
    buffer_usage = shm_toc_allocate(pcxt->toc,
                                    mul_size(sizeof(BufferUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_BUFFER_USAGE, buffer_usage);
    lps->buffer_usage = buffer_usage;
    wal_usage = shm_toc_allocate(pcxt->toc,
                                 mul_size(sizeof(WalUsage), pcxt->nworkers));
    shm_toc_insert(pcxt->toc, PARALLEL_VACUUM_KEY_WAL_USAGE, wal_usage);
    lps->wal_usage = wal_usage;

    /* Store query string for workers */
    if (debug_query_string)
    {
        char       *sharedquery;

        sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
        memcpy(sharedquery, debug_query_string, querylen + 1);
        sharedquery[querylen] = '\0';
        shm_toc_insert(pcxt->toc,
                       PARALLEL_VACUUM_KEY_QUERY_TEXT, sharedquery);
    }

    pfree(can_parallel_vacuum);
    return lps;
}

compute_max_dead_tuples()

static long compute_max_dead_tuples ( BlockNumber  relblocks, bool  hasindex  )

static

Definition at line 3457 of file vacuumlazy.c.

References autovacuum_work_mem, IsAutoVacuumWorkerProcess(), LAZY_ALLOC_TUPLES, maintenance_work_mem, Max, MaxAllocSize, MAXDEADTUPLES, MaxHeapTuplesPerPage, and Min.

Referenced by begin_parallel_vacuum(), and lazy_space_alloc().

{
    long        maxtuples;
    int         vac_work_mem = IsAutoVacuumWorkerProcess() &&
    autovacuum_work_mem != -1 ?
    autovacuum_work_mem : maintenance_work_mem;

    if (hasindex)
    {
        maxtuples = MAXDEADTUPLES(vac_work_mem * 1024L);
        maxtuples = Min(maxtuples, INT_MAX);
        maxtuples = Min(maxtuples, MAXDEADTUPLES(MaxAllocSize));

        /* curious coding here to ensure the multiplication can't overflow */
        if ((BlockNumber) (maxtuples / LAZY_ALLOC_TUPLES) > relblocks)
            maxtuples = relblocks * LAZY_ALLOC_TUPLES;

        /* stay sane if small maintenance_work_mem */
        maxtuples = Max(maxtuples, MaxHeapTuplesPerPage);
    }
    else
        maxtuples = MaxHeapTuplesPerPage;

    return maxtuples;
}

compute_parallel_vacuum_workers()

static int compute_parallel_vacuum_workers ( LVRelState *  vacrel, int  nrequested, bool *  can_parallel_vacuum  )

static

Definition at line 3751 of file vacuumlazy.c.

References IndexAmRoutine::amparallelvacuumoptions, idx(), LVRelState::indrels, IsUnderPostmaster, Max, max_parallel_maintenance_workers, Min, min_parallel_index_scan_size, LVRelState::nindexes, RelationData::rd_indam, RelationGetNumberOfBlocks, VACUUM_OPTION_NO_PARALLEL, VACUUM_OPTION_PARALLEL_BULKDEL, VACUUM_OPTION_PARALLEL_CLEANUP, and VACUUM_OPTION_PARALLEL_COND_CLEANUP.

Referenced by begin_parallel_vacuum().

{
    int         nindexes_parallel = 0;
    int         nindexes_parallel_bulkdel = 0;
    int         nindexes_parallel_cleanup = 0;
    int         parallel_workers;

    /*
     * We don't allow performing parallel operation in standalone backend or
     * when parallelism is disabled.
     */
    if (!IsUnderPostmaster || max_parallel_maintenance_workers == 0)
        return 0;

    /*
     * Compute the number of indexes that can participate in parallel vacuum.
     */
    for (int idx = 0; idx < vacrel->nindexes; idx++)
    {
        Relation    indrel = vacrel->indrels[idx];
        uint8       vacoptions = indrel->rd_indam->amparallelvacuumoptions;

        if (vacoptions == VACUUM_OPTION_NO_PARALLEL ||
            RelationGetNumberOfBlocks(indrel) < min_parallel_index_scan_size)
            continue;

        can_parallel_vacuum[idx] = true;

        if ((vacoptions & VACUUM_OPTION_PARALLEL_BULKDEL) != 0)
            nindexes_parallel_bulkdel++;
        if (((vacoptions & VACUUM_OPTION_PARALLEL_CLEANUP) != 0) ||
            ((vacoptions & VACUUM_OPTION_PARALLEL_COND_CLEANUP) != 0))
            nindexes_parallel_cleanup++;
    }

    nindexes_parallel = Max(nindexes_parallel_bulkdel,
                            nindexes_parallel_cleanup);

    /* The leader process takes one index */
    nindexes_parallel--;

    /* No index supports parallel vacuum */
    if (nindexes_parallel <= 0)
        return 0;

    /* Compute the parallel degree */
    parallel_workers = (nrequested > 0) ?
        Min(nrequested, nindexes_parallel) : nindexes_parallel;

    /* Cap by max_parallel_maintenance_workers */
    parallel_workers = Min(parallel_workers, max_parallel_maintenance_workers);

    return parallel_workers;
}

count_nondeletable_pages()

static BlockNumber count_nondeletable_pages ( LVRelState *  vacrel, bool *  lock_waiter_detected  )

static

Definition at line 3319 of file vacuumlazy.c.

References AccessExclusiveLock, LVRelState::bstrategy, buf, BUFFER_LOCK_SHARE, BufferGetPage, CHECK_FOR_INTERRUPTS, elevel, ereport, errmsg(), FirstOffsetNumber, INSTR_TIME_GET_MICROSEC, INSTR_TIME_SET_CURRENT, INSTR_TIME_SUBTRACT, InvalidBlockNumber, ItemIdIsUsed, LockBuffer(), LockHasWaitersRelation(), MAIN_FORKNUM, LVRelState::nonempty_pages, OffsetNumberNext, PageGetItemId, PageGetMaxOffsetNumber, PageIsEmpty, PageIsNew, PREFETCH_SIZE, PrefetchBuffer(), RBM_NORMAL, ReadBufferExtended(), LVRelState::rel, LVRelState::rel_pages, LVRelState::relname, StaticAssertStmt, UnlockReleaseBuffer(), and VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL.

Referenced by lazy_truncate_heap().

{
    BlockNumber blkno;
    BlockNumber prefetchedUntil;
    instr_time  starttime;

    /* Initialize the starttime if we check for conflicting lock requests */
    INSTR_TIME_SET_CURRENT(starttime);

    /*
     * Start checking blocks at what we believe relation end to be and move
     * backwards.  (Strange coding of loop control is needed because blkno is
     * unsigned.)  To make the scan faster, we prefetch a few blocks at a time
     * in forward direction, so that OS-level readahead can kick in.
     */
    blkno = vacrel->rel_pages;
    StaticAssertStmt((PREFETCH_SIZE & (PREFETCH_SIZE - 1)) == 0,
                     "prefetch size must be power of 2");
    prefetchedUntil = InvalidBlockNumber;
    while (blkno > vacrel->nonempty_pages)
    {
        Buffer      buf;
        Page        page;
        OffsetNumber offnum,
                    maxoff;
        bool        hastup;

        /*
         * Check if another process requests a lock on our relation. We are
         * holding an AccessExclusiveLock here, so they will be waiting. We
         * only do this once per VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL, and we
         * only check if that interval has elapsed once every 32 blocks to
         * keep the number of system calls and actual shared lock table
         * lookups to a minimum.
         */
        if ((blkno % 32) == 0)
        {
            instr_time  currenttime;
            instr_time  elapsed;

            INSTR_TIME_SET_CURRENT(currenttime);
            elapsed = currenttime;
            INSTR_TIME_SUBTRACT(elapsed, starttime);
            if ((INSTR_TIME_GET_MICROSEC(elapsed) / 1000)
                >= VACUUM_TRUNCATE_LOCK_CHECK_INTERVAL)
            {
                if (LockHasWaitersRelation(vacrel->rel, AccessExclusiveLock))
                {
                    ereport(elevel,
                            (errmsg("\"%s\": suspending truncate due to conflicting lock request",
                                    vacrel->relname)));

                    *lock_waiter_detected = true;
                    return blkno;
                }
                starttime = currenttime;
            }
        }

        /*
         * We don't insert a vacuum delay point here, because we have an
         * exclusive lock on the table which we want to hold for as short a
         * time as possible.  We still need to check for interrupts however.
         */
        CHECK_FOR_INTERRUPTS();

        blkno--;

        /* If we haven't prefetched this lot yet, do so now. */
        if (prefetchedUntil > blkno)
        {
            BlockNumber prefetchStart;
            BlockNumber pblkno;

            prefetchStart = blkno & ~(PREFETCH_SIZE - 1);
            for (pblkno = prefetchStart; pblkno <= blkno; pblkno++)
            {
                PrefetchBuffer(vacrel->rel, MAIN_FORKNUM, pblkno);
                CHECK_FOR_INTERRUPTS();
            }
            prefetchedUntil = prefetchStart;
        }

        buf = ReadBufferExtended(vacrel->rel, MAIN_FORKNUM, blkno, RBM_NORMAL,
                                 vacrel->bstrategy);

        /* In this phase we only need shared access to the buffer */
        LockBuffer(buf, BUFFER_LOCK_SHARE);

        page = BufferGetPage(buf);

        if (PageIsNew(page) || PageIsEmpty(page))
        {
            UnlockReleaseBuffer(buf);
            continue;
        }

        hastup = false;
        maxoff = PageGetMaxOffsetNumber(page);
        for (offnum = FirstOffsetNumber;
             offnum <= maxoff;
             offnum = OffsetNumberNext(offnum))
        {
            ItemId      itemid;

            itemid = PageGetItemId(page, offnum);

            /*
             * Note: any non-unused item should be taken as a reason to keep
             * this page.  Even an LP_DEAD item makes truncation unsafe, since
             * we must not have cleaned out its index entries.
             */
            if (ItemIdIsUsed(itemid))
            {
                hastup = true;
                break;          /* can stop scanning */
            }
        }                       /* scan along page */

        UnlockReleaseBuffer(buf);

        /* Done scanning if we found a tuple here */
        if (hastup)
            return blkno + 1;
    }

    /*
     * If we fall out of the loop, all the previously-thought-to-be-empty
     * pages still are; we need not bother to look at the last known-nonempty
     * page.
     */
    return vacrel->nonempty_pages;
}

do_parallel_lazy_cleanup_all_indexes()

static void do_parallel_lazy_cleanup_all_indexes ( LVRelState *  vacrel )

static

Definition at line 2667 of file vacuumlazy.c.

References do_parallel_vacuum_or_cleanup(), LVShared::estimated_count, LVShared::first_time, LVShared::for_cleanup, LVRelState::lps, LVParallelState::lvshared, LVRelState::new_rel_tuples, LVParallelState::nindexes_parallel_cleanup, LVParallelState::nindexes_parallel_condcleanup, LVRelState::num_index_scans, LVRelState::rel_pages, LVShared::reltuples, and LVRelState::tupcount_pages.

Referenced by lazy_cleanup_all_indexes().

{
    int         nworkers;

    /*
     * If parallel vacuum is active we perform index cleanup with parallel
     * workers.
     *
     * Tell parallel workers to do index cleanup.
     */
    vacrel->lps->lvshared->for_cleanup = true;
    vacrel->lps->lvshared->first_time = (vacrel->num_index_scans == 0);

    /*
     * Now we can provide a better estimate of total number of surviving
     * tuples (we assume indexes are more interested in that than in the
     * number of nominally live tuples).
     */
    vacrel->lps->lvshared->reltuples = vacrel->new_rel_tuples;
    vacrel->lps->lvshared->estimated_count =
        (vacrel->tupcount_pages < vacrel->rel_pages);

    /* Determine the number of parallel workers to launch */
    if (vacrel->lps->lvshared->first_time)
        nworkers = vacrel->lps->nindexes_parallel_cleanup +
            vacrel->lps->nindexes_parallel_condcleanup;
    else
        nworkers = vacrel->lps->nindexes_parallel_cleanup;

    do_parallel_vacuum_or_cleanup(vacrel, nworkers);
}

do_parallel_lazy_vacuum_all_indexes()

static void do_parallel_lazy_vacuum_all_indexes ( LVRelState *  vacrel )

static

Definition at line 2646 of file vacuumlazy.c.

References do_parallel_vacuum_or_cleanup(), LVShared::estimated_count, LVShared::first_time, LVShared::for_cleanup, LVRelState::lps, LVParallelState::lvshared, LVParallelState::nindexes_parallel_bulkdel, LVRelState::old_live_tuples, and LVShared::reltuples.

Referenced by lazy_vacuum_all_indexes().

{
    /* Tell parallel workers to do index vacuuming */
    vacrel->lps->lvshared->for_cleanup = false;
    vacrel->lps->lvshared->first_time = false;

    /*
     * We can only provide an approximate value of num_heap_tuples in vacuum
     * cases.
     */
    vacrel->lps->lvshared->reltuples = vacrel->old_live_tuples;
    vacrel->lps->lvshared->estimated_count = true;

    do_parallel_vacuum_or_cleanup(vacrel,
                                  vacrel->lps->nindexes_parallel_bulkdel);
}

do_parallel_processing()

static void do_parallel_processing ( LVRelState *  vacrel, LVShared *  lvshared  )

static

Definition at line 2819 of file vacuumlazy.c.

References idx(), LVShared::idx, LVRelState::indrels, LVRelState::indstats, LVRelState::nindexes, parallel_process_one_index(), parallel_processing_is_safe(), parallel_stats_for_idx(), pg_atomic_add_fetch_u32(), pg_atomic_fetch_add_u32(), pg_atomic_sub_fetch_u32(), and VacuumActiveNWorkers.

Referenced by do_parallel_vacuum_or_cleanup(), and parallel_vacuum_main().

{
    /*
     * Increment the active worker count if we are able to launch any worker.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);

    /* Loop until all indexes are vacuumed */
    for (;;)
    {
        int         idx;
        LVSharedIndStats *shared_istat;
        Relation    indrel;
        IndexBulkDeleteResult *istat;

        /* Get an index number to process */
        idx = pg_atomic_fetch_add_u32(&(lvshared->idx), 1);

        /* Done for all indexes? */
        if (idx >= vacrel->nindexes)
            break;

        /* Get the index statistics of this index from DSM */
        shared_istat = parallel_stats_for_idx(lvshared, idx);

        /* Skip indexes not participating in parallelism */
        if (shared_istat == NULL)
            continue;

        indrel = vacrel->indrels[idx];

        /*
         * Skip processing indexes that are unsafe for workers (these are
         * processed in do_serial_processing_for_unsafe_indexes() by leader)
         */
        if (!parallel_processing_is_safe(indrel, lvshared))
            continue;

        /* Do vacuum or cleanup of the index */
        istat = (vacrel->indstats[idx]);
        vacrel->indstats[idx] = parallel_process_one_index(indrel, istat,
                                                           lvshared,
                                                           shared_istat,
                                                           vacrel);
    }

    /*
     * We have completed the index vacuum so decrement the active worker
     * count.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

do_parallel_vacuum_or_cleanup()

static void do_parallel_vacuum_or_cleanup ( LVRelState *  vacrel, int  nworkers  )

static

Definition at line 2706 of file vacuumlazy.c.

References LVShared::active_nworkers, Assert, LVParallelState::buffer_usage, LVShared::cost_balance, do_parallel_processing(), do_serial_processing_for_unsafe_indexes(), elevel, ereport, errmsg(), LVShared::for_cleanup, i, LVShared::idx, InstrAccumParallelQuery(), IsParallelWorker, LaunchParallelWorkers(), LVRelState::lps, LVParallelState::lvshared, Min, ngettext, LVRelState::nindexes, LVRelState::num_index_scans, ParallelContext::nworkers, ParallelContext::nworkers_launched, ParallelVacuumIsActive, LVParallelState::pcxt, pg_atomic_read_u32(), pg_atomic_write_u32(), ReinitializeParallelDSM(), ReinitializeParallelWorkers(), VacuumActiveNWorkers, VacuumCostBalance, VacuumCostBalanceLocal, VacuumSharedCostBalance, WaitForParallelWorkersToFinish(), and LVParallelState::wal_usage.

Referenced by do_parallel_lazy_cleanup_all_indexes(), and do_parallel_lazy_vacuum_all_indexes().

{
    LVParallelState *lps = vacrel->lps;

    Assert(!IsParallelWorker());
    Assert(ParallelVacuumIsActive(vacrel));
    Assert(vacrel->nindexes > 0);

    /* The leader process will participate */
    nworkers--;

    /*
     * It is possible that parallel context is initialized with fewer workers
     * than the number of indexes that need a separate worker in the current
     * phase, so we need to consider it.  See compute_parallel_vacuum_workers.
     */
    nworkers = Min(nworkers, lps->pcxt->nworkers);

    /* Setup the shared cost-based vacuum delay and launch workers */
    if (nworkers > 0)
    {
        if (vacrel->num_index_scans > 0)
        {
            /* Reset the parallel index processing counter */
            pg_atomic_write_u32(&(lps->lvshared->idx), 0);

            /* Reinitialize the parallel context to relaunch parallel workers */
            ReinitializeParallelDSM(lps->pcxt);
        }

        /*
         * Set up shared cost balance and the number of active workers for
         * vacuum delay.  We need to do this before launching workers as
         * otherwise, they might not see the updated values for these
         * parameters.
         */
        pg_atomic_write_u32(&(lps->lvshared->cost_balance), VacuumCostBalance);
        pg_atomic_write_u32(&(lps->lvshared->active_nworkers), 0);

        /*
         * The number of workers can vary between bulkdelete and cleanup
         * phase.
         */
        ReinitializeParallelWorkers(lps->pcxt, nworkers);

        LaunchParallelWorkers(lps->pcxt);

        if (lps->pcxt->nworkers_launched > 0)
        {
            /*
             * Reset the local cost values for leader backend as we have
             * already accumulated the remaining balance of heap.
             */
            VacuumCostBalance = 0;
            VacuumCostBalanceLocal = 0;

            /* Enable shared cost balance for leader backend */
            VacuumSharedCostBalance = &(lps->lvshared->cost_balance);
            VacuumActiveNWorkers = &(lps->lvshared->active_nworkers);
        }

        if (lps->lvshared->for_cleanup)
            ereport(elevel,
                    (errmsg(ngettext("launched %d parallel vacuum worker for index cleanup (planned: %d)",
                                     "launched %d parallel vacuum workers for index cleanup (planned: %d)",
                                     lps->pcxt->nworkers_launched),
                            lps->pcxt->nworkers_launched, nworkers)));
        else
            ereport(elevel,
                    (errmsg(ngettext("launched %d parallel vacuum worker for index vacuuming (planned: %d)",
                                     "launched %d parallel vacuum workers for index vacuuming (planned: %d)",
                                     lps->pcxt->nworkers_launched),
                            lps->pcxt->nworkers_launched, nworkers)));
    }

    /* Process the indexes that can be processed by only leader process */
    do_serial_processing_for_unsafe_indexes(vacrel, lps->lvshared);

    /*
     * Join as a parallel worker.  The leader process alone processes all the
     * indexes in the case where no workers are launched.
     */
    do_parallel_processing(vacrel, lps->lvshared);

    /*
     * Next, accumulate buffer and WAL usage.  (This must wait for the workers
     * to finish, or we might get incomplete data.)
     */
    if (nworkers > 0)
    {
        /* Wait for all vacuum workers to finish */
        WaitForParallelWorkersToFinish(lps->pcxt);

        for (int i = 0; i < lps->pcxt->nworkers_launched; i++)
            InstrAccumParallelQuery(&lps->buffer_usage[i], &lps->wal_usage[i]);
    }

    /*
     * Carry the shared balance value to heap scan and disable shared costing
     */
    if (VacuumSharedCostBalance)
    {
        VacuumCostBalance = pg_atomic_read_u32(VacuumSharedCostBalance);
        VacuumSharedCostBalance = NULL;
        VacuumActiveNWorkers = NULL;
    }
}

do_serial_processing_for_unsafe_indexes()

static void do_serial_processing_for_unsafe_indexes ( LVRelState *  vacrel, LVShared *  lvshared  )

static

Definition at line 2879 of file vacuumlazy.c.

References Assert, idx(), LVRelState::indrels, LVRelState::indstats, IsParallelWorker, LVRelState::nindexes, parallel_process_one_index(), parallel_processing_is_safe(), parallel_stats_for_idx(), pg_atomic_add_fetch_u32(), pg_atomic_sub_fetch_u32(), and VacuumActiveNWorkers.

Referenced by do_parallel_vacuum_or_cleanup().

{
    Assert(!IsParallelWorker());

    /*
     * Increment the active worker count if we are able to launch any worker.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_add_fetch_u32(VacuumActiveNWorkers, 1);

    for (int idx = 0; idx < vacrel->nindexes; idx++)
    {
        LVSharedIndStats *shared_istat;
        Relation    indrel;
        IndexBulkDeleteResult *istat;

        shared_istat = parallel_stats_for_idx(lvshared, idx);

        /* Skip already-complete indexes */
        if (shared_istat != NULL)
            continue;

        indrel = vacrel->indrels[idx];

        /*
         * We're only here for the unsafe indexes
         */
        if (parallel_processing_is_safe(indrel, lvshared))
            continue;

        /* Do vacuum or cleanup of the index */
        istat = (vacrel->indstats[idx]);
        vacrel->indstats[idx] = parallel_process_one_index(indrel, istat,
                                                           lvshared,
                                                           shared_istat,
                                                           vacrel);
    }

    /*
     * We have completed the index vacuum so decrement the active worker
     * count.
     */
    if (VacuumActiveNWorkers)
        pg_atomic_sub_fetch_u32(VacuumActiveNWorkers, 1);
}

end_parallel_vacuum()

static void end_parallel_vacuum ( LVRelState *  vacrel )

static

Definition at line 4044 of file vacuumlazy.c.

References Assert, DestroyParallelContext(), ExitParallelMode(), idx(), LVRelState::indstats, IsParallelWorker, LVSharedIndStats::istat, LVRelState::lps, LVParallelState::lvshared, LVRelState::nindexes, palloc0(), parallel_stats_for_idx(), LVParallelState::pcxt, pfree(), and LVSharedIndStats::updated.

Referenced by lazy_space_free().

{
    IndexBulkDeleteResult **indstats = vacrel->indstats;
    LVParallelState *lps = vacrel->lps;
    int         nindexes = vacrel->nindexes;

    Assert(!IsParallelWorker());

    /* Copy the updated statistics */
    for (int idx = 0; idx < nindexes; idx++)
    {
        LVSharedIndStats *shared_istat;

        shared_istat = parallel_stats_for_idx(lps->lvshared, idx);

        /*
         * Skip unused slot.  The statistics of this index are already stored
         * in local memory.
         */
        if (shared_istat == NULL)
            continue;

        if (shared_istat->updated)
        {
            indstats[idx] = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
            memcpy(indstats[idx], &(shared_istat->istat), sizeof(IndexBulkDeleteResult));
        }
        else
            indstats[idx] = NULL;
    }

    DestroyParallelContext(lps->pcxt);
    ExitParallelMode();

    /* Deactivate parallel vacuum */
    pfree(lps);
    vacrel->lps = NULL;
}

heap_page_is_all_visible()

static bool heap_page_is_all_visible ( LVRelState *  vacrel, Buffer  buf, TransactionId *  visibility_cutoff_xid, bool *  all_frozen  )

static

Definition at line 3624 of file vacuumlazy.c.

References Assert, BufferGetBlockNumber(), BufferGetPage, elog, ERROR, FirstOffsetNumber, heap_tuple_needs_eventual_freeze(), HEAPTUPLE_DEAD, HEAPTUPLE_DELETE_IN_PROGRESS, HEAPTUPLE_INSERT_IN_PROGRESS, HEAPTUPLE_LIVE, HEAPTUPLE_RECENTLY_DEAD, HeapTupleHeaderGetXmin, HeapTupleHeaderXminCommitted, HeapTupleSatisfiesVacuum(), InvalidOffsetNumber, InvalidTransactionId, ItemIdGetLength, ItemIdIsDead, ItemIdIsNormal, ItemIdIsRedirected, ItemIdIsUsed, ItemPointerSet, LVRelState::offnum, OffsetNumberNext, LVRelState::OldestXmin, PageGetItem, PageGetItemId, PageGetMaxOffsetNumber, LVRelState::rel, RelationGetRelid, HeapTupleData::t_data, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, TransactionIdFollows(), and TransactionIdPrecedes().

Referenced by lazy_scan_prune(), and lazy_vacuum_heap_page().

{
    Page        page = BufferGetPage(buf);
    BlockNumber blockno = BufferGetBlockNumber(buf);
    OffsetNumber offnum,
                maxoff;
    bool        all_visible = true;

    *visibility_cutoff_xid = InvalidTransactionId;
    *all_frozen = true;

    /*
     * This is a stripped down version of the line pointer scan in
     * lazy_scan_heap(). So if you change anything here, also check that code.
     */
    maxoff = PageGetMaxOffsetNumber(page);
    for (offnum = FirstOffsetNumber;
         offnum <= maxoff && all_visible;
         offnum = OffsetNumberNext(offnum))
    {
        ItemId      itemid;
        HeapTupleData tuple;

        /*
         * Set the offset number so that we can display it along with any
         * error that occurred while processing this tuple.
         */
        vacrel->offnum = offnum;
        itemid = PageGetItemId(page, offnum);

        /* Unused or redirect line pointers are of no interest */
        if (!ItemIdIsUsed(itemid) || ItemIdIsRedirected(itemid))
            continue;

        ItemPointerSet(&(tuple.t_self), blockno, offnum);

        /*
         * Dead line pointers can have index pointers pointing to them. So
         * they can't be treated as visible
         */
        if (ItemIdIsDead(itemid))
        {
            all_visible = false;
            *all_frozen = false;
            break;
        }

        Assert(ItemIdIsNormal(itemid));

        tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
        tuple.t_len = ItemIdGetLength(itemid);
        tuple.t_tableOid = RelationGetRelid(vacrel->rel);

        switch (HeapTupleSatisfiesVacuum(&tuple, vacrel->OldestXmin, buf))
        {
            case HEAPTUPLE_LIVE:
                {
                    TransactionId xmin;

                    /* Check comments in lazy_scan_heap. */
                    if (!HeapTupleHeaderXminCommitted(tuple.t_data))
                    {
                        all_visible = false;
                        *all_frozen = false;
                        break;
                    }

                    /*
                     * The inserter definitely committed. But is it old enough
                     * that everyone sees it as committed?
                     */
                    xmin = HeapTupleHeaderGetXmin(tuple.t_data);
                    if (!TransactionIdPrecedes(xmin, vacrel->OldestXmin))
                    {
                        all_visible = false;
                        *all_frozen = false;
                        break;
                    }

                    /* Track newest xmin on page. */
                    if (TransactionIdFollows(xmin, *visibility_cutoff_xid))
                        *visibility_cutoff_xid = xmin;

                    /* Check whether this tuple is already frozen or not */
                    if (all_visible && *all_frozen &&
                        heap_tuple_needs_eventual_freeze(tuple.t_data))
                        *all_frozen = false;
                }
                break;

            case HEAPTUPLE_DEAD:
            case HEAPTUPLE_RECENTLY_DEAD:
            case HEAPTUPLE_INSERT_IN_PROGRESS:
            case HEAPTUPLE_DELETE_IN_PROGRESS:
                {
                    all_visible = false;
                    *all_frozen = false;
                    break;
                }
            default:
                elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
                break;
        }
    }                           /* scan along page */

    /* Clear the offset information once we have processed the given page. */
    vacrel->offnum = InvalidOffsetNumber;

    return all_visible;
}

heap_vacuum_rel()

void heap_vacuum_rel	(	Relation	rel,
		VacuumParams *	params,
		BufferAccessStrategy	bstrategy
	)

Definition at line 484 of file vacuumlazy.c.

References _, appendStringInfo(), appendStringInfoChar(), appendStringInfoString(), ErrorContextCallback::arg, Assert, LVRelState::bstrategy, buf, ErrorContextCallback::callback, LVRelState::consider_bypass_optimization, StringInfoData::data, DEBUG2, LVRelState::do_index_cleanup, LVRelState::do_index_vacuuming, LVRelState::do_rel_truncate, elevel, ereport, errmsg_internal(), error_context_stack, LVRelState::failsafe_active, VacuumParams::freeze_min_age, VacuumParams::freeze_table_age, LVRelState::FreezeLimit, LVRelState::frozenskipped_pages, get_database_name(), get_namespace_name(), GetCurrentTimestamp(), i, VacuumParams::index_cleanup, LVRelState::indname, LVRelState::indrels, LVRelState::indstats, INFO, initStringInfo(), InvalidMultiXactId, InvalidOffsetNumber, InvalidTransactionId, VacuumParams::is_wraparound, IsAutoVacuumWorkerProcess(), lazy_scan_heap(), lazy_truncate_heap(), LOG, VacuumParams::log_min_duration, LVRelState::lpdead_item_pages, LVRelState::lpdead_items, Max, VacuumParams::multixact_freeze_min_age, VacuumParams::multixact_freeze_table_age, LVRelState::MultiXactCutoff, MultiXactIdPrecedesOrEquals(), MyDatabaseId, LVRelState::new_dead_tuples, LVRelState::new_live_tuples, LVRelState::new_rel_tuples, LVRelState::nindexes, NoLock, LVRelState::nonempty_pages, LVRelState::num_index_scans, IndexBulkDeleteResult::num_pages, LVRelState::old_live_tuples, LVRelState::old_rel_pages, LVRelState::OldestXmin, VacuumParams::options, IndexBulkDeleteResult::pages_deleted, IndexBulkDeleteResult::pages_free, IndexBulkDeleteResult::pages_newly_deleted, LVRelState::pages_removed, palloc(), palloc0(), pfree(), pg_rusage_init(), pg_rusage_show(), pgstat_progress_end_command(), pgstat_progress_start_command(), pgstat_progress_update_param(), pgstat_report_vacuum(), pgStatBlockReadTime, pgStatBlockWriteTime, pgWalUsage, LVRelState::phase, LVRelState::pinskipped_pages, ErrorContextCallback::previous, PROGRESS_COMMAND_VACUUM, PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_FINAL_CLEANUP, pstrdup(), RelationData::rd_rel, LVRelState::rel, LVRelState::rel_pages, RelationGetNamespace, RelationGetRelationName, RelationGetRelid, LVRelState::relfrozenxid, LVRelState::relminmxid, LVRelState::relname, LVRelState::relnamespace, RowExclusiveLock, LVRelState::scanned_pages, should_attempt_truncation(), TimestampDifference(), TimestampDifferenceExceeds(), track_io_timing, TransactionIdPrecedesOrEquals(), VacuumParams::truncate, LVRelState::tuples_deleted, update_vacuum_error_info(), vac_close_indexes(), vac_open_indexes(), vac_update_relstats(), VACOPT_DISABLE_PAGE_SKIPPING, VACOPT_VERBOSE, VACOPTVALUE_AUTO, VACOPTVALUE_DISABLED, VACOPTVALUE_ENABLED, VACOPTVALUE_UNSPECIFIED, VACUUM_ERRCB_PHASE_TRUNCATE, VACUUM_ERRCB_PHASE_UNKNOWN, vacuum_error_callback(), vacuum_set_xid_limits(), VacuumPageDirty, VacuumPageHit, VacuumPageMiss, visibilitymap_count(), WalUsage::wal_bytes, WalUsage::wal_fpi, WalUsage::wal_records, and WalUsageAccumDiff().

Referenced by SampleHeapTupleVisible().

{
    LVRelState *vacrel;
    PGRUsage    ru0;
    TimestampTz starttime = 0;
    WalUsage    walusage_start = pgWalUsage;
    WalUsage    walusage = {0, 0, 0};
    long        secs;
    int         usecs;
    double      read_rate,
                write_rate;
    bool        aggressive;     /* should we scan all unfrozen pages? */
    bool        scanned_all_unfrozen;   /* actually scanned all such pages? */
    char      **indnames = NULL;
    TransactionId xidFullScanLimit;
    MultiXactId mxactFullScanLimit;
    BlockNumber new_rel_pages;
    BlockNumber new_rel_allvisible;
    double      new_live_tuples;
    TransactionId new_frozen_xid;
    MultiXactId new_min_multi;
    ErrorContextCallback errcallback;
    PgStat_Counter startreadtime = 0;
    PgStat_Counter startwritetime = 0;
    TransactionId OldestXmin;
    TransactionId FreezeLimit;
    MultiXactId MultiXactCutoff;

    /* measure elapsed time iff autovacuum logging requires it */
    if (IsAutoVacuumWorkerProcess() && params->log_min_duration >= 0)
    {
        pg_rusage_init(&ru0);
        starttime = GetCurrentTimestamp();
        if (track_io_timing)
        {
            startreadtime = pgStatBlockReadTime;
            startwritetime = pgStatBlockWriteTime;
        }
    }

    if (params->options & VACOPT_VERBOSE)
        elevel = INFO;
    else
        elevel = DEBUG2;

    pgstat_progress_start_command(PROGRESS_COMMAND_VACUUM,
                                  RelationGetRelid(rel));

    vacuum_set_xid_limits(rel,
                          params->freeze_min_age,
                          params->freeze_table_age,
                          params->multixact_freeze_min_age,
                          params->multixact_freeze_table_age,
                          &OldestXmin, &FreezeLimit, &xidFullScanLimit,
                          &MultiXactCutoff, &mxactFullScanLimit);

    /*
     * We request an aggressive scan if the table's frozen Xid is now older
     * than or equal to the requested Xid full-table scan limit; or if the
     * table's minimum MultiXactId is older than or equal to the requested
     * mxid full-table scan limit; or if DISABLE_PAGE_SKIPPING was specified.
     */
    aggressive = TransactionIdPrecedesOrEquals(rel->rd_rel->relfrozenxid,
                                               xidFullScanLimit);
    aggressive |= MultiXactIdPrecedesOrEquals(rel->rd_rel->relminmxid,
                                              mxactFullScanLimit);
    if (params->options & VACOPT_DISABLE_PAGE_SKIPPING)
        aggressive = true;

    vacrel = (LVRelState *) palloc0(sizeof(LVRelState));

    /* Set up high level stuff about rel */
    vacrel->rel = rel;
    vac_open_indexes(vacrel->rel, RowExclusiveLock, &vacrel->nindexes,
                     &vacrel->indrels);
    vacrel->failsafe_active = false;
    vacrel->consider_bypass_optimization = true;

    /*
     * The index_cleanup param either disables index vacuuming and cleanup or
     * forces it to go ahead when we would otherwise apply the index bypass
     * optimization.  The default is 'auto', which leaves the final decision
     * up to lazy_vacuum().
     *
     * The truncate param allows user to avoid attempting relation truncation,
     * though it can't force truncation to happen.
     */
    Assert(params->index_cleanup != VACOPTVALUE_UNSPECIFIED);
    Assert(params->truncate != VACOPTVALUE_UNSPECIFIED &&
           params->truncate != VACOPTVALUE_AUTO);
    vacrel->do_index_vacuuming = true;
    vacrel->do_index_cleanup = true;
    vacrel->do_rel_truncate = (params->truncate != VACOPTVALUE_DISABLED);
    if (params->index_cleanup == VACOPTVALUE_DISABLED)
    {
        /* Force disable index vacuuming up-front */
        vacrel->do_index_vacuuming = false;
        vacrel->do_index_cleanup = false;
    }
    else if (params->index_cleanup == VACOPTVALUE_ENABLED)
    {
        /* Force index vacuuming.  Note that failsafe can still bypass. */
        vacrel->consider_bypass_optimization = false;
    }
    else
    {
        /* Default/auto, make all decisions dynamically */
        Assert(params->index_cleanup == VACOPTVALUE_AUTO);
    }

    vacrel->bstrategy = bstrategy;
    vacrel->old_rel_pages = rel->rd_rel->relpages;
    vacrel->old_live_tuples = rel->rd_rel->reltuples;
    vacrel->relfrozenxid = rel->rd_rel->relfrozenxid;
    vacrel->relminmxid = rel->rd_rel->relminmxid;

    /* Set cutoffs for entire VACUUM */
    vacrel->OldestXmin = OldestXmin;
    vacrel->FreezeLimit = FreezeLimit;
    vacrel->MultiXactCutoff = MultiXactCutoff;

    vacrel->relnamespace = get_namespace_name(RelationGetNamespace(rel));
    vacrel->relname = pstrdup(RelationGetRelationName(rel));
    vacrel->indname = NULL;
    vacrel->phase = VACUUM_ERRCB_PHASE_UNKNOWN;

    /* Save index names iff autovacuum logging requires it */
    if (IsAutoVacuumWorkerProcess() && params->log_min_duration >= 0 &&
        vacrel->nindexes > 0)
    {
        indnames = palloc(sizeof(char *) * vacrel->nindexes);
        for (int i = 0; i < vacrel->nindexes; i++)
            indnames[i] =
                pstrdup(RelationGetRelationName(vacrel->indrels[i]));
    }

    /*
     * Setup error traceback support for ereport().  The idea is to set up an
     * error context callback to display additional information on any error
     * during a vacuum.  During different phases of vacuum (heap scan, heap
     * vacuum, index vacuum, index clean up, heap truncate), we update the
     * error context callback to display appropriate information.
     *
     * Note that the index vacuum and heap vacuum phases may be called
     * multiple times in the middle of the heap scan phase.  So the old phase
     * information is restored at the end of those phases.
     */
    errcallback.callback = vacuum_error_callback;
    errcallback.arg = vacrel;
    errcallback.previous = error_context_stack;
    error_context_stack = &errcallback;

    /* Do the vacuuming */
    lazy_scan_heap(vacrel, params, aggressive);

    /* Done with indexes */
    vac_close_indexes(vacrel->nindexes, vacrel->indrels, NoLock);

    /*
     * Compute whether we actually scanned the all unfrozen pages. If we did,
     * we can adjust relfrozenxid and relminmxid.
     *
     * NB: We need to check this before truncating the relation, because that
     * will change ->rel_pages.
     */
    if ((vacrel->scanned_pages + vacrel->frozenskipped_pages)
        < vacrel->rel_pages)
    {
        Assert(!aggressive);
        scanned_all_unfrozen = false;
    }
    else
        scanned_all_unfrozen = true;

    /*
     * Optionally truncate the relation.
     */
    if (should_attempt_truncation(vacrel))
    {
        /*
         * Update error traceback information.  This is the last phase during
         * which we add context information to errors, so we don't need to
         * revert to the previous phase.
         */
        update_vacuum_error_info(vacrel, NULL, VACUUM_ERRCB_PHASE_TRUNCATE,
                                 vacrel->nonempty_pages,
                                 InvalidOffsetNumber);
        lazy_truncate_heap(vacrel);
    }

    /* Pop the error context stack */
    error_context_stack = errcallback.previous;

    /* Report that we are now doing final cleanup */
    pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
                                 PROGRESS_VACUUM_PHASE_FINAL_CLEANUP);

    /*
     * Update statistics in pg_class.
     *
     * In principle new_live_tuples could be -1 indicating that we (still)
     * don't know the tuple count.  In practice that probably can't happen,
     * since we'd surely have scanned some pages if the table is new and
     * nonempty.
     *
     * For safety, clamp relallvisible to be not more than what we're setting
     * relpages to.
     *
     * Also, don't change relfrozenxid/relminmxid if we skipped any pages,
     * since then we don't know for certain that all tuples have a newer xmin.
     */
    new_rel_pages = vacrel->rel_pages;
    new_live_tuples = vacrel->new_live_tuples;

    visibilitymap_count(rel, &new_rel_allvisible, NULL);
    if (new_rel_allvisible > new_rel_pages)
        new_rel_allvisible = new_rel_pages;

    new_frozen_xid = scanned_all_unfrozen ? FreezeLimit : InvalidTransactionId;
    new_min_multi = scanned_all_unfrozen ? MultiXactCutoff : InvalidMultiXactId;

    vac_update_relstats(rel,
                        new_rel_pages,
                        new_live_tuples,
                        new_rel_allvisible,
                        vacrel->nindexes > 0,
                        new_frozen_xid,
                        new_min_multi,
                        false);

    /*
     * Report results to the stats collector, too.
     *
     * Deliberately avoid telling the stats collector about LP_DEAD items that
     * remain in the table due to VACUUM bypassing index and heap vacuuming.
     * ANALYZE will consider the remaining LP_DEAD items to be dead tuples. It
     * seems like a good idea to err on the side of not vacuuming again too
     * soon in cases where the failsafe prevented significant amounts of heap
     * vacuuming.
     */
    pgstat_report_vacuum(RelationGetRelid(rel),
                         rel->rd_rel->relisshared,
                         Max(new_live_tuples, 0),
                         vacrel->new_dead_tuples);
    pgstat_progress_end_command();

    /* and log the action if appropriate */
    if (IsAutoVacuumWorkerProcess() && params->log_min_duration >= 0)
    {
        TimestampTz endtime = GetCurrentTimestamp();

        if (params->log_min_duration == 0 ||
            TimestampDifferenceExceeds(starttime, endtime,
                                       params->log_min_duration))
        {
            StringInfoData buf;
            char       *msgfmt;

            TimestampDifference(starttime, endtime, &secs, &usecs);

            memset(&walusage, 0, sizeof(WalUsage));
            WalUsageAccumDiff(&walusage, &pgWalUsage, &walusage_start);

            read_rate = 0;
            write_rate = 0;
            if ((secs > 0) || (usecs > 0))
            {
                read_rate = (double) BLCKSZ * VacuumPageMiss / (1024 * 1024) /
                    (secs + usecs / 1000000.0);
                write_rate = (double) BLCKSZ * VacuumPageDirty / (1024 * 1024) /
                    (secs + usecs / 1000000.0);
            }

            /*
             * This is pretty messy, but we split it up so that we can skip
             * emitting individual parts of the message when not applicable.
             */
            initStringInfo(&buf);
            if (params->is_wraparound)
            {
                /*
                 * While it's possible for a VACUUM to be both is_wraparound
                 * and !aggressive, that's just a corner-case -- is_wraparound
                 * implies aggressive.  Produce distinct output for the corner
                 * case all the same, just in case.
                 */
                if (aggressive)
                    msgfmt = _("automatic aggressive vacuum to prevent wraparound of table \"%s.%s.%s\": index scans: %d\n");
                else
                    msgfmt = _("automatic vacuum to prevent wraparound of table \"%s.%s.%s\": index scans: %d\n");
            }
            else
            {
                if (aggressive)
                    msgfmt = _("automatic aggressive vacuum of table \"%s.%s.%s\": index scans: %d\n");
                else
                    msgfmt = _("automatic vacuum of table \"%s.%s.%s\": index scans: %d\n");
            }
            appendStringInfo(&buf, msgfmt,
                             get_database_name(MyDatabaseId),
                             vacrel->relnamespace,
                             vacrel->relname,
                             vacrel->num_index_scans);
            appendStringInfo(&buf, _("pages: %u removed, %u remain, %u skipped due to pins, %u skipped frozen\n"),
                             vacrel->pages_removed,
                             vacrel->rel_pages,
                             vacrel->pinskipped_pages,
                             vacrel->frozenskipped_pages);
            appendStringInfo(&buf,
                             _("tuples: %lld removed, %lld remain, %lld are dead but not yet removable, oldest xmin: %u\n"),
                             (long long) vacrel->tuples_deleted,
                             (long long) vacrel->new_rel_tuples,
                             (long long) vacrel->new_dead_tuples,
                             OldestXmin);
            appendStringInfo(&buf,
                             _("buffer usage: %lld hits, %lld misses, %lld dirtied\n"),
                             (long long) VacuumPageHit,
                             (long long) VacuumPageMiss,
                             (long long) VacuumPageDirty);
            if (vacrel->rel_pages > 0)
            {
                BlockNumber orig_rel_pages;

                if (vacrel->do_index_vacuuming)
                {
                    msgfmt = _(" %u pages from table (%.2f%% of total) had %lld dead item identifiers removed\n");

                    if (vacrel->nindexes == 0 || vacrel->num_index_scans == 0)
                        appendStringInfoString(&buf, _("index scan not needed:"));
                    else
                        appendStringInfoString(&buf, _("index scan needed:"));
                }
                else
                {
                    msgfmt = _(" %u pages from table (%.2f%% of total) have %lld dead item identifiers\n");

                    if (!vacrel->failsafe_active)
                        appendStringInfoString(&buf, _("index scan bypassed:"));
                    else
                        appendStringInfoString(&buf, _("index scan bypassed by failsafe:"));
                }
                orig_rel_pages = vacrel->rel_pages + vacrel->pages_removed;
                appendStringInfo(&buf, msgfmt,
                                 vacrel->lpdead_item_pages,
                                 100.0 * vacrel->lpdead_item_pages / orig_rel_pages,
                                 (long long) vacrel->lpdead_items);
            }
            for (int i = 0; i < vacrel->nindexes; i++)
            {
                IndexBulkDeleteResult *istat = vacrel->indstats[i];

                if (!istat)
                    continue;

                appendStringInfo(&buf,
                                 _("index \"%s\": pages: %u in total, %u newly deleted, %u currently deleted, %u reusable\n"),
                                 indnames[i],
                                 istat->num_pages,
                                 istat->pages_newly_deleted,
                                 istat->pages_deleted,
                                 istat->pages_free);
            }
            appendStringInfo(&buf, _("avg read rate: %.3f MB/s, avg write rate: %.3f MB/s\n"),
                             read_rate, write_rate);
            if (track_io_timing)
            {
                appendStringInfoString(&buf, _("I/O Timings:"));
                if (pgStatBlockReadTime - startreadtime > 0)
                    appendStringInfo(&buf, _(" read=%.3f"),
                                     (double) (pgStatBlockReadTime - startreadtime) / 1000);
                if (pgStatBlockWriteTime - startwritetime > 0)
                    appendStringInfo(&buf, _(" write=%.3f"),
                                     (double) (pgStatBlockWriteTime - startwritetime) / 1000);
                appendStringInfoChar(&buf, '\n');
            }
            appendStringInfo(&buf, _("system usage: %s\n"), pg_rusage_show(&ru0));
            appendStringInfo(&buf,
                             _("WAL usage: %lld records, %lld full page images, %llu bytes"),
                             (long long) walusage.wal_records,
                             (long long) walusage.wal_fpi,
                             (unsigned long long) walusage.wal_bytes);

            ereport(LOG,
                    (errmsg_internal("%s", buf.data)));
            pfree(buf.data);
        }
    }

    /* Cleanup index statistics and index names */
    for (int i = 0; i < vacrel->nindexes; i++)
    {
        if (vacrel->indstats[i])
            pfree(vacrel->indstats[i]);

        if (indnames && indnames[i])
            pfree(indnames[i]);
    }
}

lazy_check_needs_freeze()

static bool lazy_check_needs_freeze ( Buffer  buf, bool *  hastup, LVRelState *  vacrel  )

static

Definition at line 2540 of file vacuumlazy.c.

References BufferGetPage, FirstOffsetNumber, LVRelState::FreezeLimit, heap_tuple_needs_freeze(), InvalidOffsetNumber, ItemIdIsNormal, ItemIdIsUsed, LVRelState::MultiXactCutoff, LVRelState::offnum, OffsetNumberNext, PageGetItem, PageGetItemId, PageGetMaxOffsetNumber, PageIsEmpty, and PageIsNew.

Referenced by lazy_scan_heap().

{
    Page        page = BufferGetPage(buf);
    OffsetNumber offnum,
                maxoff;
    HeapTupleHeader tupleheader;

    *hastup = false;

    /*
     * New and empty pages, obviously, don't contain tuples. We could make
     * sure that the page is registered in the FSM, but it doesn't seem worth
     * waiting for a cleanup lock just for that, especially because it's
     * likely that the pin holder will do so.
     */
    if (PageIsNew(page) || PageIsEmpty(page))
        return false;

    maxoff = PageGetMaxOffsetNumber(page);
    for (offnum = FirstOffsetNumber;
         offnum <= maxoff;
         offnum = OffsetNumberNext(offnum))
    {
        ItemId      itemid;

        /*
         * Set the offset number so that we can display it along with any
         * error that occurred while processing this tuple.
         */
        vacrel->offnum = offnum;
        itemid = PageGetItemId(page, offnum);

        /* this should match hastup test in count_nondeletable_pages() */
        if (ItemIdIsUsed(itemid))
            *hastup = true;

        /* dead and redirect items never need freezing */
        if (!ItemIdIsNormal(itemid))
            continue;

        tupleheader = (HeapTupleHeader) PageGetItem(page, itemid);

        if (heap_tuple_needs_freeze(tupleheader, vacrel->FreezeLimit,
                                    vacrel->MultiXactCutoff, buf))
            break;
    }                           /* scan along page */

    /* Clear the offset information once we have processed the given page. */
    vacrel->offnum = InvalidOffsetNumber;

    return (offnum <= maxoff);
}

lazy_check_wraparound_failsafe()

static bool lazy_check_wraparound_failsafe ( LVRelState *  vacrel )

static

Definition at line 2606 of file vacuumlazy.c.

References LVRelState::do_index_cleanup, LVRelState::do_index_vacuuming, LVRelState::do_rel_truncate, ereport, errdetail(), errhint(), errmsg(), LVRelState::failsafe_active, get_database_name(), MyDatabaseId, LVRelState::num_index_scans, LVRelState::relfrozenxid, LVRelState::relminmxid, LVRelState::relname, LVRelState::relnamespace, unlikely, vacuum_xid_failsafe_check(), VacuumCostActive, VacuumCostBalance, and WARNING.

Referenced by lazy_scan_heap(), and lazy_vacuum_all_indexes().

{
    /* Don't warn more than once per VACUUM */
    if (vacrel->failsafe_active)
        return true;

    if (unlikely(vacuum_xid_failsafe_check(vacrel->relfrozenxid,
                                           vacrel->relminmxid)))
    {
        vacrel->failsafe_active = true;

        /* Disable index vacuuming, index cleanup, and heap rel truncation */
        vacrel->do_index_vacuuming = false;
        vacrel->do_index_cleanup = false;
        vacrel->do_rel_truncate = false;

        ereport(WARNING,
                (errmsg("bypassing nonessential maintenance of table \"%s.%s.%s\" as a failsafe after %d index scans",
                        get_database_name(MyDatabaseId),
                        vacrel->relnamespace,
                        vacrel->relname,
                        vacrel->num_index_scans),
                 errdetail("table's relfrozenxid or relminmxid is too far in the past"),
                 errhint("Consider increasing configuration parameter \"maintenance_work_mem\" or \"autovacuum_work_mem\".\n"
                         "You might also need to consider other ways for VACUUM to keep up with the allocation of transaction IDs.")));

        /* Stop applying cost limits from this point on */
        VacuumCostActive = false;
        VacuumCostBalance = 0;

        return true;
    }

    return false;
}

lazy_cleanup_all_indexes()

static void lazy_cleanup_all_indexes ( LVRelState *  vacrel )

static

Definition at line 2986 of file vacuumlazy.c.

References Assert, do_parallel_lazy_cleanup_all_indexes(), idx(), LVRelState::indrels, LVRelState::indstats, IsParallelWorker, lazy_cleanup_one_index(), LVRelState::new_rel_tuples, LVRelState::nindexes, ParallelVacuumIsActive, pgstat_progress_update_param(), PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_INDEX_CLEANUP, LVRelState::rel_pages, and LVRelState::tupcount_pages.

Referenced by lazy_scan_heap().

{
    Assert(!IsParallelWorker());
    Assert(vacrel->nindexes > 0);

    /* Report that we are now cleaning up indexes */
    pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
                                 PROGRESS_VACUUM_PHASE_INDEX_CLEANUP);

    if (!ParallelVacuumIsActive(vacrel))
    {
        double      reltuples = vacrel->new_rel_tuples;
        bool        estimated_count =
        vacrel->tupcount_pages < vacrel->rel_pages;

        for (int idx = 0; idx < vacrel->nindexes; idx++)
        {
            Relation    indrel = vacrel->indrels[idx];
            IndexBulkDeleteResult *istat = vacrel->indstats[idx];

            vacrel->indstats[idx] =
                lazy_cleanup_one_index(indrel, istat, reltuples,
                                       estimated_count, vacrel);
        }
    }
    else
    {
        /* Outsource everything to parallel variant */
        do_parallel_lazy_cleanup_all_indexes(vacrel);
    }
}

lazy_cleanup_one_index()

static IndexBulkDeleteResult * lazy_cleanup_one_index ( Relation  indrel, IndexBulkDeleteResult *  istat, double  reltuples, bool  estimated_count, LVRelState *  vacrel  )

static

Definition at line 3085 of file vacuumlazy.c.

References IndexVacuumInfo::analyze_only, Assert, LVRelState::bstrategy, elevel, ereport, errdetail(), errmsg(), IndexVacuumInfo::estimated_count, IndexVacuumInfo::index, index_vacuum_cleanup(), LVRelState::indname, InvalidBlockNumber, InvalidOffsetNumber, IndexVacuumInfo::message_level, IndexVacuumInfo::num_heap_tuples, pfree(), pg_rusage_init(), pg_rusage_show(), pstrdup(), RelationGetRelationName, IndexVacuumInfo::report_progress, restore_vacuum_error_info(), IndexVacuumInfo::strategy, update_vacuum_error_info(), and VACUUM_ERRCB_PHASE_INDEX_CLEANUP.

Referenced by lazy_cleanup_all_indexes(), and parallel_process_one_index().

{
    IndexVacuumInfo ivinfo;
    PGRUsage    ru0;
    LVSavedErrInfo saved_err_info;

    pg_rusage_init(&ru0);

    ivinfo.index = indrel;
    ivinfo.analyze_only = false;
    ivinfo.report_progress = false;
    ivinfo.estimated_count = estimated_count;
    ivinfo.message_level = elevel;

    ivinfo.num_heap_tuples = reltuples;
    ivinfo.strategy = vacrel->bstrategy;

    /*
     * Update error traceback information.
     *
     * The index name is saved during this phase and restored immediately
     * after this phase.  See vacuum_error_callback.
     */
    Assert(vacrel->indname == NULL);
    vacrel->indname = pstrdup(RelationGetRelationName(indrel));
    update_vacuum_error_info(vacrel, &saved_err_info,
                             VACUUM_ERRCB_PHASE_INDEX_CLEANUP,
                             InvalidBlockNumber, InvalidOffsetNumber);

    istat = index_vacuum_cleanup(&ivinfo, istat);

    if (istat)
    {
        ereport(elevel,
                (errmsg("index \"%s\" now contains %.0f row versions in %u pages",
                        RelationGetRelationName(indrel),
                        (istat)->num_index_tuples,
                        (istat)->num_pages),
                 errdetail("%.0f index row versions were removed.\n"
                           "%u index pages were newly deleted.\n"
                           "%u index pages are currently deleted, of which %u are currently reusable.\n"
                           "%s.",
                           (istat)->tuples_removed,
                           (istat)->pages_newly_deleted,
                           (istat)->pages_deleted, (istat)->pages_free,
                           pg_rusage_show(&ru0))));
    }

    /* Revert to the previous phase information for error traceback */
    restore_vacuum_error_info(vacrel, &saved_err_info);
    pfree(vacrel->indname);
    vacrel->indname = NULL;

    return istat;
}

lazy_scan_heap()

static void lazy_scan_heap ( LVRelState *  vacrel, VacuumParams *  params, bool  aggressive  )

static

Definition at line 913 of file vacuumlazy.c.

References _, LVPagePruneState::all_frozen, LVPagePruneState::all_visible, appendStringInfo(), Assert, LVRelState::blkno, LVRelState::bstrategy, buf, BUFFER_LOCK_SHARE, BUFFER_LOCK_UNLOCK, BufferGetPage, BufferIsValid, ConditionalLockBufferForCleanup(), LVRelState::consider_bypass_optimization, StringInfoData::data, LVRelState::dead_tuples, LVRelState::do_index_cleanup, LVRelState::do_index_vacuuming, elevel, elog, END_CRIT_SECTION, ereport, errdetail_internal(), errmsg(), FAILSAFE_EVERY_PAGES, FORCE_CHECK_PAGE, FreeSpaceMapVacuumRange(), LVRelState::frozenskipped_pages, GetRecordedFreeSpace(), GlobalVisTestFor(), LVPagePruneState::has_lpdead_items, LVPagePruneState::hastup, LVRelState::indstats, initStringInfo(), InvalidBlockNumber, InvalidBuffer, InvalidOffsetNumber, InvalidTransactionId, InvalidXLogRecPtr, lazy_check_needs_freeze(), lazy_check_wraparound_failsafe(), lazy_cleanup_all_indexes(), lazy_scan_prune(), lazy_space_alloc(), lazy_space_free(), lazy_vacuum(), lazy_vacuum_heap_page(), LVRelState::live_tuples, LockBuffer(), LockBufferForCleanup(), log_newpage_buffer(), LVRelState::lpdead_item_pages, LVRelState::lpdead_items, MAIN_FORKNUM, MarkBufferDirty(), Max, LVDeadTuples::max_tuples, MaxHeapTuplesPerPage, LVRelState::new_dead_tuples, LVRelState::new_live_tuples, LVRelState::new_rel_tuples, ngettext, LVRelState::nindexes, LVRelState::nonempty_pages, LVRelState::num_index_scans, LVDeadTuples::num_tuples, LVRelState::num_tuples, VacuumParams::nworkers, LVRelState::OldestXmin, VacuumParams::options, PageClearAllVisible, PageGetHeapFreeSpace(), PageGetLSN, PageIsAllVisible, PageIsEmpty, PageIsNew, LVRelState::pages_removed, PageSetAllVisible, palloc0(), pfree(), pg_rusage_init(), pg_rusage_show(), pgstat_progress_update_multi_param(), pgstat_progress_update_param(), LVRelState::pinskipped_pages, PROGRESS_VACUUM_HEAP_BLKS_SCANNED, PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, PROGRESS_VACUUM_MAX_DEAD_TUPLES, PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_SCAN_HEAP, PROGRESS_VACUUM_TOTAL_HEAP_BLKS, RBM_NORMAL, ReadBufferExtended(), RecordPageWithFreeSpace(), LVRelState::rel, LVRelState::rel_pages, RelationGetNumberOfBlocks, RelationNeedsWAL, ReleaseBuffer(), LVRelState::relname, LVRelState::relnamespace, LVRelState::scanned_pages, SizeOfPageHeaderData, SKIP_PAGES_THRESHOLD, START_CRIT_SECTION, LVRelState::tupcount_pages, LVRelState::tuples_deleted, UnlockReleaseBuffer(), update_index_statistics(), update_vacuum_error_info(), vac_estimate_reltuples(), VACOPT_DISABLE_PAGE_SKIPPING, vacuum_delay_point(), VACUUM_ERRCB_PHASE_SCAN_HEAP, VACUUM_FSM_EVERY_PAGES, LVPagePruneState::visibility_cutoff_xid, VISIBILITYMAP_ALL_FROZEN, VISIBILITYMAP_ALL_VISIBLE, visibilitymap_clear(), visibilitymap_get_status(), visibilitymap_pin(), visibilitymap_set(), VISIBILITYMAP_VALID_BITS, VM_ALL_FROZEN, VM_ALL_VISIBLE, and WARNING.

Referenced by heap_vacuum_rel().

{
    LVDeadTuples *dead_tuples;
    BlockNumber nblocks,
                blkno,
                next_unskippable_block,
                next_failsafe_block,
                next_fsm_block_to_vacuum;
    PGRUsage    ru0;
    Buffer      vmbuffer = InvalidBuffer;
    bool        skipping_blocks;
    StringInfoData buf;
    const int   initprog_index[] = {
        PROGRESS_VACUUM_PHASE,
        PROGRESS_VACUUM_TOTAL_HEAP_BLKS,
        PROGRESS_VACUUM_MAX_DEAD_TUPLES
    };
    int64       initprog_val[3];
    GlobalVisState *vistest;

    pg_rusage_init(&ru0);

    if (aggressive)
        ereport(elevel,
                (errmsg("aggressively vacuuming \"%s.%s\"",
                        vacrel->relnamespace,
                        vacrel->relname)));
    else
        ereport(elevel,
                (errmsg("vacuuming \"%s.%s\"",
                        vacrel->relnamespace,
                        vacrel->relname)));

    nblocks = RelationGetNumberOfBlocks(vacrel->rel);
    next_unskippable_block = 0;
    next_failsafe_block = 0;
    next_fsm_block_to_vacuum = 0;
    vacrel->rel_pages = nblocks;
    vacrel->scanned_pages = 0;
    vacrel->pinskipped_pages = 0;
    vacrel->frozenskipped_pages = 0;
    vacrel->tupcount_pages = 0;
    vacrel->pages_removed = 0;
    vacrel->lpdead_item_pages = 0;
    vacrel->nonempty_pages = 0;

    /* Initialize instrumentation counters */
    vacrel->num_index_scans = 0;
    vacrel->tuples_deleted = 0;
    vacrel->lpdead_items = 0;
    vacrel->new_dead_tuples = 0;
    vacrel->num_tuples = 0;
    vacrel->live_tuples = 0;

    vistest = GlobalVisTestFor(vacrel->rel);

    vacrel->indstats = (IndexBulkDeleteResult **)
        palloc0(vacrel->nindexes * sizeof(IndexBulkDeleteResult *));

    /*
     * Before beginning scan, check if it's already necessary to apply
     * failsafe
     */
    lazy_check_wraparound_failsafe(vacrel);

    /*
     * Allocate the space for dead tuples.  Note that this handles parallel
     * VACUUM initialization as part of allocating shared memory space used
     * for dead_tuples.
     */
    lazy_space_alloc(vacrel, params->nworkers, nblocks);
    dead_tuples = vacrel->dead_tuples;

    /* Report that we're scanning the heap, advertising total # of blocks */
    initprog_val[0] = PROGRESS_VACUUM_PHASE_SCAN_HEAP;
    initprog_val[1] = nblocks;
    initprog_val[2] = dead_tuples->max_tuples;
    pgstat_progress_update_multi_param(3, initprog_index, initprog_val);

    /*
     * Except when aggressive is set, we want to skip pages that are
     * all-visible according to the visibility map, but only when we can skip
     * at least SKIP_PAGES_THRESHOLD consecutive pages.  Since we're reading
     * sequentially, the OS should be doing readahead for us, so there's no
     * gain in skipping a page now and then; that's likely to disable
     * readahead and so be counterproductive. Also, skipping even a single
     * page means that we can't update relfrozenxid, so we only want to do it
     * if we can skip a goodly number of pages.
     *
     * When aggressive is set, we can't skip pages just because they are
     * all-visible, but we can still skip pages that are all-frozen, since
     * such pages do not need freezing and do not affect the value that we can
     * safely set for relfrozenxid or relminmxid.
     *
     * Before entering the main loop, establish the invariant that
     * next_unskippable_block is the next block number >= blkno that we can't
     * skip based on the visibility map, either all-visible for a regular scan
     * or all-frozen for an aggressive scan.  We set it to nblocks if there's
     * no such block.  We also set up the skipping_blocks flag correctly at
     * this stage.
     *
     * Note: The value returned by visibilitymap_get_status could be slightly
     * out-of-date, since we make this test before reading the corresponding
     * heap page or locking the buffer.  This is OK.  If we mistakenly think
     * that the page is all-visible or all-frozen when in fact the flag's just
     * been cleared, we might fail to vacuum the page.  It's easy to see that
     * skipping a page when aggressive is not set is not a very big deal; we
     * might leave some dead tuples lying around, but the next vacuum will
     * find them.  But even when aggressive *is* set, it's still OK if we miss
     * a page whose all-frozen marking has just been cleared.  Any new XIDs
     * just added to that page are necessarily newer than the GlobalXmin we
     * computed, so they'll have no effect on the value to which we can safely
     * set relfrozenxid.  A similar argument applies for MXIDs and relminmxid.
     *
     * We will scan the table's last page, at least to the extent of
     * determining whether it has tuples or not, even if it should be skipped
     * according to the above rules; except when we've already determined that
     * it's not worth trying to truncate the table.  This avoids having
     * lazy_truncate_heap() take access-exclusive lock on the table to attempt
     * a truncation that just fails immediately because there are tuples in
     * the last page.  This is worth avoiding mainly because such a lock must
     * be replayed on any hot standby, where it can be disruptive.
     */
    if ((params->options & VACOPT_DISABLE_PAGE_SKIPPING) == 0)
    {
        while (next_unskippable_block < nblocks)
        {
            uint8       vmstatus;

            vmstatus = visibilitymap_get_status(vacrel->rel,
                                                next_unskippable_block,
                                                &vmbuffer);
            if (aggressive)
            {
                if ((vmstatus & VISIBILITYMAP_ALL_FROZEN) == 0)
                    break;
            }
            else
            {
                if ((vmstatus & VISIBILITYMAP_ALL_VISIBLE) == 0)
                    break;
            }
            vacuum_delay_point();
            next_unskippable_block++;
        }
    }

    if (next_unskippable_block >= SKIP_PAGES_THRESHOLD)
        skipping_blocks = true;
    else
        skipping_blocks = false;

    for (blkno = 0; blkno < nblocks; blkno++)
    {
        Buffer      buf;
        Page        page;
        bool        all_visible_according_to_vm = false;
        LVPagePruneState prunestate;

        /*
         * Consider need to skip blocks.  See note above about forcing
         * scanning of last page.
         */
#define FORCE_CHECK_PAGE() \
        (blkno == nblocks - 1 && should_attempt_truncation(vacrel))

        pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED, blkno);

        update_vacuum_error_info(vacrel, NULL, VACUUM_ERRCB_PHASE_SCAN_HEAP,
                                 blkno, InvalidOffsetNumber);

        if (blkno == next_unskippable_block)
        {
            /* Time to advance next_unskippable_block */
            next_unskippable_block++;
            if ((params->options & VACOPT_DISABLE_PAGE_SKIPPING) == 0)
            {
                while (next_unskippable_block < nblocks)
                {
                    uint8       vmskipflags;

                    vmskipflags = visibilitymap_get_status(vacrel->rel,
                                                           next_unskippable_block,
                                                           &vmbuffer);
                    if (aggressive)
                    {
                        if ((vmskipflags & VISIBILITYMAP_ALL_FROZEN) == 0)
                            break;
                    }
                    else
                    {
                        if ((vmskipflags & VISIBILITYMAP_ALL_VISIBLE) == 0)
                            break;
                    }
                    vacuum_delay_point();
                    next_unskippable_block++;
                }
            }

            /*
             * We know we can't skip the current block.  But set up
             * skipping_blocks to do the right thing at the following blocks.
             */
            if (next_unskippable_block - blkno > SKIP_PAGES_THRESHOLD)
                skipping_blocks = true;
            else
                skipping_blocks = false;

            /*
             * Normally, the fact that we can't skip this block must mean that
             * it's not all-visible.  But in an aggressive vacuum we know only
             * that it's not all-frozen, so it might still be all-visible.
             */
            if (aggressive && VM_ALL_VISIBLE(vacrel->rel, blkno, &vmbuffer))
                all_visible_according_to_vm = true;
        }
        else
        {
            /*
             * The current block is potentially skippable; if we've seen a
             * long enough run of skippable blocks to justify skipping it, and
             * we're not forced to check it, then go ahead and skip.
             * Otherwise, the page must be at least all-visible if not
             * all-frozen, so we can set all_visible_according_to_vm = true.
             */
            if (skipping_blocks && !FORCE_CHECK_PAGE())
            {
                /*
                 * Tricky, tricky.  If this is in aggressive vacuum, the page
                 * must have been all-frozen at the time we checked whether it
                 * was skippable, but it might not be any more.  We must be
                 * careful to count it as a skipped all-frozen page in that
                 * case, or else we'll think we can't update relfrozenxid and
                 * relminmxid.  If it's not an aggressive vacuum, we don't
                 * know whether it was all-frozen, so we have to recheck; but
                 * in this case an approximate answer is OK.
                 */
                if (aggressive || VM_ALL_FROZEN(vacrel->rel, blkno, &vmbuffer))
                    vacrel->frozenskipped_pages++;
                continue;
            }
            all_visible_according_to_vm = true;
        }

        vacuum_delay_point();

        /*
         * Regularly check if wraparound failsafe should trigger.
         *
         * There is a similar check inside lazy_vacuum_all_indexes(), but
         * relfrozenxid might start to look dangerously old before we reach
         * that point.  This check also provides failsafe coverage for the
         * one-pass strategy, and the two-pass strategy with the index_cleanup
         * param set to 'off'.
         */
        if (blkno - next_failsafe_block >= FAILSAFE_EVERY_PAGES)
        {
            lazy_check_wraparound_failsafe(vacrel);
            next_failsafe_block = blkno;
        }

        /*
         * Consider if we definitely have enough space to process TIDs on page
         * already.  If we are close to overrunning the available space for
         * dead-tuple TIDs, pause and do a cycle of vacuuming before we tackle
         * this page.
         */
        if ((dead_tuples->max_tuples - dead_tuples->num_tuples) < MaxHeapTuplesPerPage &&
            dead_tuples->num_tuples > 0)
        {
            /*
             * Before beginning index vacuuming, we release any pin we may
             * hold on the visibility map page.  This isn't necessary for
             * correctness, but we do it anyway to avoid holding the pin
             * across a lengthy, unrelated operation.
             */
            if (BufferIsValid(vmbuffer))
            {
                ReleaseBuffer(vmbuffer);
                vmbuffer = InvalidBuffer;
            }

            /* Remove the collected garbage tuples from table and indexes */
            vacrel->consider_bypass_optimization = false;
            lazy_vacuum(vacrel);

            /*
             * Vacuum the Free Space Map to make newly-freed space visible on
             * upper-level FSM pages.  Note we have not yet processed blkno.
             */
            FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum,
                                    blkno);
            next_fsm_block_to_vacuum = blkno;

            /* Report that we are once again scanning the heap */
            pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
                                         PROGRESS_VACUUM_PHASE_SCAN_HEAP);
        }

        /*
         * Set up visibility map page as needed.
         *
         * Pin the visibility map page in case we need to mark the page
         * all-visible.  In most cases this will be very cheap, because we'll
         * already have the correct page pinned anyway.  However, it's
         * possible that (a) next_unskippable_block is covered by a different
         * VM page than the current block or (b) we released our pin and did a
         * cycle of index vacuuming.
         */
        visibilitymap_pin(vacrel->rel, blkno, &vmbuffer);

        buf = ReadBufferExtended(vacrel->rel, MAIN_FORKNUM, blkno,
                                 RBM_NORMAL, vacrel->bstrategy);

        /*
         * We need buffer cleanup lock so that we can prune HOT chains and
         * defragment the page.
         */
        if (!ConditionalLockBufferForCleanup(buf))
        {
            bool        hastup;

            /*
             * If we're not performing an aggressive scan to guard against XID
             * wraparound, and we don't want to forcibly check the page, then
             * it's OK to skip vacuuming pages we get a lock conflict on. They
             * will be dealt with in some future vacuum.
             */
            if (!aggressive && !FORCE_CHECK_PAGE())
            {
                ReleaseBuffer(buf);
                vacrel->pinskipped_pages++;
                continue;
            }

            /*
             * Read the page with share lock to see if any xids on it need to
             * be frozen.  If not we just skip the page, after updating our
             * scan statistics.  If there are some, we wait for cleanup lock.
             *
             * We could defer the lock request further by remembering the page
             * and coming back to it later, or we could even register
             * ourselves for multiple buffers and then service whichever one
             * is received first.  For now, this seems good enough.
             *
             * If we get here with aggressive false, then we're just forcibly
             * checking the page, and so we don't want to insist on getting
             * the lock; we only need to know if the page contains tuples, so
             * that we can update nonempty_pages correctly.  It's convenient
             * to use lazy_check_needs_freeze() for both situations, though.
             */
            LockBuffer(buf, BUFFER_LOCK_SHARE);
            if (!lazy_check_needs_freeze(buf, &hastup, vacrel))
            {
                UnlockReleaseBuffer(buf);
                vacrel->scanned_pages++;
                vacrel->pinskipped_pages++;
                if (hastup)
                    vacrel->nonempty_pages = blkno + 1;
                continue;
            }
            if (!aggressive)
            {
                /*
                 * Here, we must not advance scanned_pages; that would amount
                 * to claiming that the page contains no freezable tuples.
                 */
                UnlockReleaseBuffer(buf);
                vacrel->pinskipped_pages++;
                if (hastup)
                    vacrel->nonempty_pages = blkno + 1;
                continue;
            }
            LockBuffer(buf, BUFFER_LOCK_UNLOCK);
            LockBufferForCleanup(buf);
            /* drop through to normal processing */
        }

        /*
         * By here we definitely have enough dead_tuples space for whatever
         * LP_DEAD tids are on this page, we have the visibility map page set
         * up in case we need to set this page's all_visible/all_frozen bit,
         * and we have a super-exclusive lock.  Any tuples on this page are
         * now sure to be "counted" by this VACUUM.
         *
         * One last piece of preamble needs to take place before we can prune:
         * we need to consider new and empty pages.
         */
        vacrel->scanned_pages++;
        vacrel->tupcount_pages++;

        page = BufferGetPage(buf);

        if (PageIsNew(page))
        {
            /*
             * All-zeroes pages can be left over if either a backend extends
             * the relation by a single page, but crashes before the newly
             * initialized page has been written out, or when bulk-extending
             * the relation (which creates a number of empty pages at the tail
             * end of the relation, but enters them into the FSM).
             *
             * Note we do not enter the page into the visibilitymap. That has
             * the downside that we repeatedly visit this page in subsequent
             * vacuums, but otherwise we'll never not discover the space on a
             * promoted standby. The harm of repeated checking ought to
             * normally not be too bad - the space usually should be used at
             * some point, otherwise there wouldn't be any regular vacuums.
             *
             * Make sure these pages are in the FSM, to ensure they can be
             * reused. Do that by testing if there's any space recorded for
             * the page. If not, enter it. We do so after releasing the lock
             * on the heap page, the FSM is approximate, after all.
             */
            UnlockReleaseBuffer(buf);

            if (GetRecordedFreeSpace(vacrel->rel, blkno) == 0)
            {
                Size        freespace = BLCKSZ - SizeOfPageHeaderData;

                RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
            }
            continue;
        }

        if (PageIsEmpty(page))
        {
            Size        freespace = PageGetHeapFreeSpace(page);

            /*
             * Empty pages are always all-visible and all-frozen (note that
             * the same is currently not true for new pages, see above).
             */
            if (!PageIsAllVisible(page))
            {
                START_CRIT_SECTION();

                /* mark buffer dirty before writing a WAL record */
                MarkBufferDirty(buf);

                /*
                 * It's possible that another backend has extended the heap,
                 * initialized the page, and then failed to WAL-log the page
                 * due to an ERROR.  Since heap extension is not WAL-logged,
                 * recovery might try to replay our record setting the page
                 * all-visible and find that the page isn't initialized, which
                 * will cause a PANIC.  To prevent that, check whether the
                 * page has been previously WAL-logged, and if not, do that
                 * now.
                 */
                if (RelationNeedsWAL(vacrel->rel) &&
                    PageGetLSN(page) == InvalidXLogRecPtr)
                    log_newpage_buffer(buf, true);

                PageSetAllVisible(page);
                visibilitymap_set(vacrel->rel, blkno, buf, InvalidXLogRecPtr,
                                  vmbuffer, InvalidTransactionId,
                                  VISIBILITYMAP_ALL_VISIBLE | VISIBILITYMAP_ALL_FROZEN);
                END_CRIT_SECTION();
            }

            UnlockReleaseBuffer(buf);
            RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
            continue;
        }

        /*
         * Prune and freeze tuples.
         *
         * Accumulates details of remaining LP_DEAD line pointers on page in
         * dead tuple list.  This includes LP_DEAD line pointers that we
         * pruned ourselves, as well as existing LP_DEAD line pointers that
         * were pruned some time earlier.  Also considers freezing XIDs in the
         * tuple headers of remaining items with storage.
         */
        lazy_scan_prune(vacrel, buf, blkno, page, vistest, &prunestate);

        Assert(!prunestate.all_visible || !prunestate.has_lpdead_items);

        /* Remember the location of the last page with nonremovable tuples */
        if (prunestate.hastup)
            vacrel->nonempty_pages = blkno + 1;

        if (vacrel->nindexes == 0)
        {
            /*
             * Consider the need to do page-at-a-time heap vacuuming when
             * using the one-pass strategy now.
             *
             * The one-pass strategy will never call lazy_vacuum().  The steps
             * performed here can be thought of as the one-pass equivalent of
             * a call to lazy_vacuum().
             */
            if (prunestate.has_lpdead_items)
            {
                Size        freespace;

                lazy_vacuum_heap_page(vacrel, blkno, buf, 0, &vmbuffer);

                /* Forget the now-vacuumed tuples */
                dead_tuples->num_tuples = 0;

                /*
                 * Periodically perform FSM vacuuming to make newly-freed
                 * space visible on upper FSM pages.  Note we have not yet
                 * performed FSM processing for blkno.
                 */
                if (blkno - next_fsm_block_to_vacuum >= VACUUM_FSM_EVERY_PAGES)
                {
                    FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum,
                                            blkno);
                    next_fsm_block_to_vacuum = blkno;
                }

                /*
                 * Now perform FSM processing for blkno, and move on to next
                 * page.
                 *
                 * Our call to lazy_vacuum_heap_page() will have considered if
                 * it's possible to set all_visible/all_frozen independently
                 * of lazy_scan_prune().  Note that prunestate was invalidated
                 * by lazy_vacuum_heap_page() call.
                 */
                freespace = PageGetHeapFreeSpace(page);

                UnlockReleaseBuffer(buf);
                RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
                continue;
            }

            /*
             * There was no call to lazy_vacuum_heap_page() because pruning
             * didn't encounter/create any LP_DEAD items that needed to be
             * vacuumed.  Prune state has not been invalidated, so proceed
             * with prunestate-driven visibility map and FSM steps (just like
             * the two-pass strategy).
             */
            Assert(dead_tuples->num_tuples == 0);
        }

        /*
         * Handle setting visibility map bit based on what the VM said about
         * the page before pruning started, and using prunestate
         */
        if (!all_visible_according_to_vm && prunestate.all_visible)
        {
            uint8       flags = VISIBILITYMAP_ALL_VISIBLE;

            if (prunestate.all_frozen)
                flags |= VISIBILITYMAP_ALL_FROZEN;

            /*
             * It should never be the case that the visibility map page is set
             * while the page-level bit is clear, but the reverse is allowed
             * (if checksums are not enabled).  Regardless, set both bits so
             * that we get back in sync.
             *
             * NB: If the heap page is all-visible but the VM bit is not set,
             * we don't need to dirty the heap page.  However, if checksums
             * are enabled, we do need to make sure that the heap page is
             * dirtied before passing it to visibilitymap_set(), because it
             * may be logged.  Given that this situation should only happen in
             * rare cases after a crash, it is not worth optimizing.
             */
            PageSetAllVisible(page);
            MarkBufferDirty(buf);
            visibilitymap_set(vacrel->rel, blkno, buf, InvalidXLogRecPtr,
                              vmbuffer, prunestate.visibility_cutoff_xid,
                              flags);
        }

        /*
         * As of PostgreSQL 9.2, the visibility map bit should never be set if
         * the page-level bit is clear.  However, it's possible that the bit
         * got cleared after we checked it and before we took the buffer
         * content lock, so we must recheck before jumping to the conclusion
         * that something bad has happened.
         */
        else if (all_visible_according_to_vm && !PageIsAllVisible(page)
                 && VM_ALL_VISIBLE(vacrel->rel, blkno, &vmbuffer))
        {
            elog(WARNING, "page is not marked all-visible but visibility map bit is set in relation \"%s\" page %u",
                 vacrel->relname, blkno);
            visibilitymap_clear(vacrel->rel, blkno, vmbuffer,
                                VISIBILITYMAP_VALID_BITS);
        }

        /*
         * It's possible for the value returned by
         * GetOldestNonRemovableTransactionId() to move backwards, so it's not
         * wrong for us to see tuples that appear to not be visible to
         * everyone yet, while PD_ALL_VISIBLE is already set. The real safe
         * xmin value never moves backwards, but
         * GetOldestNonRemovableTransactionId() is conservative and sometimes
         * returns a value that's unnecessarily small, so if we see that
         * contradiction it just means that the tuples that we think are not
         * visible to everyone yet actually are, and the PD_ALL_VISIBLE flag
         * is correct.
         *
         * There should never be dead tuples on a page with PD_ALL_VISIBLE
         * set, however.
         */
        else if (prunestate.has_lpdead_items && PageIsAllVisible(page))
        {
            elog(WARNING, "page containing dead tuples is marked as all-visible in relation \"%s\" page %u",
                 vacrel->relname, blkno);
            PageClearAllVisible(page);
            MarkBufferDirty(buf);
            visibilitymap_clear(vacrel->rel, blkno, vmbuffer,
                                VISIBILITYMAP_VALID_BITS);
        }

        /*
         * If the all-visible page is all-frozen but not marked as such yet,
         * mark it as all-frozen.  Note that all_frozen is only valid if
         * all_visible is true, so we must check both.
         */
        else if (all_visible_according_to_vm && prunestate.all_visible &&
                 prunestate.all_frozen &&
                 !VM_ALL_FROZEN(vacrel->rel, blkno, &vmbuffer))
        {
            /*
             * We can pass InvalidTransactionId as the cutoff XID here,
             * because setting the all-frozen bit doesn't cause recovery
             * conflicts.
             */
            visibilitymap_set(vacrel->rel, blkno, buf, InvalidXLogRecPtr,
                              vmbuffer, InvalidTransactionId,
                              VISIBILITYMAP_ALL_FROZEN);
        }

        /*
         * Final steps for block: drop super-exclusive lock, record free space
         * in the FSM
         */
        if (prunestate.has_lpdead_items && vacrel->do_index_vacuuming)
        {
            /*
             * Wait until lazy_vacuum_heap_rel() to save free space.  This
             * doesn't just save us some cycles; it also allows us to record
             * any additional free space that lazy_vacuum_heap_page() will
             * make available in cases where it's possible to truncate the
             * page's line pointer array.
             *
             * Note: It's not in fact 100% certain that we really will call
             * lazy_vacuum_heap_rel() -- lazy_vacuum() might yet opt to skip
             * index vacuuming (and so must skip heap vacuuming).  This is
             * deemed okay because it only happens in emergencies, or when
             * there is very little free space anyway. (Besides, we start
             * recording free space in the FSM once index vacuuming has been
             * abandoned.)
             *
             * Note: The one-pass (no indexes) case is only supposed to make
             * it this far when there were no LP_DEAD items during pruning.
             */
            Assert(vacrel->nindexes > 0);
            UnlockReleaseBuffer(buf);
        }
        else
        {
            Size        freespace = PageGetHeapFreeSpace(page);

            UnlockReleaseBuffer(buf);
            RecordPageWithFreeSpace(vacrel->rel, blkno, freespace);
        }
    }

    /* report that everything is now scanned */
    pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_SCANNED, blkno);

    /* Clear the block number information */
    vacrel->blkno = InvalidBlockNumber;

    /* now we can compute the new value for pg_class.reltuples */
    vacrel->new_live_tuples = vac_estimate_reltuples(vacrel->rel, nblocks,
                                                     vacrel->tupcount_pages,
                                                     vacrel->live_tuples);

    /*
     * Also compute the total number of surviving heap entries.  In the
     * (unlikely) scenario that new_live_tuples is -1, take it as zero.
     */
    vacrel->new_rel_tuples =
        Max(vacrel->new_live_tuples, 0) + vacrel->new_dead_tuples;

    /*
     * Release any remaining pin on visibility map page.
     */
    if (BufferIsValid(vmbuffer))
    {
        ReleaseBuffer(vmbuffer);
        vmbuffer = InvalidBuffer;
    }

    /* If any tuples need to be deleted, perform final vacuum cycle */
    if (dead_tuples->num_tuples > 0)
        lazy_vacuum(vacrel);

    /*
     * Vacuum the remainder of the Free Space Map.  We must do this whether or
     * not there were indexes, and whether or not we bypassed index vacuuming.
     */
    if (blkno > next_fsm_block_to_vacuum)
        FreeSpaceMapVacuumRange(vacrel->rel, next_fsm_block_to_vacuum, blkno);

    /* report all blocks vacuumed */
    pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, blkno);

    /* Do post-vacuum cleanup */
    if (vacrel->nindexes > 0 && vacrel->do_index_cleanup)
        lazy_cleanup_all_indexes(vacrel);

    /*
     * Free resources managed by lazy_space_alloc().  (We must end parallel
     * mode/free shared memory before updating index statistics.  We cannot
     * write while in parallel mode.)
     */
    lazy_space_free(vacrel);

    /* Update index statistics */
    if (vacrel->nindexes > 0 && vacrel->do_index_cleanup)
        update_index_statistics(vacrel);

    /*
     * If table has no indexes and at least one heap pages was vacuumed, make
     * log report that lazy_vacuum_heap_rel would've made had there been
     * indexes (having indexes implies using the two pass strategy).
     *
     * We deliberately don't do this in the case where there are indexes but
     * index vacuuming was bypassed.  We make a similar report at the point
     * that index vacuuming is bypassed, but that's actually quite different
     * in one important sense: it shows information about work we _haven't_
     * done.
     *
     * log_autovacuum output does things differently; it consistently presents
     * information about LP_DEAD items for the VACUUM as a whole.  We always
     * report on each round of index and heap vacuuming separately, though.
     */
    if (vacrel->nindexes == 0 && vacrel->lpdead_item_pages > 0)
        ereport(elevel,
                (errmsg("\"%s\": removed %lld dead item identifiers in %u pages",
                        vacrel->relname, (long long) vacrel->lpdead_items,
                        vacrel->lpdead_item_pages)));

    initStringInfo(&buf);
    appendStringInfo(&buf,
                     _("%lld dead row versions cannot be removed yet, oldest xmin: %u\n"),
                     (long long) vacrel->new_dead_tuples, vacrel->OldestXmin);
    appendStringInfo(&buf, ngettext("%u page removed.\n",
                                    "%u pages removed.\n",
                                    vacrel->pages_removed),
                     vacrel->pages_removed);
    appendStringInfo(&buf, ngettext("Skipped %u page due to buffer pins, ",
                                    "Skipped %u pages due to buffer pins, ",
                                    vacrel->pinskipped_pages),
                     vacrel->pinskipped_pages);
    appendStringInfo(&buf, ngettext("%u frozen page.\n",
                                    "%u frozen pages.\n",
                                    vacrel->frozenskipped_pages),
                     vacrel->frozenskipped_pages);
    appendStringInfo(&buf, _("%s."), pg_rusage_show(&ru0));

    ereport(elevel,
            (errmsg("\"%s\": found %lld removable, %lld nonremovable row versions in %u out of %u pages",
                    vacrel->relname,
                    (long long) vacrel->tuples_deleted,
                    (long long) vacrel->num_tuples, vacrel->scanned_pages,
                    nblocks),
             errdetail_internal("%s", buf.data)));
    pfree(buf.data);
}

lazy_scan_prune()

static void lazy_scan_prune ( LVRelState *  vacrel, Buffer  buf, BlockNumber  blkno, Page  page, GlobalVisState *  vistest, LVPagePruneState *  prunestate  )

static

Definition at line 1706 of file vacuumlazy.c.

References LVPagePruneState::all_frozen, LVPagePruneState::all_visible, Assert, LVRelState::dead_tuples, elog, END_CRIT_SECTION, ERROR, FirstOffsetNumber, LVRelState::FreezeLimit, FrozenTransactionId, LVPagePruneState::has_lpdead_items, LVPagePruneState::hastup, heap_execute_freeze_tuple(), heap_page_is_all_visible(), heap_page_prune(), heap_prepare_freeze_tuple(), HEAPTUPLE_DEAD, HEAPTUPLE_DELETE_IN_PROGRESS, HEAPTUPLE_INSERT_IN_PROGRESS, HEAPTUPLE_LIVE, HEAPTUPLE_RECENTLY_DEAD, HeapTupleHeaderGetXmin, HeapTupleHeaderXminCommitted, HeapTupleSatisfiesVacuum(), i, InvalidOffsetNumber, InvalidTransactionId, ItemIdGetLength, ItemIdIsDead, ItemIdIsNormal, ItemIdIsRedirected, ItemIdIsUsed, ItemPointerSet, ItemPointerSetBlockNumber, ItemPointerSetOffsetNumber, LVDeadTuples::itemptrs, LVRelState::live_tuples, log_heap_freeze(), LVRelState::lpdead_item_pages, LVRelState::lpdead_items, MarkBufferDirty(), LVDeadTuples::max_tuples, MaxHeapTuplesPerPage, LVRelState::MultiXactCutoff, LVRelState::new_dead_tuples, LVDeadTuples::num_tuples, LVRelState::num_tuples, LVRelState::offnum, xl_heap_freeze_tuple::offset, OffsetNumberNext, LVRelState::OldestXmin, PageGetItem, PageGetItemId, PageGetMaxOffsetNumber, PageSetLSN, pgstat_progress_update_param(), PROGRESS_VACUUM_NUM_DEAD_TUPLES, LVRelState::rel, RelationGetRelid, RelationNeedsWAL, LVRelState::relfrozenxid, LVRelState::relminmxid, START_CRIT_SECTION, HeapTupleData::t_data, HeapTupleData::t_len, HeapTupleData::t_self, HeapTupleData::t_tableOid, TransactionIdFollows(), TransactionIdPrecedes(), LVRelState::tuples_deleted, unlikely, and LVPagePruneState::visibility_cutoff_xid.

Referenced by lazy_scan_heap().

{
    Relation    rel = vacrel->rel;
    OffsetNumber offnum,
                maxoff;
    ItemId      itemid;
    HeapTupleData tuple;
    HTSV_Result res;
    int         tuples_deleted,
                lpdead_items,
                new_dead_tuples,
                num_tuples,
                live_tuples;
    int         nfrozen;
    OffsetNumber deadoffsets[MaxHeapTuplesPerPage];
    xl_heap_freeze_tuple frozen[MaxHeapTuplesPerPage];

    maxoff = PageGetMaxOffsetNumber(page);

retry:

    /* Initialize (or reset) page-level counters */
    tuples_deleted = 0;
    lpdead_items = 0;
    new_dead_tuples = 0;
    num_tuples = 0;
    live_tuples = 0;

    /*
     * Prune all HOT-update chains in this page.
     *
     * We count tuples removed by the pruning step as tuples_deleted.  Its
     * final value can be thought of as the number of tuples that have been
     * deleted from the table.  It should not be confused with lpdead_items;
     * lpdead_items's final value can be thought of as the number of tuples
     * that were deleted from indexes.
     */
    tuples_deleted = heap_page_prune(rel, buf, vistest,
                                     InvalidTransactionId, 0, false,
                                     &vacrel->offnum);

    /*
     * Now scan the page to collect LP_DEAD items and check for tuples
     * requiring freezing among remaining tuples with storage
     */
    prunestate->hastup = false;
    prunestate->has_lpdead_items = false;
    prunestate->all_visible = true;
    prunestate->all_frozen = true;
    prunestate->visibility_cutoff_xid = InvalidTransactionId;
    nfrozen = 0;

    for (offnum = FirstOffsetNumber;
         offnum <= maxoff;
         offnum = OffsetNumberNext(offnum))
    {
        bool        tuple_totally_frozen;

        /*
         * Set the offset number so that we can display it along with any
         * error that occurred while processing this tuple.
         */
        vacrel->offnum = offnum;
        itemid = PageGetItemId(page, offnum);

        if (!ItemIdIsUsed(itemid))
            continue;

        /* Redirect items mustn't be touched */
        if (ItemIdIsRedirected(itemid))
        {
            prunestate->hastup = true;  /* page won't be truncatable */
            continue;
        }

        /*
         * LP_DEAD items are processed outside of the loop.
         *
         * Note that we deliberately don't set hastup=true in the case of an
         * LP_DEAD item here, which is not how lazy_check_needs_freeze() or
         * count_nondeletable_pages() do it -- they only consider pages empty
         * when they only have LP_UNUSED items, which is important for
         * correctness.
         *
         * Our assumption is that any LP_DEAD items we encounter here will
         * become LP_UNUSED inside lazy_vacuum_heap_page() before we actually
         * call count_nondeletable_pages().  In any case our opinion of
         * whether or not a page 'hastup' (which is how our caller sets its
         * vacrel->nonempty_pages value) is inherently race-prone.  It must be
         * treated as advisory/unreliable, so we might as well be slightly
         * optimistic.
         */
        if (ItemIdIsDead(itemid))
        {
            deadoffsets[lpdead_items++] = offnum;
            prunestate->all_visible = false;
            prunestate->has_lpdead_items = true;
            continue;
        }

        Assert(ItemIdIsNormal(itemid));

        ItemPointerSet(&(tuple.t_self), blkno, offnum);
        tuple.t_data = (HeapTupleHeader) PageGetItem(page, itemid);
        tuple.t_len = ItemIdGetLength(itemid);
        tuple.t_tableOid = RelationGetRelid(rel);

        /*
         * DEAD tuples are almost always pruned into LP_DEAD line pointers by
         * heap_page_prune(), but it's possible that the tuple state changed
         * since heap_page_prune() looked.  Handle that here by restarting.
         * (See comments at the top of function for a full explanation.)
         */
        res = HeapTupleSatisfiesVacuum(&tuple, vacrel->OldestXmin, buf);

        if (unlikely(res == HEAPTUPLE_DEAD))
            goto retry;

        /*
         * The criteria for counting a tuple as live in this block need to
         * match what analyze.c's acquire_sample_rows() does, otherwise VACUUM
         * and ANALYZE may produce wildly different reltuples values, e.g.
         * when there are many recently-dead tuples.
         *
         * The logic here is a bit simpler than acquire_sample_rows(), as
         * VACUUM can't run inside a transaction block, which makes some cases
         * impossible (e.g. in-progress insert from the same transaction).
         *
         * We treat LP_DEAD items a little differently, too -- we don't count
         * them as dead_tuples at all (we only consider new_dead_tuples).  The
         * outcome is no different because we assume that any LP_DEAD items we
         * encounter here will become LP_UNUSED inside lazy_vacuum_heap_page()
         * before we report anything to the stats collector. (Cases where we
         * bypass index vacuuming will violate our assumption, but the overall
         * impact of that should be negligible.)
         */
        switch (res)
        {
            case HEAPTUPLE_LIVE:

                /*
                 * Count it as live.  Not only is this natural, but it's also
                 * what acquire_sample_rows() does.
                 */
                live_tuples++;

                /*
                 * Is the tuple definitely visible to all transactions?
                 *
                 * NB: Like with per-tuple hint bits, we can't set the
                 * PD_ALL_VISIBLE flag if the inserter committed
                 * asynchronously. See SetHintBits for more info. Check that
                 * the tuple is hinted xmin-committed because of that.
                 */
                if (prunestate->all_visible)
                {
                    TransactionId xmin;

                    if (!HeapTupleHeaderXminCommitted(tuple.t_data))
                    {
                        prunestate->all_visible = false;
                        break;
                    }

                    /*
                     * The inserter definitely committed. But is it old enough
                     * that everyone sees it as committed?
                     */
                    xmin = HeapTupleHeaderGetXmin(tuple.t_data);
                    if (!TransactionIdPrecedes(xmin, vacrel->OldestXmin))
                    {
                        prunestate->all_visible = false;
                        break;
                    }

                    /* Track newest xmin on page. */
                    if (TransactionIdFollows(xmin, prunestate->visibility_cutoff_xid))
                        prunestate->visibility_cutoff_xid = xmin;
                }
                break;
            case HEAPTUPLE_RECENTLY_DEAD:

                /*
                 * If tuple is recently deleted then we must not remove it
                 * from relation.  (We only remove items that are LP_DEAD from
                 * pruning.)
                 */
                new_dead_tuples++;
                prunestate->all_visible = false;
                break;
            case HEAPTUPLE_INSERT_IN_PROGRESS:

                /*
                 * We do not count these rows as live, because we expect the
                 * inserting transaction to update the counters at commit, and
                 * we assume that will happen only after we report our
                 * results.  This assumption is a bit shaky, but it is what
                 * acquire_sample_rows() does, so be consistent.
                 */
                prunestate->all_visible = false;
                break;
            case HEAPTUPLE_DELETE_IN_PROGRESS:
                /* This is an expected case during concurrent vacuum */
                prunestate->all_visible = false;

                /*
                 * Count such rows as live.  As above, we assume the deleting
                 * transaction will commit and update the counters after we
                 * report.
                 */
                live_tuples++;
                break;
            default:
                elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result");
                break;
        }

        /*
         * Non-removable tuple (i.e. tuple with storage).
         *
         * Check tuple left behind after pruning to see if needs to be frozen
         * now.
         */
        num_tuples++;
        prunestate->hastup = true;
        if (heap_prepare_freeze_tuple(tuple.t_data,
                                      vacrel->relfrozenxid,
                                      vacrel->relminmxid,
                                      vacrel->FreezeLimit,
                                      vacrel->MultiXactCutoff,
                                      &frozen[nfrozen],
                                      &tuple_totally_frozen))
        {
            /* Will execute freeze below */
            frozen[nfrozen++].offset = offnum;
        }

        /*
         * If tuple is not frozen (and not about to become frozen) then caller
         * had better not go on to set this page's VM bit
         */
        if (!tuple_totally_frozen)
            prunestate->all_frozen = false;
    }

    /*
     * We have now divided every item on the page into either an LP_DEAD item
     * that will need to be vacuumed in indexes later, or a LP_NORMAL tuple
     * that remains and needs to be considered for freezing now (LP_UNUSED and
     * LP_REDIRECT items also remain, but are of no further interest to us).
     */
    vacrel->offnum = InvalidOffsetNumber;

    /*
     * Consider the need to freeze any items with tuple storage from the page
     * first (arbitrary)
     */
    if (nfrozen > 0)
    {
        Assert(prunestate->hastup);

        /*
         * At least one tuple with storage needs to be frozen -- execute that
         * now.
         *
         * If we need to freeze any tuples we'll mark the buffer dirty, and
         * write a WAL record recording the changes.  We must log the changes
         * to be crash-safe against future truncation of CLOG.
         */
        START_CRIT_SECTION();

        MarkBufferDirty(buf);

        /* execute collected freezes */
        for (int i = 0; i < nfrozen; i++)
        {
            HeapTupleHeader htup;

            itemid = PageGetItemId(page, frozen[i].offset);
            htup = (HeapTupleHeader) PageGetItem(page, itemid);

            heap_execute_freeze_tuple(htup, &frozen[i]);
        }

        /* Now WAL-log freezing if necessary */
        if (RelationNeedsWAL(vacrel->rel))
        {
            XLogRecPtr  recptr;

            recptr = log_heap_freeze(vacrel->rel, buf, vacrel->FreezeLimit,
                                     frozen, nfrozen);
            PageSetLSN(page, recptr);
        }

        END_CRIT_SECTION();
    }

    /*
     * The second pass over the heap can also set visibility map bits, using
     * the same approach.  This is important when the table frequently has a
     * few old LP_DEAD items on each page by the time we get to it (typically
     * because past opportunistic pruning operations freed some non-HOT
     * tuples).
     *
     * VACUUM will call heap_page_is_all_visible() during the second pass over
     * the heap to determine all_visible and all_frozen for the page -- this
     * is a specialized version of the logic from this function.  Now that
     * we've finished pruning and freezing, make sure that we're in total
     * agreement with heap_page_is_all_visible() using an assertion.
     */
#ifdef USE_ASSERT_CHECKING
    /* Note that all_frozen value does not matter when !all_visible */
    if (prunestate->all_visible)
    {
        TransactionId cutoff;
        bool        all_frozen;

        if (!heap_page_is_all_visible(vacrel, buf, &cutoff, &all_frozen))
            Assert(false);

        Assert(lpdead_items == 0);
        Assert(prunestate->all_frozen == all_frozen);

        /*
         * It's possible that we froze tuples and made the page's XID cutoff
         * (for recovery conflict purposes) FrozenTransactionId.  This is okay
         * because visibility_cutoff_xid will be logged by our caller in a
         * moment.
         */
        Assert(cutoff == FrozenTransactionId ||
               cutoff == prunestate->visibility_cutoff_xid);
    }
#endif

    /*
     * Now save details of the LP_DEAD items from the page in the dead_tuples
     * array.  Also record that page has dead items in per-page prunestate.
     */
    if (lpdead_items > 0)
    {
        LVDeadTuples *dead_tuples = vacrel->dead_tuples;
        ItemPointerData tmp;

        Assert(!prunestate->all_visible);
        Assert(prunestate->has_lpdead_items);

        vacrel->lpdead_item_pages++;

        ItemPointerSetBlockNumber(&tmp, blkno);

        for (int i = 0; i < lpdead_items; i++)
        {
            ItemPointerSetOffsetNumber(&tmp, deadoffsets[i]);
            dead_tuples->itemptrs[dead_tuples->num_tuples++] = tmp;
        }

        Assert(dead_tuples->num_tuples <= dead_tuples->max_tuples);
        pgstat_progress_update_param(PROGRESS_VACUUM_NUM_DEAD_TUPLES,
                                     dead_tuples->num_tuples);
    }

    /* Finally, add page-local counts to whole-VACUUM counts */
    vacrel->tuples_deleted += tuples_deleted;
    vacrel->lpdead_items += lpdead_items;
    vacrel->new_dead_tuples += new_dead_tuples;
    vacrel->num_tuples += num_tuples;
    vacrel->live_tuples += live_tuples;
}

lazy_space_alloc()

static void lazy_space_alloc ( LVRelState *  vacrel, int  nworkers, BlockNumber  relblocks  )

static

Definition at line 3489 of file vacuumlazy.c.

References begin_parallel_vacuum(), compute_max_dead_tuples(), LVRelState::dead_tuples, LVRelState::do_index_vacuuming, ereport, errmsg(), LVRelState::lps, LVDeadTuples::max_tuples, LVRelState::nindexes, LVDeadTuples::num_tuples, palloc(), ParallelVacuumIsActive, LVRelState::rel, RelationUsesLocalBuffers, LVRelState::relname, SizeOfDeadTuples, and WARNING.

Referenced by lazy_scan_heap().

{
    LVDeadTuples *dead_tuples;
    long        maxtuples;

    /*
     * Initialize state for a parallel vacuum.  As of now, only one worker can
     * be used for an index, so we invoke parallelism only if there are at
     * least two indexes on a table.
     */
    if (nworkers >= 0 && vacrel->nindexes > 1 && vacrel->do_index_vacuuming)
    {
        /*
         * Since parallel workers cannot access data in temporary tables, we
         * can't perform parallel vacuum on them.
         */
        if (RelationUsesLocalBuffers(vacrel->rel))
        {
            /*
             * Give warning only if the user explicitly tries to perform a
             * parallel vacuum on the temporary table.
             */
            if (nworkers > 0)
                ereport(WARNING,
                        (errmsg("disabling parallel option of vacuum on \"%s\" --- cannot vacuum temporary tables in parallel",
                                vacrel->relname)));
        }
        else
            vacrel->lps = begin_parallel_vacuum(vacrel, nblocks, nworkers);

        /* If parallel mode started, we're done */
        if (ParallelVacuumIsActive(vacrel))
            return;
    }

    maxtuples = compute_max_dead_tuples(nblocks, vacrel->nindexes > 0);

    dead_tuples = (LVDeadTuples *) palloc(SizeOfDeadTuples(maxtuples));
    dead_tuples->num_tuples = 0;
    dead_tuples->max_tuples = (int) maxtuples;

    vacrel->dead_tuples = dead_tuples;
}

lazy_space_free()

static void lazy_space_free ( LVRelState *  vacrel )

static

Definition at line 3537 of file vacuumlazy.c.

References end_parallel_vacuum(), and ParallelVacuumIsActive.

Referenced by lazy_scan_heap().

{
    if (!ParallelVacuumIsActive(vacrel))
        return;

    /*
     * End parallel mode before updating index statistics as we cannot write
     * during parallel mode.
     */
    end_parallel_vacuum(vacrel);
}

lazy_tid_reaped()

static bool lazy_tid_reaped ( ItemPointer  itemptr, void *  state  )

static

Definition at line 3557 of file vacuumlazy.c.

References itemptr_encode(), LVDeadTuples::itemptrs, LVDeadTuples::num_tuples, and vac_cmp_itemptr().

Referenced by lazy_vacuum_one_index().

{
    LVDeadTuples *dead_tuples = (LVDeadTuples *) state;
    int64       litem,
                ritem,
                item;
    ItemPointer res;

    litem = itemptr_encode(&dead_tuples->itemptrs[0]);
    ritem = itemptr_encode(&dead_tuples->itemptrs[dead_tuples->num_tuples - 1]);
    item = itemptr_encode(itemptr);

    /*
     * Doing a simple bound check before bsearch() is useful to avoid the
     * extra cost of bsearch(), especially if dead tuples on the heap are
     * concentrated in a certain range.  Since this function is called for
     * every index tuple, it pays to be really fast.
     */
    if (item < litem || item > ritem)
        return false;

    res = (ItemPointer) bsearch((void *) itemptr,
                                (void *) dead_tuples->itemptrs,
                                dead_tuples->num_tuples,
                                sizeof(ItemPointerData),
                                vac_cmp_itemptr);

    return (res != NULL);
}

lazy_truncate_heap()

static void lazy_truncate_heap ( LVRelState *  vacrel )

static

Definition at line 3186 of file vacuumlazy.c.

References AccessExclusiveLock, LVRelState::blkno, CHECK_FOR_INTERRUPTS, ConditionalLockRelation(), count_nondeletable_pages(), elevel, ereport, errdetail_internal(), errmsg(), MyLatch, LVRelState::nonempty_pages, LVRelState::pages_removed, pg_rusage_init(), pg_rusage_show(), pgstat_progress_update_param(), PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_TRUNCATE, LVRelState::rel, LVRelState::rel_pages, RelationGetNumberOfBlocks, RelationTruncate(), LVRelState::relname, ResetLatch(), UnlockRelation(), VACUUM_TRUNCATE_LOCK_TIMEOUT, VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL, WAIT_EVENT_VACUUM_TRUNCATE, WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by heap_vacuum_rel().

{
    BlockNumber old_rel_pages = vacrel->rel_pages;
    BlockNumber new_rel_pages;
    bool        lock_waiter_detected;
    int         lock_retry;

    /* Report that we are now truncating */
    pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
                                 PROGRESS_VACUUM_PHASE_TRUNCATE);

    /*
     * Loop until no more truncating can be done.
     */
    do
    {
        PGRUsage    ru0;

        pg_rusage_init(&ru0);

        /*
         * We need full exclusive lock on the relation in order to do
         * truncation. If we can't get it, give up rather than waiting --- we
         * don't want to block other backends, and we don't want to deadlock
         * (which is quite possible considering we already hold a lower-grade
         * lock).
         */
        lock_waiter_detected = false;
        lock_retry = 0;
        while (true)
        {
            if (ConditionalLockRelation(vacrel->rel, AccessExclusiveLock))
                break;

            /*
             * Check for interrupts while trying to (re-)acquire the exclusive
             * lock.
             */
            CHECK_FOR_INTERRUPTS();

            if (++lock_retry > (VACUUM_TRUNCATE_LOCK_TIMEOUT /
                                VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL))
            {
                /*
                 * We failed to establish the lock in the specified number of
                 * retries. This means we give up truncating.
                 */
                ereport(elevel,
                        (errmsg("\"%s\": stopping truncate due to conflicting lock request",
                                vacrel->relname)));
                return;
            }

            (void) WaitLatch(MyLatch,
                             WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
                             VACUUM_TRUNCATE_LOCK_WAIT_INTERVAL,
                             WAIT_EVENT_VACUUM_TRUNCATE);
            ResetLatch(MyLatch);
        }

        /*
         * Now that we have exclusive lock, look to see if the rel has grown
         * whilst we were vacuuming with non-exclusive lock.  If so, give up;
         * the newly added pages presumably contain non-deletable tuples.
         */
        new_rel_pages = RelationGetNumberOfBlocks(vacrel->rel);
        if (new_rel_pages != old_rel_pages)
        {
            /*
             * Note: we intentionally don't update vacrel->rel_pages with the
             * new rel size here.  If we did, it would amount to assuming that
             * the new pages are empty, which is unlikely. Leaving the numbers
             * alone amounts to assuming that the new pages have the same
             * tuple density as existing ones, which is less unlikely.
             */
            UnlockRelation(vacrel->rel, AccessExclusiveLock);
            return;
        }

        /*
         * Scan backwards from the end to verify that the end pages actually
         * contain no tuples.  This is *necessary*, not optional, because
         * other backends could have added tuples to these pages whilst we
         * were vacuuming.
         */
        new_rel_pages = count_nondeletable_pages(vacrel, &lock_waiter_detected);
        vacrel->blkno = new_rel_pages;

        if (new_rel_pages >= old_rel_pages)
        {
            /* can't do anything after all */
            UnlockRelation(vacrel->rel, AccessExclusiveLock);
            return;
        }

        /*
         * Okay to truncate.
         */
        RelationTruncate(vacrel->rel, new_rel_pages);

        /*
         * We can release the exclusive lock as soon as we have truncated.
         * Other backends can't safely access the relation until they have
         * processed the smgr invalidation that smgrtruncate sent out ... but
         * that should happen as part of standard invalidation processing once
         * they acquire lock on the relation.
         */
        UnlockRelation(vacrel->rel, AccessExclusiveLock);

        /*
         * Update statistics.  Here, it *is* correct to adjust rel_pages
         * without also touching reltuples, since the tuple count wasn't
         * changed by the truncation.
         */
        vacrel->pages_removed += old_rel_pages - new_rel_pages;
        vacrel->rel_pages = new_rel_pages;

        ereport(elevel,
                (errmsg("\"%s\": truncated %u to %u pages",
                        vacrel->relname,
                        old_rel_pages, new_rel_pages),
                 errdetail_internal("%s",
                                    pg_rusage_show(&ru0))));
        old_rel_pages = new_rel_pages;
    } while (new_rel_pages > vacrel->nonempty_pages && lock_waiter_detected);
}

lazy_vacuum()

static void lazy_vacuum ( LVRelState *  vacrel )

static

Definition at line 2094 of file vacuumlazy.c.

References Assert, BYPASS_THRESHOLD_PAGES, LVRelState::consider_bypass_optimization, LVRelState::dead_tuples, LVRelState::do_index_cleanup, LVRelState::do_index_vacuuming, elevel, ereport, errmsg(), LVRelState::failsafe_active, IsParallelWorker, lazy_vacuum_all_indexes(), lazy_vacuum_heap_rel(), LVRelState::lpdead_item_pages, LVRelState::lpdead_items, MAXDEADTUPLES, LVRelState::nindexes, LVRelState::num_index_scans, LVDeadTuples::num_tuples, LVRelState::rel_pages, and LVRelState::relname.

Referenced by lazy_scan_heap().

{
    bool        bypass;

    /* Should not end up here with no indexes */
    Assert(vacrel->nindexes > 0);
    Assert(!IsParallelWorker());
    Assert(vacrel->lpdead_item_pages > 0);

    if (!vacrel->do_index_vacuuming)
    {
        Assert(!vacrel->do_index_cleanup);
        vacrel->dead_tuples->num_tuples = 0;
        return;
    }

    /*
     * Consider bypassing index vacuuming (and heap vacuuming) entirely.
     *
     * We currently only do this in cases where the number of LP_DEAD items
     * for the entire VACUUM operation is close to zero.  This avoids sharp
     * discontinuities in the duration and overhead of successive VACUUM
     * operations that run against the same table with a fixed workload.
     * Ideally, successive VACUUM operations will behave as if there are
     * exactly zero LP_DEAD items in cases where there are close to zero.
     *
     * This is likely to be helpful with a table that is continually affected
     * by UPDATEs that can mostly apply the HOT optimization, but occasionally
     * have small aberrations that lead to just a few heap pages retaining
     * only one or two LP_DEAD items.  This is pretty common; even when the
     * DBA goes out of their way to make UPDATEs use HOT, it is practically
     * impossible to predict whether HOT will be applied in 100% of cases.
     * It's far easier to ensure that 99%+ of all UPDATEs against a table use
     * HOT through careful tuning.
     */
    bypass = false;
    if (vacrel->consider_bypass_optimization && vacrel->rel_pages > 0)
    {
        BlockNumber threshold;

        Assert(vacrel->num_index_scans == 0);
        Assert(vacrel->lpdead_items == vacrel->dead_tuples->num_tuples);
        Assert(vacrel->do_index_vacuuming);
        Assert(vacrel->do_index_cleanup);

        /*
         * This crossover point at which we'll start to do index vacuuming is
         * expressed as a percentage of the total number of heap pages in the
         * table that are known to have at least one LP_DEAD item.  This is
         * much more important than the total number of LP_DEAD items, since
         * it's a proxy for the number of heap pages whose visibility map bits
         * cannot be set on account of bypassing index and heap vacuuming.
         *
         * We apply one further precautionary test: the space currently used
         * to store the TIDs (TIDs that now all point to LP_DEAD items) must
         * not exceed 32MB.  This limits the risk that we will bypass index
         * vacuuming again and again until eventually there is a VACUUM whose
         * dead_tuples space is not CPU cache resident.
         *
         * We don't take any special steps to remember the LP_DEAD items (such
         * as counting them in new_dead_tuples report to the stats collector)
         * when the optimization is applied.  Though the accounting used in
         * analyze.c's acquire_sample_rows() will recognize the same LP_DEAD
         * items as dead rows in its own stats collector report, that's okay.
         * The discrepancy should be negligible.  If this optimization is ever
         * expanded to cover more cases then this may need to be reconsidered.
         */
        threshold = (double) vacrel->rel_pages * BYPASS_THRESHOLD_PAGES;
        bypass = (vacrel->lpdead_item_pages < threshold &&
                  vacrel->lpdead_items < MAXDEADTUPLES(32L * 1024L * 1024L));
    }

    if (bypass)
    {
        /*
         * There are almost zero TIDs.  Behave as if there were precisely
         * zero: bypass index vacuuming, but do index cleanup.
         *
         * We expect that the ongoing VACUUM operation will finish very
         * quickly, so there is no point in considering speeding up as a
         * failsafe against wraparound failure. (Index cleanup is expected to
         * finish very quickly in cases where there were no ambulkdelete()
         * calls.)
         */
        vacrel->do_index_vacuuming = false;
        ereport(elevel,
                (errmsg("\"%s\": index scan bypassed: %u pages from table (%.2f%% of total) have %lld dead item identifiers",
                        vacrel->relname, vacrel->lpdead_item_pages,
                        100.0 * vacrel->lpdead_item_pages / vacrel->rel_pages,
                        (long long) vacrel->lpdead_items)));
    }
    else if (lazy_vacuum_all_indexes(vacrel))
    {
        /*
         * We successfully completed a round of index vacuuming.  Do related
         * heap vacuuming now.
         */
        lazy_vacuum_heap_rel(vacrel);
    }
    else
    {
        /*
         * Failsafe case.
         *
         * we attempted index vacuuming, but didn't finish a full round/full
         * index scan.  This happens when relfrozenxid or relminmxid is too
         * far in the past.
         *
         * From this point on the VACUUM operation will do no further index
         * vacuuming or heap vacuuming.  This VACUUM operation won't end up
         * back here again.
         */
        Assert(vacrel->failsafe_active);
    }

    /*
     * Forget the LP_DEAD items that we just vacuumed (or just decided to not
     * vacuum)
     */
    vacrel->dead_tuples->num_tuples = 0;
}

lazy_vacuum_all_indexes()

static bool lazy_vacuum_all_indexes ( LVRelState *  vacrel )

static

Definition at line 2225 of file vacuumlazy.c.

References Assert, LVRelState::dead_tuples, LVRelState::do_index_cleanup, LVRelState::do_index_vacuuming, do_parallel_lazy_vacuum_all_indexes(), LVRelState::failsafe_active, idx(), LVRelState::indrels, LVRelState::indstats, IsParallelWorker, lazy_check_wraparound_failsafe(), lazy_vacuum_one_index(), LVRelState::lpdead_items, MultiXactIdIsValid, LVRelState::nindexes, LVRelState::num_index_scans, LVDeadTuples::num_tuples, LVRelState::old_live_tuples, ParallelVacuumIsActive, pgstat_progress_update_param(), PROGRESS_VACUUM_NUM_INDEX_VACUUMS, PROGRESS_VACUUM_PHASE, PROGRESS_VACUUM_PHASE_VACUUM_INDEX, LVRelState::relfrozenxid, LVRelState::relminmxid, and TransactionIdIsNormal.

Referenced by lazy_vacuum().

{
    bool        allindexes = true;

    Assert(!IsParallelWorker());
    Assert(vacrel->nindexes > 0);
    Assert(vacrel->do_index_vacuuming);
    Assert(vacrel->do_index_cleanup);
    Assert(TransactionIdIsNormal(vacrel->relfrozenxid));
    Assert(MultiXactIdIsValid(vacrel->relminmxid));

    /* Precheck for XID wraparound emergencies */
    if (lazy_check_wraparound_failsafe(vacrel))
    {
        /* Wraparound emergency -- don't even start an index scan */
        return false;
    }

    /* Report that we are now vacuuming indexes */
    pgstat_progress_update_param(PROGRESS_VACUUM_PHASE,
                                 PROGRESS_VACUUM_PHASE_VACUUM_INDEX);

    if (!ParallelVacuumIsActive(vacrel))
    {
        for (int idx = 0; idx < vacrel->nindexes; idx++)
        {
            Relation    indrel = vacrel->indrels[idx];
            IndexBulkDeleteResult *istat = vacrel->indstats[idx];

            vacrel->indstats[idx] =
                lazy_vacuum_one_index(indrel, istat, vacrel->old_live_tuples,
                                      vacrel);

            if (lazy_check_wraparound_failsafe(vacrel))
            {
                /* Wraparound emergency -- end current index scan */
                allindexes = false;
                break;
            }
        }
    }
    else
    {
        /* Outsource everything to parallel variant */
        do_parallel_lazy_vacuum_all_indexes(vacrel);

        /*
         * Do a postcheck to consider applying wraparound failsafe now.  Note
         * that parallel VACUUM only gets the precheck and this postcheck.
         */
        if (lazy_check_wraparound_failsafe(vacrel))
            allindexes = false;
    }

    /*
     * We delete all LP_DEAD items from the first heap pass in all indexes on
     * each call here (except calls where we choose to do the failsafe). This
     * makes the next call to lazy_vacuum_heap_rel() safe (except in the event
     * of the failsafe triggering, which prevents the next call from taking
     * place).
     */
    Assert(vacrel->num_index_scans > 0 ||
           vacrel->dead_tuples->num_tuples == vacrel->lpdead_items);
    Assert(allindexes || vacrel->failsafe_active);

    /*
     * Increase and report the number of index scans.
     *
     * We deliberately include the case where we started a round of bulk
     * deletes that we weren't able to finish due to the failsafe triggering.
     */
    vacrel->num_index_scans++;
    pgstat_progress_update_param(PROGRESS_VACUUM_NUM_INDEX_VACUUMS,
                                 vacrel->num_index_scans);

    return allindexes;
}

lazy_vacuum_heap_page()

static int lazy_vacuum_heap_page ( LVRelState *  vacrel, BlockNumber  blkno, Buffer  buffer, int  tupindex, Buffer *  vmbuffer  )

static

Definition at line 2418 of file vacuumlazy.c.

References Assert, BufferGetPage, BufferIsValid, LVRelState::dead_tuples, LVRelState::do_index_vacuuming, END_CRIT_SECTION, heap_page_is_all_visible(), InvalidOffsetNumber, InvalidXLogRecPtr, ItemIdHasStorage, ItemIdIsDead, ItemIdSetUnused, ItemPointerGetBlockNumber, ItemPointerGetOffsetNumber, LVDeadTuples::itemptrs, MarkBufferDirty(), MaxHeapTuplesPerPage, LVRelState::nindexes, LVDeadTuples::num_tuples, xl_heap_vacuum::nunused, PageGetItemId, PageIsAllVisible, PageSetAllVisible, PageSetLSN, PageTruncateLinePointerArray(), pgstat_progress_update_param(), PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, REGBUF_STANDARD, LVRelState::rel, RelationNeedsWAL, restore_vacuum_error_info(), SizeOfHeapVacuum, START_CRIT_SECTION, update_vacuum_error_info(), VACUUM_ERRCB_PHASE_VACUUM_HEAP, VISIBILITYMAP_ALL_FROZEN, VISIBILITYMAP_ALL_VISIBLE, visibilitymap_get_status(), visibilitymap_set(), XLOG_HEAP2_VACUUM, XLogBeginInsert(), XLogInsert(), XLogRegisterBufData(), XLogRegisterBuffer(), and XLogRegisterData().

Referenced by lazy_scan_heap(), and lazy_vacuum_heap_rel().

{
    LVDeadTuples *dead_tuples = vacrel->dead_tuples;
    Page        page = BufferGetPage(buffer);
    OffsetNumber unused[MaxHeapTuplesPerPage];
    int         uncnt = 0;
    TransactionId visibility_cutoff_xid;
    bool        all_frozen;
    LVSavedErrInfo saved_err_info;

    Assert(vacrel->nindexes == 0 || vacrel->do_index_vacuuming);

    pgstat_progress_update_param(PROGRESS_VACUUM_HEAP_BLKS_VACUUMED, blkno);

    /* Update error traceback information */
    update_vacuum_error_info(vacrel, &saved_err_info,
                             VACUUM_ERRCB_PHASE_VACUUM_HEAP, blkno,
                             InvalidOffsetNumber);

    START_CRIT_SECTION();

    for (; tupindex < dead_tuples->num_tuples; tupindex++)
    {
        BlockNumber tblk;
        OffsetNumber toff;
        ItemId      itemid;

        tblk = ItemPointerGetBlockNumber(&dead_tuples->itemptrs[tupindex]);
        if (tblk != blkno)
            break;              /* past end of tuples for this block */
        toff = ItemPointerGetOffsetNumber(&dead_tuples->itemptrs[tupindex]);
        itemid = PageGetItemId(page, toff);

        Assert(ItemIdIsDead(itemid) && !ItemIdHasStorage(itemid));
        ItemIdSetUnused(itemid);
        unused[uncnt++] = toff;
    }

    Assert(uncnt > 0);

    /* Attempt to truncate line pointer array now */
    PageTruncateLinePointerArray(page);

    /*
     * Mark buffer dirty before we write WAL.
     */
    MarkBufferDirty(buffer);

    /* XLOG stuff */
    if (RelationNeedsWAL(vacrel->rel))
    {
        xl_heap_vacuum xlrec;
        XLogRecPtr  recptr;

        xlrec.nunused = uncnt;

        XLogBeginInsert();
        XLogRegisterData((char *) &xlrec, SizeOfHeapVacuum);

        XLogRegisterBuffer(0, buffer, REGBUF_STANDARD);
        XLogRegisterBufData(0, (char *) unused, uncnt * sizeof(OffsetNumber));

        recptr = XLogInsert(RM_HEAP2_ID, XLOG_HEAP2_VACUUM);

        PageSetLSN(page, recptr);
    }

    /*
     * End critical section, so we safely can do visibility tests (which
     * possibly need to perform IO and allocate memory!). If we crash now the
     * page (including the corresponding vm bit) might not be marked all
     * visible, but that's fine. A later vacuum will fix that.
     */
    END_CRIT_SECTION();

    /*
     * Now that we have removed the LD_DEAD items from the page, once again
     * check if the page has become all-visible.  The page is already marked
     * dirty, exclusively locked, and, if needed, a full page image has been
     * emitted.
     */
    if (heap_page_is_all_visible(vacrel, buffer, &visibility_cutoff_xid,
                                 &all_frozen))
        PageSetAllVisible(page);

    /*
     * All the changes to the heap page have been done. If the all-visible
     * flag is now set, also set the VM all-visible bit (and, if possible, the
     * all-frozen bit) unless this has already been done previously.
     */
    if (PageIsAllVisible(page))
    {
        uint8       flags = 0;
        uint8       vm_status = visibilitymap_get_status(vacrel->rel,
                                                         blkno, vmbuffer);

        /* Set the VM all-frozen bit to flag, if needed */
        if ((vm_status & VISIBILITYMAP_ALL_VISIBLE) == 0)
            flags |= VISIBILITYMAP_ALL_VISIBLE;
        if ((vm_status & VISIBILITYMAP_ALL_FROZEN) == 0 && all_frozen)
            flags |= VISIBILITYMAP_ALL_FROZEN;

        Assert(BufferIsValid(*vmbuffer));
        if (flags != 0)
            visibilitymap_set(vacrel->rel, blkno, buffer, InvalidXLogRecPtr,
                              *vmbuffer, visibility_cutoff_xid, flags);
    }

    /* Revert to the previous phase information for error traceback */
    restore_vacuum_error_info(vacrel, &saved_err_info);
    return tupindex;
}