agora inbox for pgsql-hackers@postgresql.orghelp / color / mirror / Atom feed
[PATCH v09 7/9] Introduce repack_max_xlock_time configuration variable. 13+ messages / 2 participants [nested] [flat]
* [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-02-26 08:17 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 133 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 14 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 292 insertions(+), 20 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index e55700f35b..758cf6849d 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -10888,6 +10888,37 @@ dynamic_library_path = 'C:\tools\postgresql;H:\my_project\lib;$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e351..0c250689d1 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 5876a96e79..beec45b18e 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1004,7 +1004,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 1ff0fcd1d9..a5790d77b5 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -108,6 +110,15 @@ RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; #define REPACK_CONCURRENT_IN_PROGRESS_MSG \ "relation \"%s\" is already being processed by REPACK CONCURRENTLY" +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -197,7 +208,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -214,13 +226,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -3016,7 +3030,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -3054,6 +3069,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -3096,7 +3114,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; HeapTuple tup_old = NULL; @@ -3126,6 +3145,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -3250,10 +3272,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it + * now. However, make sure the loop does not break if tup_old was set + * in the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3456,11 +3490,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3469,10 +3507,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3484,7 +3531,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3494,6 +3541,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3654,6 +3723,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, RelReopenInfo *rri = NULL; int nrel; Relation *ind_refs_all, *ind_refs_p; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3733,7 +3804,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Release the locks that allowed concurrent data changes, in order to @@ -3855,9 +3927,38 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index ad25cbb39c..a1a19f2cbd 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,6 +39,7 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" #include "commands/tablespace.h" #include "commands/trigger.h" @@ -2814,6 +2815,19 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop( + "The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 5362ff8051..b59f8aae7e 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -744,6 +744,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ef3cb55751..f5600bf4f6 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -154,7 +156,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void can_repack_concurrently(Relation rel); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed61..f2728d9422 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98..0f45f9d254 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v08-0008-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH v09 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-02-26 08:17 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 133 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 16 ++- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 293 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index bdcefa8140b..f6f248080ea 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11213,6 +11213,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 485d22b9488..9bb37eb83fb 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 1dafca4531f..0a2bacbb1df 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -108,6 +110,15 @@ RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; #define REPACK_CONCURRENT_IN_PROGRESS_MSG \ "relation \"%s\" is already being processed by REPACK CONCURRENTLY" +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -197,7 +208,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -214,13 +226,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -3021,7 +3035,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -3059,6 +3074,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -3101,7 +3119,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; HeapTuple tup_old = NULL; @@ -3131,6 +3150,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -3255,10 +3277,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it + * now. However, make sure the loop does not break if tup_old was set + * in the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3462,11 +3496,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3475,10 +3513,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3490,7 +3537,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3500,6 +3547,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3660,6 +3729,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, RelReopenInfo *rri = NULL; int nrel; Relation *ind_refs_all, *ind_refs_p; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3739,7 +3810,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Release the locks that allowed concurrent data changes, in order to @@ -3861,9 +3933,38 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 97cfd6e5a82..0ba416a1982 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2837,6 +2838,19 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop( + "The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 9f31e4071c7..d25c67f7047 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -761,6 +761,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ef3cb557516..f5600bf4f62 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -154,7 +156,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void can_repack_concurrently(Relation rel); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.39.5 --v2l46qwn2nggemiy Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v09-0008-Enable-logical-decoding-transiently-only-for-REP.patch" ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-02-26 08:17 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 133 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 14 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 292 insertions(+), 20 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index e55700f35b..758cf6849d 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -10888,6 +10888,37 @@ dynamic_library_path = 'C:\tools\postgresql;H:\my_project\lib;$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e351..0c250689d1 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 5876a96e79..beec45b18e 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1004,7 +1004,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 1ff0fcd1d9..a5790d77b5 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -108,6 +110,15 @@ RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; #define REPACK_CONCURRENT_IN_PROGRESS_MSG \ "relation \"%s\" is already being processed by REPACK CONCURRENTLY" +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -197,7 +208,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -214,13 +226,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -3016,7 +3030,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -3054,6 +3069,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -3096,7 +3114,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; HeapTuple tup_old = NULL; @@ -3126,6 +3145,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -3250,10 +3272,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it + * now. However, make sure the loop does not break if tup_old was set + * in the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3456,11 +3490,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3469,10 +3507,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3484,7 +3531,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3494,6 +3541,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3654,6 +3723,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, RelReopenInfo *rri = NULL; int nrel; Relation *ind_refs_all, *ind_refs_p; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3733,7 +3804,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Release the locks that allowed concurrent data changes, in order to @@ -3855,9 +3927,38 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index ad25cbb39c..a1a19f2cbd 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,6 +39,7 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" #include "commands/tablespace.h" #include "commands/trigger.h" @@ -2814,6 +2815,19 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop( + "The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 5362ff8051..b59f8aae7e 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -744,6 +744,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ef3cb55751..f5600bf4f6 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -154,7 +156,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void can_repack_concurrently(Relation rel); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed61..f2728d9422 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98..0f45f9d254 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v08-0008-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH v09 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-02-26 08:17 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 133 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 16 ++- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 293 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index bdcefa8140b..f6f248080ea 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11213,6 +11213,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 485d22b9488..9bb37eb83fb 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 1dafca4531f..0a2bacbb1df 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -108,6 +110,15 @@ RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; #define REPACK_CONCURRENT_IN_PROGRESS_MSG \ "relation \"%s\" is already being processed by REPACK CONCURRENTLY" +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -197,7 +208,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -214,13 +226,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -3021,7 +3035,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -3059,6 +3074,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -3101,7 +3119,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; HeapTuple tup_old = NULL; @@ -3131,6 +3150,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -3255,10 +3277,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it + * now. However, make sure the loop does not break if tup_old was set + * in the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3462,11 +3496,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3475,10 +3513,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3490,7 +3537,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3500,6 +3547,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3660,6 +3729,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, RelReopenInfo *rri = NULL; int nrel; Relation *ind_refs_all, *ind_refs_p; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3739,7 +3810,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Release the locks that allowed concurrent data changes, in order to @@ -3861,9 +3933,38 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 97cfd6e5a82..0ba416a1982 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2837,6 +2838,19 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop( + "The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 9f31e4071c7..d25c67f7047 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -761,6 +761,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ef3cb557516..f5600bf4f62 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -154,7 +156,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void can_repack_concurrently(Relation rel); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.39.5 --v2l46qwn2nggemiy Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v09-0008-Enable-logical-decoding-transiently-only-for-REP.patch" ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH v10 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-03-25 12:42 Álvaro Herrera <alvherre@alvh.no-ip.org> 0 siblings, 0 replies; 13+ messages in thread From: Álvaro Herrera @ 2025-03-25 12:42 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 294 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 69fc93dffc4..4ab14939387 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11205,6 +11205,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index ea1d6f299b3..850708c7830 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index fa455e99d4a..c272ed03cb9 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -109,6 +111,15 @@ RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; #define REPACK_CONCURRENT_IN_PROGRESS_MSG \ "relation \"%s\" is already being processed by REPACK CONCURRENTLY" +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -198,7 +209,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -215,13 +227,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -3039,7 +3053,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -3077,6 +3092,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -3119,7 +3137,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -3151,6 +3170,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -3275,10 +3297,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3482,11 +3516,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3495,10 +3533,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3510,7 +3557,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3520,6 +3567,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3684,6 +3753,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, int nrel; Relation *ind_refs_all, *ind_refs_p; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3764,7 +3835,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Release the locks that allowed concurrent data changes, in order to @@ -3889,9 +3961,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 989825d3a9c..4e695246fc5 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2837,6 +2838,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 0b9e3066bde..bb256414142 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -763,6 +763,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index cad10a02bd0..268c3098512 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -154,7 +156,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void can_repack_concurrently(Relation rel); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.39.5 --w2lc3gf2kw4enm6y Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v10-0008-Enable-logical-decoding-transiently-only-for-REP.patch" ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-03-31 13:47 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-03-31 13:47 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 294 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 0d02e21a1ab..f596eef7bcf 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11183,6 +11183,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index ea1d6f299b3..850708c7830 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index a48e25deb5f..3f07b42d615 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -103,6 +105,15 @@ static Oid repacked_rel = InvalidOid; RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -149,7 +160,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -166,13 +178,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2597,7 +2611,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2627,6 +2642,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2667,7 +2685,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2698,6 +2717,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2822,10 +2844,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3029,11 +3063,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3042,10 +3080,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3057,7 +3104,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3067,6 +3114,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3228,6 +3297,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3308,7 +3379,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3396,9 +3468,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 4eaeca89f2c..533fd40d383 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2827,6 +2828,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index ff56a1f0732..bc0217161ec 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -763,6 +763,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ab1d9fc25dc..be283c70fce 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -153,7 +155,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, ClusterCommand cmd); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v11-0008-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-03-31 13:47 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-03-31 13:47 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 294 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 0d02e21a1ab..f596eef7bcf 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11183,6 +11183,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index ea1d6f299b3..850708c7830 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index a48e25deb5f..3f07b42d615 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -103,6 +105,15 @@ static Oid repacked_rel = InvalidOid; RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -149,7 +160,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -166,13 +178,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2597,7 +2611,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2627,6 +2642,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2667,7 +2685,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2698,6 +2717,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2822,10 +2844,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3029,11 +3063,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3042,10 +3080,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3057,7 +3104,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3067,6 +3114,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3228,6 +3297,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3308,7 +3379,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3396,9 +3468,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 4eaeca89f2c..533fd40d383 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2827,6 +2828,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index ff56a1f0732..bc0217161ec 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -763,6 +763,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ab1d9fc25dc..be283c70fce 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -153,7 +155,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, ClusterCommand cmd); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v11-0008-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. @ 2025-04-01 11:48 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-04-01 11:48 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 9 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 294 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index fea683cb49c..d1fc39b98be 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11190,6 +11190,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9ee640e3517..0c250689d13 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -188,7 +188,14 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU (<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. + 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. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index ea1d6f299b3..850708c7830 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -1008,7 +1008,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 3895013ca9d..085649716ac 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -103,6 +105,15 @@ static Oid repacked_rel = InvalidOid; RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -149,7 +160,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -166,13 +178,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2598,7 +2612,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2628,6 +2643,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2668,7 +2686,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2699,6 +2718,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2823,10 +2845,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -3030,11 +3064,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -3043,10 +3081,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -3058,7 +3105,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -3068,6 +3115,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -3229,6 +3298,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3309,7 +3380,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3397,9 +3469,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 4eaeca89f2c..533fd40d383 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -39,8 +39,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2827,6 +2828,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index ff56a1f0732..bc0217161ec 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -763,6 +763,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index ab1d9fc25dc..be283c70fce 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -153,7 +155,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, ClusterCommand cmd); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index 49a736ed617..f2728d94222 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index 5aa8983f98d..0f45f9d2544 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -127,6 +127,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -138,3 +166,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v12-0008-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. @ 2025-04-11 09:13 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-04-11 09:13 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 5 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 290 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index c1674c22cb2..c8119c8fd81 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11234,6 +11234,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index c837e4614f3..7e44fa636ac 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -195,7 +195,10 @@ INDEX <replaceable class="parameter">index_name</replaceable> ] 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. + <literal>ACCESS EXCLUSIVE</literal> lock is being held. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index 9d55004305f..19d2b33f88f 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -986,7 +986,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 058b750a0ed..cc765b88d52 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -89,6 +91,15 @@ typedef struct RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -132,7 +143,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -148,13 +160,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2351,7 +2365,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2381,6 +2396,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2421,7 +2439,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2451,6 +2470,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2551,10 +2573,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -2736,11 +2770,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -2749,10 +2787,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -2764,7 +2811,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -2772,6 +2819,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -2933,6 +3002,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3028,7 +3099,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3124,9 +3196,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock"); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 60b12446a1c..fb5f1b5e11e 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -42,8 +42,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2839,6 +2840,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 34826d01380..b239c58cfc1 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -769,6 +769,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 0a7e72bc74a..4914f217267 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -134,7 +136,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index f919087ca5b..02967ed9d48 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index a17064462ce..d0fa38dd8cd 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -130,6 +130,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -141,3 +169,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v13-0007-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. @ 2025-06-09 10:00 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-06-09 10:00 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 5 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 290 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 021153b2a5f..fc8df5b6f3d 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11234,6 +11234,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9c089a6b3d7..e1313f40599 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -192,7 +192,10 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU 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. + <literal>ACCESS EXCLUSIVE</literal> lock is being held. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index c829c06f769..03e722347a1 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -986,7 +986,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 04dcfc900de..0ebc7eacad9 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -89,6 +91,15 @@ typedef struct RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -132,7 +143,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -148,13 +160,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2352,7 +2366,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2382,6 +2397,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2422,7 +2440,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2452,6 +2471,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2552,10 +2574,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -2737,11 +2771,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -2750,10 +2788,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -2765,7 +2812,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -2773,6 +2820,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -2934,6 +3003,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3029,7 +3100,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3125,9 +3197,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock", NULL); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index f04bfedb2fd..dac58787f30 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -42,8 +42,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2839,6 +2840,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 341f88adc87..42d32a2c198 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -765,6 +765,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 0a7e72bc74a..4914f217267 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -134,7 +136,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index f919087ca5b..02967ed9d48 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index a17064462ce..d0fa38dd8cd 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -130,6 +130,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -141,3 +169,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v14-0007-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. @ 2025-06-09 10:00 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-06-09 10:00 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 5 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 290 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 021153b2a5f..fc8df5b6f3d 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11234,6 +11234,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9c089a6b3d7..e1313f40599 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -192,7 +192,10 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU 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. + <literal>ACCESS EXCLUSIVE</literal> lock is being held. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index c829c06f769..03e722347a1 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -986,7 +986,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 04dcfc900de..0ebc7eacad9 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -89,6 +91,15 @@ typedef struct RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -132,7 +143,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -148,13 +160,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2352,7 +2366,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2382,6 +2397,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2422,7 +2440,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2452,6 +2471,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2552,10 +2574,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -2737,11 +2771,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -2750,10 +2788,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -2765,7 +2812,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -2773,6 +2820,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -2934,6 +3003,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3029,7 +3100,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3125,9 +3197,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock", NULL); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index f04bfedb2fd..dac58787f30 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -42,8 +42,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2839,6 +2840,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 341f88adc87..42d32a2c198 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -765,6 +765,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 0a7e72bc74a..4914f217267 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -134,7 +136,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index f919087ca5b..02967ed9d48 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index a17064462ce..d0fa38dd8cd 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -130,6 +130,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -141,3 +169,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v14-0007-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. @ 2025-06-30 17:41 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-06-30 17:41 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 5 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 290 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 59a0874528a..c0529005c78 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11239,6 +11239,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9c089a6b3d7..e1313f40599 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -192,7 +192,10 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU 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. + <literal>ACCESS EXCLUSIVE</literal> lock is being held. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index c829c06f769..03e722347a1 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -986,7 +986,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index abbbfc99036..37f69f369eb 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -89,6 +91,15 @@ typedef struct RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -132,7 +143,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -148,13 +160,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2352,7 +2366,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2382,6 +2397,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2422,7 +2440,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2452,6 +2471,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2552,10 +2574,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -2737,11 +2771,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -2750,10 +2788,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -2765,7 +2812,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -2773,6 +2820,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -2934,6 +3003,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3029,7 +3100,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3125,9 +3197,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock", NULL); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 511dc32d519..6a373a8b65a 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -42,8 +42,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2839,6 +2840,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 341f88adc87..42d32a2c198 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -765,6 +765,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 0a7e72bc74a..4914f217267 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -134,7 +136,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index f919087ca5b..02967ed9d48 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index a17064462ce..d0fa38dd8cd 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -130,6 +130,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -141,3 +169,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.47.1 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v15-0007-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
* [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. @ 2025-06-30 17:41 Antonin Houska <ah@cybertec.at> 0 siblings, 0 replies; 13+ messages in thread From: Antonin Houska @ 2025-06-30 17:41 UTC (permalink / raw) When executing REPACK CONCURRENTLY, we need the AccessExclusiveLock to swap the relation files and that should require pretty short time. However, on a busy system, other backends might change non-negligible amount of data in the table while we are waiting for the lock. Since these changes must be applied to the new storage before the swap, the time we eventually hold the lock might become non-negligible too. If the user is worried about this situation, he can set repack_max_xlock_time to the maximum time for which the exclusive lock may be held. If this amount of time is not sufficient to complete the REPACK CONCURRENTLY command, ERROR is raised and the command is canceled. --- doc/src/sgml/config.sgml | 31 ++++ doc/src/sgml/ref/repack.sgml | 5 +- src/backend/access/heap/heapam_handler.c | 3 +- src/backend/commands/cluster.c | 135 +++++++++++++++--- src/backend/utils/misc/guc_tables.c | 15 +- src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/commands/cluster.h | 5 +- .../injection_points/expected/repack.out | 74 +++++++++- .../injection_points/specs/repack.spec | 42 ++++++ 9 files changed, 290 insertions(+), 21 deletions(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 59a0874528a..c0529005c78 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -11239,6 +11239,37 @@ dynamic_library_path = '/usr/local/lib/postgresql:$libdir' </listitem> </varlistentry> + <varlistentry id="guc-repack-max-xclock-time" xreflabel="repack_max_xlock_time"> + <term><varname>repack_max_xlock_time</varname> (<type>integer</type>) + <indexterm> + <primary><varname>repack_max_xlock_time</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + This is the maximum amount of time to hold an exclusive lock on a + table by <command>REPACK</command> with + the <literal>CONCURRENTLY</literal> option. Typically, these commands + should not need the lock for longer time + than <command>TRUNCATE</command> does. However, additional time might + be needed if the system is too busy. (See <xref linkend="sql-repack"/> + for explanation how the <literal>CONCURRENTLY</literal> option works.) + </para> + + <para> + If you want to restrict the lock time, set this variable to the + highest acceptable value. If it appears during the processing that + additional time is needed to release the lock, the command will be + cancelled. + </para> + + <para> + The default value is 0, which means that the lock is not released + until the concurrent data changes are processed. + </para> + </listitem> + </varlistentry> + </variablelist> </sect1> diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 9c089a6b3d7..e1313f40599 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -192,7 +192,10 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU 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. + <literal>ACCESS EXCLUSIVE</literal> lock is being held. If you are + worried about this situation, set + the <link linkend="guc-repack-max-xclock-time"><varname>repack_max_xlock_time</varname></link> + configuration parameter to a value that your applications can tolerate. </para> <para> diff --git a/src/backend/access/heap/heapam_handler.c b/src/backend/access/heap/heapam_handler.c index c829c06f769..03e722347a1 100644 --- a/src/backend/access/heap/heapam_handler.c +++ b/src/backend/access/heap/heapam_handler.c @@ -986,7 +986,8 @@ heapam_relation_copy_for_cluster(Relation OldHeap, Relation NewHeap, 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); + repack_decode_concurrent_changes(decoding_ctx, end_of_wal, + NULL); end_of_wal_prev = end_of_wal; } } diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index abbbfc99036..37f69f369eb 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -17,6 +17,8 @@ */ #include "postgres.h" +#include <sys/time.h> + #include "access/amapi.h" #include "access/heapam.h" #include "access/multixact.h" @@ -89,6 +91,15 @@ typedef struct RelFileLocator repacked_rel_locator = {.relNumber = InvalidOid}; RelFileLocator repacked_rel_toast_locator = {.relNumber = InvalidOid}; +/* + * The maximum time to hold AccessExclusiveLock during the final + * processing. Note that only the execution time of + * process_concurrent_changes() is included here. The very last steps like + * swap_relation_files() shouldn't get blocked and it'd be wrong to consider + * them a reason to abort otherwise completed processing. + */ +int repack_max_xlock_time = 0; + /* * Everything we need to call ExecInsertIndexTuples(). */ @@ -132,7 +143,8 @@ static LogicalDecodingContext *setup_logical_decoding(Oid relid, static HeapTuple get_changed_tuple(char *change); static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, ScanKey key, int nkeys, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); static void apply_concurrent_insert(Relation rel, ConcurrentChange *change, HeapTuple tup, IndexInsertState *iistate, TupleTableSlot *index_slot); @@ -148,13 +160,15 @@ static HeapTuple find_target_tuple(Relation rel, ScanKey key, int nkeys, IndexInsertState *iistate, TupleTableSlot *ident_slot, IndexScanDesc *scan_p); -static void process_concurrent_changes(LogicalDecodingContext *ctx, +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, int ident_key_nentries, - IndexInsertState *iistate); + IndexInsertState *iistate, + struct timeval *must_complete); +static bool processing_time_elapsed(struct timeval *must_complete); static IndexInsertState *get_index_insert_state(Relation relation, Oid ident_index_id); static ScanKey build_identity_key(Oid ident_idx_oid, Relation rel_src, @@ -2352,7 +2366,8 @@ get_changed_tuple(char *change) */ void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal) + XLogRecPtr end_of_wal, + struct timeval *must_complete) { RepackDecodingState *dstate; ResourceOwner resowner_old; @@ -2382,6 +2397,9 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, if (record != NULL) LogicalDecodingProcessRecord(ctx, ctx->reader); + if (processing_time_elapsed(must_complete)) + break; + /* * If WAL segment boundary has been crossed, inform the decoding * system that the catalog_xmin can advance. (We can confirm more @@ -2422,7 +2440,8 @@ repack_decode_concurrent_changes(LogicalDecodingContext *ctx, */ static void apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, - ScanKey key, int nkeys, IndexInsertState *iistate) + ScanKey key, int nkeys, IndexInsertState *iistate, + struct timeval *must_complete) { TupleTableSlot *index_slot, *ident_slot; @@ -2452,6 +2471,9 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, CHECK_FOR_INTERRUPTS(); + Assert(dstate->nchanges > 0); + dstate->nchanges--; + /* Get the change from the single-column tuple. */ tup_change = ExecFetchSlotHeapTuple(dstate->tsslot, false, &shouldFree); heap_deform_tuple(tup_change, dstate->tupdesc_change, values, isnull); @@ -2552,10 +2574,22 @@ apply_concurrent_changes(RepackDecodingState *dstate, Relation rel, /* TTSOpsMinimalTuple has .get_heap_tuple==NULL. */ Assert(shouldFree); pfree(tup_change); + + /* + * If there is a limit on the time of completion, check it now. + * However, make sure the loop does not break if tup_old was set in + * the previous iteration. In such a case we could not resume the + * processing in the next call. + */ + if (must_complete && tup_old == NULL && + processing_time_elapsed(must_complete)) + /* The next call will process the remaining changes. */ + break; } - tuplestore_clear(dstate->tstore); - dstate->nchanges = 0; + /* If we could not apply all the changes, the next call will do. */ + if (dstate->nchanges == 0) + tuplestore_clear(dstate->tstore); /* Cleanup. */ ExecDropSingleTupleTableSlot(index_slot); @@ -2737,11 +2771,15 @@ find_target_tuple(Relation rel, ScanKey key, int nkeys, HeapTuple tup_key, * Decode and apply concurrent changes. * * Pass rel_src iff its reltoastrelid is needed. + * + * Returns true if must_complete is NULL or if managed to complete by the time + * *must_complete indicates. */ -static void +static bool process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, Relation rel_dst, Relation rel_src, ScanKey ident_key, - int ident_key_nentries, IndexInsertState *iistate) + int ident_key_nentries, IndexInsertState *iistate, + struct timeval *must_complete) { RepackDecodingState *dstate; @@ -2750,10 +2788,19 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, dstate = (RepackDecodingState *) ctx->output_writer_private; - repack_decode_concurrent_changes(ctx, end_of_wal); + repack_decode_concurrent_changes(ctx, end_of_wal, must_complete); + if (processing_time_elapsed(must_complete)) + /* Caller is responsible for applying the changes. */ + return false; + + /* + * *must_complete not reached, so there are really no changes. (It's + * possible to see no changes just because not enough time was left for + * the decoding.) + */ if (dstate->nchanges == 0) - return; + return true; PG_TRY(); { @@ -2765,7 +2812,7 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = rel_src->rd_rel->reltoastrelid; apply_concurrent_changes(dstate, rel_dst, ident_key, - ident_key_nentries, iistate); + ident_key_nentries, iistate, must_complete); } PG_FINALLY(); { @@ -2773,6 +2820,28 @@ process_concurrent_changes(LogicalDecodingContext *ctx, XLogRecPtr end_of_wal, rel_dst->rd_toastoid = InvalidOid; } PG_END_TRY(); + + /* + * apply_concurrent_changes() does check the processing time, so if some + * changes are left, we ran out of time. + */ + return dstate->nchanges == 0; +} + +/* + * Check if the current time is beyond *must_complete. + */ +static bool +processing_time_elapsed(struct timeval *must_complete) +{ + struct timeval now; + + if (must_complete == NULL) + return false; + + gettimeofday(&now, NULL); + + return timercmp(&now, must_complete, >); } static IndexInsertState * @@ -2934,6 +3003,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, Relation *ind_refs, *ind_refs_p; int nind; + struct timeval t_end; + struct timeval *t_end_ptr = NULL; /* Like in cluster_rel(). */ lockmode_old = ShareUpdateExclusiveLock; @@ -3029,7 +3100,8 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, */ process_concurrent_changes(ctx, end_of_wal, NewHeap, swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + ident_key, ident_key_nentries, iistate, + NULL); /* * Acquire AccessExclusiveLock on the table, its TOAST relation (if there @@ -3125,9 +3197,40 @@ rebuild_relation_finish_concurrent(Relation NewHeap, Relation OldHeap, end_of_wal = GetFlushRecPtr(NULL); /* Apply the concurrent changes again. */ - process_concurrent_changes(ctx, end_of_wal, NewHeap, - swap_toast_by_content ? OldHeap : NULL, - ident_key, ident_key_nentries, iistate); + + /* + * This time we have the exclusive lock on the table, so make sure that + * repack_max_xlock_time is not exceeded. + */ + if (repack_max_xlock_time > 0) + { + int64 usec; + struct timeval t_start; + + gettimeofday(&t_start, NULL); + /* Add the whole seconds. */ + t_end.tv_sec = t_start.tv_sec + repack_max_xlock_time / 1000; + /* Add the rest, expressed in microseconds. */ + usec = t_start.tv_usec + 1000 * (repack_max_xlock_time % 1000); + /* The number of microseconds could have overflown. */ + t_end.tv_sec += usec / USECS_PER_SEC; + t_end.tv_usec = usec % USECS_PER_SEC; + t_end_ptr = &t_end; + } + + /* + * During testing, stop here to simulate excessive processing time. + */ + INJECTION_POINT("repack-concurrently-after-lock", NULL); + + if (!process_concurrent_changes(ctx, end_of_wal, NewHeap, + swap_toast_by_content ? OldHeap : NULL, + ident_key, ident_key_nentries, iistate, + t_end_ptr)) + ereport(ERROR, + (errmsg("could not process concurrent data changes in time"), + errhint("Please consider adjusting \"repack_max_xlock_time\"."))); + /* Remember info about rel before closing OldHeap */ relpersistence = OldHeap->rd_rel->relpersistence; diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 511dc32d519..6a373a8b65a 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -42,8 +42,9 @@ #include "catalog/namespace.h" #include "catalog/storage.h" #include "commands/async.h" -#include "commands/extension.h" +#include "commands/cluster.h" #include "commands/event_trigger.h" +#include "commands/extension.h" #include "commands/tablespace.h" #include "commands/trigger.h" #include "commands/user.h" @@ -2839,6 +2840,18 @@ struct config_int ConfigureNamesInt[] = 1600000000, 0, 2100000000, NULL, NULL, NULL }, + { + {"repack_max_xlock_time", PGC_USERSET, LOCK_MANAGEMENT, + gettext_noop("Maximum time for REPACK CONCURRENTLY to keep table locked."), + gettext_noop("The table is locked in exclusive mode during the final stage of processing. " + "If the lock time exceeds this value, error is raised and the lock is " + "released. Set to zero if you don't care how long the lock can be held."), + GUC_UNIT_MS + }, + &repack_max_xlock_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, /* * See also CheckRequiredParameterValues() if this parameter changes diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 341f88adc87..42d32a2c198 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -765,6 +765,7 @@ autovacuum_worker_slots = 16 # autovacuum worker slots to allocate #lock_timeout = 0 # in milliseconds, 0 is disabled #idle_in_transaction_session_timeout = 0 # in milliseconds, 0 is disabled #idle_session_timeout = 0 # in milliseconds, 0 is disabled +#repack_max_xlock_time = 0 #bytea_output = 'hex' # hex, escape #xmlbinary = 'base64' #xmloption = 'content' diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 0a7e72bc74a..4914f217267 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -59,6 +59,8 @@ typedef enum ClusterCommand extern RelFileLocator repacked_rel_locator; extern RelFileLocator repacked_rel_toast_locator; +extern PGDLLIMPORT int repack_max_xlock_time; + typedef enum { CHANGE_INSERT, @@ -134,7 +136,8 @@ extern void check_index_is_clusterable(Relation OldHeap, Oid indexOid, LOCKMODE lockmode); extern void mark_index_clustered(Relation rel, Oid indexOid, bool is_internal); extern void repack_decode_concurrent_changes(LogicalDecodingContext *ctx, - XLogRecPtr end_of_wal); + XLogRecPtr end_of_wal, + struct timeval *must_complete); extern Oid make_new_heap(Oid OIDOldHeap, Oid NewTableSpace, Oid NewAccessMethod, char relpersistence, LOCKMODE lockmode); extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, diff --git a/src/test/modules/injection_points/expected/repack.out b/src/test/modules/injection_points/expected/repack.out index f919087ca5b..02967ed9d48 100644 --- a/src/test/modules/injection_points/expected/repack.out +++ b/src/test/modules/injection_points/expected/repack.out @@ -1,4 +1,4 @@ -Parsed test spec with 2 sessions +Parsed test spec with 4 sessions starting permutation: wait_before_lock change_existing change_new change_subxact1 change_subxact2 check2 wakeup_before_lock check1 injection_points_attach @@ -111,3 +111,75 @@ injection_points_detach (1 row) +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + + +starting permutation: wait_after_lock after_lock_delay wakeup_after_lock +injection_points_attach +----------------------- + +(1 row) + +step wait_after_lock: + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; + <waiting ...> +step after_lock_delay: + SELECT pg_sleep(1.5); + +pg_sleep +-------- + +(1 row) + +step wakeup_after_lock: + SELECT injection_points_wakeup('repack-concurrently-after-lock'); + +injection_points_wakeup +----------------------- + +(1 row) + +step wait_after_lock: <... completed> +ERROR: could not process concurrent data changes in time +injection_points_detach +----------------------- + +(1 row) + +injection_points_detach +----------------------- + +(1 row) + diff --git a/src/test/modules/injection_points/specs/repack.spec b/src/test/modules/injection_points/specs/repack.spec index a17064462ce..d0fa38dd8cd 100644 --- a/src/test/modules/injection_points/specs/repack.spec +++ b/src/test/modules/injection_points/specs/repack.spec @@ -130,6 +130,34 @@ step wakeup_before_lock SELECT injection_points_wakeup('repack-concurrently-before-lock'); } +session s3 +setup +{ + SELECT injection_points_set_local(); + SELECT injection_points_attach('repack-concurrently-after-lock', 'wait'); + SET repack_max_xlock_time TO '1s'; +} +# Perform the initial load, lock the table in exclusive mode and wait. s4 will +# cancel the waiting. +step wait_after_lock +{ + REPACK CONCURRENTLY repack_test USING INDEX repack_test_pkey; +} +teardown +{ + SELECT injection_points_detach('repack-concurrently-after-lock'); +} + +session s4 +step wakeup_after_lock +{ + SELECT injection_points_wakeup('repack-concurrently-after-lock'); +} +step after_lock_delay +{ + SELECT pg_sleep(1.5); +} + # Test if data changes introduced while one session is performing REPACK # CONCURRENTLY find their way into the table. permutation @@ -141,3 +169,17 @@ permutation check2 wakeup_before_lock check1 + +# Test the repack_max_xlock_time configuration variable. +# +# First, cancel waiting on the injection point immediately. That way, REPACK +# should complete. +permutation + wait_after_lock + wakeup_after_lock +# Second, cancel the waiting with a delay that violates +# repack_max_xlock_time. +permutation + wait_after_lock + after_lock_delay + wakeup_after_lock -- 2.47.1 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v15-0007-Enable-logical-decoding-transiently-only-for-REPACK-.patch ^ permalink raw reply [nested|flat] 13+ messages in thread
end of thread, other threads:[~2025-06-30 17:41 UTC | newest] Thread overview: 13+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2025-02-26 08:17 [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-02-26 08:17 [PATCH v09 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-02-26 08:17 [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-02-26 08:17 [PATCH v09 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-03-25 12:42 [PATCH v10 7/9] Introduce repack_max_xlock_time configuration variable. Álvaro Herrera <alvherre@alvh.no-ip.org> 2025-03-31 13:47 [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-03-31 13:47 [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-04-01 11:48 [PATCH 7/9] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-04-11 09:13 [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-06-09 10:00 [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-06-09 10:00 [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-06-30 17:41 [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at> 2025-06-30 17:41 [PATCH 6/7] Introduce repack_max_xlock_time configuration variable. Antonin Houska <ah@cybertec.at>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox