repack.h File Reference

#include <signal.h>
#include "nodes/parsenodes.h"
#include "parser/parse_node.h"
#include "storage/lockdefs.h"
#include "utils/relcache.h"

Include dependency graph for repack.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	ClusterParams

Macros
#define	CLUOPT_VERBOSE   0x01 /* print progress info */
#define	CLUOPT_RECHECK   0x02 /* recheck relation state */
#define	CLUOPT_RECHECK_ISCLUSTERED
#define	CLUOPT_ANALYZE   0x08 /* do an ANALYZE */
#define	CLUOPT_CONCURRENT   0x10 /* allow concurrent data changes */

Typedefs
typedef struct ClusterParams	ClusterParams

Functions
void	ExecRepack (ParseState *pstate, RepackStmt *stmt, bool isTopLevel)
void	cluster_rel (RepackCommand command, Relation OldHeap, Oid indexOid, ClusterParams *params, bool isTopLevel)
void	check_index_is_clusterable (Relation OldHeap, Oid indexOid, LOCKMODE lockmode)
void	mark_index_clustered (Relation rel, Oid indexOid, bool is_internal)
Oid	make_new_heap (Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode)
void	finish_heap_swap (Oid OIDOldHeap, Oid OIDNewHeap, bool is_system_catalog, bool swap_toast_by_content, bool check_constraints, bool is_internal, bool reindex, TransactionId frozenXid, MultiXactId cutoffMulti, char newrelpersistence)
void	HandleRepackMessageInterrupt (void)
void	ProcessRepackMessages (void)
void	RepackWorkerMain (Datum main_arg)
bool	AmRepackWorker (void)

Variables
PGDLLIMPORT volatile sig_atomic_t	RepackMessagePending

Macro Definition Documentation

CLUOPT_ANALYZE

#define CLUOPT_ANALYZE   0x08 /* do an ANALYZE */

Definition at line 28 of file repack.h.

CLUOPT_CONCURRENT

#define CLUOPT_CONCURRENT   0x10 /* allow concurrent data changes */

Definition at line 29 of file repack.h.

CLUOPT_RECHECK

#define CLUOPT_RECHECK   0x02 /* recheck relation state */

Definition at line 26 of file repack.h.

CLUOPT_RECHECK_ISCLUSTERED

#define CLUOPT_RECHECK_ISCLUSTERED

Value:

                                         0x04 /* recheck relation state for
                                         * indisclustered */

Definition at line 27 of file repack.h.

{
    uint32      options;        /* bitmask of CLUOPT_* */
} ClusterParams;
 
extern PGDLLIMPORT volatile sig_atomic_t RepackMessagePending;
 
 
extern void ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel);
 
extern void cluster_rel(RepackCommand command, Relation OldHeap, Oid indexOid,
                        ClusterParams *params, bool isTopLevel);
extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid,
                                       LOCKMODE lockmode);
extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal);
 
extern Oid  make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod,
                          char relpersistence, LOCKMODE lockmode);
extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap,
                             bool is_system_catalog,
                             bool swap_toast_by_content,
                             bool check_constraints,
                             bool is_internal,
                             bool reindex,
                             TransactionId frozenXid,
                             MultiXactId cutoffMulti,
                             char newrelpersistence);
 
extern void HandleRepackMessageInterrupt(void);
extern void ProcessRepackMessages(void);
 
/* in repack_worker.c */
extern void RepackWorkerMain(Datum main_arg);
extern bool AmRepackWorker(void);
 
#endif                          /* REPACK_H */

CLUOPT_VERBOSE

#define CLUOPT_VERBOSE   0x01 /* print progress info */

Definition at line 25 of file repack.h.

Typedef Documentation

ClusterParams

typedef struct ClusterParams ClusterParams

Function Documentation

AmRepackWorker()

bool AmRepackWorker ( void  )

extern

Definition at line 183 of file repack_worker.c.

{
    return am_repack_worker;
}

References am_repack_worker.

Referenced by mq_putmessage().

check_index_is_clusterable()

void check_index_is_clusterable ( Relation  OldHeap, Oid  indexOid, LOCKMODE  lockmode  )

extern

Definition at line 751 of file repack.c.

{
    Relation    OldIndex;
 
    OldIndex = index_open(indexOid, lockmode);
 
    /*
     * Check that index is in fact an index on the given relation
     */
    if (OldIndex->rd_index == NULL ||
        OldIndex->rd_index->indrelid != RelationGetRelid(OldHeap))
        ereport(ERROR,
                (errcode(ERRCODE_WRONG_OBJECT_TYPE),
                 errmsg("\"%s\" is not an index for table \"%s\"",
                        RelationGetRelationName(OldIndex),
                        RelationGetRelationName(OldHeap))));
 
    /* Index AM must allow clustering */
    if (!OldIndex->rd_indam->amclusterable)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot cluster on index \"%s\" because access method does not support clustering",
                        RelationGetRelationName(OldIndex))));
 
    /*
     * Disallow clustering on incomplete indexes (those that might not index
     * every row of the relation).  We could relax this by making a separate
     * seqscan pass over the table to copy the missing rows, but that seems
     * expensive and tedious.
     */
    if (!heap_attisnull(OldIndex->rd_indextuple, Anum_pg_index_indpred, NULL))
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot cluster on partial index \"%s\"",
                        RelationGetRelationName(OldIndex))));
 
    /*
     * Disallow if index is left over from a failed CREATE INDEX CONCURRENTLY;
     * it might well not contain entries for every heap row, or might not even
     * be internally consistent.  (But note that we don't check indcheckxmin;
     * the worst consequence of following broken HOT chains would be that we
     * might put recently-dead tuples out-of-order in the new table, and there
     * is little harm in that.)
     */
    if (!OldIndex->rd_index->indisvalid)
        ereport(ERROR,
                (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                 errmsg("cannot cluster on invalid index \"%s\"",
                        RelationGetRelationName(OldIndex))));
 
    /* Drop relcache refcnt on OldIndex, but keep lock */
    index_close(OldIndex, NoLock);
}

References ereport, errcode(), errmsg, ERROR, fb(), heap_attisnull(), index_close(), index_open(), NoLock, RelationGetRelationName, and RelationGetRelid.

Referenced by ATExecClusterOn(), cluster_rel(), ExecRepack(), and process_single_relation().

cluster_rel()

void cluster_rel ( RepackCommand  command, Relation  OldHeap, Oid  indexOid, ClusterParams *  params, bool  isTopLevel  )

extern

Definition at line 502 of file repack.c.

{
    Oid         tableOid = RelationGetRelid(OldHeap);
    Relation    index;
    LOCKMODE    lmode;
    Oid         save_userid;
    int         save_sec_context;
    int         save_nestlevel;
    bool        verbose = ((params->options & CLUOPT_VERBOSE) != 0);
    bool        recheck = ((params->options & CLUOPT_RECHECK) != 0);
    bool        concurrent = ((params->options & CLUOPT_CONCURRENT) != 0);
    Oid         ident_idx = InvalidOid;
 
    /* Determine the lock mode to use. */
    lmode = RepackLockLevel(concurrent);
 
    /*
     * Check some preconditions in the concurrent case.  This also obtains the
     * replica index OID.
     */
    if (concurrent)
        check_concurrent_repack_requirements(OldHeap, &ident_idx);
 
    /* Check for user-requested abort. */
    CHECK_FOR_INTERRUPTS();
 
    pgstat_progress_start_command(PROGRESS_COMMAND_REPACK, tableOid);
    pgstat_progress_update_param(PROGRESS_REPACK_COMMAND, cmd);
 
    /*
     * Switch to the table owner's userid, so that any index functions are run
     * as that user.  Also lock down security-restricted operations and
     * arrange to make GUC variable changes local to this command.
     */
    GetUserIdAndSecContext(&save_userid, &save_sec_context);
    SetUserIdAndSecContext(OldHeap->rd_rel->relowner,
                           save_sec_context | SECURITY_RESTRICTED_OPERATION);
    save_nestlevel = NewGUCNestLevel();
    RestrictSearchPath();
 
    /*
     * Recheck that the relation is still what it was when we started.
     *
     * Note that it's critical to skip this in single-relation CLUSTER;
     * otherwise, we would reject an attempt to cluster using a
     * not-previously-clustered index.
     */
    if (recheck &&
        !cluster_rel_recheck(cmd, OldHeap, indexOid, save_userid,
                             lmode, params->options))
        goto out;
 
    /*
     * We allow repacking shared catalogs only when not using an index. It
     * would work to use an index in most respects, but the index would only
     * get marked as indisclustered in the current database, leading to
     * unexpected behavior if CLUSTER were later invoked in another database.
     */
    if (OidIsValid(indexOid) && OldHeap->rd_rel->relisshared)
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
        /*- translator: first %s is name of a SQL command, eg. REPACK */
                errmsg("cannot execute %s on a shared catalog",
                       RepackCommandAsString(cmd)));
 
    /*
     * The CONCURRENTLY case should have been rejected earlier because it does
     * not support system catalogs.
     */
    Assert(!(OldHeap->rd_rel->relisshared && concurrent));
 
    /*
     * Don't process temp tables of other backends ... their local buffer
     * manager is not going to cope.
     */
    if (RELATION_IS_OTHER_TEMP(OldHeap))
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
        /*- translator: first %s is name of a SQL command, eg. REPACK */
                errmsg("cannot execute %s on temporary tables of other sessions",
                       RepackCommandAsString(cmd)));
 
    /*
     * Also check for active uses of the relation in the current transaction,
     * including open scans and pending AFTER trigger events.
     */
    CheckTableNotInUse(OldHeap, RepackCommandAsString(cmd));
 
    /* Check heap and index are valid to cluster on */
    if (OidIsValid(indexOid))
    {
        /* verify the index is good and lock it */
        check_index_is_clusterable(OldHeap, indexOid, lmode);
        /* also open it */
        index = index_open(indexOid, NoLock);
    }
    else
        index = NULL;
 
    /*
     * When allow_system_table_mods is turned off, we disallow repacking a
     * catalog on a particular index unless that's already the clustered index
     * for that catalog.
     *
     * XXX We don't check for this in CLUSTER, because it's historically been
     * allowed.
     */
    if (cmd != REPACK_COMMAND_CLUSTER &&
        !allowSystemTableMods && OidIsValid(indexOid) &&
        IsCatalogRelation(OldHeap) && !index->rd_index->indisclustered)
        ereport(ERROR,
                errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                errmsg("permission denied: \"%s\" is a system catalog",
                       RelationGetRelationName(OldHeap)),
                errdetail("System catalogs can only be clustered by the index they're already clustered on, if any, unless \"%s\" is enabled.",
                          "allow_system_table_mods"));
 
    /*
     * Quietly ignore the request if this is a materialized view which has not
     * been populated from its query. No harm is done because there is no data
     * to deal with, and we don't want to throw an error if this is part of a
     * multi-relation request -- for example, CLUSTER was run on the entire
     * database.
     */
    if (OldHeap->rd_rel->relkind == RELKIND_MATVIEW &&
        !RelationIsPopulated(OldHeap))
    {
        if (index)
            index_close(index, lmode);
        relation_close(OldHeap, lmode);
        goto out;
    }
 
    Assert(OldHeap->rd_rel->relkind == RELKIND_RELATION ||
           OldHeap->rd_rel->relkind == RELKIND_MATVIEW ||
           OldHeap->rd_rel->relkind == RELKIND_TOASTVALUE);
 
    /*
     * All predicate locks on the tuples or pages are about to be made
     * invalid, because we move tuples around.  Promote them to relation
     * locks.  Predicate locks on indexes will be promoted when they are
     * reindexed.
     *
     * During concurrent processing, the heap as well as its indexes stay in
     * operation, so we postpone this step until they are locked using
     * AccessExclusiveLock near the end of the processing.
     */
    if (!concurrent)
        TransferPredicateLocksToHeapRelation(OldHeap);
 
    /* rebuild_relation does all the dirty work */
    PG_TRY();
    {
        rebuild_relation(OldHeap, index, verbose, ident_idx);
    }
    PG_FINALLY();
    {
        if (concurrent)
        {
            /*
             * Since during normal operation the worker was already asked to
             * exit, stopping it explicitly is especially important on ERROR.
             * However it still seems a good practice to make sure that the
             * worker never survives the REPACK command.
             */
            stop_repack_decoding_worker();
        }
    }
    PG_END_TRY();
 
    /* rebuild_relation closes OldHeap, and index if valid */
 
