/
githubmirror
/
postgres
Обзор
Документация
Войти
/
githubmirror
/
postgres
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
src/backend/commands/repack.c
3 954 строки
122 KB
Álvaro Herrera
Move code to get_tables_to_repack_partitioned
20 июл 2026, 13:12
Не верифицирован
20 июл 2026, 13:12
ad6ae52
Код
Авторство
О чём код?
/*------------------------------------------------------------------------- * * repack.c * REPACK a table; formerly known as CLUSTER. VACUUM FULL also uses * parts of this code. * * There are two somewhat different ways to rewrite a table. In non- * concurrent mode, it's easy: take AccessExclusiveLock, create a new * transient relation, copy the tuples over to the relfilenode of the new * relation, swap the relfilenodes, then drop the old relation. * * In concurrent mode, we lock the table with only ShareUpdateExclusiveLock, * then do an initial copy as above. However, while the tuples are being * copied, concurrent transactions could modify the table. To cope with those * changes, we rely on logical decoding to obtain them from WAL. A bgworker * consumes WAL while the initial copy is ongoing (to prevent excessive WAL * from being reserved), and accumulates the changes in a file. Once the * initial copy is complete, we read the changes from the file and re-apply * them on the new heap. Then we upgrade our ShareUpdateExclusiveLock to * AccessExclusiveLock and swap the relfilenodes. This way, the time we hold * a strong lock on the table is much reduced, and the bloat is eliminated. * * * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group * Portions Copyright (c) 1994-5, Regents of the University of California * * * IDENTIFICATION * src/backend/commands/repack.c * *------------------------------------------------------------------------- */ #include "postgres.h" #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" #include "access/relscan.h" #include "access/tableam.h" #include "access/toast_internals.h" #include "access/transam.h" #include "access/xact.h" #include "access/xlog.h" #include "catalog/catalog.h" #include "catalog/dependency.h" #include "catalog/heap.h" #include "catalog/index.h" #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/pg_am.h" #include "catalog/pg_attrdef.h" #include "catalog/pg_constraint.h" #include "catalog/pg_inherits.h" #include "catalog/toasting.h" #include "commands/defrem.h" #include "commands/progress.h" #include "commands/repack.h" #include "commands/repack_internal.h" #include "commands/tablecmds.h" #include "commands/vacuum.h" #include "executor/executor.h" #include "libpq/pqformat.h" #include "libpq/pqmq.h" #include "miscadmin.h" #include "optimizer/optimizer.h" #include "parser/parse_relation.h" #include "pgstat.h" #include "replication/logicalrelation.h" #include "storage/bufmgr.h" #include "storage/ipc.h" #include "storage/lmgr.h" #include "storage/predicate.h" #include "storage/proc.h" #include "utils/acl.h" #include "utils/fmgroids.h" #include "utils/guc.h" #include "utils/injection_point.h" #include "utils/inval.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/pg_rusage.h" #include "utils/relmapper.h" #include "utils/snapmgr.h" #include "utils/syscache.h" #include "utils/wait_event_types.h" /* * This struct is used to pass around the information on tables to be * clustered. We need this so we can make a list of them when invoked without * a specific table/index pair. */ typedef struct { Oid tableOid; Oid indexOid; } RelToCluster; /* * The first file exported by the decoding worker must contain a snapshot, the * following ones contain the data changes. */ #define WORKER_FILE_SNAPSHOT 0 /* * Information needed to apply concurrent data changes. */ typedef struct ChangeContext { /* The relation the changes are applied to. */ Relation cc_rel; /* Needed to update indexes of cc_rel. */ ResultRelInfo *cc_rri; EState *cc_estate; /* * Existing tuples to UPDATE and DELETE are located via this index. We * keep the scankey in partially initialized state to avoid repeated work. * sk_argument is completed on the fly. */ Relation cc_ident_index; ScanKey cc_ident_key; int cc_ident_key_nentries; /* The latest column we need to deform to have the tuple identity */ AttrNumber cc_last_key_attno; /* Sequential number of the file containing the changes. */ int cc_file_seq; } ChangeContext; /* * Backend-local information to control the decoding worker. */ typedef struct DecodingWorker { /* The worker. */ BackgroundWorkerHandle *handle; /* DecodingWorkerShared is in this segment. */ dsm_segment *seg; /* Handle of the error queue. */ shm_mq_handle *error_mqh; } DecodingWorker; /* Pointer to currently running decoding worker. */ static DecodingWorker *decoding_worker = NULL; /* * Is there a message sent by a repack worker that the backend needs to * receive? */ volatile sig_atomic_t RepackMessagePending = false; static LOCKMODE RepackLockLevel(bool concurrent); static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, Oid userid, LOCKMODE lmode, int options); static void check_concurrent_repack_requirements(Relation rel, Oid *ident_idx_p); static void rebuild_relation(Relation OldHeap, Relation index, bool verbose, Oid ident_idx); static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, Snapshot snapshot, bool verbose, bool *pSwapToastByContent, TransactionId *pFreezeXid, MultiXactId *pCutoffMulti); static List *get_tables_to_repack(RepackCommand cmd, bool usingindex, MemoryContext permcxt); static List *get_tables_to_repack_partitioned(RepackStmt *stmt, Relation rel, MemoryContext permcxt); static bool repack_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid); static void apply_concurrent_changes(BufFile *file, ChangeContext *chgcxt); static void apply_concurrent_insert(Relation rel, TupleTableSlot *slot, ChangeContext *chgcxt); static void apply_concurrent_update(Relation rel, TupleTableSlot *spilled_tuple, TupleTableSlot *ondisk_tuple, ChangeContext *chgcxt); static void apply_concurrent_delete(Relation rel, TupleTableSlot *slot); static void restore_tuple(BufFile *file, Relation relation, TupleTableSlot *slot); static void adjust_toast_pointers(Relation relation, TupleTableSlot *dest, TupleTableSlot *src); static bool find_target_tuple(Relation rel, ChangeContext *chgcxt, TupleTableSlot *locator, TupleTableSlot *retrieved); static bool identity_key_equal(ChangeContext *chgcxt, TupleTableSlot *locator, TupleTableSlot *candidate); static void process_concurrent_changes(XLogRecPtr end_of_wal, ChangeContext *chgcxt, bool done); static void initialize_change_context(ChangeContext *chgcxt, Relation relation, Oid ident_index_id); static void release_change_context(ChangeContext *chgcxt); static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Oid identIdx, TransactionId frozenXid, MultiXactId cutoffMulti); static List *build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes); static void copy_index_constraints(Relation old_index, Oid new_index_id, Oid new_heap_id); static void copy_attribute_defaults(Oid old_heap_oid, Oid new_heap_oid); static Relation process_single_relation(RepackStmt *stmt, LOCKMODE lockmode, bool isTopLevel, ClusterParams *params); static Oid determine_clustered_index(Relation rel, bool usingindex, const char *indexname); static void start_repack_decoding_worker(Oid relid); static void stop_repack_decoding_worker(void); static void stop_repack_decoding_worker_cb(int code, Datum arg); static Snapshot get_initial_snapshot(DecodingWorker *worker); static void ProcessRepackMessage(StringInfo msg); static const char *RepackCommandAsString(RepackCommand cmd); /* * The repack code allows for processing multiple tables at once. Because * of this, we cannot just run everything on a single transaction, or we * would be forced to acquire exclusive locks on all the tables being * clustered, simultaneously --- very likely leading to deadlock. * * To solve this we follow a similar strategy to VACUUM code, processing each * relation in a separate transaction. For this to work, we need to: * * - provide a separate memory context so that we can pass information in * a way that survives across transactions * - start a new transaction every time a new relation is clustered * - check for validity of the information on to-be-clustered relations, * as someone might have deleted a relation behind our back, or * clustered one on a different index * - end the transaction * * The single-relation case does not have any such overhead. * * We also allow a relation to be repacked following an index, but without * naming a specific one. In that case, the indisclustered bit will be * looked up, and an ERROR will be thrown if no so-marked index is found. */ void ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) { ClusterParams params = {0}; Relation rel = NULL; MemoryContext repack_context; LOCKMODE lockmode; List *rtcs; bool verbose = false; bool analyze = false; bool concurrently = false; /* Parse option list */ foreach_node(DefElem, opt, stmt->params) { if (strcmp(opt->defname, "verbose") == 0) verbose = defGetBoolean(opt); else if (strcmp(opt->defname, "analyze") == 0 || strcmp(opt->defname, "analyse") == 0) analyze = defGetBoolean(opt); else if (strcmp(opt->defname, "concurrently") == 0) { if (stmt->command != REPACK_COMMAND_REPACK) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("CONCURRENTLY option not supported for %s", RepackCommandAsString(stmt->command))); concurrently = defGetBoolean(opt); } else ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), errmsg("unrecognized %s option \"%s\"", RepackCommandAsString(stmt->command), opt->defname), parser_errposition(pstate, opt->location)); } params.options |= (verbose ? CLUOPT_VERBOSE : 0) | (analyze ? CLUOPT_ANALYZE : 0) | (concurrently ? CLUOPT_CONCURRENT : 0); /* 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, ¶ms); 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("%s is not supported for partitioned tables", "REPACK (CONCURRENTLY)"), errhint("Consider running the command on individual partitions.")); } else ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("%s requires an explicit table name", "REPACK (CONCURRENTLY)")); } /* * 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. Note that we don't acquire any locks on these tables, so * the returned list must be treated with suspicion. */ if (rel == NULL) { Assert(stmt->indexname == NULL); rtcs = get_tables_to_repack(stmt->command, stmt->usingindex, repack_context); params.options |= CLUOPT_RECHECK_ISCLUSTERED; } else { rtcs = get_tables_to_repack_partitioned(stmt, rel, repack_context); rel = NULL; /* clobber no longer valid pointer */ } /* 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. It may have been dropped or replaced with * something different, in which case silently skip it. */ rel = try_relation_open(rtc->tableOid, lockmode); if (rel == NULL) { CommitTransactionCommand(); continue; } if (rel->rd_rel->relkind != RELKIND_RELATION && rel->rd_rel->relkind != RELKIND_MATVIEW) { relation_close(rel, lockmode); CommitTransactionCommand(); continue; } /* functions in indexes may want a snapshot set */ PushActiveSnapshot(GetTransactionSnapshot()); /* Process this table */ cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms, 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); } /* * In the non-concurrent case, we obtain AccessExclusiveLock throughout the * operation to avoid any lock-upgrade hazards. In the concurrent case, we * grab ShareUpdateExclusiveLock (just like VACUUM) for most of the * processing and only acquire AccessExclusiveLock at the end, to swap the * relation -- supposedly for a short time. */ static LOCKMODE RepackLockLevel(bool concurrent) { if (concurrent) return ShareUpdateExclusiveLock; else return AccessExclusiveLock; } /* * cluster_rel * * This clusters the table by creating a new, clustered table and * swapping the relfilenumbers of the new table and the old table, so * the OID of the original table is preserved. Thus we do not lose * GRANT, inheritance nor references to this table. * * Indexes are rebuilt too, via REINDEX. Since we are effectively bulk-loading * the new table, it's better to create the indexes afterwards than to fill * them incrementally while we load the table. * * If indexOid is InvalidOid, the table will be rewritten in physical order * instead of index order. * * Note that, in the concurrent case, the function releases the lock at some * point, in order to get AccessExclusiveLock for the final steps (i.e. to * swap the relation files). To make things simpler, the caller should expect * OldHeap to be closed on return, regardless CLUOPT_CONCURRENT. (The * AccessExclusiveLock is kept till the end of the transaction.) * * 'cmd' indicates which command is being executed, to be used for error * messages. */ void cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, ClusterParams *params, bool isTopLevel) { 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, and closes OldHeap and index, * if valid. * * In concurrent mode, make sure the worker terminates; normally it does * so by itself, but a PG_ENSURE_ERROR_CLEANUP callback ensures that this * happens even in case this backend dies early on a FATAL exit. Normal * mode doesn't need that overhead. */ if (concurrent) { PG_ENSURE_ERROR_CLEANUP(stop_repack_decoding_worker_cb, 0); { rebuild_relation(OldHeap, index, verbose, ident_idx); } PG_END_ENSURE_ERROR_CLEANUP(stop_repack_decoding_worker_cb, 0); stop_repack_decoding_worker(); } else rebuild_relation(OldHeap, index, verbose, ident_idx); 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(); } /* * Check if the table (and its index) still meets the requirements of * cluster_rel(). */ static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, Oid userid, LOCKMODE lmode, int options) { Oid tableOid = RelationGetRelid(OldHeap); Assert(CheckRelationLockedByMe(OldHeap, lmode, false)); /* Check that the user still has privileges for the relation */ if (!repack_is_permitted_for_relation(cmd, tableOid, userid)) { relation_close(OldHeap, lmode); return false; } /* * Silently skip a temp table for a remote session. Only doing this check * in the "recheck" case is appropriate (which currently means somebody is * executing a database-wide CLUSTER or on a partitioned table), because * there is another check in cluster() which will stop any attempt to * cluster remote temp tables by name. There is another check in * cluster_rel which is redundant, but we leave it for extra safety. */ if (RELATION_IS_OTHER_TEMP(OldHeap)) { relation_close(OldHeap, lmode); return false; } if (OidIsValid(indexOid)) { /* * Check that the index still exists */ if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(indexOid))) { relation_close(OldHeap, lmode); return false; } /* * Check that the index is still the one with indisclustered set, if * needed. */ if ((options & CLUOPT_RECHECK_ISCLUSTERED) != 0 && !get_index_isclustered(indexOid)) { relation_close(OldHeap, lmode); return false; } } return true; } /* * Verify that the specified heap and index are valid to cluster on * * Side effect: obtains lock on the index. The caller may * in some cases already have a lock of the same strength on the table, but * not in all cases so we can't rely on the table-level lock for * protection here. */ void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode) { 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); } /* * mark_index_clustered: mark the specified index as the one clustered on * * With indexOid == InvalidOid, will mark all indexes of rel not-clustered. */ void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal) { 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); } /* * Check if the CONCURRENTLY option is legal for the relation. * * *Ident_idx_p receives OID of the identity index. */ static void check_concurrent_repack_requirements(Relation rel, Oid *ident_idx_p) { char relpersistence, replident; Oid ident_idx; if (wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("cannot execute %s in this configuration", "REPACK (CONCURRENTLY)"), errdetail("%s requires \"wal_level\" to be set to \"replica\" or higher.", "REPACK (CONCURRENTLY)")); /* Data changes in system relations are not logically decoded. */ if (IsCatalogRelation(rel)) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errhint("%s is not supported for catalog relations.", "REPACK (CONCURRENTLY)")); /* * reorderbuffer.c does not seem to handle processing of TOAST relation * alone. */ if (IsToastRelation(rel)) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errhint("%s is not supported for TOAST relations.", "REPACK (CONCURRENTLY)")); relpersistence = rel->rd_rel->relpersistence; if (relpersistence != RELPERSISTENCE_PERMANENT) ereport(ERROR, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errhint("%s is only allowed for permanent relations.", "REPACK (CONCURRENTLY)")); /* * With NOTHING, WAL does not contain the old tuple; FULL is not yet * supported. */ replident = rel->rd_rel->relreplident; if (replident == REPLICA_IDENTITY_NOTHING || replident == REPLICA_IDENTITY_FULL) ereport(ERROR, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errdetail("%s does not support tables with %s.", "REPACK (CONCURRENTLY)", replident == REPLICA_IDENTITY_NOTHING ? "REPLICA IDENTITY NOTHING" : "REPLICA IDENTITY FULL")); /* * Obtain the replica identity index -- either one that has been set * explicitly, or a non-deferrable primary key. If none of these cases * apply, the table cannot be repacked concurrently. It might be possible * to have repack work with a FULL replica identity; however that requires * more work and is not implemented yet. */ ident_idx = GetRelationIdentityOrPK(rel); if (!OidIsValid(ident_idx)) { /* This special case warrants its own error message */ if (OidIsValid(rel->rd_pkindex) && rel->rd_ispkdeferrable) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errdetail("%s does not support deferrable primary keys.", "REPACK (CONCURRENTLY)"), errhint("Use ALTER TABLE ... REPLICA IDENTITY USING INDEX to designate another index as replica identity.")); ereport(ERROR, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("cannot execute %s on relation \"%s\"", "REPACK (CONCURRENTLY)", RelationGetRelationName(rel)), errhint("Relation \"%s\" has no identity index.", RelationGetRelationName(rel))); } *ident_idx_p = ident_idx; } /* * rebuild_relation: rebuild an existing relation in index or physical order * * OldHeap: table to rebuild. See cluster_rel() for comments on the required * lock strength. * * index: index to cluster by, or NULL to rewrite in physical order. * * ident_idx: identity index, to handle replaying of concurrent data changes * to the new heap. InvalidOid if there's no CONCURRENTLY option. * * On entry, heap and index (if one is given) must be open, and the * appropriate lock held on them -- AccessExclusiveLock for exclusive * processing and ShareUpdateExclusiveLock for concurrent processing. * * On exit, they are closed, but still locked with AccessExclusiveLock. * (The function handles the lock upgrade if 'concurrent' is true.) */ static void rebuild_relation(Relation OldHeap, Relation index, bool verbose, Oid ident_idx) { Oid tableOid = RelationGetRelid(OldHeap); Oid accessMethod = OldHeap->rd_rel->relam; Oid tableSpace = OldHeap->rd_rel->reltablespace; Oid OIDNewHeap; Relation NewHeap; char relpersistence; bool swap_toast_by_content; TransactionId frozenXid; MultiXactId cutoffMulti; bool concurrent = OidIsValid(ident_idx); Snapshot snapshot = NULL; #if USE_ASSERT_CHECKING LOCKMODE lmode; lmode = RepackLockLevel(concurrent); Assert(CheckRelationLockedByMe(OldHeap, lmode, false)); Assert(index == NULL || CheckRelationLockedByMe(index, lmode, false)); #endif if (concurrent) { /* * The worker needs to be member of the locking group we're the leader * of. We ought to become the leader before the worker starts. The * worker will join the group as soon as it starts. * * This is to make sure that the deadlock described below is * detectable by deadlock.c: if the worker waits for a transaction to * complete and we are waiting for the worker output, then effectively * we (i.e. this backend) are waiting for that transaction. */ BecomeLockGroupLeader(); /* * Start the worker that decodes data changes applied while we're * copying the table contents. * * Note that the worker has to wait for all transactions with XID * already assigned to finish. If some of those transactions is * waiting for a lock conflicting with ShareUpdateExclusiveLock on our * table (e.g. it runs CREATE INDEX), we can end up in a deadlock. * Not sure this risk is worth unlocking/locking the table (and its * clustering index) and checking again if it's still eligible for * REPACK CONCURRENTLY. */ start_repack_decoding_worker(tableOid); /* * Wait until the worker has the initial snapshot and retrieve it. */ snapshot = get_initial_snapshot(decoding_worker); PushActiveSnapshot(snapshot); } /* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */ if (index != NULL) mark_index_clustered(OldHeap, RelationGetRelid(index), true); /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; /* * Create the transient table that will receive the re-ordered data. * * OldHeap is already locked, so no need to lock it again. make_new_heap * obtains AccessExclusiveLock on the new heap and its toast table. */ OIDNewHeap = make_new_heap(tableOid, tableSpace, accessMethod, relpersistence, NoLock); Assert(CheckRelationOidLockedByMe(OIDNewHeap, AccessExclusiveLock, false)); NewHeap = table_open(OIDNewHeap, NoLock); /* * In concurrent mode, create a copy of the attribute defaults on the temp * table, which the executor needs when replaying concurrent data changes. */ if (concurrent) copy_attribute_defaults(tableOid, OIDNewHeap); /* Copy the heap data into the new table in the desired order */ copy_table_data(NewHeap, OldHeap, index, snapshot, verbose, &swap_toast_by_content, &frozenXid, &cutoffMulti); /* The historic snapshot won't be needed anymore. */ if (snapshot) { PopActiveSnapshot(); UpdateActiveSnapshotCommandId(); } if (concurrent) { Assert(!swap_toast_by_content); /* * Close the index, but keep the lock. Both heaps will be closed by * the following call. */ if (index) index_close(index, NoLock); rebuild_relation_finish_concurrent(NewHeap, OldHeap, ident_idx, frozenXid, cutoffMulti); pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_FINAL_CLEANUP); } else { bool is_system_catalog = IsSystemRelation(OldHeap); /* Close relcache entries, but keep lock until transaction commit */ table_close(OldHeap, NoLock); if (index) index_close(index, NoLock); /* * Close the new relation so it can be dropped as soon as the storage * is swapped. The relation is not visible to others, so no need to * unlock it explicitly. */ table_close(NewHeap, NoLock); /* * Swap the physical files of the target and transient tables, then * rebuild the target's indexes and throw away the transient table. */ finish_heap_swap(tableOid, OIDNewHeap, is_system_catalog, swap_toast_by_content, false, true, true, /* reindex */ frozenXid, cutoffMulti, relpersistence); } } /* * Create the transient table that will be filled with new data during * CLUSTER, ALTER TABLE, and similar operations. The transient table * duplicates the logical structure of the OldHeap; but will have the * specified physical storage properties NewTableSpace, NewAccessMethod, and * relpersistence. * * After this, the caller should load the new heap with transferred/modified * data, then call finish_heap_swap to complete the operation. */ Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode) { 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; } /* * Do the physical copying of table data. * * 'snapshot' and 'decoding_ctx': see table_relation_copy_for_cluster(). Pass * iff concurrent processing is required. * * There are three output parameters: * *pSwapToastByContent is set true if toast tables must be swapped by content. * *pFreezeXid receives the TransactionId used as freeze cutoff point. * *pCutoffMulti receives the MultiXactId used as a cutoff point. */ static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, Snapshot snapshot, bool verbose, bool *pSwapToastByContent, TransactionId *pFreezeXid, MultiXactId *pCutoffMulti) { Relation relRelation; HeapTuple reltup; Form_pg_class relform; TupleDesc oldTupDesc PG_USED_FOR_ASSERTS_ONLY; TupleDesc newTupDesc PG_USED_FOR_ASSERTS_ONLY; VacuumParams params; struct VacuumCutoffs cutoffs; bool use_sort; double num_tuples = 0, tups_vacuumed = 0, tups_recently_dead = 0; BlockNumber num_pages; int elevel = verbose ? INFO : DEBUG2; PGRUsage ru0; char *nspname; bool concurrent = snapshot != NULL; LOCKMODE lmode; lmode = RepackLockLevel(concurrent); pg_rusage_init(&ru0); /* Store a copy of the namespace name for logging purposes */ nspname = get_namespace_name(RelationGetNamespace(OldHeap)); /* * Their tuple descriptors should be exactly alike, but here we only need * assume that they have the same number of columns. */ oldTupDesc = RelationGetDescr(OldHeap); newTupDesc = RelationGetDescr(NewHeap); Assert(newTupDesc->natts == oldTupDesc->natts); /* * If the OldHeap has a toast table, get lock on the toast table to keep * it from being vacuumed. This is needed because autovacuum processes * toast tables independently of their main tables, with no lock on the * latter. If an autovacuum were to start on the toast table after we * compute our OldestXmin below, it would use a later OldestXmin, and then * possibly remove as DEAD toast tuples belonging to main tuples we think * are only RECENTLY_DEAD. Then we'd fail while trying to copy those * tuples. * * We don't need to open the toast relation here, just lock it. The lock * will be held till end of transaction. */ if (OldHeap->rd_rel->reltoastrelid) LockRelationOid(OldHeap->rd_rel->reltoastrelid, lmode); /* * If both tables have TOAST tables, perform toast swap by content. It is * possible that the old table has a toast table but the new one doesn't, * if toastable columns have been dropped. In that case we have to do * swap by links. This is okay because swap by content is only essential * for system catalogs, and we don't support schema changes for them. */ if (OldHeap->rd_rel->reltoastrelid && NewHeap->rd_rel->reltoastrelid && !concurrent) { *pSwapToastByContent = true; /* * When doing swap by content, any toast pointers written into NewHeap * must use the old toast table's OID, because that's where the toast * data will eventually be found. Set this up by setting rd_toastoid. * This also tells toast_save_datum() to preserve the toast value * OIDs, which we want so as not to invalidate toast pointers in * system catalog caches, and to avoid making multiple copies of a * single toast value. * * Note that we must hold NewHeap open until we are done writing data, * since the relcache will not guarantee to remember this setting once * the relation is closed. Also, this technique depends on the fact * that no one will try to read from the NewHeap until after we've * finished writing it and swapping the rels --- otherwise they could * follow the toast pointers to the wrong place. (It would actually * work for values copied over from the old toast table, but not for * any values that we toast which were previously not toasted.) * * This would not work with CONCURRENTLY because we may need to delete * TOASTed tuples from the new heap. With this hack, we'd delete them * from the old heap. */ NewHeap->rd_toastoid = OldHeap->rd_rel->reltoastrelid; } else *pSwapToastByContent = false; /* * Compute xids used to freeze and weed out dead tuples and multixacts. * Since we're going to rewrite the whole table anyway, there's no reason * not to be aggressive about this. */ memset(¶ms, 0, sizeof(VacuumParams)); vacuum_get_cutoffs(OldHeap, ¶ms, &cutoffs); /* * FreezeXid will become the table's new relfrozenxid, and that mustn't go * backwards, so take the max. */ { TransactionId relfrozenxid = OldHeap->rd_rel->relfrozenxid; if (TransactionIdIsValid(relfrozenxid) && TransactionIdPrecedes(cutoffs.FreezeLimit, relfrozenxid)) cutoffs.FreezeLimit = relfrozenxid; } /* * MultiXactCutoff, similarly, shouldn't go backwards either. */ { MultiXactId relminmxid = OldHeap->rd_rel->relminmxid; if (MultiXactIdIsValid(relminmxid) && MultiXactIdPrecedes(cutoffs.MultiXactCutoff, relminmxid)) cutoffs.MultiXactCutoff = relminmxid; } /* * Decide whether to use an indexscan or seqscan-and-optional-sort to scan * the OldHeap. We know how to use a sort to duplicate the ordering of a * btree index, and will use seqscan-and-sort for that case if the planner * tells us it's cheaper. Otherwise, always indexscan if an index is * provided, else plain seqscan. */ if (OldIndex != NULL && OldIndex->rd_rel->relam == BTREE_AM_OID) use_sort = plan_cluster_use_sort(RelationGetRelid(OldHeap), RelationGetRelid(OldIndex)); else use_sort = false; /* Log what we're doing */ if (OldIndex != NULL && !use_sort) ereport(elevel, errmsg("repacking \"%s.%s\" using index scan on \"%s\"", nspname, RelationGetRelationName(OldHeap), RelationGetRelationName(OldIndex))); else if (use_sort) ereport(elevel, errmsg("repacking \"%s.%s\" using sequential scan and sort", nspname, RelationGetRelationName(OldHeap))); else ereport(elevel, errmsg("repacking \"%s.%s\" in physical order", nspname, RelationGetRelationName(OldHeap))); /* * Hand off the actual copying to AM specific function, the generic code * cannot know how to deal with visibility across AMs. Note that this * routine is allowed to set FreezeXid / MultiXactCutoff to different * values (e.g. because the AM doesn't use freezing). */ table_relation_copy_for_cluster(OldHeap, NewHeap, OldIndex, use_sort, cutoffs.OldestXmin, snapshot, &cutoffs.FreezeLimit, &cutoffs.MultiXactCutoff, &num_tuples, &tups_vacuumed, &tups_recently_dead); /* return selected values to caller, get set as relfrozenxid/minmxid */ *pFreezeXid = cutoffs.FreezeLimit; *pCutoffMulti = cutoffs.MultiXactCutoff; /* * Reset rd_toastoid just to be tidy --- it shouldn't be looked at again. * In the CONCURRENTLY case, we need to set it again before applying the * concurrent changes. */ NewHeap->rd_toastoid = InvalidOid; num_pages = RelationGetNumberOfBlocks(NewHeap); /* Log what we did */ ereport(elevel, (errmsg("\"%s.%s\": found %.0f removable, %.0f nonremovable row versions in %u pages", nspname, RelationGetRelationName(OldHeap), tups_vacuumed, num_tuples, RelationGetNumberOfBlocks(OldHeap)), errdetail("%.0f dead row versions cannot be removed yet.\n" "%s.", tups_recently_dead, pg_rusage_show(&ru0)))); /* Update pg_class to reflect the correct values of pages and tuples. */ relRelation = table_open(RelationRelationId, RowExclusiveLock); reltup = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(RelationGetRelid(NewHeap))); if (!HeapTupleIsValid(reltup)) elog(ERROR, "cache lookup failed for relation %u", RelationGetRelid(NewHeap)); relform = (Form_pg_class) GETSTRUCT(reltup); relform->relpages = num_pages; relform->reltuples = num_tuples; /* Don't update the stats for pg_class. See swap_relation_files. */ if (RelationGetRelid(OldHeap) != RelationRelationId) CatalogTupleUpdate(relRelation, &reltup->t_self, reltup); else CacheInvalidateRelcacheByTuple(reltup); /* Clean up. */ heap_freetuple(reltup); table_close(relRelation, RowExclusiveLock); /* Make the update visible */ CommandCounterIncrement(); } /* * Swap the physical files of two given relations. * * We swap the physical identity (reltablespace, relfilenumber) while keeping * the same logical identities of the two relations. relpersistence is also * swapped, which is critical since it determines where buffers live for each * relation. * * We can swap associated TOAST data in either of two ways: recursively swap * the physical content of the toast tables (and their indexes), or swap the * TOAST links in the given relations' pg_class entries. The former is needed * to manage rewrites of shared catalogs (where we cannot change the pg_class * links) while the latter is the only way to handle cases in which a toast * table is added or removed altogether. * * Additionally, the first relation is marked with relfrozenxid set to * frozenXid. It seems a bit ugly to have this here, but the caller would * have to do it anyway, so having it here saves a heap_update. Note: in * the swap-toast-links case, we assume we don't need to change the toast * table's relfrozenxid: the new version of the toast table should already * have relfrozenxid set to RecentXmin, which is good enough. * * Lastly, if r2 and its toast table and toast index (if any) are mapped, * their OIDs are emitted into mapped_tables[]. This is hacky but beats * having to look the information up again later in finish_heap_swap. */ static void swap_relation_files(Oid r1, Oid r2, bool target_is_pg_class, bool swap_toast_by_content, bool is_internal, TransactionId frozenXid, MultiXactId cutoffMulti, Oid *mapped_tables) { Relation relRelation; HeapTuple reltup1, reltup2; Form_pg_class relform1, relform2; RelFileNumber relfilenumber1, relfilenumber2; RelFileNumber swaptemp; char swptmpchr; Oid relam1, relam2; /* We need writable copies of both pg_class tuples. */ relRelation = table_open(RelationRelationId, RowExclusiveLock); reltup1 = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(r1)); if (!HeapTupleIsValid(reltup1)) elog(ERROR, "cache lookup failed for relation %u", r1); relform1 = (Form_pg_class) GETSTRUCT(reltup1); reltup2 = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(r2)); if (!HeapTupleIsValid(reltup2)) elog(ERROR, "cache lookup failed for relation %u", r2); relform2 = (Form_pg_class) GETSTRUCT(reltup2); relfilenumber1 = relform1->relfilenode; relfilenumber2 = relform2->relfilenode; relam1 = relform1->relam; relam2 = relform2->relam; if (RelFileNumberIsValid(relfilenumber1) && RelFileNumberIsValid(relfilenumber2)) { /* * Normal non-mapped relations: swap relfilenumbers, reltablespaces, * relpersistence */ Assert(!target_is_pg_class); swaptemp = relform1->relfilenode; relform1->relfilenode = relform2->relfilenode; relform2->relfilenode = swaptemp; swaptemp = relform1->reltablespace; relform1->reltablespace = relform2->reltablespace; relform2->reltablespace = swaptemp; swaptemp = relform1->relam; relform1->relam = relform2->relam; relform2->relam = swaptemp; swptmpchr = relform1->relpersistence; relform1->relpersistence = relform2->relpersistence; relform2->relpersistence = swptmpchr; /* Also swap toast links, if we're swapping by links */ if (!swap_toast_by_content) { swaptemp = relform1->reltoastrelid; relform1->reltoastrelid = relform2->reltoastrelid; relform2->reltoastrelid = swaptemp; } } else { /* * Mapped-relation case. Here we have to swap the relation mappings * instead of modifying the pg_class columns. Both must be mapped. */ if (RelFileNumberIsValid(relfilenumber1) || RelFileNumberIsValid(relfilenumber2)) elog(ERROR, "cannot swap mapped relation \"%s\" with non-mapped relation", NameStr(relform1->relname)); /* * We can't change the tablespace nor persistence of a mapped rel, and * we can't handle toast link swapping for one either, because we must * not apply any critical changes to its pg_class row. These cases * should be prevented by upstream permissions tests, so these checks * are non-user-facing emergency backstop. */ if (relform1->reltablespace != relform2->reltablespace) elog(ERROR, "cannot change tablespace of mapped relation \"%s\"", NameStr(relform1->relname)); if (relform1->relpersistence != relform2->relpersistence) elog(ERROR, "cannot change persistence of mapped relation \"%s\"", NameStr(relform1->relname)); if (relform1->relam != relform2->relam) elog(ERROR, "cannot change access method of mapped relation \"%s\"", NameStr(relform1->relname)); if (!swap_toast_by_content && (relform1->reltoastrelid || relform2->reltoastrelid)) elog(ERROR, "cannot swap toast by links for mapped relation \"%s\"", NameStr(relform1->relname)); /* * Fetch the mappings --- shouldn't fail, but be paranoid */ relfilenumber1 = RelationMapOidToFilenumber(r1, relform1->relisshared); if (!RelFileNumberIsValid(relfilenumber1)) elog(ERROR, "could not find relation mapping for relation \"%s\", OID %u", NameStr(relform1->relname), r1); relfilenumber2 = RelationMapOidToFilenumber(r2, relform2->relisshared); if (!RelFileNumberIsValid(relfilenumber2)) elog(ERROR, "could not find relation mapping for relation \"%s\", OID %u", NameStr(relform2->relname), r2); /* * Send replacement mappings to relmapper. Note these won't actually * take effect until CommandCounterIncrement. */ RelationMapUpdateMap(r1, relfilenumber2, relform1->relisshared, false); RelationMapUpdateMap(r2, relfilenumber1, relform2->relisshared, false); /* Pass OIDs of mapped r2 tables back to caller */ *mapped_tables++ = r2; } /* * Recognize that rel1's relfilenumber (swapped from rel2) is new in this * subtransaction. The rel2 storage (swapped from rel1) may or may not be * new. */ { Relation rel1, rel2; rel1 = relation_open(r1, NoLock); rel2 = relation_open(r2, NoLock); rel2->rd_createSubid = rel1->rd_createSubid; rel2->rd_newRelfilelocatorSubid = rel1->rd_newRelfilelocatorSubid; rel2->rd_firstRelfilelocatorSubid = rel1->rd_firstRelfilelocatorSubid; RelationAssumeNewRelfilelocator(rel1); relation_close(rel1, NoLock); relation_close(rel2, NoLock); } /* * In the case of a shared catalog, these next few steps will only affect * our own database's pg_class row; but that's okay, because they are all * noncritical updates. That's also an important fact for the case of a * mapped catalog, because it's possible that we'll commit the map change * and then fail to commit the pg_class update. */ /* set rel1's frozen Xid and minimum MultiXid */ if (relform1->relkind != RELKIND_INDEX) { Assert(!TransactionIdIsValid(frozenXid) || TransactionIdIsNormal(frozenXid)); relform1->relfrozenxid = frozenXid; relform1->relminmxid = cutoffMulti; } /* swap size statistics too, since new rel has freshly-updated stats */ { int32 swap_pages; float4 swap_tuples; int32 swap_allvisible; int32 swap_allfrozen; swap_pages = relform1->relpages; relform1->relpages = relform2->relpages; relform2->relpages = swap_pages; swap_tuples = relform1->reltuples; relform1->reltuples = relform2->reltuples; relform2->reltuples = swap_tuples; swap_allvisible = relform1->relallvisible; relform1->relallvisible = relform2->relallvisible; relform2->relallvisible = swap_allvisible; swap_allfrozen = relform1->relallfrozen; relform1->relallfrozen = relform2->relallfrozen; relform2->relallfrozen = swap_allfrozen; } /* * Update the tuples in pg_class --- unless the target relation of the * swap is pg_class itself. In that case, there is zero point in making * changes because we'd be updating the old data that we're about to throw * away. Because the real work being done here for a mapped relation is * just to change the relation map settings, it's all right to not update * the pg_class rows in this case. The most important changes will instead * performed later, in finish_heap_swap() itself. */ if (!target_is_pg_class) { CatalogIndexState indstate; indstate = CatalogOpenIndexes(relRelation); CatalogTupleUpdateWithInfo(relRelation, &reltup1->t_self, reltup1, indstate); CatalogTupleUpdateWithInfo(relRelation, &reltup2->t_self, reltup2, indstate); CatalogCloseIndexes(indstate); } else { /* no update ... but we do still need relcache inval */ CacheInvalidateRelcacheByTuple(reltup1); CacheInvalidateRelcacheByTuple(reltup2); } /* * Now that pg_class has been updated with its relevant information for * the swap, update the dependency of the relations to point to their new * table AM, if it has changed. */ if (relam1 != relam2) { if (changeDependencyFor(RelationRelationId, r1, AccessMethodRelationId, relam1, relam2) != 1) elog(ERROR, "could not change access method dependency for relation \"%s.%s\"", get_namespace_name(get_rel_namespace(r1)), get_rel_name(r1)); if (changeDependencyFor(RelationRelationId, r2, AccessMethodRelationId, relam2, relam1) != 1) elog(ERROR, "could not change access method dependency for relation \"%s.%s\"", get_namespace_name(get_rel_namespace(r2)), get_rel_name(r2)); } /* * Post alter hook for modified relations. The change to r2 is always * internal, but r1 depends on the invocation context. */ InvokeObjectPostAlterHookArg(RelationRelationId, r1, 0, InvalidOid, is_internal); InvokeObjectPostAlterHookArg(RelationRelationId, r2, 0, InvalidOid, true); /* * If we have toast tables associated with the relations being swapped, * deal with them too. */ if (relform1->reltoastrelid || relform2->reltoastrelid) { if (swap_toast_by_content) { if (relform1->reltoastrelid && relform2->reltoastrelid) { /* Recursively swap the contents of the toast tables */ swap_relation_files(relform1->reltoastrelid, relform2->reltoastrelid, target_is_pg_class, swap_toast_by_content, is_internal, frozenXid, cutoffMulti, mapped_tables); } else { /* caller messed up */ elog(ERROR, "cannot swap toast files by content when there's only one"); } } else { /* * We swapped the ownership links, so we need to change dependency * data to match. * * NOTE: it is possible that only one table has a toast table. * * NOTE: at present, a TOAST table's only dependency is the one on * its owning table. If more are ever created, we'd need to use * something more selective than deleteDependencyRecordsFor() to * get rid of just the link we want. */ ObjectAddress baseobject, toastobject; long count; /* * We disallow this case for system catalogs, to avoid the * possibility that the catalog we're rebuilding is one of the * ones the dependency changes would change. It's too late to be * making any data changes to the target catalog. */ if (IsSystemClass(r1, relform1)) elog(ERROR, "cannot swap toast files by links for system catalogs"); /* Delete old dependencies */ if (relform1->reltoastrelid) { count = deleteDependencyRecordsFor(RelationRelationId, relform1->reltoastrelid, false); if (count != 1) elog(ERROR, "expected one dependency record for TOAST table, found %ld", count); } if (relform2->reltoastrelid) { count = deleteDependencyRecordsFor(RelationRelationId, relform2->reltoastrelid, false); if (count != 1) elog(ERROR, "expected one dependency record for TOAST table, found %ld", count); } /* Register new dependencies */ baseobject.classId = RelationRelationId; baseobject.objectSubId = 0; toastobject.classId = RelationRelationId; toastobject.objectSubId = 0; if (relform1->reltoastrelid) { baseobject.objectId = r1; toastobject.objectId = relform1->reltoastrelid; recordDependencyOn(&toastobject, &baseobject, DEPENDENCY_INTERNAL); } if (relform2->reltoastrelid) { baseobject.objectId = r2; toastobject.objectId = relform2->reltoastrelid; recordDependencyOn(&toastobject, &baseobject, DEPENDENCY_INTERNAL); } } } /* * If we're swapping two toast tables by content, do the same for their * valid index. The swap can actually be safely done only if the relations * have indexes. */ if (swap_toast_by_content && relform1->relkind == RELKIND_TOASTVALUE && relform2->relkind == RELKIND_TOASTVALUE) { Oid toastIndex1, toastIndex2; /* Get valid index for each relation */ toastIndex1 = toast_get_valid_index(r1, AccessExclusiveLock); toastIndex2 = toast_get_valid_index(r2, AccessExclusiveLock); swap_relation_files(toastIndex1, toastIndex2, target_is_pg_class, swap_toast_by_content, is_internal, InvalidTransactionId, InvalidMultiXactId, mapped_tables); } /* Clean up. */ heap_freetuple(reltup1); heap_freetuple(reltup2); table_close(relRelation, RowExclusiveLock); } /* * Remove the transient table that was built by make_new_heap, and finish * cleaning up (including rebuilding all indexes on the old heap). */ 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) { 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); } } /* * Determine which relations to process, when REPACK/CLUSTER is called * without specifying a table name. The exact process depends on whether * USING INDEX was given or not, and in any case we only return tables and * materialized views that the current user has privileges to repack/cluster. * * If USING INDEX was given, we scan pg_index to find those that have * indisclustered set; if it was not given, scan pg_class and return all * tables. * * Return it as a list of RelToCluster in the given memory context. */ static List * get_tables_to_repack(RepackCommand cmd, bool usingindex, MemoryContext permcxt) { Relation catalog; TableScanDesc scan; HeapTuple tuple; List *rtcs = NIL; if (usingindex) { ScanKeyData entry; /* * For USING INDEX, scan pg_index to find those with indisclustered. * * Note we don't obtain lock of any kind on the index, which means the * index or its owning table could be gone or change at any point. We * have to be extra careful when examining catalog state for them. */ catalog = table_open(IndexRelationId, AccessShareLock); ScanKeyInit(&entry, Anum_pg_index_indisclustered, BTEqualStrategyNumber, F_BOOLEQ, BoolGetDatum(true)); scan = table_beginscan_catalog(catalog, 1, &entry); while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL) { RelToCluster *rtc; Form_pg_index index; HeapTuple classtup; Oid relnamespace; char relpersistence; MemoryContext oldcxt; index = (Form_pg_index) GETSTRUCT(tuple); classtup = SearchSysCache1(RELOID, ObjectIdGetDatum(index->indrelid)); if (!HeapTupleIsValid(classtup)) continue; relnamespace = ((Form_pg_class) GETSTRUCT(classtup))->relnamespace; relpersistence = ((Form_pg_class) GETSTRUCT(classtup))->relpersistence; ReleaseSysCache(classtup); /* Skip temp relations belonging to other sessions */ if (relpersistence == RELPERSISTENCE_TEMP && !isTempOrTempToastNamespace(relnamespace)) continue; /* noisily skip rels which the user can't process */ if (!repack_is_permitted_for_relation(cmd, index->indrelid, GetUserId())) continue; /* Use a permanent memory context for the result list */ oldcxt = MemoryContextSwitchTo(permcxt); rtc = palloc_object(RelToCluster); rtc->tableOid = index->indrelid; rtc->indexOid = index->indexrelid; rtcs = lappend(rtcs, rtc); MemoryContextSwitchTo(oldcxt); } } else { catalog = table_open(RelationRelationId, AccessShareLock); scan = table_beginscan_catalog(catalog, 0, NULL); while ((tuple = heap_getnext(scan, ForwardScanDirection)) != NULL) { RelToCluster *rtc; Form_pg_class class; MemoryContext oldcxt; class = (Form_pg_class) GETSTRUCT(tuple); /* Can only process plain tables and matviews */ if (class->relkind != RELKIND_RELATION && class->relkind != RELKIND_MATVIEW) continue; /* Skip temp relations belonging to other sessions */ if (class->relpersistence == RELPERSISTENCE_TEMP && !isTempOrTempToastNamespace(class->relnamespace)) continue; /* noisily skip rels which the user can't process */ if (!repack_is_permitted_for_relation(cmd, class->oid, GetUserId())) continue; /* Use a permanent memory context for the result list */ oldcxt = MemoryContextSwitchTo(permcxt); rtc = palloc_object(RelToCluster); rtc->tableOid = class->oid; rtc->indexOid = InvalidOid; rtcs = lappend(rtcs, rtc); MemoryContextSwitchTo(oldcxt); } } table_endscan(scan); table_close(catalog, AccessShareLock); return rtcs; } /* * Determine relations to process, when REPACK/CLUSTER is called with a * partitioning table; that is, a list of its leaf partitions. That table has * already been opened by caller and is passed as 'rel'. It is closed and * unlocked here before return, so caller should clobber its pointer to avoid * confusion. * * Return it as a list of RelToCluster. * * XXX we don't support CONCURRENTLY for partitioned tables yet. */ static List * get_tables_to_repack_partitioned(RepackStmt *stmt, Relation rel, MemoryContext permcxt) { Oid relid; bool rel_is_index; List *inhoids; List *rtcs = NIL; Assert(rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE); Assert(CheckRelationLockedByMe(rel, AccessExclusiveLock, false)); /* * We find the list of tables by looking for inheritors. If USING INDEX * was given, look for inheritors of that index, whose name we resolve * now. * * Otherwise we look for inheritors of the table itself. */ 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; } /* * Do not lock the children until they're processed. Note that we do hold * a lock on the parent partitioned table. */ inhoids = find_all_inheritors(relid, NoLock, NULL); foreach_oid(child_oid, inhoids) { Oid table_oid, index_oid; RelToCluster *rtc; MemoryContext oldcxt; if (rel_is_index) { /* consider only leaf indexes */ if (get_rel_relkind(child_oid) != RELKIND_INDEX) continue; /* * Although we do have a lock on some ancestor partitioned index, * we may not have one on the immediate parent, so this lookup may * still return invalid. */ table_oid = IndexGetRelation(child_oid, true); if (!OidIsValid(table_oid)) continue; index_oid = child_oid; } else { /* consider only leaf relations */ if (get_rel_relkind(child_oid) != RELKIND_RELATION) continue; table_oid = child_oid; index_oid = InvalidOid; } /* * It's possible that the user does not have privileges to CLUSTER the * leaf partition despite having them on the partitioned table. Skip * if so. */ if (!repack_is_permitted_for_relation(stmt->command, table_oid, GetUserId())) continue; /* Use a permanent memory context for the result list */ oldcxt = MemoryContextSwitchTo(permcxt); rtc = palloc_object(RelToCluster); rtc->tableOid = table_oid; rtc->indexOid = index_oid; rtcs = lappend(rtcs, rtc); MemoryContextSwitchTo(oldcxt); } /* close parent relation, releasing lock on it */ table_close(rel, AccessExclusiveLock); return rtcs; } /* * Return whether userid has privileges to execute REPACK on relid. * * Caller may not have a lock on the relation, so it could have been * dropped concurrently. In that case, silently return false. * * If the relation does exist but the user doesn't have the required * privs, emit a WARNING and return false. Otherwise, return true. */ static bool repack_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid) { bool is_missing = false; AclResult result; char *relname; Assert(cmd == REPACK_COMMAND_CLUSTER || cmd == REPACK_COMMAND_REPACK); result = pg_class_aclcheck_ext(relid, userid, ACL_MAINTAIN, &is_missing); if (is_missing) return false; if (result == ACLCHECK_OK) return true; /* * The relation can also be dropped after we tested its ACL and before we * read its relname, so be careful here. */ relname = get_rel_name(relid); if (relname != NULL) { ereport(WARNING, errmsg("permission denied to execute %s on \"%s\", skipping it", RepackCommandAsString(cmd), relname)); pfree(relname); } return false; } /* * Given a RepackStmt with an indicated relation name, resolve the relation * name, obtain lock on it, then determine what to do based on the relation * type: if it's table and not partitioned, repack it as indicated (using an * existing clustered index, or following the given one), and return NULL. * * On the other hand, if the table is partitioned, do nothing further and * instead return the opened and locked relcache entry, so that caller can * process the partitions using the multiple-table handling code. In this * case, if an index name is given, it's up to the caller to resolve it. */ static Relation process_single_relation(RepackStmt *stmt, LOCKMODE lockmode, bool isTopLevel, ClusterParams *params) { Relation rel; Oid tableOid; Assert(stmt->relation != NULL); Assert(stmt->command == REPACK_COMMAND_CLUSTER || stmt->command == REPACK_COMMAND_REPACK); /* * Make sure ANALYZE is specified if a column list is present. */ if ((params->options & CLUOPT_ANALYZE) == 0 && stmt->relation->va_cols != NIL) ereport(ERROR, errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("ANALYZE option must be specified when a column list is provided")); /* Find, lock, and check permissions on the table. */ tableOid = RangeVarGetRelidExtended(stmt->relation->relation, lockmode, 0, RangeVarCallbackMaintainsTable, NULL); rel = table_open(tableOid, NoLock); /* * Reject clustering a remote temp table ... their local buffer manager is * not going to cope. */ if (RELATION_IS_OTHER_TEMP(rel)) 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(stmt->command))); /* * For partitioned tables, let caller handle this. Otherwise, process it * here and we're done. */ if (rel->rd_rel->relkind == RELKIND_PARTITIONED_TABLE) return rel; else { Oid indexOid = InvalidOid; indexOid = determine_clustered_index(rel, stmt->usingindex, stmt->indexname); if (OidIsValid(indexOid)) check_index_is_clusterable(rel, indexOid, lockmode); cluster_rel(stmt->command, rel, indexOid, params, isTopLevel); /* * Do an analyze, if requested. We close the transaction and start a * new one, so that we don't hold the stronger lock for longer than * needed. */ if (params->options & CLUOPT_ANALYZE) { VacuumParams vac_params = {0}; PopActiveSnapshot(); CommitTransactionCommand(); StartTransactionCommand(); PushActiveSnapshot(GetTransactionSnapshot()); vac_params.options |= VACOPT_ANALYZE; if (params->options & CLUOPT_VERBOSE) vac_params.options |= VACOPT_VERBOSE; analyze_rel(tableOid, NULL, &vac_params, stmt->relation->va_cols, true, NULL); PopActiveSnapshot(); CommandCounterIncrement(); } return NULL; } } /* * Given a relation and the usingindex/indexname options in a * REPACK USING INDEX or CLUSTER command, return the OID of the * index to use for clustering the table. * * Caller must hold lock on the relation so that the set of indexes * doesn't change, and must call check_index_is_clusterable. */ static Oid determine_clustered_index(Relation rel, bool usingindex, const char *indexname) { Oid indexOid; if (indexname == NULL && usingindex) { /* * If USING INDEX with no name is given, find a clustered index, or * error out if none. */ indexOid = InvalidOid; foreach_oid(idxoid, RelationGetIndexList(rel)) { if (get_index_isclustered(idxoid)) { indexOid = idxoid; break; } } if (!OidIsValid(indexOid)) ereport(ERROR, errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("there is no previously clustered index for table \"%s\"", RelationGetRelationName(rel))); } else if (indexname != NULL) { /* An index was specified; obtain its OID. */ indexOid = get_relname_relid(indexname, rel->rd_rel->relnamespace); if (!OidIsValid(indexOid)) ereport(ERROR, errcode(ERRCODE_UNDEFINED_OBJECT), errmsg("index \"%s\" for table \"%s\" does not exist", indexname, RelationGetRelationName(rel))); } else indexOid = InvalidOid; return indexOid; } static const char * RepackCommandAsString(RepackCommand cmd) { switch (cmd) { case REPACK_COMMAND_REPACK: return "REPACK"; case REPACK_COMMAND_VACUUMFULL: return "VACUUM"; case REPACK_COMMAND_CLUSTER: return "CLUSTER"; } return "???"; /* keep compiler quiet */ } /* * Apply all the changes stored in 'file'. */ static void apply_concurrent_changes(BufFile *file, ChangeContext *chgcxt) { ConcurrentChangeKind kind = '\0'; Relation rel = chgcxt->cc_rel; TupleTableSlot *spilled_tuple; TupleTableSlot *old_update_tuple; TupleTableSlot *ondisk_tuple; bool have_old_tuple = false; MemoryContext oldcxt; spilled_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), &TTSOpsVirtual); ondisk_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), table_slot_callbacks(rel)); old_update_tuple = MakeSingleTupleTableSlot(RelationGetDescr(rel), &TTSOpsVirtual); oldcxt = MemoryContextSwitchTo(GetPerTupleMemoryContext(chgcxt->cc_estate)); while (true) { size_t nread; ConcurrentChangeKind prevkind = kind; CHECK_FOR_INTERRUPTS(); nread = BufFileReadMaybeEOF(file, &kind, 1, true); if (nread == 0) /* done with the file? */ break; /* * If this is the old tuple for an update, read it into the tuple slot * and go to the next one. The update itself will be executed on the * next iteration, when we receive the NEW tuple. */ if (kind == CHANGE_UPDATE_OLD) { restore_tuple(file, rel, old_update_tuple); have_old_tuple = true; continue; } /* * Just before an UPDATE or DELETE, we must update the command * counter, because the change could refer to a tuple that we have * just inserted; and before an INSERT, we have to do this also if the * previous command was either update or delete. * * With this approach we don't spend so many CCIs for long strings of * only INSERTs, which can't affect one another. */ if (kind == CHANGE_UPDATE_NEW || kind == CHANGE_DELETE || (kind == CHANGE_INSERT && (prevkind == CHANGE_UPDATE_NEW || prevkind == CHANGE_DELETE))) { CommandCounterIncrement(); UpdateActiveSnapshotCommandId(); } /* * Now restore the tuple into the slot and execute the change. */ restore_tuple(file, rel, spilled_tuple); if (kind == CHANGE_INSERT) { apply_concurrent_insert(rel, spilled_tuple, chgcxt); } else if (kind == CHANGE_DELETE) { bool found; /* Find the tuple to be deleted */ found = find_target_tuple(rel, chgcxt, spilled_tuple, ondisk_tuple); if (!found) elog(ERROR, "could not find target tuple"); apply_concurrent_delete(rel, ondisk_tuple); } else if (kind == CHANGE_UPDATE_NEW) { TupleTableSlot *key; bool found; if (have_old_tuple) key = old_update_tuple; else key = spilled_tuple; /* Find the tuple to be updated or deleted. */ found = find_target_tuple(rel, chgcxt, key, ondisk_tuple); if (!found) elog(ERROR, "could not find target tuple"); /* * If 'tup' contains TOAST pointers, they point to the old * relation's toast. Copy the corresponding TOAST pointers for the * new relation from the existing tuple. (The fact that we * received a TOAST pointer here implies that the attribute hasn't * changed.) */ adjust_toast_pointers(rel, spilled_tuple, ondisk_tuple); apply_concurrent_update(rel, spilled_tuple, ondisk_tuple, chgcxt); ExecClearTuple(old_update_tuple); have_old_tuple = false; } else elog(ERROR, "unrecognized kind of change: %d", kind); ResetPerTupleExprContext(chgcxt->cc_estate); } /* Cleanup. */ ExecDropSingleTupleTableSlot(spilled_tuple); ExecDropSingleTupleTableSlot(ondisk_tuple); ExecDropSingleTupleTableSlot(old_update_tuple); MemoryContextSwitchTo(oldcxt); } /* * Apply an insert from the spill of concurrent changes to the new copy of the * table. */ static void apply_concurrent_insert(Relation rel, TupleTableSlot *slot, ChangeContext *chgcxt) { /* Put the tuple in the table, but make sure it won't be decoded */ table_tuple_insert(rel, slot, GetCurrentCommandId(true), TABLE_INSERT_NO_LOGICAL, NULL); /* Update indexes with this new tuple. */ ExecInsertIndexTuples(chgcxt->cc_rri, chgcxt->cc_estate, 0, slot, NIL, NULL); pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_INSERTED, 1); } /* * Apply an update from the spill of concurrent changes to the new copy of the * table. */ static void apply_concurrent_update(Relation rel, TupleTableSlot *spilled_tuple, TupleTableSlot *ondisk_tuple, ChangeContext *chgcxt) { LockTupleMode lockmode; TM_FailureData tmfd; TU_UpdateIndexes update_indexes; TM_Result res; /* * Carry out the update, skipping logical decoding for it. */ res = table_tuple_update(rel, &(ondisk_tuple->tts_tid), spilled_tuple, GetCurrentCommandId(true), TABLE_UPDATE_NO_LOGICAL, InvalidSnapshot, InvalidSnapshot, false, &tmfd, &lockmode, &update_indexes); if (res != TM_Ok) ereport(ERROR, errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), errmsg("could not apply concurrent %s on relation \"%s\"", "UPDATE", RelationGetRelationName(rel))); if (update_indexes != TU_None) { uint32 flags = EIIT_IS_UPDATE; if (update_indexes == TU_Summarizing) flags |= EIIT_ONLY_SUMMARIZING; ExecInsertIndexTuples(chgcxt->cc_rri, chgcxt->cc_estate, flags, spilled_tuple, NIL, NULL); } pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_UPDATED, 1); } static void apply_concurrent_delete(Relation rel, TupleTableSlot *slot) { TM_Result res; TM_FailureData tmfd; /* * Delete tuple from the new heap, skipping logical decoding for it. */ res = table_tuple_delete(rel, &(slot->tts_tid), GetCurrentCommandId(true), TABLE_DELETE_NO_LOGICAL, InvalidSnapshot, InvalidSnapshot, false, &tmfd); if (res != TM_Ok) ereport(ERROR, errcode(ERRCODE_T_R_SERIALIZATION_FAILURE), errmsg("could not apply concurrent %s on relation \"%s\"", "DELETE", RelationGetRelationName(rel))); pgstat_progress_incr_param(PROGRESS_REPACK_HEAP_TUPLES_DELETED, 1); } /* * Read tuple from file and put it in the input slot. All memory is allocated * in the current memory context; caller is responsible for freeing it as * appropriate. * * External attributes are stored in separate memory chunks, in order to avoid * exceeding MaxAllocSize - that could happen if the individual attributes are * smaller than MaxAllocSize but the whole tuple is bigger. */ static void restore_tuple(BufFile *file, Relation relation, TupleTableSlot *slot) { uint32 t_len; HeapTuple tup; int natt_ext; /* Read the tuple. */ BufFileReadExact(file, &t_len, sizeof(t_len)); tup = (HeapTuple) palloc(HEAPTUPLESIZE + t_len); tup->t_data = (HeapTupleHeader) ((char *) tup + HEAPTUPLESIZE); BufFileReadExact(file, tup->t_data, t_len); tup->t_len = t_len; ItemPointerSetInvalid(&tup->t_self); tup->t_tableOid = RelationGetRelid(relation); /* * Put the tuple we read in a slot. This deforms it, so that we can hack * the external attributes in place. */ ExecForceStoreHeapTuple(tup, slot, false); /* * Next, read any attributes we stored separately into the tts_values * array elements expecting them, if any. This matches * repack_store_change. */ BufFileReadExact(file, &natt_ext, sizeof(natt_ext)); if (natt_ext > 0) { TupleDesc desc = slot->tts_tupleDescriptor; for (int i = 0; i < desc->natts; i++) { CompactAttribute *attr = TupleDescCompactAttr(desc, i); varlena *varlen; uint64 chunk_header; void *value; Size varlensz; if (attr->attisdropped || attr->attlen != -1) continue; if (slot_attisnull(slot, i + 1)) continue; varlen = (varlena *) DatumGetPointer(slot->tts_values[i]); if (!VARATT_IS_EXTERNAL_INDIRECT(varlen)) continue; slot_getsomeattrs(slot, i + 1); BufFileReadExact(file, &chunk_header, VARHDRSZ); varlensz = VARSIZE_ANY(&chunk_header); value = palloc(varlensz); memcpy(value, &chunk_header, VARHDRSZ); BufFileReadExact(file, (char *) value + VARHDRSZ, varlensz - VARHDRSZ); slot->tts_values[i] = PointerGetDatum(value); natt_ext--; if (natt_ext < 0) elog(ERROR, "insufficient number of attributes stored separately"); } if (natt_ext != 0) elog(ERROR, "unexpected number of attributes stored separately (%d remaining)", natt_ext); } } /* * Adjust 'dest' replacing any EXTERNAL_ONDISK toast pointers with the * corresponding ones from 'src'. */ static void adjust_toast_pointers(Relation relation, TupleTableSlot *dest, TupleTableSlot *src) { TupleDesc desc = dest->tts_tupleDescriptor; for (int i = 0; i < desc->natts; i++) { CompactAttribute *attr = TupleDescCompactAttr(desc, i); varlena *varlena_dst; if (attr->attisdropped) continue; if (attr->attlen != -1) continue; if (slot_attisnull(dest, i + 1)) continue; slot_getsomeattrs(dest, i + 1); varlena_dst = (varlena *) DatumGetPointer(dest->tts_values[i]); if (!VARATT_IS_EXTERNAL_ONDISK(varlena_dst)) continue; slot_getsomeattrs(src, i + 1); dest->tts_values[i] = src->tts_values[i]; } } /* * Find the tuple to be updated or deleted by the given data change, whose * tuple has already been loaded into locator. * * If the tuple is found, put it in retrieved and return true. If the tuple is * not found, return false. */ static bool find_target_tuple(Relation rel, ChangeContext *chgcxt, TupleTableSlot *locator, TupleTableSlot *retrieved) { Form_pg_index idx = chgcxt->cc_ident_index->rd_index; IndexScanDesc scan; bool retval = false; /* * Scan key is passed by caller, so it does not have to be constructed * multiple times. Key entries have all fields initialized, except for * sk_argument. * * Use the incoming tuple to finalize the scan key. */ for (int i = 0; i < chgcxt->cc_ident_key_nentries; i++) { ScanKey entry = &chgcxt->cc_ident_key[i]; AttrNumber attno = idx->indkey.values[i]; entry->sk_argument = locator->tts_values[attno - 1]; Assert(!locator->tts_isnull[attno - 1]); } /* XXX no instrumentation for now */ scan = index_beginscan(rel, chgcxt->cc_ident_index, GetActiveSnapshot(), NULL, chgcxt->cc_ident_key_nentries, 0, 0); index_rescan(scan, chgcxt->cc_ident_key, chgcxt->cc_ident_key_nentries, NULL, 0); while (index_getnext_slot(scan, ForwardScanDirection, retrieved)) { /* Be wary of temporal constraints */ if (scan->xs_recheck && !identity_key_equal(chgcxt, locator, retrieved)) { CHECK_FOR_INTERRUPTS(); continue; } retval = true; break; } index_endscan(scan); return retval; } /* * Check whether the candidate tuple matches the locator tuple on all replica * identity key columns, using the same equality operators as the identity * index scan. The locator tuple has already been loaded into cc_ident_key. * * This is needed to filter lossy index matches, such as GiST multirange scans * used for temporal constraints. */ static bool identity_key_equal(ChangeContext *chgcxt, TupleTableSlot *locator, TupleTableSlot *candidate) { slot_getsomeattrs(locator, chgcxt->cc_last_key_attno); slot_getsomeattrs(candidate, chgcxt->cc_last_key_attno); for (int i = 0; i < chgcxt->cc_ident_key_nentries; i++) { ScanKey entry = &chgcxt->cc_ident_key[i]; AttrNumber attno = chgcxt->cc_ident_index->rd_index->indkey.values[i]; Assert(attno > 0); if (locator->tts_isnull[attno - 1] != candidate->tts_isnull[attno - 1]) return false; if (locator->tts_isnull[attno - 1]) continue; if (!DatumGetBool(FunctionCall2Coll(&entry->sk_func, entry->sk_collation, candidate->tts_values[attno - 1], entry->sk_argument))) return false; } return true; } /* * Decode and apply concurrent changes, up to (and including) the record whose * LSN is 'end_of_wal'. * * XXX the names "process_concurrent_changes" and "apply_concurrent_changes" * are far too similar to each other. */ static void process_concurrent_changes(XLogRecPtr end_of_wal, ChangeContext *chgcxt, bool done) { DecodingWorkerShared *shared; char fname[MAXPGPATH]; BufFile *file; pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_CATCH_UP); /* Ask the worker for the file. */ shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg); SpinLockAcquire(&shared->mutex); shared->lsn_upto = end_of_wal; shared->done = done; SpinLockRelease(&shared->mutex); /* * The worker needs to finish processing of the current WAL record. Even * if it's idle, it'll need to close the output file. Thus we're likely to * wait, so prepare for sleep. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) { int last_exported; SpinLockAcquire(&shared->mutex); last_exported = shared->last_exported; SpinLockRelease(&shared->mutex); /* * Has the worker exported the file we are waiting for? */ if (last_exported == chgcxt->cc_file_seq) break; ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); } ConditionVariableCancelSleep(); /* Open the file. */ DecodingWorkerFileName(fname, shared->relid, chgcxt->cc_file_seq); file = BufFileOpenFileSet(&shared->sfs.fs, fname, O_RDONLY, false); apply_concurrent_changes(file, chgcxt); BufFileClose(file); /* Get ready for the next file. */ chgcxt->cc_file_seq++; } /* * Initialize the ChangeContext struct for the given relation, with * the given index as identity index. */ static void initialize_change_context(ChangeContext *chgcxt, Relation relation, Oid ident_index_id) { chgcxt->cc_rel = relation; /* Only initialize fields needed by ExecInsertIndexTuples(). */ chgcxt->cc_estate = CreateExecutorState(); /* * Set up a range table for the executor, containing our repacked table as * its only member. */ { RangeTblEntry *rte; TupleDesc desc = RelationGetDescr(relation); List *perminfos = NIL; Bitmapset *updatedCols = NULL; RTEPermissionInfo *perminfo; /* * For our use, the RTE only needs to have perminfoindex initialized, * but there's no reason to not set the fields whose values we have at * hand. */ rte = makeNode(RangeTblEntry); rte->rtekind = RTE_RELATION; rte->relid = RelationGetRelid(relation); rte->relkind = RelationGetForm(relation)->relkind; /* Create the RTEPermissionInfo instance (and set ->perminfoindex). */ addRTEPermissionInfo(&perminfos, rte); /* * Initialize updatedCols to show that all columns are updated. This * is of course not necessarily true, and we cannot know this early; * but this is only used by ExecInsertIndexTuples to flag index * updates with no logical value changes, so if it's wrong, nothing * terribly bad happens. We may want to improve this someday though. * * Don't claim that dropped columns are changed though. */ for (int i = 0; i < desc->natts; i++) { CompactAttribute *attr = TupleDescCompactAttr(desc, i); if (attr->attisdropped) continue; updatedCols = bms_add_member(updatedCols, i + 1 - FirstLowInvalidHeapAttributeNumber); } /* install updatedCols in the right place */ perminfo = getRTEPermissionInfo(perminfos, rte); perminfo->updatedCols = updatedCols; /* finally we can initialize the range table proper */ ExecInitRangeTable(chgcxt->cc_estate, list_make1(rte), perminfos, bms_make_singleton(1)); } /* Set up our ResultRelInfo to use for index updates */ chgcxt->cc_rri = makeNode(ResultRelInfo); InitResultRelInfo(chgcxt->cc_rri, relation, 1, NULL, 0); ExecOpenIndices(chgcxt->cc_rri, false); /* * The table's relcache entry already has the relcache entry for the * identity index; find that. */ chgcxt->cc_ident_index = NULL; for (int i = 0; i < chgcxt->cc_rri->ri_NumIndices; i++) { Relation ind_rel; ind_rel = chgcxt->cc_rri->ri_IndexRelationDescs[i]; if (ind_rel->rd_id == ident_index_id) { chgcxt->cc_ident_index = ind_rel; break; } } if (chgcxt->cc_ident_index == NULL) elog(ERROR, "could not find identity index"); /* Set up for scanning said identity index */ { Form_pg_index indexForm; indexForm = chgcxt->cc_ident_index->rd_index; chgcxt->cc_ident_key_nentries = indexForm->indnkeyatts; chgcxt->cc_ident_key = (ScanKey) palloc_array(ScanKeyData, indexForm->indnkeyatts); for (int i = 0; i < indexForm->indnkeyatts; i++) { ScanKey entry; Oid opfamily, opcintype, opno, opcode; StrategyNumber eq_strategy; entry = &chgcxt->cc_ident_key[i]; opfamily = chgcxt->cc_ident_index->rd_opfamily[i]; opcintype = chgcxt->cc_ident_index->rd_opcintype[i]; eq_strategy = IndexAmTranslateCompareType(COMPARE_EQ, chgcxt->cc_ident_index->rd_rel->relam, opfamily, false); if (eq_strategy == InvalidStrategy) elog(ERROR, "could not find equality strategy for index operator family %u for type %u", opfamily, opcintype); opno = get_opfamily_member(opfamily, opcintype, opcintype, eq_strategy); if (!OidIsValid(opno)) elog(ERROR, "missing operator %d(%u,%u) in opfamily %u", eq_strategy, opcintype, opcintype, opfamily); opcode = get_opcode(opno); if (!OidIsValid(opcode)) elog(ERROR, "missing oprcode for operator %u", opno); /* Initialize everything but argument. */ ScanKeyInit(entry, i + 1, eq_strategy, opcode, (Datum) 0); entry->sk_collation = chgcxt->cc_ident_index->rd_indcollation[i]; } } /* Determine the last column we must deform to read the identity */ chgcxt->cc_last_key_attno = InvalidAttrNumber; for (int i = 0; i < chgcxt->cc_ident_key_nentries; i++) { AttrNumber attno = chgcxt->cc_ident_index->rd_index->indkey.values[i]; Assert(attno > 0); chgcxt->cc_last_key_attno = Max(chgcxt->cc_last_key_attno, attno); } chgcxt->cc_file_seq = WORKER_FILE_SNAPSHOT + 1; } /* * Free up resources taken by a ChangeContext. */ static void release_change_context(ChangeContext *chgcxt) { ExecCloseIndices(chgcxt->cc_rri); FreeExecutorState(chgcxt->cc_estate); /* XXX are these pfrees necessary? */ pfree(chgcxt->cc_rri); pfree(chgcxt->cc_ident_key); } /* * The final steps of rebuild_relation() for concurrent processing. * * On entry, NewHeap is locked in AccessExclusiveLock mode. OldHeap and its * clustering index (if one is passed) are still locked in a mode that allows * concurrent data changes. On exit, both tables and their indexes are closed, * but locked in AccessExclusiveLock mode. */ static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Oid identIdx, TransactionId frozenXid, MultiXactId cutoffMulti) { List *ind_oids_new; Oid old_table_oid = RelationGetRelid(OldHeap); Oid new_table_oid = RelationGetRelid(NewHeap); List *ind_oids_old = RelationGetIndexList(OldHeap); ListCell *lc, *lc2; char relpersistence; bool is_system_catalog; Oid ident_idx_new; XLogRecPtr end_of_wal; List *indexrels; ChangeContext chgcxt; Assert(CheckRelationLockedByMe(OldHeap, ShareUpdateExclusiveLock, false)); Assert(CheckRelationLockedByMe(NewHeap, AccessExclusiveLock, false)); /* * Unlike the exclusive case, we build new indexes for the new relation * rather than swapping the storage and reindexing the old relation. The * point is that the index build can take some time, so we do it before we * get AccessExclusiveLock on the old heap and therefore we cannot swap * the heap storage yet. * * index_create() will lock the new indexes using AccessExclusiveLock - no * need to change that. At the same time, we use ShareUpdateExclusiveLock * to lock the existing indexes - that should be enough to prevent others * from changing them while we're repacking the relation. The lock on * table should prevent others from changing the index column list, but * might not be enough for commands like ALTER INDEX ... SET ... (Those * are not necessarily dangerous, but can make user confused if the * changes they do get lost due to REPACK.) */ ind_oids_new = build_new_indexes(NewHeap, OldHeap, ind_oids_old); /* * The identity index in the new relation appears in the same relative * position as the corresponding index in the old relation. Find it. */ ident_idx_new = InvalidOid; foreach_oid(ind_old, ind_oids_old) { if (identIdx == ind_old) { int pos = foreach_current_index(ind_old); if (list_length(ind_oids_new) <= pos) elog(ERROR, "list of new indexes too short"); ident_idx_new = list_nth_oid(ind_oids_new, pos); break; } } if (!OidIsValid(ident_idx_new)) elog(ERROR, "could not find index matching \"%s\" at the new relation", get_rel_name(identIdx)); /* Gather information to apply concurrent changes. */ initialize_change_context(&chgcxt, NewHeap, ident_idx_new); /* * During testing, wait for another backend to perform concurrent data * changes which we will process below. */ INJECTION_POINT("repack-concurrently-before-lock", NULL); /* * Flush all WAL records inserted so far (possibly except for the last * incomplete page; see GetInsertRecPtr), to minimize the amount of data * we need to flush while holding exclusive lock on the source table. */ XLogFlush(GetXLogInsertEndRecPtr()); end_of_wal = GetFlushRecPtr(NULL); /* * Apply concurrent changes first time, to minimize the time we need to * hold AccessExclusiveLock. (Quite some amount of WAL could have been * written during the data copying and index creation.) */ process_concurrent_changes(end_of_wal, &chgcxt, false); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there * is one), all its indexes, so that we can swap the files. */ LockRelationOid(old_table_oid, AccessExclusiveLock); /* * Lock all indexes now, not only the clustering one: all indexes need to * have their files swapped. While doing that, store their relation * references in a zero-terminated array, to handle predicate locks below. */ indexrels = NIL; foreach_oid(ind_oid, ind_oids_old) { Relation index; index = index_open(ind_oid, AccessExclusiveLock); /* * Some things about the index may have changed before we locked the * index, such as ALTER INDEX RENAME. We don't need to do anything * here to absorb those changes in the new index. */ indexrels = lappend(indexrels, index); } /* * Lock the OldHeap's TOAST relation exclusively - again, the lock is * needed to swap the files. */ if (OidIsValid(OldHeap->rd_rel->reltoastrelid)) LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); /* * Tuples and pages of the old heap will be gone, but the heap will stay. */ TransferPredicateLocksToHeapRelation(OldHeap); foreach_ptr(RelationData, index, indexrels) { TransferPredicateLocksToHeapRelation(index); index_close(index, NoLock); } list_free(indexrels); /* * Flush WAL again, to make sure that all changes committed while we were * waiting for the exclusive lock are available for decoding. */ XLogFlush(GetXLogInsertEndRecPtr()); end_of_wal = GetFlushRecPtr(NULL); /* * Apply the concurrent changes again. Indicate that the decoding worker * won't be needed anymore. */ process_concurrent_changes(end_of_wal, &chgcxt, true); /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; is_system_catalog = IsSystemRelation(OldHeap); pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_SWAP_REL_FILES); /* * Even ShareUpdateExclusiveLock should have prevented others from * creating / dropping indexes (even using the CONCURRENTLY option), so we * do not need to check whether the lists match. */ forboth(lc, ind_oids_old, lc2, ind_oids_new) { Oid ind_old = lfirst_oid(lc); Oid ind_new = lfirst_oid(lc2); Oid mapped_tables[4] = {0}; swap_relation_files(ind_old, ind_new, (old_table_oid == RelationRelationId), false, /* swap_toast_by_content */ true, InvalidTransactionId, InvalidMultiXactId, mapped_tables); #ifdef USE_ASSERT_CHECKING /* * Concurrent processing is not supported for system relations, so * there should be no mapped tables. */ for (int i = 0; i < 4; i++) Assert(!OidIsValid(mapped_tables[i])); #endif } /* The new indexes must be visible for deletion. */ CommandCounterIncrement(); /* Close the old heap but keep lock until transaction commit. */ table_close(OldHeap, NoLock); /* Close the new heap. (We didn't have to open its indexes). */ table_close(NewHeap, NoLock); /* Cleanup what we don't need anymore. (And close the identity index.) */ release_change_context(&chgcxt); /* * Swap the relations and their TOAST relations and TOAST indexes. This * also drops the new relation and its indexes. * * (System catalogs are currently not supported.) */ Assert(!is_system_catalog); finish_heap_swap(old_table_oid, new_table_oid, is_system_catalog, false, /* swap_toast_by_content */ false, true, false, /* reindex */ frozenXid, cutoffMulti, relpersistence); } /* * Build indexes on NewHeap according to those on OldHeap. * * OldIndexes is the list of index OIDs on OldHeap. The contained indexes end * up locked using ShareUpdateExclusiveLock. * * A list of OIDs of the corresponding indexes created on NewHeap is * returned. The order of items does match, so we can use these arrays to swap * index storage. */ static List * build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes) { List *result = NIL; pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_REBUILD_INDEX); foreach_oid(oldindex, OldIndexes) { Oid newindex; char *newName; Relation ind; ind = index_open(oldindex, ShareUpdateExclusiveLock); newName = ChooseRelationName(get_rel_name(oldindex), NULL, "repacknew", get_rel_namespace(ind->rd_index->indrelid), false); newindex = index_create_copy(NewHeap, INDEX_CREATE_SUPPRESS_PROGRESS, oldindex, ind->rd_rel->reltablespace, newName); copy_index_constraints(ind, newindex, RelationGetRelid(NewHeap)); result = lappend_oid(result, newindex); index_close(ind, NoLock); } return result; } /* * Create a transient copy of a constraint -- supported by a transient * copy of the index that supports the original constraint. * * When repacking a table that contains exclusion constraints, the executor * relies on these constraints being properly catalogued. These copies are * to support that. * * We don't need the constraints for anything else (the original constraints * will be there once repack completes), so we add pg_depend entries so that * they are dropped when the transient table is dropped. */ static void copy_index_constraints(Relation old_index, Oid new_index_id, Oid new_heap_id) { ScanKeyData skey; Relation rel; TupleDesc desc; SysScanDesc scan; HeapTuple tup; ObjectAddress objrel; rel = table_open(ConstraintRelationId, RowExclusiveLock); ObjectAddressSet(objrel, RelationRelationId, new_heap_id); /* * Retrieve the constraints supported by the old index and create an * identical one that points to the new index. */ ScanKeyInit(&skey, Anum_pg_constraint_conrelid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(old_index->rd_index->indrelid)); scan = systable_beginscan(rel, ConstraintRelidTypidNameIndexId, true, NULL, 1, &skey); desc = RelationGetDescr(rel); while (HeapTupleIsValid(tup = systable_getnext(scan))) { Form_pg_constraint conform = (Form_pg_constraint) GETSTRUCT(tup); Oid oid; Datum values[Natts_pg_constraint] = {0}; bool nulls[Natts_pg_constraint] = {0}; bool replaces[Natts_pg_constraint] = {0}; HeapTuple new_tup; ObjectAddress objcon; if (conform->conindid != RelationGetRelid(old_index)) continue; oid = GetNewOidWithIndex(rel, ConstraintOidIndexId, Anum_pg_constraint_oid); values[Anum_pg_constraint_oid - 1] = ObjectIdGetDatum(oid); replaces[Anum_pg_constraint_oid - 1] = true; values[Anum_pg_constraint_conrelid - 1] = ObjectIdGetDatum(new_heap_id); replaces[Anum_pg_constraint_conrelid - 1] = true; values[Anum_pg_constraint_conindid - 1] = ObjectIdGetDatum(new_index_id); replaces[Anum_pg_constraint_conindid - 1] = true; new_tup = heap_modify_tuple(tup, desc, values, nulls, replaces); /* Insert it into the catalog. */ CatalogTupleInsert(rel, new_tup); /* Create a dependency so it's removed when we drop the new heap. */ ObjectAddressSet(objcon, ConstraintRelationId, oid); recordDependencyOn(&objcon, &objrel, DEPENDENCY_AUTO); } systable_endscan(scan); table_close(rel, RowExclusiveLock); CommandCounterIncrement(); } /* * Create a transient copy of attribute defaults. * * When repacking a table that has stored generated columns, the executor * relies on these entries to generate the values for them during apply of * concurrent operations. These copies are there to support that. * * We don't need the defaults for anything else, so we add pg_depend entries * so that they are dropped when the transient table is dropped. */ static void copy_attribute_defaults(Oid old_heap_oid, Oid new_heap_oid) { ScanKeyData skey; Relation rel; Relation att_rel; SysScanDesc scan; HeapTuple def_tup; ObjectAddress objrel; rel = table_open(AttrDefaultRelationId, RowExclusiveLock); att_rel = table_open(AttributeRelationId, RowExclusiveLock); ObjectAddressSet(objrel, RelationRelationId, new_heap_oid); ScanKeyInit(&skey, Anum_pg_attrdef_adrelid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(old_heap_oid)); scan = systable_beginscan(rel, AttrDefaultIndexId, true, NULL, 1, &skey); while (HeapTupleIsValid(def_tup = systable_getnext(scan))) { Form_pg_attrdef adform; Oid oid; Datum def_values[Natts_pg_attrdef]; bool def_nulls[Natts_pg_attrdef]; bool def_replaces[Natts_pg_attrdef] = {0}; Datum att_values[Natts_pg_attribute]; bool att_nulls[Natts_pg_attribute]; bool att_replaces[Natts_pg_attribute] = {0}; HeapTuple new_def_tup, att_tup, new_att_tup; ObjectAddress objad; adform = (Form_pg_attrdef) GETSTRUCT(def_tup); Assert(adform->adrelid == old_heap_oid); /* * Insert a new tuple that's identical to the existing one, other than * its OID and the relation it refers to. */ oid = GetNewOidWithIndex(rel, AttrDefaultOidIndexId, Anum_pg_attrdef_oid); def_values[Anum_pg_attrdef_oid - 1] = ObjectIdGetDatum(oid); def_nulls[Anum_pg_attrdef_oid - 1] = false; def_replaces[Anum_pg_attrdef_oid - 1] = true; def_values[Anum_pg_attrdef_adrelid - 1] = ObjectIdGetDatum(new_heap_oid); def_nulls[Anum_pg_attrdef_adrelid - 1] = false; def_replaces[Anum_pg_attrdef_adrelid - 1] = true; new_def_tup = heap_modify_tuple(def_tup, RelationGetDescr(rel), def_values, def_nulls, def_replaces); CatalogTupleInsert(rel, new_def_tup); /* Set atthasdef for this attribute in the transient table */ att_tup = SearchSysCache2(ATTNUM, ObjectIdGetDatum(new_heap_oid), ObjectIdGetDatum(adform->adnum)); if (!HeapTupleIsValid(att_tup)) elog(ERROR, "cache lookup failed for attribute %d of relation %u", adform->adnum, new_heap_oid); att_values[Anum_pg_attribute_atthasdef - 1] = BoolGetDatum(true); att_nulls[Anum_pg_attribute_atthasdef - 1] = false; att_replaces[Anum_pg_attribute_atthasdef - 1] = true; new_att_tup = heap_modify_tuple(att_tup, RelationGetDescr(att_rel), att_values, att_nulls, att_replaces); CatalogTupleUpdate(att_rel, &new_att_tup->t_self, new_att_tup); ReleaseSysCache(att_tup); /* Add a pg_depend record so it's removed with the transient table */ ObjectAddressSet(objad, AttrDefaultRelationId, oid); recordDependencyOn(&objad, &objrel, DEPENDENCY_AUTO); } systable_endscan(scan); table_close(rel, RowExclusiveLock); table_close(att_rel, RowExclusiveLock); CommandCounterIncrement(); } /* * Try to start a background worker to perform logical decoding of data * changes applied to relation while REPACK CONCURRENTLY is copying its * contents to a new table. */ static void start_repack_decoding_worker(Oid relid) { Size size; DecodingWorkerShared *shared; shm_mq *mq; BackgroundWorker bgw; decoding_worker = palloc0_object(DecodingWorker); /* Setup shared memory. */ size = BUFFERALIGN(offsetof(DecodingWorkerShared, error_queue)) + BUFFERALIGN(REPACK_ERROR_QUEUE_SIZE); decoding_worker->seg = dsm_create(size, 0); shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg); shared->initialized = false; shared->lsn_upto = InvalidXLogRecPtr; shared->done = false; SharedFileSetInit(&shared->sfs, decoding_worker->seg); shared->last_exported = -1; SpinLockInit(&shared->mutex); shared->dbid = MyDatabaseId; /* * This is the UserId set in cluster_rel(). Security context shouldn't be * needed for decoding worker. */ shared->roleid = GetUserId(); shared->relid = relid; ConditionVariableInit(&shared->cv); shared->backend_proc = MyProc; shared->backend_pid = MyProcPid; shared->backend_proc_number = MyProcNumber; mq = shm_mq_create((char *) BUFFERALIGN(shared->error_queue), REPACK_ERROR_QUEUE_SIZE); shm_mq_set_receiver(mq, MyProc); decoding_worker->error_mqh = shm_mq_attach(mq, decoding_worker->seg, NULL); memset(&bgw, 0, sizeof(bgw)); snprintf(bgw.bgw_name, BGW_MAXLEN, "REPACK decoding worker for relation \"%s\"", get_rel_name(relid)); snprintf(bgw.bgw_type, BGW_MAXLEN, "REPACK decoding worker"); bgw.bgw_flags = BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION; bgw.bgw_start_time = BgWorkerStart_RecoveryFinished; bgw.bgw_restart_time = BGW_NEVER_RESTART; snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres"); snprintf(bgw.bgw_function_name, BGW_MAXLEN, "RepackWorkerMain"); bgw.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(decoding_worker->seg)); bgw.bgw_notify_pid = MyProcPid; if (!RegisterDynamicBackgroundWorker(&bgw, &decoding_worker->handle)) ereport(ERROR, errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED), errmsg("out of background worker slots"), errhint("You might need to increase \"%s\".", "max_worker_processes")); /* * The decoding setup must be done before the caller can have XID assigned * for any reason, otherwise the worker might end up in a deadlock, * waiting for the caller's transaction to end. Therefore wait here until * the worker indicates that it has the logical decoding initialized. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) { bool initialized; SpinLockAcquire(&shared->mutex); initialized = shared->initialized; SpinLockRelease(&shared->mutex); if (initialized) break; ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); } ConditionVariableCancelSleep(); } /* * Stop the decoding worker and cleanup the related resources. * * The worker stops on its own when it knows there is no more work to do, but * we need to stop it explicitly at least on ERROR in the launching backend. */ static void stop_repack_decoding_worker(void) { /* Nothing to do if no worker was set up. */ if (decoding_worker == NULL) return; /* Terminate the worker process, if one is running. */ if (decoding_worker->handle != NULL) { BgwHandleStatus status; TerminateBackgroundWorker(decoding_worker->handle); /* The worker should really exit before the REPACK command does. */ HOLD_INTERRUPTS(); status = WaitForBackgroundWorkerShutdown(decoding_worker->handle); RESUME_INTERRUPTS(); if (status == BGWH_POSTMASTER_DIED) ereport(FATAL, errcode(ERRCODE_ADMIN_SHUTDOWN), errmsg("postmaster exited during REPACK command")); } /* * Now detach from our shared memory segment. In error cases there might * still be messages from the worker in the queue, which ProcessInterrupts * would try to read; this is pointless (and causes an assertion failure), * so set the global pointer to NULL to have ProcessRepackMessages ignore * them. * * We must also cancel the current sleep, if one is still set up. This is * critical because the CV lives in the DSM that we're about to detach, so * if we omit it, later automatic cleanup tries to clear freed memory. */ if (decoding_worker->error_mqh != NULL) shm_mq_detach(decoding_worker->error_mqh); ConditionVariableCancelSleep(); if (decoding_worker->seg != NULL) dsm_detach(decoding_worker->seg); pfree(decoding_worker); decoding_worker = NULL; } /* stop_repack_decoding_worker, wrapped as a before_shmem_exit callback */ static void stop_repack_decoding_worker_cb(int code, Datum arg) { stop_repack_decoding_worker(); } /* * Get the initial snapshot from the decoding worker. */ static Snapshot get_initial_snapshot(DecodingWorker *worker) { DecodingWorkerShared *shared; char fname[MAXPGPATH]; BufFile *file; Size snap_size; char *snap_space; Snapshot snapshot; shared = (DecodingWorkerShared *) dsm_segment_address(worker->seg); /* * The worker needs to initialize the logical decoding, which usually * takes some time. Therefore it makes sense to prepare for the sleep * first. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) { int last_exported; SpinLockAcquire(&shared->mutex); last_exported = shared->last_exported; SpinLockRelease(&shared->mutex); /* * Has the worker exported the file we are waiting for? */ if (last_exported == WORKER_FILE_SNAPSHOT) break; ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); } ConditionVariableCancelSleep(); /* Read the snapshot from a file. */ DecodingWorkerFileName(fname, shared->relid, WORKER_FILE_SNAPSHOT); file = BufFileOpenFileSet(&shared->sfs.fs, fname, O_RDONLY, false); BufFileReadExact(file, &snap_size, sizeof(snap_size)); snap_space = (char *) palloc(snap_size); BufFileReadExact(file, snap_space, snap_size); BufFileClose(file); /* Restore it. */ snapshot = RestoreSnapshot(snap_space); pfree(snap_space); return snapshot; } /* * Generate worker's file name into 'fname', which must be of size MAXPGPATH. * If relations of the same 'relid' happen to be processed at the same time, * they must be from different databases and therefore different backends must * be involved. */ void DecodingWorkerFileName(char *fname, Oid relid, uint32 seq) { /* The PID is already present in the fileset name, so we needn't add it */ snprintf(fname, MAXPGPATH, "%u-%u", relid, seq); } /* * Handle receipt of an interrupt indicating a repack worker message. * * Note: this is called within a signal handler! All we can do is set * a flag that will cause the next CHECK_FOR_INTERRUPTS() to invoke * ProcessRepackMessages(). */ void HandleRepackMessageInterrupt(void) { InterruptPending = true; RepackMessagePending = true; SetLatch(MyLatch); } /* * Process any queued protocol messages received from the repack worker. */ void ProcessRepackMessages(void) { 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(); } /* * Process a single protocol message received from a single parallel worker. */ static void ProcessRepackMessage(StringInfo msg) { char msgtype; msgtype = pq_getmsgbyte(msg); switch (msgtype) { case PqMsg_ErrorResponse: case PqMsg_NoticeResponse: { ErrorData edata; /* Parse ErrorResponse or NoticeResponse. */ pq_parse_errornotice(msg, &edata); /* Death of a worker isn't enough justification for suicide. */ edata.elevel = Min(edata.elevel, ERROR); /* * Add a context line to show that this is a message * propagated from the worker. Otherwise, it can sometimes be * confusing to understand what actually happened. */ if (edata.context) edata.context = psprintf("%s\n%s", edata.context, _("REPACK decoding worker")); else edata.context = pstrdup(_("REPACK decoding worker")); /* Rethrow error or print notice. */ ThrowErrorData(&edata); break; } default: { elog(ERROR, "unrecognized message type received from decoding worker: %c (message length %d bytes)", msgtype, msg->len); } } }