agora inbox for [email protected]help / color / mirror / Atom feed
From: Álvaro Herrera <[email protected]> Subject: [PATCH v20 5/6] Add CONCURRENTLY option to REPACK command. Date: Sat, 30 Aug 2025 19:13:38 +0200 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. Author: Antonin Houska <[email protected]> --- 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 | 227 ++- src/backend/access/heap/rewriteheap.c | 6 +- src/backend/access/transam/xact.c | 11 +- src/backend/catalog/system_views.sql | 30 +- src/backend/commands/cluster.c | 1677 +++++++++++++++-- src/backend/commands/matview.c | 2 +- src/backend/commands/tablecmds.c | 1 + src/backend/commands/vacuum.c | 12 +- src/backend/meson.build | 1 + src/backend/replication/logical/decode.c | 83 + src/backend/replication/logical/snapbuild.c | 20 + .../replication/pgoutput_repack/Makefile | 32 + .../replication/pgoutput_repack/meson.build | 18 + .../pgoutput_repack/pgoutput_repack.c | 288 +++ src/backend/storage/ipc/ipci.c | 1 + .../storage/lmgr/generate-lwlocknames.pl | 2 +- src/backend/utils/cache/relcache.c | 1 + src/backend/utils/time/snapmgr.c | 3 +- src/bin/psql/tab-complete.in.c | 25 +- src/include/access/heapam.h | 9 +- src/include/access/heapam_xlog.h | 2 + src/include/access/tableam.h | 10 + src/include/commands/cluster.h | 91 +- src/include/commands/progress.h | 23 +- src/include/replication/snapbuild.h | 1 + src/include/storage/lockdefs.h | 4 +- src/include/utils/snapmgr.h | 2 + src/test/modules/injection_points/Makefile | 5 +- .../injection_points/expected/repack.out | 113 ++ .../modules/injection_points/logical.conf | 1 + src/test/modules/injection_points/meson.build | 4 + .../injection_points/specs/repack.spec | 143 ++ src/test/regress/expected/rules.out | 29 +- src/tools/pgindent/typedefs.list | 4 + 39 files changed, 2821 insertions(+), 273 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/logical.conf create mode 100644 src/test/modules/injection_points/specs/repack.spec diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 12e103d319d..61c0197555f 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -6074,14 +6074,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> @@ -6162,6 +6183,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 + phase. + </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 049ee75a4ba..0f5c34af542 100644 --- a/doc/src/sgml/mvcc.sgml +++ b/doc/src/sgml/mvcc.sgml @@ -1833,15 +1833,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 fd9d89f8aaa..ff5ce48de55 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -27,6 +27,7 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re VERBOSE [ <replaceable class="parameter">boolean</replaceable> ] ANALYSE | ANALYZE + CONCURRENTLY </synopsis> </refsynopsisdiv> @@ -49,7 +50,8 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re 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> @@ -62,7 +64,8 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re 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"> @@ -179,6 +182,128 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] [ <re </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 + 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 e3e7307ef5f..f9a4fe3faed 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 wal_logical); #ifdef USE_ASSERT_CHECKING static void check_lock_if_inplace_updateable_rel(Relation relation, ItemPointer otid, @@ -2780,7 +2781,7 @@ xmax_infomask_changed(uint16 new_infomask, uint16 old_infomask) TM_Result heap_delete(Relation relation, ItemPointer tid, CommandId cid, Snapshot crosscheck, bool wait, - TM_FailureData *tmfd, bool changingPart) + TM_FailureData *tmfd, bool changingPart, bool wal_logical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -3027,7 +3028,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 = wal_logical ? + ExtractReplicaIdentity(relation, &tp, true, &old_key_copied) : NULL; /* * If this is the first possibly-multixact-able operation in the current @@ -3117,6 +3119,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 (!wal_logical) + xlrec.flags |= XLH_DELETE_NO_LOGICAL; + XLogBeginInsert(); XLogRegisterData(&xlrec, SizeOfHeapDelete); @@ -3209,7 +3220,8 @@ simple_heap_delete(Relation relation, ItemPointer tid) result = heap_delete(relation, tid, GetCurrentCommandId(true), InvalidSnapshot, true /* wait for commit */ , - &tmfd, false /* changingPart */ ); + &tmfd, false, /* changingPart */ + true /* wal_logical */ ); switch (result) { case TM_SelfModified: @@ -3250,7 +3262,7 @@ TM_Result heap_update(Relation relation, ItemPointer otid, HeapTuple newtup, CommandId cid, Snapshot crosscheck, bool wait, TM_FailureData *tmfd, LockTupleMode *lockmode, - TU_UpdateIndexes *update_indexes) + TU_UpdateIndexes *update_indexes, bool wal_logical) { TM_Result result; TransactionId xid = GetCurrentTransactionId(); @@ -4143,7 +4155,8 @@ l2: newbuf, &oldtup, heaptup, old_key_tuple, all_visible_cleared, - all_visible_cleared_new); + all_visible_cleared_new, + wal_logical); if (newbuf != buffer) { PageSetLSN(BufferGetPage(newbuf), recptr); @@ -4501,7 +4514,8 @@ simple_heap_update(Relation relation, ItemPointer 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 /* wal_logical */ ); switch (result) { case TM_SelfModified: @@ -8842,7 +8856,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 wal_logical) { xl_heap_update xlrec; xl_heap_header xlhdr; @@ -8853,7 +8868,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) && + wal_logical; bool init; int bufflags; diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 79f9de5d760..d03084768e0 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 = (Datum *) palloc(natts * sizeof(Datum)); isnull = (bool *) palloc(natts * sizeof(bool)); - /* 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; @@ -785,6 +801,7 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, HeapTuple tuple; Buffer buf; bool isdead; + HTSV_Result vis; CHECK_FOR_INTERRUPTS(); @@ -837,70 +854,84 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, tuple = ExecFetchSlotHeapTuple(slot, false, NULL); buf = hslot->buffer; - LockBuffer(buf, BUFFER_LOCK_SHARE); - - switch (HeapTupleSatisfiesVacuum(tuple, OldestXmin, buf)) + /* + * Regarding CONCURRENTLY, see the comments on MVCC snapshot above. + */ + if (!concurrent) { - case HEAPTUPLE_DEAD: - /* Definitely dead */ - isdead = true; - break; - case HEAPTUPLE_RECENTLY_DEAD: - *tups_recently_dead += 1; - /* fall through */ - case HEAPTUPLE_LIVE: - /* Live or recently dead, must copy it */ - isdead = false; - break; - case HEAPTUPLE_INSERT_IN_PROGRESS: + LockBuffer(buf, BUFFER_LOCK_SHARE); - /* - * 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 ((vis = 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; + /* fall through */ + 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. Also, there's no exclusive lock + * during concurrent processing. Give a warning if neither + * case applies; but in any case we had better copy it. + */ + if (!is_system_catalog && !concurrent && + !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 && !concurrent && + !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; } - continue; + + 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; + } + + LockBuffer(buf, BUFFER_LOCK_UNLOCK); + continue; + } + + /* + * In the concurrent case, we have a copy of the tuple, so we + * don't worry whether the source tuple will be deleted / updated + * after we release the lock. + */ + LockBuffer(buf, BUFFER_LOCK_UNLOCK); } *num_tuples += 1; @@ -919,7 +950,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]; @@ -934,6 +965,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) @@ -977,7 +1033,7 @@ 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); } @@ -985,7 +1041,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, } /* Write out any remaining tuples, and fsync if needed */ - end_heap_rewrite(rwstate); + if (rwstate) + end_heap_rewrite(rwstate); /* Clean up */ pfree(values); @@ -2376,6 +2433,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, @@ -2398,8 +2459,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 e6d2b5fced1..6aa2ed214f2 100644 --- a/src/backend/access/heap/rewriteheap.c +++ b/src/backend/access/heap/rewriteheap.c @@ -617,9 +617,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/access/transam/xact.c b/src/backend/access/transam/xact.c index b46e7e9c2a6..5670f2bfbde 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -215,6 +215,7 @@ typedef struct TransactionStateData bool parallelChildXact; /* is any parent transaction parallel? */ bool chain; /* start a new block after this one */ bool topXidLogged; /* for a subxact: is top-level XID logged? */ + bool internal; /* for a subxact: launched internally? */ struct TransactionStateData *parent; /* back link to parent */ } TransactionStateData; @@ -4735,6 +4736,7 @@ BeginInternalSubTransaction(const char *name) /* Normal subtransaction start */ PushTransaction(); s = CurrentTransactionState; /* changed by push */ + s->internal = true; /* * Savepoint names, like the TransactionState block itself, live @@ -5251,7 +5253,13 @@ AbortSubTransaction(void) LWLockReleaseAll(); pgstat_report_wait_end(); - pgstat_progress_end_command(); + + /* + * Internal subtransacion might be used by an user command, in which case + * the command outlives the subtransaction. + */ + if (!s->internal) + pgstat_progress_end_command(); pgaio_error_cleanup(); @@ -5468,6 +5476,7 @@ PushTransaction(void) s->parallelModeLevel = 0; s->parallelChildXact = (p->parallelModeLevel != 0 || p->parallelChildXact); s->topXidLogged = false; + s->internal = false; CurrentTransactionState = s; diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index b2b7b10c2be..a92ac78ad9e 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1266,16 +1266,17 @@ CREATE VIEW pg_stat_progress_cluster 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' + -- 5 is 'catch-up', but that should not appear here. + 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 cluster_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.param8 AS heap_blks_total, + S.param9 AS heap_blks_scanned, + S.param10 AS index_rebuild_count FROM pg_stat_get_progress_info('CLUSTER') AS S LEFT JOIN pg_database D ON S.datid = D.oid; @@ -1291,16 +1292,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; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 8b64f9e6795..511b2bb6c43 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -25,6 +25,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" @@ -32,6 +36,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" @@ -39,15 +44,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" @@ -67,13 +78,45 @@ 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. + */ +RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; +RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; + +/* + * Everything we need to call ExecInsertIndexTuples(). + */ +typedef struct IndexInsertState +{ + ResultRelInfo *rri; + EState *estate; + + Relation ident_index; +} IndexInsertState; + +/* The WAL segment being decoded. */ +static XLogSegNo repack_current_segment = 0; + + static bool cluster_rel_recheck(RepackCommand cmd, Relation OldHeap, - Oid indexOid, Oid userid, int options); + Oid indexOid, Oid userid, LOCKMODE lmode, + int options); +static void check_repack_concurrently_requirements(Relation rel); static void rebuild_relation(RepackCommand cmd, bool usingindex, - Relation OldHeap, Relation index, bool verbose); + Relation OldHeap, Relation index, Oid userid, + 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,12 +124,61 @@ static List *get_tables_to_repack_partitioned(RepackCommand cmd, Oid relid, bool rel_is_index); 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, + const char *slotname, + TupleDesc tupdesc); +static HeapTuple get_changed_tuple(char *change); +static void apply_concurrent_changes(RepackDecodingState *dstate, + Relation rel, ScanKey key, int nkeys, + IndexInsertState *iistate); +static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, + HeapTuple tup, IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_update(Relation rel, HeapTuple tup, + HeapTuple tup_target, + ConcurrentChange *change, + IndexInsertState *iistate, + TupleTableSlot *index_slot); +static void apply_concurrent_delete(Relation rel, HeapTuple tup_target, + ConcurrentChange *change); +static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, + HeapTuple tup_key, + IndexInsertState *iistate, + TupleTableSlot *ident_slot, + IndexScanDesc *scan_p); +static void process_concurrent_changes(LogicalDecodingContext *ctx, + XLogRecPtr end_of_wal, + Relation rel_dst, + Relation rel_src, + ScanKey ident_key, + int ident_key_nentries, + IndexInsertState *iistate); +static IndexInsertState *get_index_insert_state(Relation relation, + Oid ident_index_id); +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, + Relation cl_index, + LogicalDecodingContext *ctx, + bool swap_toast_by_content, + 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); +#define REPL_PLUGIN_NAME "pgoutput_repack" + static const char * RepackCommandAsString(RepackCommand cmd) { @@ -95,7 +187,7 @@ RepackCommandAsString(RepackCommand cmd) case REPACK_COMMAND_REPACK: return "REPACK"; case REPACK_COMMAND_VACUUMFULL: - return "VACUUM"; + return "VACUUM (FULL)"; case REPACK_COMMAND_CLUSTER: return "CLUSTER"; } @@ -132,6 +224,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) ClusterParams params = {0}; Relation rel = NULL; MemoryContext repack_context; + LOCKMODE lockmode; List *rtcs; /* Parse option list */ @@ -142,6 +235,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), @@ -151,13 +254,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; } @@ -169,10 +284,29 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) errmsg("cannot ANALYZE multiple tables")); /* - * 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)); @@ -252,7 +386,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(); @@ -264,7 +398,7 @@ ExecRepack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel) /* Process this table */ cluster_rel(stmt->command, stmt->usingindex, - rel, rtc->indexOid, ¶ms); + rel, rtc->indexOid, ¶ms, isTopLevel); /* cluster_rel closes the relation, but keeps lock */ PopActiveSnapshot(); @@ -293,22 +427,55 @@ 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, bool usingindex, - Relation OldHeap, Oid indexOid, ClusterParams *params) + 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); - Relation index; + bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); - Assert(CheckRelationLockedByMe(OldHeap, AccessExclusiveLock, false)); + /* + * Check that the correct lock is held. 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(); @@ -351,11 +518,13 @@ cluster_rel(RepackCommand cmd, bool usingindex, * 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) - if (!cluster_rel_recheck(cmd, OldHeap, indexOid, save_userid, - params->options)) - goto out; + 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 @@ -369,6 +538,12 @@ cluster_rel(RepackCommand cmd, bool usingindex, 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. @@ -404,7 +579,7 @@ cluster_rel(RepackCommand cmd, bool usingindex, 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); } @@ -421,7 +596,9 @@ cluster_rel(RepackCommand cmd, bool usingindex, 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; } @@ -434,11 +611,35 @@ cluster_rel(RepackCommand cmd, bool usingindex, * 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(cmd, usingindex, 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(cmd, usingindex, OldHeap, index, save_userid, + verbose, concurrent); + } + PG_FINALLY(); + { + if (concurrent) + end_concurrent_repack(); + } + PG_END_TRY(); + /* rebuild_relation closes OldHeap, and index if valid */ out: @@ -457,14 +658,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; } @@ -478,7 +679,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; } @@ -489,7 +690,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; } @@ -500,7 +701,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; } } @@ -641,19 +842,89 @@ mark_index_clustered(Relation rel, Oid indexOid, bool is_internal) table_close(pg_index, RowExclusiveLock); } +/* + * Check if the CONCURRENTLY option is legal for the relation. + */ +static void +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. + * 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 - * AccessExclusiveLock held on them. - * On exit, they are closed, but locks on them are not released. + * 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(RepackCommand cmd, bool usingindex, - Relation OldHeap, Relation index, bool verbose) + Relation OldHeap, Relation index, Oid userid, + bool verbose, bool concurrent) { Oid tableOid = RelationGetRelid(OldHeap); Oid accessMethod = OldHeap->rd_rel->relam; @@ -661,13 +932,55 @@ rebuild_relation(RepackCommand cmd, bool usingindex, Oid OIDNewHeap; Relation NewHeap; char relpersistence; - bool is_system_catalog; bool swap_toast_by_content; TransactionId frozenXid; MultiXactId cutoffMulti; + NameData slotname; + LogicalDecodingContext *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(!usingindex || CheckRelationLockedByMe(index, lmode, false)); +#endif + + if (concurrent) + { + TupleDesc tupdesc; + + /* + * REPACK CONCURRENTLY is not allowed in a transaction block, so this + * should never fire. + */ + Assert(GetTopTransactionIdIfAny() == InvalidTransactionId); + + /* + * A single backend should not execute multiple REPACK commands at a + * time, so use PID to make the slot unique. + */ + snprintf(NameStr(slotname), NAMEDATALEN, "repack_%d", MyProcPid); + + tupdesc = CreateTupleDescCopy(RelationGetDescr(OldHeap)); + + /* + * 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. + */ + ctx = setup_logical_decoding(tableOid, NameStr(slotname), tupdesc); + + snapshot = SnapBuildInitialSnapshotForRepack(ctx->snapshot_builder); + PushActiveSnapshot(snapshot); + } /* for CLUSTER or REPACK USING INDEX, mark the index as the one to use */ if (usingindex) @@ -675,7 +988,6 @@ rebuild_relation(RepackCommand cmd, bool usingindex, /* 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. @@ -691,30 +1003,67 @@ rebuild_relation(RepackCommand cmd, bool usingindex, 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, ctx, verbose, &swap_toast_by_content, &frozenXid, &cutoffMulti); + /* The historic snapshot won't be needed anymore. */ + if (snapshot) + PopActiveSnapshot(); - /* Close relcache entries, but keep lock until transaction commit */ - table_close(OldHeap, NoLock); - if (index) - index_close(index, NoLock); + if (concurrent) + { + /* + * Push a snapshot that we will use to find old versions of rows when + * processing concurrent UPDATE and DELETE commands. (That snapshot + * should also be used by index expressions.) + */ + PushActiveSnapshot(GetTransactionSnapshot()); - /* - * 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); + /* + * Make sure we can find the tuples just inserted when applying DML + * commands on top of those. + */ + CommandCounterIncrement(); + UpdateActiveSnapshotCommandId(); - /* - * 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, index, + ctx, swap_toast_by_content, + frozenXid, cutoffMulti); + PopActiveSnapshot(); + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + PROGRESS_REPACK_PHASE_FINAL_CLEANUP); + + /* Done with decoding. */ + cleanup_logical_decoding(ctx); + ReplicationSlotRelease(); + ReplicationSlotDrop(NameStr(slotname), false); + } + 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, + frozenXid, cutoffMulti, + relpersistence); + } } @@ -849,15 +1198,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