out:
    /* Roll back any GUC changes executed by index functions */
    AtEOXact_GUC(false, save_nestlevel);
 
    /* Restore userid and security context */
    SetUserIdAndSecContext(save_userid, save_sec_context);
 
    pgstat_progress_end_command();
}

References allowSystemTableMods, Assert, AtEOXact_GUC(), check_concurrent_repack_requirements(), CHECK_FOR_INTERRUPTS, check_index_is_clusterable(), CheckTableNotInUse(), CLUOPT_CONCURRENT, CLUOPT_RECHECK, CLUOPT_VERBOSE, cluster_rel_recheck(), ereport, errcode(), errdetail(), errmsg, ERROR, fb(), GetUserIdAndSecContext(), index_close(), index_open(), InvalidOid, IsCatalogRelation(), NewGUCNestLevel(), NoLock, OidIsValid, ClusterParams::options, PG_END_TRY, PG_FINALLY, PG_TRY, pgstat_progress_end_command(), pgstat_progress_start_command(), pgstat_progress_update_param(), PROGRESS_COMMAND_REPACK, PROGRESS_REPACK_COMMAND, rebuild_relation(), relation_close(), RELATION_IS_OTHER_TEMP, RelationGetRelationName, RelationGetRelid, RelationIsPopulated, REPACK_COMMAND_CLUSTER, RepackCommandAsString(), RepackLockLevel(), RestrictSearchPath(), SECURITY_RESTRICTED_OPERATION, SetUserIdAndSecContext(), stop_repack_decoding_worker(), TransferPredicateLocksToHeapRelation(), and verbose.

Referenced by ExecRepack(), process_single_relation(), and vacuum_rel().

ExecRepack()

void ExecRepack ( ParseState *  pstate, RepackStmt *  stmt, bool  isTopLevel  )

extern

Definition at line 237 of file repack.c.

{
    ClusterParams params = {0};
    Relation    rel = NULL;
    MemoryContext repack_context;
    LOCKMODE    lockmode;
    List       *rtcs;
 
    /* Parse option list */
    foreach_node(DefElem, opt, stmt->params)
    {
        if (strcmp(opt->defname, "verbose") == 0)
            params.options |= defGetBoolean(opt) ? CLUOPT_VERBOSE : 0;
        else if (strcmp(opt->defname, "analyze") == 0 ||
                 strcmp(opt->defname, "analyse") == 0)
            params.options |= defGetBoolean(opt) ? CLUOPT_ANALYZE : 0;
        else if (strcmp(opt->defname, "concurrently") == 0 &&
                 defGetBoolean(opt))
        {
            if (stmt->command != REPACK_COMMAND_REPACK)
                ereport(ERROR,
                        errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                        errmsg("CONCURRENTLY option not supported for %s",
                               RepackCommandAsString(stmt->command)));
            params.options |= CLUOPT_CONCURRENT;
        }
        else
            ereport(ERROR,
                    errcode(ERRCODE_SYNTAX_ERROR),
                    errmsg("unrecognized %s option \"%s\"",
                           RepackCommandAsString(stmt->command),
                           opt->defname),
                    parser_errposition(pstate, opt->location));
    }
 
    /* Determine the lock mode to use. */
    lockmode = RepackLockLevel((params.options & CLUOPT_CONCURRENT) != 0);
 
    if ((params.options & CLUOPT_CONCURRENT) != 0)
    {
        /*
         * Make sure we're not in a transaction block.
         *
         * The reason is that repack_setup_logical_decoding() could wait
         * indefinitely for our XID to complete. (The deadlock detector would
         * not recognize it because we'd be waiting for ourselves, i.e. no
         * real lock conflict.) It would be possible to run in a transaction
         * block if we had no XID, but this restriction is simpler for users
         * to understand and we don't lose any functionality.
         */
        PreventInTransactionBlock(isTopLevel, "REPACK (CONCURRENTLY)");
    }
 
    /*
     * If a single relation is specified, process it and we're done ... unless
     * the relation is a partitioned table, in which case we fall through.
     */
    if (stmt->relation != NULL)
    {
        rel = process_single_relation(stmt, lockmode, isTopLevel, &params);
        if (rel == NULL)
            return;             /* all done */
    }
 
    /*
     * Don't allow ANALYZE in the multiple-relation case for now.  Maybe we
     * can add support for this later.
     */
    if (params.options & CLUOPT_ANALYZE)
        ereport(ERROR,
                errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                errmsg("cannot execute %s on multiple tables",
                       "REPACK (ANALYZE)"));
 
    /*
     * By here, we know we are in a multi-table situation.
     *
     * Concurrent processing is currently considered rather special (e.g. in
     * terms of resources consumed) so it is not performed in bulk.
     */
    if (params.options & CLUOPT_CONCURRENT)
    {
        if (rel != NULL)
        {
            Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("REPACK (CONCURRENTLY) is not supported for partitioned tables"),
                    errhint("Consider running the command on individual partitions."));
        }
        else
            ereport(ERROR,
                    errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                    errmsg("REPACK (CONCURRENTLY) requires an explicit table name"));
    }
 
    /*
     * In order to avoid holding locks for too long, we want to process each
     * table in its own transaction.  This forces us to disallow running
     * inside a user transaction block.
     */
    PreventInTransactionBlock(isTopLevel, RepackCommandAsString(stmt->command));
 
    /* Also, we need a memory context to hold our list of relations */
    repack_context = AllocSetContextCreate(PortalContext,
                                           "Repack",
                                           ALLOCSET_DEFAULT_SIZES);
 
    /*
     * Since we open a new transaction for each relation, we have to check
     * that the relation still is what we think it is.
     *
     * In single-transaction CLUSTER, we don't need the overhead.
     */
    params.options |= CLUOPT_RECHECK;
 
    /*
     * If we don't have a relation yet, determine a relation list.  If we do,
     * then it must be a partitioned table, and we want to process its
     * partitions.
     */
    if (rel == NULL)
    {
        Assert(stmt->indexname == NULL);
        rtcs = get_tables_to_repack(stmt->command, stmt->usingindex,
                                    repack_context);
        params.options |= CLUOPT_RECHECK_ISCLUSTERED;
    }
    else
    {
        Oid         relid;
        bool        rel_is_index;
 
        Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE);
 
        /*
         * If USING INDEX was specified, resolve the index name now and pass
         * it down.
         */
        if (stmt->usingindex)
        {
            /*
             * If no index name was specified when repacking a partitioned
             * table, punt for now.  Maybe we can improve this later.
             */
            if (!stmt->indexname)
            {
                if (stmt->command == REPACK_COMMAND_CLUSTER)
                    ereport(ERROR,
                            errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                            errmsg("there is no previously clustered index for table \"%s\"",
                                   RelationGetRelationName(rel)));
                else
                    ereport(ERROR,
                            errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                    /*- translator: first %s is name of a SQL command, eg. REPACK */
                            errmsg("cannot execute %s on partitioned table \"%s\" USING INDEX with no index name",
                                   RepackCommandAsString(stmt->command),
                                   RelationGetRelationName(rel)));
            }
 
            relid = determine_clustered_index(rel, stmt->usingindex,
                                              stmt->indexname);
            if (!OidIsValid(relid))
                elog(ERROR, "unable to determine index to cluster on");
            check_index_is_clusterable(rel, relid, AccessExclusiveLock);
 
            rel_is_index = true;
        }
        else
        {
            relid = RelationGetRelid(rel);
            rel_is_index = false;
        }
 
        rtcs = get_tables_to_repack_partitioned(stmt->command,
                                                relid, rel_is_index,
                                                repack_context);
 
        /* close parent relation, releasing lock on it */
        table_close(rel, AccessExclusiveLock);
        rel = NULL;
    }
 
    /* Commit to get out of starting transaction */
    PopActiveSnapshot();
    CommitTransactionCommand();
 
    /* Cluster the tables, each in a separate transaction */
    Assert(rel == NULL);
    foreach_ptr(RelToCluster, rtc, rtcs)
    {
        /* Start a new transaction for each relation. */
        StartTransactionCommand();
 
        /*
         * Open the target table, coping with the case where it has been
         * dropped.
         */
        rel = try_table_open(rtc->tableOid, lockmode);
        if (rel == NULL)
        {
            CommitTransactionCommand();
            continue;
        }
 
        /* functions in indexes may want a snapshot set */
        PushActiveSnapshot(GetTransactionSnapshot());
 
        /* Process this table */
        cluster_rel(stmt->command, rel, rtc->indexOid, &params, isTopLevel);
        /* cluster_rel closes the relation, but keeps lock */
 
        PopActiveSnapshot();
        CommitTransactionCommand();
    }
 
    /* Start a new transaction for the cleanup work. */
    StartTransactionCommand();
 
    /* Clean up working storage */
    MemoryContextDelete(repack_context);
}

