agora inbox for [email protected]help / color / mirror / Atom feed
From: Antonin Houska <[email protected]> Subject: [PATCH v35 4/5] Add CONCURRENTLY option to REPACK command. Date: Mon, 16 Feb 2026 20:49:23 +0100 The REPACK command copies the relation data into a new file, creates new indexes and eventually swaps the files. To make sure that the old file does not change during the copying, the relation is locked in an exclusive mode, which prevents applications from both reading and writing. (To keep the data consistent, we'd only need to prevent the applications from writing, but even reading needs to be blocked before we can swap the files - otherwise some applications could continue using the old file. Since we should not request a stronger lock without releasing the weaker one first, we acquire the exclusive lock in the beginning and keep it till the end of the processing.) This patch introduces an alternative workflow, which only requires the exclusive lock when the relation (and index) files are being swapped. (Supposedly, the swapping should be pretty fast.) On the other hand, when we copy the data to the new file, we allow applications to read from the relation and even to write to it. First, we scan the relation using a "historic snapshot", and insert all the tuples satisfying this snapshot into the new file. Second, logical decoding is used to capture the data changes done by applications during the copying (i.e. changes that do not satisfy the historic snapshot mentioned above), and those are applied to the new file before we acquire the exclusive lock that we need to swap the files. (Of course, more data changes can take place while we are waiting for the lock - these will be applied to the new file after we have acquired the lock, before we swap the files.) Since the logical decoding system, during its startup, waits until all the transactions which already have XID assigned have finished, there is a risk of deadlock if a transaction that already changed anything in the database tries to acquire a conflicting lock on the table REPACK CONCURRENTLY is working on. As an example, consider transaction running CREATE INDEX command on the table that is being REPACKed CONCURRENTLY. On the other hand, DML commands (INSERT, UPDATE, DELETE) are not a problem as their lock does not conflict with REPACK CONCURRENTLY. The current approach is that we accept the risk. If we tried to avoid it, it'd be necessary to unlock the table before the logical decoding is setup and lock it again afterwards. Such temporary unlocking would imply re-checking if the table still meets all the requirements for REPACK CONCURRENTLY. Like the existing implementation of REPACK, the variant with the CONCURRENTLY option also requires an extra space for the new relation and index files (which coexist with the old files for some time). In addition, the CONCURRENTLY option might introduce a lag in releasing WAL segments for archiving / recycling. This is due to the decoding of the data changes done by applications concurrently. When copying the table contents into the new file, we check the lag periodically. If it exceeds the size of a WAL segment, we decode all the available WAL before resuming the copying. (Of course, the changes are not applied until the whole table contents is copied.) A background worker might be a better approach for the decoding - let's consider implementing it in the future. The WAL records produced by running DML commands on the new relation do not contain enough information to be processed by the logical decoding system. All we need from the new relation is the file (relfilenode), while the actual relation is eventually dropped. Thus there is no point in replaying the DMLs anywhere. --- doc/src/sgml/monitoring.sgml | 37 +- doc/src/sgml/mvcc.sgml | 12 +- doc/src/sgml/ref/repack.sgml | 129 +- src/Makefile | 1 + src/backend/access/heap/heapam.c | 34 +- src/backend/access/heap/heapam_handler.c | 260 ++- src/backend/access/heap/rewriteheap.c | 6 +- src/backend/catalog/system_views.sql | 19 +- src/backend/commands/cluster.c | 1640 +++++++++++++++-- src/backend/commands/matview.c | 1 + src/backend/commands/tablecmds.c | 1 + src/backend/commands/vacuum.c | 12 +- src/backend/meson.build | 1 + src/backend/replication/logical/decode.c | 37 +- src/backend/replication/logical/snapbuild.c | 21 + .../replication/pgoutput_repack/Makefile | 32 + .../replication/pgoutput_repack/meson.build | 18 + .../pgoutput_repack/pgoutput_repack.c | 240 +++ .../storage/lmgr/generate-lwlocknames.pl | 2 +- src/backend/utils/time/snapmgr.c | 3 +- src/bin/psql/tab-complete.in.c | 4 +- src/include/access/heapam.h | 5 +- src/include/access/heapam_xlog.h | 2 + src/include/access/tableam.h | 10 + src/include/commands/cluster.h | 88 +- src/include/commands/progress.h | 17 +- src/include/replication/snapbuild.h | 1 + src/include/storage/lockdefs.h | 4 +- src/include/utils/snapmgr.h | 2 + src/test/modules/injection_points/Makefile | 2 + .../injection_points/expected/repack.out | 113 ++ .../expected/repack_toast.out | 64 + src/test/modules/injection_points/meson.build | 2 + .../injection_points/specs/repack.spec | 142 ++ .../injection_points/specs/repack_toast.spec | 105 ++ src/test/regress/expected/rules.out | 19 +- src/tools/pgindent/typedefs.list | 5 + 37 files changed, 2830 insertions(+), 261 deletions(-) create mode 100644 src/backend/replication/pgoutput_repack/Makefile create mode 100644 src/backend/replication/pgoutput_repack/meson.build create mode 100644 src/backend/replication/pgoutput_repack/pgoutput_repack.c create mode 100644 src/test/modules/injection_points/expected/repack.out create mode 100644 src/test/modules/injection_points/expected/repack_toast.out create mode 100644 src/test/modules/injection_points/specs/repack.spec create mode 100644 src/test/modules/injection_points/specs/repack_toast.spec diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 71c92ed53ef..ec97197461a 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -6239,14 +6239,35 @@ FROM pg_stat_get_backend_idset() AS backendid; <row> <entry role="catalog_table_entry"><para role="column_definition"> - <structfield>heap_tuples_written</structfield> <type>bigint</type> + <structfield>heap_tuples_inserted</structfield> <type>bigint</type> </para> <para> - Number of heap tuples written. + Number of heap tuples inserted. This counter only advances when the phase is <literal>seq scanning heap</literal>, - <literal>index scanning heap</literal> - or <literal>writing new heap</literal>. + <literal>index scanning heap</literal>, + <literal>writing new heap</literal> + or <literal>catch-up</literal>. + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>heap_tuples_updated</structfield> <type>bigint</type> + </para> + <para> + Number of heap tuples updated. + This counter only advances when the phase is <literal>catch-up</literal>. + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>heap_tuples_deleted</structfield> <type>bigint</type> + </para> + <para> + Number of heap tuples deleted. + This counter only advances when the phase is <literal>catch-up</literal>. </para></entry> </row> @@ -6327,6 +6348,14 @@ FROM pg_stat_get_backend_idset() AS backendid; <command>REPACK</command> is currently writing the new heap. </entry> </row> + <row> + <entry><literal>catch-up</literal></entry> + <entry> + <command>REPACK CONCURRENTLY</command> is currently processing the DML + commands that other transactions executed during any of the preceding + phases. + </entry> + </row> <row> <entry><literal>swapping relation files</literal></entry> <entry> diff --git a/doc/src/sgml/mvcc.sgml b/doc/src/sgml/mvcc.sgml index e775260936a..241caeb3593 100644 --- a/doc/src/sgml/mvcc.sgml +++ b/doc/src/sgml/mvcc.sgml @@ -1845,15 +1845,17 @@ SELECT pg_advisory_lock(q.id) FROM <title>Caveats</title> <para> - Some DDL commands, currently only <link linkend="sql-truncate"><command>TRUNCATE</command></link> and the - table-rewriting forms of <link linkend="sql-altertable"><command>ALTER TABLE</command></link>, are not + Some commands, currently only <link linkend="sql-truncate"><command>TRUNCATE</command></link>, the + table-rewriting forms of <link linkend="sql-altertable"><command>ALTER + TABLE</command></link> and <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option, are not MVCC-safe. This means that after the truncation or rewrite commits, the table will appear empty to concurrent transactions, if they are using a - snapshot taken before the DDL command committed. This will only be an + snapshot taken before the command committed. This will only be an issue for a transaction that did not access the table in question - before the DDL command started — any transaction that has done so + before the command started — any transaction that has done so would hold at least an <literal>ACCESS SHARE</literal> table lock, - which would block the DDL command until that transaction completes. + which would block the truncating or rewriting command until that transaction completes. So these commands will not cause any apparent inconsistency in the table contents for successive queries on the target table, but they could cause visible inconsistency between the contents of the target diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 25b55e2228b..8b875345687 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -28,6 +28,7 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] USING VERBOSE [ <replaceable class="parameter">boolean</replaceable> ] ANALYZE [ <replaceable class="parameter">boolean</replaceable> ] + CONCURRENTLY [ <replaceable class="parameter">boolean</replaceable> ] <phrase>and <replaceable class="parameter">table_and_columns</replaceable> is:</phrase> @@ -54,7 +55,8 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] USING processes every table and materialized view in the current database that the current user has the <literal>MAINTAIN</literal> privilege on. This form of <command>REPACK</command> cannot be executed inside a transaction - block. + block. Also, this form is not allowed if + the <literal>CONCURRENTLY</literal> option is used. </para> <para> @@ -67,7 +69,8 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] USING When a table is being repacked, an <literal>ACCESS EXCLUSIVE</literal> lock is acquired on it. This prevents any other database operations (both reads and writes) from operating on the table until the <command>REPACK</command> - is finished. + is finished. If you want to keep the table accessible during the repacking, + consider using the <literal>CONCURRENTLY</literal> option. </para> <refsect2 id="sql-repack-notes-on-clustering" xreflabel="Notes on Clustering"> @@ -195,6 +198,128 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] USING </listitem> </varlistentry> + <varlistentry> + <term><literal>CONCURRENTLY</literal></term> + <listitem> + <para> + Allow other transactions to use the table while it is being repacked. + </para> + + <para> + Internally, <command>REPACK</command> copies the contents of the table + (ignoring dead tuples) into a new file, sorted by the specified index, + and also creates a new file for each index. Then it swaps the old and + new files for the table and all the indexes, and deletes the old + files. The <literal>ACCESS EXCLUSIVE</literal> lock is needed to make + sure that the old files do not change during the processing because the + changes would get lost due to the swap. + </para> + + <para> + With the <literal>CONCURRENTLY</literal> option, the <literal>ACCESS + EXCLUSIVE</literal> lock is only acquired to swap the table and index + files. The data changes that took place during the creation of the new + table and index files are captured using logical decoding + (<xref linkend="logicaldecoding"/>) and applied before + the <literal>ACCESS EXCLUSIVE</literal> lock is requested. Thus the lock + is typically held only for the time needed to swap the files, which + should be pretty short. However, the time might still be noticeable if + too many data changes have been done to the table while + <command>REPACK</command> was waiting for the lock: those changes must + be processed just before the files are swapped, while the + <literal>ACCESS EXCLUSIVE</literal> lock is being held. + </para> + + <para> + Note that <command>REPACK</command> with the + <literal>CONCURRENTLY</literal> option does not try to order the rows + inserted into the table after the repacking started. Also + note <command>REPACK</command> might fail to complete due to DDL + commands executed on the table by other transactions during the + repacking. + </para> + + <note> + <para> + In addition to the temporary space requirements explained in + <xref linkend="sql-repack-notes-on-resources"/>, + the <literal>CONCURRENTLY</literal> option can add to the usage of + temporary space a bit more. The reason is that other transactions can + perform DML operations which cannot be applied to the new file until + <command>REPACK</command> has copied all the tuples from the old + file. Thus the tuples inserted into the old file during the copying are + also stored separately in a temporary file, so they can eventually be + applied to the new file. + </para> + + <para> + Furthermore, the data changes performed during the copying are + extracted from <link linkend="wal">write-ahead log</link> (WAL), and + this extraction (decoding) only takes place when certain amount of WAL + has been written. Therefore, WAL removal can be delayed by this + threshold. Currently the threshold is equal to the value of + the <link linkend="guc-wal-segment-size"><varname>wal_segment_size</varname></link> + configuration parameter. + </para> + </note> + + <para> + The <literal>CONCURRENTLY</literal> option cannot be used in the + following cases: + + <itemizedlist> + <listitem> + <para> + The table is <literal>UNLOGGED</literal>. + </para> + </listitem> + + <listitem> + <para> + The table is partitioned. + </para> + </listitem> + + <listitem> + <para> + The table is a system catalog or a <acronym>TOAST</acronym> table. + </para> + </listitem> + + <listitem> + <para> + <command>REPACK</command> is executed inside a transaction block. + </para> + </listitem> + + <listitem> + <para> + The <link linkend="guc-wal-level"><varname>wal_level</varname></link> + configuration parameter is less than <literal>logical</literal>. + </para> + </listitem> + + <listitem> + <para> + The <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link> + configuration parameter does not allow for creation of an additional + replication slot. + </para> + </listitem> + </itemizedlist> + </para> + + <warning> + <para> + <command>REPACK</command> with the <literal>CONCURRENTLY</literal> + option is not MVCC-safe, see <xref linkend="mvcc-caveats"/> for + details. + </para> + </warning> + + </listitem> + </varlistentry> + <varlistentry> <term><literal>VERBOSE</literal></term> <listitem> diff --git a/src/Makefile b/src/Makefile index 2f31a2f20a7..b18c9a14ffa 100644 --- a/src/Makefile +++ b/src/Makefile @@ -23,6 +23,7 @@ SUBDIRS = \ interfaces \ backend/replication/libpqwalreceiver \ backend/replication/pgoutput \ + backend/replication/pgoutput_repack \ fe_utils \ bin \ pl \ diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c index 98d53caeea8..37bbe211b6f 100644 --- a/src/backend/access/heap/heapam.c +++ b/src/backend/access/heap/heapam.c @@ -60,7 +60,8 @@ static HeapTuple heap_prepare_insert(Relation relation, HeapTuple tup, static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, Buffer newbuf, HeapTuple oldtup, HeapTuple newtup, HeapTuple old_key_tuple, - bool all_visible_cleared, bool new_all_visible_cleared); + bool all_visible_cleared, bool new_all_visible_cleared, + bool walLogical); #ifdef USE_ASSERT_CHECKING static void check_lock_if_inplace_updateable_rel(Relation relation, const ItemPointerData *otid, @@ -2841,7 +2842,7 @@ xmax_infomask_changed(uint16 new_infomask, uint16 old_infomask) TM_Result heap_delete(Relation relation, const ItemPointerData *tid, CommandId cid, Snapshot crosscheck, bool wait, - TM_FailureData *tmfd, bool changingPart) + TM_FailureData *tmfd, bool changingPart, bool walLogical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -3088,7 +3089,8 @@ l1: * Compute replica identity tuple before entering the critical section so * we don't PANIC upon a memory allocation failure. */ - old_key_tuple = ExtractReplicaIdentity(relation, &tp, true, &old_key_copied); + old_key_tuple = walLogical ? + ExtractReplicaIdentity(relation, &tp, true, &old_key_copied) : NULL; /* * If this is the first possibly-multixact-able operation in the current @@ -3178,6 +3180,15 @@ l1: xlrec.flags |= XLH_DELETE_CONTAINS_OLD_KEY; } + /* + * Unlike UPDATE, DELETE is decoded even if there is no old key, so it + * does not help to clear both XLH_DELETE_CONTAINS_OLD_TUPLE and + * XLH_DELETE_CONTAINS_OLD_KEY. Thus we need an extra flag. TODO + * Consider not decoding tuples w/o the old tuple/key instead. + */ + if (!walLogical) + xlrec.flags |= XLH_DELETE_NO_LOGICAL; + XLogBeginInsert(); XLogRegisterData(&xlrec, SizeOfHeapDelete); @@ -3270,7 +3281,8 @@ simple_heap_delete(Relation relation, const ItemPointerData *tid) result = heap_delete(relation, tid, GetCurrentCommandId(true), InvalidSnapshot, true /* wait for commit */ , - &tmfd, false /* changingPart */ ); + &tmfd, false, /* changingPart */ + true /* walLogical */ ); switch (result) { case TM_SelfModified: @@ -3311,7 +3323,7 @@ TM_Result heap_update(Relation relation, const ItemPointerData *otid, HeapTuple newtup, CommandId cid, Snapshot crosscheck, bool wait, TM_FailureData *tmfd, LockTupleMode *lockmode, - TU_UpdateIndexes *update_indexes) + TU_UpdateIndexes *update_indexes, bool walLogical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -4204,7 +4216,8 @@ l2: newbuf, &oldtup, heaptup, old_key_tuple, all_visible_cleared, - all_visible_cleared_new); + all_visible_cleared_new, + walLogical); if (newbuf != buffer) { PageSetLSN(BufferGetPage(newbuf), recptr); @@ -4562,7 +4575,8 @@ simple_heap_update(Relation relation, const ItemPointerData *otid, HeapTuple tup result = heap_update(relation, otid, tup, GetCurrentCommandId(true), InvalidSnapshot, true /* wait for commit */ , - &tmfd, &lockmode, update_indexes); + &tmfd, &lockmode, update_indexes, + true /* walLogical */ ); switch (result) { case TM_SelfModified: @@ -8918,7 +8932,8 @@ static XLogRecPtr log_heap_update(Relation reln, Buffer oldbuf, Buffer newbuf, HeapTuple oldtup, HeapTuple newtup, HeapTuple old_key_tuple, - bool all_visible_cleared, bool new_all_visible_cleared) + bool all_visible_cleared, bool new_all_visible_cleared, + bool walLogical) { xl_heap_update xlrec; xl_heap_header xlhdr; @@ -8929,7 +8944,8 @@ log_heap_update(Relation reln, Buffer oldbuf, suffixlen = 0; XLogRecPtr recptr; Page page = BufferGetPage(newbuf); - bool need_tuple_data = RelationIsLogicallyLogged(reln); + bool need_tuple_data = RelationIsLogicallyLogged(reln) && + walLogical; bool init; int bufflags; diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index e3748ff6301..41c52046e14 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -33,6 +33,7 @@ #include "catalog/index.h" #include "catalog/storage.h" #include "catalog/storage_xlog.h" +#include "commands/cluster.h" #include "commands/progress.h" #include "executor/executor.h" #include "miscadmin.h" @@ -309,7 +310,8 @@ heapam_tuple_delete(Relation relation, ItemPointer tid, CommandId cid, * the storage itself is cleaning the dead tuples by itself, it is the * time to call the index tuple deletion also. */ - return heap_delete(relation, tid, cid, crosscheck, wait, tmfd, changingPart); + return heap_delete(relation, tid, cid, crosscheck, wait, tmfd, changingPart, + true); } @@ -328,7 +330,7 @@ heapam_tuple_update(Relation relation, ItemPointer otid, TupleTableSlot *slot, tuple->t_tableOid = slot->tts_tableOid; result = heap_update(relation, otid, tuple, cid, crosscheck, wait, - tmfd, lockmode, update_indexes); + tmfd, lockmode, update_indexes, true); ItemPointerCopy(&tuple->t_self, &slot->tts_tid); /* @@ -685,13 +687,15 @@ static void heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, Relation OldIndex, bool use_sort, TransactionId OldestXmin, + Snapshot snapshot, + LogicalDecodingContext *decoding_ctx, TransactionId *xid_cutoff, MultiXactId *multi_cutoff, double *num_tuples, double *tups_vacuumed, double *tups_recently_dead) { - RewriteState rwstate; + RewriteState rwstate = NULL; IndexScanDesc indexScan; TableScanDesc tableScan; HeapScanDesc heapScan; @@ -705,6 +709,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, bool *isnull; BufferHeapTupleTableSlot *hslot; BlockNumber prev_cblock = InvalidBlockNumber; + bool concurrent = snapshot != NULL; + XLogRecPtr end_of_wal_prev = GetFlushRecPtr(NULL); /* Remember if it's a system catalog */ is_system_catalog = IsSystemRelation(OldHeap); @@ -720,9 +726,12 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, values = palloc_array(Datum, natts); isnull = palloc_array(bool, natts); - /* Initialize the rewrite operation */ - rwstate = begin_heap_rewrite(OldHeap, NewHeap, OldestXmin, *xid_cutoff, - *multi_cutoff); + /* + * Initialize the rewrite operation. + */ + if (!concurrent) + rwstate = begin_heap_rewrite(OldHeap, NewHeap, OldestXmin, + *xid_cutoff, *multi_cutoff); /* Set up sorting if wanted */ @@ -737,6 +746,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, * Prepare to scan the OldHeap. To ensure we see recently-dead tuples * that still need to be copied, we scan with SnapshotAny and use * HeapTupleSatisfiesVacuum for the visibility test. + * + * In the CONCURRENTLY case, we do regular MVCC visibility tests, using + * the snapshot passed by the caller. */ if (OldIndex != NULL && !use_sort) { @@ -753,7 +765,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tableScan = NULL; heapScan = NULL; - indexScan = index_beginscan(OldHeap, OldIndex, SnapshotAny, NULL, 0, 0); + indexScan = index_beginscan(OldHeap, OldIndex, + snapshot ? snapshot : SnapshotAny, + NULL, 0, 0); index_rescan(indexScan, NULL, 0, NULL, 0); } else @@ -762,7 +776,9 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, pgstat_progress_update_param(PROGRESS_REPACK_PHASE, PROGRESS_REPACK_PHASE_SEQ_SCAN_HEAP); - tableScan = table_beginscan(OldHeap, SnapshotAny, 0, (ScanKey) NULL); + tableScan = table_beginscan(OldHeap, + snapshot ? snapshot : SnapshotAny, + 0, (ScanKey) NULL); heapScan = (HeapScanDesc) tableScan; indexScan = NULL; @@ -838,83 +854,91 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, buf = hslot->buffer; /* - * To be able to guarantee that we can set the hint bit, acquire an - * exclusive lock on the old buffer. We need the hint bits, set in - * heapam_relation_copy_for_cluster() -> HeapTupleSatisfiesVacuum(), - * to be set, as otherwise reform_and_rewrite_tuple() -> - * rewrite_heap_tuple() will get confused. Specifically, - * rewrite_heap_tuple() checks for HEAP_XMAX_INVALID in the old tuple - * to determine whether to check the old-to-new mapping hash table. - * - * It'd be better if we somehow could avoid setting hint bits on the - * old page. One reason to use VACUUM FULL are very bloated tables - - * rewriting most of the old table during VACUUM FULL doesn't exactly - * help... + * Regarding CONCURRENTLY, see the comments on MVCC snapshot above. */ - LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); - - switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) + if (!concurrent) { - case HEAPTUPLE_DEAD: - /* Definitely dead */ - isdead = true; - break; - case HEAPTUPLE_RECENTLY_DEAD: - *tups_recently_dead += 1; - pg_fallthrough; - case HEAPTUPLE_LIVE: - /* Live or recently dead, must copy it */ - isdead = false; - break; - case HEAPTUPLE_INSERT_IN_PROGRESS: + /* + * To be able to guarantee that we can set the hint bit, acquire + * an exclusive lock on the old buffer. We need the hint bits, set + * in heapam_relation_copy_for_cluster() -> + * HeapTupleSatisfiesVacuum(), to be set, as otherwise + * reform_and_rewrite_tuple() -> rewrite_heap_tuple() will get + * confused. Specifically, rewrite_heap_tuple() checks for + * HEAP_XMAX_INVALID in the old tuple to determine whether to + * check the old-to-new mapping hash table. + * + * It'd be better if we somehow could avoid setting hint bits on + * the old page. One reason to use VACUUM FULL are very bloated + * tables - rewriting most of the old table during VACUUM FULL + * doesn't exactly help... + */ + LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE); - /* - * Since we hold exclusive lock on the relation, normally the - * only way to see this is if it was inserted earlier in our - * own transaction. However, it can happen in system - * catalogs, since we tend to release write lock before commit - * there. Give a warning if neither case applies; but in any - * case we had better copy it. - */ - if (!is_system_catalog && - !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(tuple->t_data))) - elog(WARNING, "concurrent insert in progress within table \"%s\"", - RelationGetRelationName(OldHeap)); - /* treat as live */ - isdead = false; - break; - case HEAPTUPLE_DELETE_IN_PROGRESS: - - /* - * Similar situation to INSERT_IN_PROGRESS case. - */ - if (!is_system_catalog && - !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetUpdateXid(tuple->t_data))) - elog(WARNING, "concurrent delete in progress within table \"%s\"", - RelationGetRelationName(OldHeap)); - /* treat as recently dead */ - *tups_recently_dead += 1; - isdead = false; - break; - default: - elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result"); - isdead = false; /* keep compiler quiet */ - break; - } - - LockBuffer(buf, BUFFER_LOCK_UNLOCK); - - if (isdead) - { - *tups_vacuumed += 1; - /* heap rewrite module still needs to see it... */ - if (rewrite_heap_dead_tuple(rwstate, tuple)) + switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) { - /* A previous recently-dead tuple is now known dead */ - *tups_vacuumed += 1; - *tups_recently_dead -= 1; + case HEAPTUPLE_DEAD: + /* Definitely dead */ + isdead = true; + break; + case HEAPTUPLE_RECENTLY_DEAD: + *tups_recently_dead += 1; + pg_fallthrough; + case HEAPTUPLE_LIVE: + /* Live or recently dead, must copy it */ + isdead = false; + break; + case HEAPTUPLE_INSERT_IN_PROGRESS: + + /* + * As long as we hold exclusive lock on the relation, + * normally the only way to see this is if it was inserted + * earlier in our own transaction. However, it can happen + * in system catalogs, since we tend to release write lock + * before commit there. Give a warning if neither case + * applies; but in any case we had better copy it. + */ + if (!is_system_catalog && + !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetXmin(tuple->t_data))) + elog(WARNING, "concurrent insert in progress within table \"%s\"", + RelationGetRelationName(OldHeap)); + /* treat as live */ + isdead = false; + break; + case HEAPTUPLE_DELETE_IN_PROGRESS: + + /* + * Similar situation to INSERT_IN_PROGRESS case. + */ + if (!is_system_catalog && + !TransactionIdIsCurrentTransactionId(HeapTupleHeaderGetUpdateXid(tuple->t_data))) + elog(WARNING, "concurrent delete in progress within table \"%s\"", + RelationGetRelationName(OldHeap)); + /* treat as recently dead */ + *tups_recently_dead += 1; + isdead = false; + break; + default: + elog(ERROR, "unexpected HeapTupleSatisfiesVacuum result"); + isdead = false; /* keep compiler quiet */ + break; + } + + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + + if (isdead) + { + *tups_vacuumed += 1; + /* heap rewrite module still needs to see it... */ + if (rewrite_heap_dead_tuple(rwstate, tuple)) + { + /* A previous recently-dead tuple is now known dead */ + *tups_vacuumed += 1; + *tups_recently_dead -= 1; + } + + continue; } - continue; } *num_tuples += 1; @@ -933,7 +957,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, { const int ct_index[] = { PROGRESS_REPACK_HEAP_TUPLES_SCANNED, - PROGRESS_REPACK_HEAP_TUPLES_WRITTEN + PROGRESS_REPACK_HEAP_TUPLES_INSERTED }; int64 ct_val[2]; @@ -948,6 +972,31 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, ct_val[1] = *num_tuples; pgstat_progress_update_multi_param(2, ct_index, ct_val); } + + /* + * Process the WAL produced by the load, as well as by other + * transactions, so that the replication slot can advance and WAL does + * not pile up. Use wal_segment_size as a threshold so that we do not + * introduce the decoding overhead too often. + * + * Of course, we must not apply the changes until the initial load has + * completed. + * + * Note that our insertions into the new table should not be decoded + * as we (intentionally) do not write the logical decoding specific + * information to WAL. + */ + if (concurrent) + { + XLogRecPtr end_of_wal; + + end_of_wal = GetFlushRecPtr(NULL); + if ((end_of_wal - end_of_wal_prev) > wal_segment_size) + { + repack_decode_concurrent_changes(decoding_ctx, end_of_wal); + end_of_wal_prev = end_of_wal; + } + } } if (indexScan != NULL) @@ -991,15 +1040,32 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, values, isnull, rwstate); /* Report n_tuples */ - pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_WRITTEN, + pgstat_progress_update_param(PROGRESS_REPACK_HEAP_TUPLES_INSERTED, n_tuples); + + /* + * Try to keep the amount of not-yet-decoded WAL small, like + * above. + */ + if (concurrent) + { + XLogRecPtr end_of_wal; + + end_of_wal = GetFlushRecPtr(NULL); + if ((end_of_wal - end_of_wal_prev) > wal_segment_size) + { + repack_decode_concurrent_changes(decoding_ctx, end_of_wal); + end_of_wal_prev = end_of_wal; + } + } } tuplesort_end(tuplesort); } /* Write out any remaining tuples, and fsync if needed */ - end_heap_rewrite(rwstate); + if (rwstate) + end_heap_rewrite(rwstate); /* Clean up */ pfree(values); @@ -2390,6 +2456,10 @@ heapam_scan_sample_next_tuple(TableScanDesc scan, SampleScanState *scanstate, * SET WITHOUT OIDS. * * So, we must reconstruct the tuple from component Datums. + * + * If rwstate=NULL, use simple_heap_insert() instead of rewriting - in that + * case we still need to deform/form the tuple. TODO Shouldn't we rename the + * function, as might not do any rewrite? */ static void reform_and_rewrite_tuple(HeapTuple tuple, @@ -2412,8 +2482,28 @@ reform_and_rewrite_tuple(HeapTuple tuple, copiedTuple = heap_form_tuple(newTupDesc, values, isnull); - /* The heap rewrite module does the rest */ - rewrite_heap_tuple(rwstate, tuple, copiedTuple); + if (rwstate) + /* The heap rewrite module does the rest */ + rewrite_heap_tuple(rwstate, tuple, copiedTuple); + else + { + /* + * Insert tuple when processing REPACK CONCURRENTLY. + * + * rewriteheap.c is not used in the CONCURRENTLY case because it'd be + * difficult to do the same in the catch-up phase (as the logical + * decoding does not provide us with sufficient visibility + * information). Thus we must use heap_insert() both during the + * catch-up and here. + * + * The following is like simple_heap_insert() except that we pass the + * flag to skip logical decoding: as soon as REPACK CONCURRENTLY swaps + * the relation files, it drops this relation, so no logical + * replication subscription should need the data. + */ + heap_insert(NewHeap, copiedTuple, GetCurrentCommandId(true), + HEAP_INSERT_NO_LOGICAL, NULL); + } heap_freetuple(copiedTuple); } diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c index 77fd48eb59e..96be7684660 100644 --- a/src/backend/access/heap/rewriteheap.c +++ b/src/backend/access/heap/rewriteheap.c @@ -620,9 +620,9 @@ raw_heap_insert(RewriteState state, HeapTuple tup) int options = HEAP_INSERT_SKIP_FSM; /* - * While rewriting the heap for VACUUM FULL / CLUSTER, make sure data - * for the TOAST table are not logically decoded. The main heap is - * WAL-logged as XLOG FPI records, which are not logically decoded. + * While rewriting the heap for REPACK, make sure data for the TOAST + * table are not logically decoded. The main heap is WAL-logged as + * XLOG FPI records, which are not logically decoded. */ options |= HEAP_INSERT_NO_LOGICAL; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 201c503548e..01e1b09493e 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1319,16 +1319,19 @@ CREATE VIEW pg_stat_progress_repack AS WHEN 2 THEN 'index scanning heap' WHEN 3 THEN 'sorting tuples' WHEN 4 THEN 'writing new heap' - WHEN 5 THEN 'swapping relation files' - WHEN 6 THEN 'rebuilding index' - WHEN 7 THEN 'performing final cleanup' + WHEN 5 THEN 'catch-up' + WHEN 6 THEN 'swapping relation files' + WHEN 7 THEN 'rebuilding index' + WHEN 8 THEN 'performing final cleanup' END AS phase, CAST(S.param3 AS oid) AS repack_index_relid, S.param4 AS heap_tuples_scanned, - S.param5 AS heap_tuples_written, - S.param6 AS heap_blks_total, - S.param7 AS heap_blks_scanned, - S.param8 AS index_rebuild_count + S.param5 AS heap_tuples_inserted, + S.param6 AS heap_tuples_updated, + S.param7 AS heap_tuples_deleted, + S.param8 AS heap_blks_total, + S.param9 AS heap_blks_scanned, + S.param10 AS index_rebuild_count FROM pg_stat_get_progress_info('REPACK') AS S LEFT JOIN pg_database D ON S.datid = D.oid; @@ -1346,7 +1349,7 @@ CREATE VIEW pg_stat_progress_cluster AS phase, repack_index_relid AS cluster_index_relid, heap_tuples_scanned, - heap_tuples_written, + heap_tuples_inserted + heap_tuples_updated AS heap_tuples_written, heap_blks_total, heap_blks_scanned, index_rebuild_count diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index d70636a6a63..6111f7985ab 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1,8 +1,23 @@ /*------------------------------------------------------------------------- * * cluster.c - * CLUSTER a table on an index. This is now also used for VACUUM FULL and - * REPACK. + * Implementation of REPACK [CONCURRENTLY], also known as CLUSTER and + * VACUUM FULL. + * + * 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. The changes + * are accumulated in a tuplestore. Once the initial copy is complete, we + * read the changes from the tuplestore 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. * * There is hardly anything left of Paul Brown's original implementation... * @@ -26,6 +41,10 @@ #include "access/toast_internals.h" #include "access/transam.h" #include "access/xact.h" +#include "access/xlog.h" +#include "access/xlog_internal.h" +#include "access/xloginsert.h" +#include "access/xlogutils.h" #include "catalog/catalog.h" #include "catalog/dependency.h" #include "catalog/heap.h" @@ -33,6 +52,7 @@ #include "catalog/namespace.h" #include "catalog/objectaccess.h" #include "catalog/pg_am.h" +#include "catalog/pg_control.h" #include "catalog/pg_inherits.h" #include "catalog/toasting.h" #include "commands/cluster.h" @@ -40,15 +60,21 @@ #include "commands/progress.h" #include "commands/tablecmds.h" #include "commands/vacuum.h" +#include "executor/executor.h" #include "miscadmin.h" #include "optimizer/optimizer.h" #include "pgstat.h" +#include "replication/decode.h" +#include "replication/logical.h" +#include "replication/snapbuild.h" #include "storage/bufmgr.h" +#include "storage/ipc.h" #include "storage/lmgr.h" #include "storage/predicate.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" @@ -68,12 +94,62 @@ typedef struct Oid indexOid; } RelToCluster; +/* + * The following definitions are used for concurrent processing. + */ + +/* + * The locators are used to avoid logical decoding of data that we do not need + * for our table. + */ +static RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; +static RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; + +/* + * Everything we need to call ExecInsertIndexTuples(). + */ +typedef struct IndexInsertState +{ + ResultRelInfo *rri; + EState *estate; +} IndexInsertState; + +/* The WAL segment being decoded. */ +static XLogSegNo repack_current_segment = 0; + +/* + * Information needed to apply concurrent data changes. + */ +typedef struct ChangeDest +{ + /* The relation the changes are applied to. */ + Relation rel; + + /* + * The following is needed to find the existing tuple if the change is + * UPDATE or DELETE. 'ident_key' should have all the fields except for + * 'sk_argument' initialized. + */ + Relation ident_index; + ScanKey ident_key; + int ident_key_nentries; + + /* Needed to update indexes of rel_dst. */ + IndexInsertState *iistate; +} ChangeDest; + static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, - Oid indexOid, Oid userid, int options); -static void rebuild_relation(Relation OldHeap, Relation index, bool verbose); + Oid indexOid, Oid userid, LOCKMODE lmode, + int options); +static void check_repack_concurrently_requirements(Relation rel); +static void rebuild_relation(Relation OldHeap, Relation index, bool verbose, + bool concurrent); static void copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, - bool verbose, bool *pSwapToastByContent, - TransactionId *pFreezeXid, MultiXactId *pCutoffMulti); + Snapshot snapshot, LogicalDecodingContext *decoding_ctx, + 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(RepackCommand cmd, @@ -81,13 +157,50 @@ static List *get_tables_to_repack_partitioned(RepackCommand cmd, MemoryContext permcxt); static bool cluster_is_permitted_for_relation(RepackCommand cmd, Oid relid, Oid userid); + +static void begin_concurrent_repack(Relation rel); +static void end_concurrent_repack(void); +static LogicalDecodingContext *setup_logical_decoding(Oid relid); +static HeapTuple get_changed_tuple(char *change); +static void apply_concurrent_changes(RepackDecodingState *dstate, + ChangeDest *dest); +static void apply_concurrent_insert(Relation rel, HeapTuple tup, + IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_update(Relation rel, HeapTuple tup, + HeapTuple tup_target, + IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_delete(Relation rel, HeapTuple tup_target); +static HeapTuple find_target_tuple(Relation rel, ChangeDest *dest, + HeapTuple tup_key, + TupleTableSlot *ident_slot); +static void process_concurrent_changes(LogicalDecodingContext *decoding_ctx, + XLogRecPtr end_of_wal, + ChangeDest *dest); +static IndexInsertState *get_index_insert_state(Relation relation, + Oid ident_index_id, + Relation *ident_index_p); +static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, + int *nentries); +static void free_index_insert_state(IndexInsertState *iistate); +static void cleanup_logical_decoding(LogicalDecodingContext *ctx); +static void rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, + LogicalDecodingContext *decoding_ctx, + TransactionId frozenXid, + MultiXactId cutoffMulti); +static List *build_new_indexes(Relation NewHeap, Relation OldHeap, List *OldIndexes); 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 const char *RepackCommandAsString(RepackCommand cmd); +#define REPL_PLUGIN_NAME "pgoutput_repack" + /* * The repack code allows for processing multiple tables at once. Because * of this, we cannot just run everything on a single transaction, or we @@ -117,6 +230,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) ClusterParams params = {0}; Relation rel = NULL; MemoryContext repack_context; + LOCKMODE lockmode; List *rtcs; /* Parse option list */ @@ -127,6 +241,16 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) else if (strcmp(opt->defname, "analyze") == 0 || strcmp(opt->defname, "analyse") == 0) params.options |= defGetBoolean(opt) ? CLUOPT_ANALYZE : 0; + else if (strcmp(opt->defname, "concurrently") == 0 && + defGetBoolean(opt)) + { + if (stmt->command != REPACK_COMMAND_REPACK) + ereport(ERROR, + errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("CONCURRENTLY option not supported for %s", + RepackCommandAsString(stmt->command))); + params.options |= CLUOPT_CONCURRENT; + } else ereport(ERROR, errcode(ERRCODE_SYNTAX_ERROR), @@ -136,13 +260,25 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) parser_errposition(pstate, opt->location)); } + /* + * Determine the lock mode expected by cluster_rel(). + * + * In the exclusive case, we obtain AccessExclusiveLock right away to + * avoid lock-upgrade hazard in the single-transaction case. In the + * CONCURRENTLY case, the AccessExclusiveLock will only be used at the end + * of processing, supposedly for very short time. Until then, we'll have + * to unlock the relation temporarily, so there's no lock-upgrade hazard. + */ + lockmode = (params.options & CLUOPT_CONCURRENT) == 0 ? + AccessExclusiveLock : ShareUpdateExclusiveLock; + /* * 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, ¶ms); + rel = process_single_relation(stmt, lockmode, isTopLevel, ¶ms); if (rel == NULL) return; /* all done */ } @@ -157,10 +293,29 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) errmsg("cannot %s multiple tables", "REPACK (ANALYZE)")); /* - * By here, we know we are in a multi-table situation. 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. + * 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, + errmsg("REPACK CONCURRENTLY not supported for partitioned tables"), + errhint("Consider running the command for individual partitions.")); + } + else + ereport(ERROR, + errmsg("REPACK CONCURRENTLY requires explicit table name")); + } + + /* + * In order to avoid holding locks for too long, we want to process each + * table in its own transaction. This forces us to disallow running + * inside a user transaction block. */ PreventInTransactionBlock(isTopLevel, RepackCommandAsString(stmt->command)); @@ -244,7 +399,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) * Open the target table, coping with the case where it has been * dropped. */ - rel = try_table_open(rtc->tableOid, AccessExclusiveLock); + rel = try_table_open(rtc->tableOid, lockmode); if (rel == NULL) { CommitTransactionCommand(); @@ -255,7 +410,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) PushActiveSnapshot(GetTransactionSnapshot()); /* Process this table */ - cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms); + cluster_rel(stmt->command, rel, rtc->indexOid, ¶ms, isTopLevel); /* cluster_rel closes the relation, but keeps lock */ PopActiveSnapshot(); @@ -284,22 +439,53 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) * 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) + 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); - Relation index; + bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); - Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false)); + /* + * The lock mode is AccessExclusiveLock for normal processing and + * ShareUpdateExclusiveLock for concurrent processing (so that SELECT, + * INSERT, UPDATE and DELETE commands work, but cluster_rel() cannot be + * called concurrently for the same relation). + */ + lmode = !concurrent ? AccessExclusiveLock : ShareUpdateExclusiveLock; + + /* There are specific requirements on concurrent processing. */ + if (concurrent) + { + /* + * Make sure we have no XID assigned, otherwise call of + * setup_logical_decoding() can cause a deadlock. + * + * The existence of transaction block actually does not imply that XID + * was already assigned, but it very likely is. We might want to check + * the result of GetCurrentTransactionIdIfAny() instead, but that + * would be less clear from user's perspective. + */ + PreventInTransactionBlock(isTopLevel, "REPACK (CONCURRENTLY)"); + + check_repack_concurrently_requirements(OldHeap); + } /* Check for user-requested abort. */ CHECK_FOR_INTERRUPTS(); @@ -325,10 +511,13 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, * If this is a single-transaction CLUSTER, we can skip these tests. We * *must* skip the one on indisclustered since it would reject an attempt * to cluster a not-previously-clustered index. + * + * XXX move [some of] these comments to where the RECHECK flag is + * determined? */ if (recheck && !cluster_rel_recheck(cmd, OldHeap, indexOid, save_userid, - params->options)) + lmode, params->options)) goto out; /* @@ -343,6 +532,12 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, errmsg("cannot run %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. @@ -363,7 +558,7 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, if (OidIsValid(indexOid)) { /* verify the index is good and lock it */ - check_index_is_clusterable(OldHeap, indexOid, AccessExclusiveLock); + check_index_is_clusterable(OldHeap, indexOid, lmode); /* also open it */ index = index_open(indexOid, NoLock); } @@ -398,7 +593,9 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, if (OldHeap->rd_rel->relkind == RELKIND_MATVIEW && !RelationIsPopulated(OldHeap)) { - relation_close(OldHeap, AccessExclusiveLock); + if (index) + index_close(index, lmode); + relation_close(OldHeap, lmode); goto out; } @@ -411,11 +608,34 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, * 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. */ - TransferPredicateLocksToHeapRelation(OldHeap); + if (!concurrent) + TransferPredicateLocksToHeapRelation(OldHeap); /* rebuild_relation does all the dirty work */ - rebuild_relation(OldHeap, index, verbose); + PG_TRY(); + { + /* + * For concurrent processing, make sure that our logical decoding + * ignores data changes of other tables than the one we are + * processing. + */ + if (concurrent) + begin_concurrent_repack(OldHeap); + + rebuild_relation(OldHeap, index, verbose, concurrent); + } + PG_FINALLY(); + { + if (concurrent) + end_concurrent_repack(); + } + PG_END_TRY(); + /* rebuild_relation closes OldHeap, and index if valid */ out: @@ -434,14 +654,14 @@ out: */ static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, - Oid userid, int options) + Oid userid, LOCKMODE lmode, int options) { Oid tableOid = RelationGetRelid(OldHeap); /* Check that the user still has privileges for the relation */ if (!cluster_is_permitted_for_relation(cmd, tableOid, userid)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -455,7 +675,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, */ if (RELATION_IS_OTHER_TEMP(OldHeap)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -466,7 +686,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, */ if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(indexOid))) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } @@ -477,7 +697,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, if ((options & CLUOPT_RECHECK_ISCLUSTERED) != 0 && !get_index_isclustered(indexOid)) { - relation_close(OldHeap, AccessExclusiveLock); + relation_close(OldHeap, lmode); return false; } } @@ -489,7 +709,7 @@ cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, Oid indexOid, * 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 AccessExclusiveLock on the table, but + * 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. */ @@ -619,17 +839,86 @@ mark_index_clustered(Relation rel, Oid indexOid, bool is_internal) } /* - * rebuild_relation: rebuild an existing relation in index or physical order - * - * OldHeap: table to rebuild. - * index: index to cluster by, or NULL to rewrite in physical order. - * - * On entry, heap and index (if one is given) must be open, and - * AccessExclusiveLock held on them. - * On exit, they are closed, but locks on them are not released. + * Check if the CONCURRENTLY option is legal for the relation. */ static void -rebuild_relation(Relation OldHeap, Relation index, bool verbose) +check_repack_concurrently_requirements(Relation rel) +{ + char relpersistence, + replident; + Oid ident_idx; + + /* Data changes in system relations are not logically decoded. */ + if (IsCatalogRelation(rel)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is not supported for catalog relations."))); + + /* + * reorderbuffer.c does not seem to handle processing of TOAST relation + * alone. + */ + if (IsToastRelation(rel)) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is not supported for TOAST relations, unless the main relation is repacked too."))); + + relpersistence = rel->rd_rel->relpersistence; + if (relpersistence != RELPERSISTENCE_PERMANENT) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("REPACK CONCURRENTLY is only allowed for permanent relations."))); + + /* With NOTHING, WAL does not contain the old tuple. */ + replident = rel->rd_rel->relreplident; + if (replident == REPLICA_IDENTITY_NOTHING) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot repack relation \"%s\"", + RelationGetRelationName(rel)), + errhint("Relation \"%s\" has insufficient replication identity.", + RelationGetRelationName(rel)))); + + /* + * Identity index is not set if the replica identity is FULL, but PK might + * exist in such a case. + */ + ident_idx = RelationGetReplicaIndex(rel); + if (!OidIsValid(ident_idx) && OidIsValid(rel->rd_pkindex)) + ident_idx = rel->rd_pkindex; + if (!OidIsValid(ident_idx)) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot process relation \"%s\"", + RelationGetRelationName(rel)), + errhint("Relation \"%s\" has no identity index.", + RelationGetRelationName(rel)))); +} + + +/* + * 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. + * + * 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, bool concurrent) { Oid tableOid = RelationGetRelid(OldHeap); Oid accessMethod = OldHeap->rd_rel->relam; @@ -637,13 +926,38 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose) Oid OIDNewHeap; Relation NewHeap; char relpersistence; - bool is_system_catalog; bool swap_toast_by_content; TransactionId frozenXid; MultiXactId cutoffMulti; + LogicalDecodingContext *decoding_ctx = NULL; + Snapshot snapshot = NULL; +#if USE_ASSERT_CHECKING + LOCKMODE lmode; - Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false) && - (index == NULL || CheckRelationLockedByMe(index, AccessExclusiveLock, false))); + lmode = concurrent ? ShareUpdateExclusiveLock : AccessExclusiveLock; + + Assert(CheckRelationLockedByMe(OldHeap, lmode, false)); + Assert(index == NULL || CheckRelationLockedByMe(index, lmode, false)); +#endif + + if (concurrent) + { + /* + * Prepare to capture the concurrent data changes. + * + * Note that this call waits 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 its still eligible for REPACK + * CONCURRENTLY. + */ + decoding_ctx = setup_logical_decoding(tableOid); + + snapshot = SnapBuildInitialSnapshotForRepack(decoding_ctx->snapshot_builder); + PushActiveSnapshot(snapshot); + } /* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */ if (index != NULL) @@ -651,7 +965,6 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose) /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; - is_system_catalog = IsSystemRelation(OldHeap); /* * Create the transient table that will receive the re-ordered data. @@ -667,30 +980,62 @@ rebuild_relation(Relation OldHeap, Relation index, bool verbose) NewHeap = table_open(OIDNewHeap, NoLock); /* Copy the heap data into the new table in the desired order */ - copy_table_data(NewHeap, OldHeap, index, verbose, + copy_table_data(NewHeap, OldHeap, index, snapshot, decoding_ctx, verbose, &swap_toast_by_content, &frozenXid, &cutoffMulti); + /* The historic snapshot won't be needed anymore. */ + if (snapshot) + { + PopActiveSnapshot(); + UpdateActiveSnapshotCommandId(); + } - /* Close relcache entries, but keep lock until transaction commit */ - table_close(OldHeap, NoLock); - if (index) - index_close(index, NoLock); + if (concurrent) + { + Assert(!swap_toast_by_content); - /* - * 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); + /* + * Close the index, but keep the lock. Both heaps will be closed by + * the following call. + */ + if (index) + index_close(index, 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, - frozenXid, cutoffMulti, - relpersistence); + rebuild_relation_finish_concurrent(NewHeap, OldHeap, decoding_ctx, + frozenXid, cutoffMulti); + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_FINAL_CLEANUP); + + /* Done with decoding. */ + cleanup_logical_decoding(decoding_ctx); + } + 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); + } } @@ -825,15 +1170,19 @@ make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, /* * 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, bool verbose, - bool *pSwapToastByContent, TransactionId *pFreezeXid, - MultiXactId *pCutoffMulti) +copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, +