References AccessExclusiveLock, ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, Assert, check_index_is_clusterable(), CLUOPT_ANALYZE, CLUOPT_CONCURRENT, CLUOPT_RECHECK, CLUOPT_RECHECK_ISCLUSTERED, CLUOPT_VERBOSE, cluster_rel(), CommitTransactionCommand(), defGetBoolean(), determine_clustered_index(), elog, ereport, errcode(), errhint(), errmsg, ERROR, fb(), foreach_node, foreach_ptr, get_tables_to_repack(), get_tables_to_repack_partitioned(), GetTransactionSnapshot(), MemoryContextDelete(), OidIsValid, ClusterParams::options, parser_errposition(), PopActiveSnapshot(), PortalContext, PreventInTransactionBlock(), process_single_relation(), PushActiveSnapshot(), RelationData::rd_rel, RelationGetRelationName, RelationGetRelid, REPACK_COMMAND_CLUSTER, REPACK_COMMAND_REPACK, RepackCommandAsString(), RepackLockLevel(), StartTransactionCommand(), stmt, table_close(), and try_table_open().

Referenced by standard_ProcessUtility().

finish_heap_swap()

void finish_heap_swap ( Oid  OIDOldHeap, Oid  OIDNewHeap, bool  is_system_catalog, bool  swap_toast_by_content, bool  check_constraints, bool  is_internal, bool  reindex, TransactionId  frozenXid, MultiXactId  cutoffMulti, char  newrelpersistence  )

extern

Definition at line 1865 of file repack.c.

{
    ObjectAddress object;
    Oid         mapped_tables[4];
    int         i;
 
    /* Report that we are now swapping relation files */
    pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
                                 PROGRESS_REPACK_PHASE_SWAP_REL_FILES);
 
    /* Zero out possible results from swapped_relation_files */
    memset(mapped_tables, 0, sizeof(mapped_tables));
 
    /*
     * Swap the contents of the heap relations (including any toast tables).
     * Also set old heap's relfrozenxid to frozenXid.
     */
    swap_relation_files(OIDOldHeap, OIDNewHeap,
                        (OIDOldHeap == RelationRelationId),
                        swap_toast_by_content, is_internal,
                        frozenXid, cutoffMulti, mapped_tables);
 
    /*
     * If it's a system catalog, queue a sinval message to flush all catcaches
     * on the catalog when we reach CommandCounterIncrement.
     */
    if (is_system_catalog)
        CacheInvalidateCatalog(OIDOldHeap);
 
    if (reindex)
    {
        int         reindex_flags;
        ReindexParams reindex_params = {0};
 
        /*
         * Rebuild each index on the relation (but not the toast table, which
         * is all-new at this point).  It is important to do this before the
         * DROP step because if we are processing a system catalog that will
         * be used during DROP, we want to have its indexes available.  There
         * is no advantage to the other order anyway because this is all
         * transactional, so no chance to reclaim disk space before commit. We
         * do not need a final CommandCounterIncrement() because
         * reindex_relation does it.
         *
         * Note: because index_build is called via reindex_relation, it will
         * never set indcheckxmin true for the indexes.  This is OK even
         * though in some sense we are building new indexes rather than
         * rebuilding existing ones, because the new heap won't contain any
         * HOT chains at all, let alone broken ones, so it can't be necessary
         * to set indcheckxmin.
         */
        reindex_flags = REINDEX_REL_SUPPRESS_INDEX_USE;
        if (check_constraints)
            reindex_flags |= REINDEX_REL_CHECK_CONSTRAINTS;
 
        /*
         * Ensure that the indexes have the same persistence as the parent
         * relation.
         */
        if (newrelpersistence == RELPERSISTENCE_UNLOGGED)
            reindex_flags |= REINDEX_REL_FORCE_INDEXES_UNLOGGED;
        else if (newrelpersistence == RELPERSISTENCE_PERMANENT)
            reindex_flags |= REINDEX_REL_FORCE_INDEXES_PERMANENT;
 
        /* Report that we are now reindexing relations */
        pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
                                     PROGRESS_REPACK_PHASE_REBUILD_INDEX);
 
        reindex_relation(NULL, OIDOldHeap, reindex_flags, &reindex_params);
    }
 
    /* Report that we are now doing clean up */
    pgstat_progress_update_param(PROGRESS_REPACK_PHASE,
                                 PROGRESS_REPACK_PHASE_FINAL_CLEANUP);
 
    /*
     * If the relation being rebuilt is pg_class, swap_relation_files()
     * couldn't update pg_class's own pg_class entry (check comments in
     * swap_relation_files()), thus relfrozenxid was not updated. That's
     * annoying because a potential reason for doing a VACUUM FULL is a
     * imminent or actual anti-wraparound shutdown.  So, now that we can
     * access the new relation using its indices, update relfrozenxid.
     * pg_class doesn't have a toast relation, so we don't need to update the
     * corresponding toast relation. Not that there's little point moving all
     * relfrozenxid updates here since swap_relation_files() needs to write to
     * pg_class for non-mapped relations anyway.
     */
    if (OIDOldHeap == RelationRelationId)
    {
        Relation    relRelation;
        HeapTuple   reltup;
        Form_pg_class relform;
 
        relRelation = table_open(RelationRelationId, RowExclusiveLock);
 
        reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(OIDOldHeap));
        if (!HeapTupleIsValid(reltup))
            elog(ERROR, "cache lookup failed for relation %u", OIDOldHeap);
        relform = (Form_pg_class) GETSTRUCT(reltup);
 
        relform->relfrozenxid = frozenXid;
        relform->relminmxid = cutoffMulti;
 
        CatalogTupleUpdate(relRelation, &reltup->t_self, reltup);
 
        table_close(relRelation, RowExclusiveLock);
    }
 
    /* Destroy new heap with old filenumber */
    object.classId = RelationRelationId;
    object.objectId = OIDNewHeap;
    object.objectSubId = 0;
 
    if (!reindex)
    {
        /*
         * Make sure the changes in pg_class are visible. This is especially
         * important if !swap_toast_by_content, so that the correct TOAST
         * relation is dropped. (reindex_relation() above did not help in this
         * case))
         */
        CommandCounterIncrement();
    }
 
    /*
     * The new relation is local to our transaction and we know nothing
     * depends on it, so DROP_RESTRICT should be OK.
     */
    performDeletion(&object, DROP_RESTRICT, PERFORM_DELETION_INTERNAL);
 
    /* performDeletion does CommandCounterIncrement at end */
 
    /*
     * Now we must remove any relation mapping entries that we set up for the
     * transient table, as well as its toast table and toast index if any. If
     * we fail to do this before commit, the relmapper will complain about new
     * permanent map entries being added post-bootstrap.
     */
    for (i = 0; OidIsValid(mapped_tables[i]); i++)
        RelationMapRemoveMapping(mapped_tables[i]);
 
    /*
     * At this point, everything is kosher except that, if we did toast swap
     * by links, the toast table's name corresponds to the transient table.
     * The name is irrelevant to the backend because it's referenced by OID,
     * but users looking at the catalogs could be confused.  Rename it to
     * prevent this problem.
     *
     * Note no lock required on the relation, because we already hold an
     * exclusive lock on it.
     */
    if (!swap_toast_by_content)
    {
        Relation    newrel;
 
        newrel = table_open(OIDOldHeap, NoLock);
        if (OidIsValid(newrel->rd_rel->reltoastrelid))
        {
            Oid         toastidx;
            char        NewToastName[NAMEDATALEN];
 
            /* Get the associated valid index to be renamed */
            toastidx = toast_get_valid_index(newrel->rd_rel->reltoastrelid,
                                             AccessExclusiveLock);
 
            /* rename the toast table ... */
            snprintf(NewToastName, NAMEDATALEN, "pg_toast_%u",
                     OIDOldHeap);
            RenameRelationInternal(newrel->rd_rel->reltoastrelid,
                                   NewToastName, true, false);
 
            /* ... and its valid index too. */
            snprintf(NewToastName, NAMEDATALEN, "pg_toast_%u_index",
                     OIDOldHeap);
 
            RenameRelationInternal(toastidx,
                                   NewToastName, true, true);
 
            /*
             * Reset the relrewrite for the toast. The command-counter
             * increment is required here as we are about to update the tuple
             * that is updated as part of RenameRelationInternal.
             */
            CommandCounterIncrement();
            ResetRelRewrite(newrel->rd_rel->reltoastrelid);
        }
        relation_close(newrel, NoLock);
    }
 
    /* if it's not a catalog table, clear any missing attribute settings */
    if (!is_system_catalog)
    {
        Relation    newrel;
 
        newrel = table_open(OIDOldHeap, NoLock);
        RelationClearMissing(newrel);
        relation_close(newrel, NoLock);
    }
}

References AccessExclusiveLock, CacheInvalidateCatalog(), CatalogTupleUpdate(), CommandCounterIncrement(), DROP_RESTRICT, elog, ERROR, fb(), GETSTRUCT(), HeapTupleIsValid, i, NAMEDATALEN, NoLock, ObjectIdGetDatum(), OidIsValid, PERFORM_DELETION_INTERNAL, performDeletion(), pgstat_progress_update_param(), PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_FINAL_CLEANUP, PROGRESS_REPACK_PHASE_REBUILD_INDEX, PROGRESS_REPACK_PHASE_SWAP_REL_FILES, REINDEX_REL_CHECK_CONSTRAINTS, REINDEX_REL_FORCE_INDEXES_PERMANENT, REINDEX_REL_FORCE_INDEXES_UNLOGGED, REINDEX_REL_SUPPRESS_INDEX_USE, reindex_relation(), relation_close(), RelationClearMissing(), RelationMapRemoveMapping(), RenameRelationInternal(), ResetRelRewrite(), RowExclusiveLock, SearchSysCacheCopy1, snprintf, swap_relation_files(), table_close(), table_open(), and toast_get_valid_index().

Referenced by ATRewriteTables(), rebuild_relation(), rebuild_relation_finish_concurrent(), and refresh_by_heap_swap().

HandleRepackMessageInterrupt()

void HandleRepackMessageInterrupt ( void  )

extern

Definition at line 3508 of file repack.c.

{
    InterruptPending = true;
    RepackMessagePending = true;
    SetLatch(MyLatch);
}

References InterruptPending, MyLatch, RepackMessagePending, and SetLatch().

Referenced by procsignal_sigusr1_handler().

make_new_heap()

Oid make_new_heap ( Oid  OIDOldHeap, Oid  NewTableSpace, Oid  NewAccessMethod, char  relpersistence, LOCKMODE  lockmode  )

extern

Definition at line 1108 of file repack.c.

{
    TupleDesc   OldHeapDesc;
    char        NewHeapName[NAMEDATALEN];
    Oid         OIDNewHeap;
    Oid         toastid;
    Relation    OldHeap;
    HeapTuple   tuple;
    Datum       reloptions;
    bool        isNull;
    Oid         namespaceid;
 
    OldHeap = table_open(OIDOldHeap, lockmode);
    OldHeapDesc = RelationGetDescr(OldHeap);
 
    /*
     * Note that the NewHeap will not receive any of the defaults or
     * constraints associated with the OldHeap; we don't need 'em, and there's
     * no reason to spend cycles inserting them into the catalogs only to
     * delete them.
     */
 
    /*
     * But we do want to use reloptions of the old heap for new heap.
     */
    tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(OIDOldHeap));
    if (!HeapTupleIsValid(tuple))
        elog(ERROR, "cache lookup failed for relation %u", OIDOldHeap);
    reloptions = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
                                 &isNull);
    if (isNull)
        reloptions = (Datum) 0;
 
    if (relpersistence == RELPERSISTENCE_TEMP)
        namespaceid = LookupCreationNamespace("pg_temp");
    else
        namespaceid = RelationGetNamespace(OldHeap);
 
    /*
     * Create the new heap, using a temporary name in the same namespace as
     * the existing table.  NOTE: there is some risk of collision with user
     * relnames.  Working around this seems more trouble than it's worth; in
     * particular, we can't create the new heap in a different namespace from
     * the old, or we will have problems with the TEMP status of temp tables.
     *
     * Note: the new heap is not a shared relation, even if we are rebuilding
     * a shared rel.  However, we do make the new heap mapped if the source is
     * mapped.  This simplifies swap_relation_files, and is absolutely
     * necessary for rebuilding pg_class, for reasons explained there.
     */
    snprintf(NewHeapName, sizeof(NewHeapName), "pg_temp_%u", OIDOldHeap);
 
    OIDNewHeap = heap_create_with_catalog(NewHeapName,
                                          namespaceid,
                                          NewTableSpace,
                                          InvalidOid,
                                          InvalidOid,
                                          InvalidOid,
                                          OldHeap->rd_rel->relowner,
                                          NewAccessMethod,
                                          OldHeapDesc,
                                          NIL,
                                          RELKIND_RELATION,
                                          relpersistence,
                                          false,
                                          RelationIsMapped(OldHeap),
                                          ONCOMMIT_NOOP,
                                          reloptions,
                                          false,
                                          true,
                                          true,
                                          OIDOldHeap,
                                          NULL);
    Assert(OIDNewHeap != InvalidOid);
 
    ReleaseSysCache(tuple);
 
    /*
     * Advance command counter so that the newly-created relation's catalog
     * tuples will be visible to table_open.
     */
    CommandCounterIncrement();
 
    /*
     * If necessary, create a TOAST table for the new relation.
     *
     * If the relation doesn't have a TOAST table already, we can't need one
     * for the new relation.  The other way around is possible though: if some
     * wide columns have been dropped, NewHeapCreateToastTable can decide that
     * no TOAST table is needed for the new table.
     *
     * Note that NewHeapCreateToastTable ends with CommandCounterIncrement, so
     * that the TOAST table will be visible for insertion.
     */
    toastid = OldHeap->rd_rel->reltoastrelid;
    if (OidIsValid(toastid))
    {
        /* keep the existing toast table's reloptions, if any */
        tuple = SearchSysCache1(RELOID, ObjectIdGetDatum(toastid));
        if (!HeapTupleIsValid(tuple))
            elog(ERROR, "cache lookup failed for relation %u", toastid);
        reloptions = SysCacheGetAttr(RELOID, tuple, Anum_pg_class_reloptions,
                                     &isNull);
        if (isNull)
            reloptions = (Datum) 0;
 
        NewHeapCreateToastTable(OIDNewHeap, reloptions, lockmode, toastid);
 
        ReleaseSysCache(tuple);
    }
 
    table_close(OldHeap, NoLock);
 
    return OIDNewHeap;
}

References Assert, CommandCounterIncrement(), elog, ERROR, fb(), heap_create_with_catalog(), HeapTupleIsValid, InvalidOid, LookupCreationNamespace(), NAMEDATALEN, NewHeapCreateToastTable(), NIL, NoLock, ObjectIdGetDatum(), OidIsValid, ONCOMMIT_NOOP, RelationGetDescr, RelationGetNamespace, RelationIsMapped, ReleaseSysCache(), SearchSysCache1(), snprintf, SysCacheGetAttr(), table_close(), and table_open().

Referenced by ATRewriteTables(), rebuild_relation(), and RefreshMatViewByOid().

mark_index_clustered()

void mark_index_clustered ( Relation  rel, Oid  indexOid, bool  is_internal  )

extern

Definition at line 811 of file repack.c.

{
    HeapTuple   indexTuple;
    Form_pg_index indexForm;
    Relation    pg_index;
    ListCell   *index;
 
    Assert(rel->rd_rel->relkind != RELKIND_PARTITIONED_TABLE);
 
    /*
     * If the index is already marked clustered, no need to do anything.
     */
    if (OidIsValid(indexOid))
    {
        if (get_index_isclustered(indexOid))
            return;
    }
 
    /*
     * Check each index of the relation and set/clear the bit as needed.
     */
    pg_index = table_open(IndexRelationId, RowExclusiveLock);
 
    foreach(index, RelationGetIndexList(rel))
    {
        Oid         thisIndexOid = lfirst_oid(index);
 
        indexTuple = SearchSysCacheCopy1(INDEXRELID,
                                         ObjectIdGetDatum(thisIndexOid));
        if (!HeapTupleIsValid(indexTuple))
            elog(ERROR, "cache lookup failed for index %u", thisIndexOid);
        indexForm = (Form_pg_index) GETSTRUCT(indexTuple);
 
        /*
         * Unset the bit if set.  We know it's wrong because we checked this
         * earlier.
         */
        if (indexForm->indisclustered)
        {
            indexForm->indisclustered = false;
            CatalogTupleUpdate(pg_index, &indexTuple->t_self, indexTuple);
        }
        else if (thisIndexOid == indexOid)
        {
            /* this was checked earlier, but let's be real sure */
            if (!indexForm->indisvalid)
                elog(ERROR, "cannot cluster on invalid index %u", indexOid);
            indexForm->indisclustered = true;
            CatalogTupleUpdate(pg_index, &indexTuple->t_self, indexTuple);
        }
 
        InvokeObjectPostAlterHookArg(IndexRelationId, thisIndexOid, 0,
                                     InvalidOid, is_internal);
 
        heap_freetuple(indexTuple);
    }
 
    table_close(pg_index, RowExclusiveLock);
}

References Assert, CatalogTupleUpdate(), elog, ERROR, fb(), Form_pg_index, get_index_isclustered(), GETSTRUCT(), heap_freetuple(), HeapTupleIsValid, InvalidOid, InvokeObjectPostAlterHookArg, lfirst_oid, ObjectIdGetDatum(), OidIsValid, RelationData::rd_rel, RelationGetIndexList(), RowExclusiveLock, SearchSysCacheCopy1, table_close(), and table_open().

Referenced by ATExecClusterOn(), ATExecDropCluster(), and rebuild_relation().

ProcessRepackMessages()

void ProcessRepackMessages ( void  )

extern

Definition at line 3519 of file repack.c.

{
    MemoryContext oldcontext;
    static MemoryContext hpm_context = NULL;
 
    /*
     * Nothing to do if we haven't launched the worker yet or have already
     * terminated it.
     */
    if (decoding_worker == NULL)
        return;
 
    /*
     * This is invoked from ProcessInterrupts(), and since some of the
     * functions it calls contain CHECK_FOR_INTERRUPTS(), there is a potential
     * for recursive calls if more signals are received while this runs.  It's
     * unclear that recursive entry would be safe, and it doesn't seem useful
     * even if it is safe, so let's block interrupts until done.
     */
    HOLD_INTERRUPTS();
 
    /*
     * Moreover, CurrentMemoryContext might be pointing almost anywhere.  We
     * don't want to risk leaking data into long-lived contexts, so let's do
     * our work here in a private context that we can reset on each use.
     */
    if (hpm_context == NULL)    /* first time through? */
        hpm_context = AllocSetContextCreate(TopMemoryContext,
                                            "ProcessRepackMessages",
                                            ALLOCSET_DEFAULT_SIZES);
    else
        MemoryContextReset(hpm_context);
 
    oldcontext = MemoryContextSwitchTo(hpm_context);
 
    /* OK to process messages.  Reset the flag saying there are more to do. */
    RepackMessagePending = false;
 
    /*
     * Read as many messages as we can from the worker, but stop when no more
     * messages can be read from the worker without blocking.
     */
    while (true)
    {
        shm_mq_result res;
        Size        nbytes;
        void       *data;
 
        res = shm_mq_receive(decoding_worker->error_mqh, &nbytes,
                             &data, true);
        if (res == SHM_MQ_WOULD_BLOCK)
            break;
        else if (res == SHM_MQ_SUCCESS)
        {
            StringInfoData msg;
 
            initStringInfo(&msg);
            appendBinaryStringInfo(&msg, data, nbytes);
            ProcessRepackMessage(&msg);
            pfree(msg.data);
        }
        else
        {
            /*
             * The decoding worker is special in that it exits as soon as it
             * has its work done. Thus the DETACHED result code is fine.
             */
            Assert(res == SHM_MQ_DETACHED);
 
            break;
        }
    }
 
    MemoryContextSwitchTo(oldcontext);
 
    /* Might as well clear the context on our way out */
    MemoryContextReset(hpm_context);
 
    RESUME_INTERRUPTS();
}

References ALLOCSET_DEFAULT_SIZES, AllocSetContextCreate, appendBinaryStringInfo(), Assert, StringInfoData::data, data, decoding_worker, DecodingWorker::error_mqh, fb(), HOLD_INTERRUPTS, initStringInfo(), MemoryContextReset(), MemoryContextSwitchTo(), pfree(), ProcessRepackMessage(), RepackMessagePending, RESUME_INTERRUPTS, SHM_MQ_DETACHED, shm_mq_receive(), SHM_MQ_SUCCESS, SHM_MQ_WOULD_BLOCK, and TopMemoryContext.

Referenced by ProcessInterrupts().

RepackWorkerMain()

void RepackWorkerMain ( Datum  main_arg )

extern

Definition at line 57 of file repack_worker.c.

{
    dsm_segment *seg;
    DecodingWorkerShared *shared;
    shm_mq     *mq;
    shm_mq_handle *mqh;
    LogicalDecodingContext *decoding_ctx;
    SharedFileSet *sfs;
    Snapshot    snapshot;
 
    am_repack_worker = true;
 
    /*
     * Override the default bgworker_die() with die() so we can use
     * CHECK_FOR_INTERRUPTS().
     */
    pqsignal(SIGTERM, die);
    BackgroundWorkerUnblockSignals();
 
    seg = dsm_attach(DatumGetUInt32(main_arg));
    if (seg == NULL)
        ereport(ERROR,
                errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
                errmsg("could not map dynamic shared memory segment"));
 
    shared = (DecodingWorkerShared *) dsm_segment_address(seg);
    shared->dsm_seg = seg;
 
    /* Arrange to signal the leader if we exit. */
    before_shmem_exit(RepackWorkerShutdown, PointerGetDatum(shared));
 
    /*
     * Join locking group - see the comments around the call of
     * start_repack_decoding_worker().
     */
    if (!BecomeLockGroupMember(shared->backend_proc, shared->backend_pid))
        return;                 /* The leader is not running anymore. */
 
    /*
     * Setup a queue to send error messages to the backend that launched this
     * worker.
     */
    mq = (shm_mq *) (char *) BUFFERALIGN(shared->error_queue);
    shm_mq_set_sender(mq, MyProc);
    mqh = shm_mq_attach(mq, seg, NULL);
    pq_redirect_to_shm_mq(seg, mqh);
    pq_set_parallel_leader(shared->backend_pid,
                           shared->backend_proc_number);
 
    /* Connect to the database. */
    BackgroundWorkerInitializeConnectionByOid(shared->dbid, shared->roleid, 0);
 
    /*
     * Transaction is needed to open relation, and it also provides us with a
     * resource owner.
     */
    StartTransactionCommand();
 
    shared = (DecodingWorkerShared *) dsm_segment_address(seg);
 
    /*
     * Not sure the spinlock is needed here - the backend should not change
     * anything in the shared memory until we have serialized the snapshot.
     */
    SpinLockAcquire(&shared->mutex);
    Assert(!XLogRecPtrIsValid(shared->lsn_upto));
    sfs = &shared->sfs;
    SpinLockRelease(&shared->mutex);
 
    SharedFileSetAttach(sfs, seg);
 
    /*
     * Prepare to capture the concurrent data changes ourselves.
     */
    decoding_ctx = repack_setup_logical_decoding(shared->relid);
 
    /* Announce that we're ready. */
    SpinLockAcquire(&shared->mutex);
    shared->initialized = true;
    SpinLockRelease(&shared->mutex);
    ConditionVariableSignal(&shared->cv);
 
    /* There doesn't seem to a nice API to set these */
    XactIsoLevel = XACT_REPEATABLE_READ;
    XactReadOnly = true;
 
    /* Build the initial snapshot and export it. */
    snapshot = SnapBuildInitialSnapshot(decoding_ctx->snapshot_builder);
    export_initial_snapshot(snapshot, shared);
 
    /*
     * Only historic snapshots should be used now. Do not let us restrict the
     * progress of xmin horizon.
     */
    InvalidateCatalogSnapshot();
 
    for (;;)
    {
        bool        stop = decode_concurrent_changes(decoding_ctx, shared);
 
        if (stop)
            break;
 
    }
 
    /* Cleanup. */
    repack_cleanup_logical_decoding(decoding_ctx);
    CommitTransactionCommand();
}

References am_repack_worker, Assert, DecodingWorkerShared::backend_pid, DecodingWorkerShared::backend_proc, DecodingWorkerShared::backend_proc_number, BackgroundWorkerInitializeConnectionByOid(), BackgroundWorkerUnblockSignals(), BecomeLockGroupMember(), before_shmem_exit(), BUFFERALIGN, CommitTransactionCommand(), ConditionVariableSignal(), DecodingWorkerShared::cv, DatumGetUInt32(), DecodingWorkerShared::dbid, decode_concurrent_changes(), die, dsm_attach(), DecodingWorkerShared::dsm_seg, dsm_segment_address(), ereport, errcode(), errmsg, ERROR, DecodingWorkerShared::error_queue, export_initial_snapshot(), fb(), DecodingWorkerShared::initialized, InvalidateCatalogSnapshot(), DecodingWorkerShared::lsn_upto, DecodingWorkerShared::mutex, MyProc, PointerGetDatum(), pq_redirect_to_shm_mq(), pq_set_parallel_leader(), pqsignal, DecodingWorkerShared::relid, repack_cleanup_logical_decoding(), repack_setup_logical_decoding(), RepackWorkerShutdown(), DecodingWorkerShared::roleid, DecodingWorkerShared::sfs, SharedFileSetAttach(), shm_mq_attach(), shm_mq_set_sender(), SnapBuildInitialSnapshot(), SpinLockAcquire(), SpinLockRelease(), StartTransactionCommand(), XACT_REPEATABLE_READ, XactIsoLevel, XactReadOnly, and XLogRecPtrIsValid.

Variable Documentation

RepackMessagePending

PGDLLIMPORT volatile sig_atomic_t RepackMessagePending

extern

Definition at line 146 of file repack.c.

Referenced by HandleRepackMessageInterrupt(), ProcessInterrupts(), and ProcessRepackMessages().