From 7ad8b35fb1ebc54445048ccee5db9f0f03b120de Mon Sep 17 00:00:00 2001 From: Sami Imseih Date: Fri, 18 Sep 2026 21:12:32 +0000 Subject: [PATCH v4 2/2] Add REPACK progress phases for logical decoding setup REPACK (CONCURRENTLY) reports everything it does before the heap scan as "initializing". This includes enabling logical decoding and then initializing it, which reads WAL until the logical decoding snapshot becomes consistent and therefore has to wait for the transactions that have a transaction ID assigned to end. That wait is unbounded, and it is hard to tell apart from the rest of the setup when it is all reported as one phase. Add a phase for enabling logical decoding, which only has work to do if logical decoding is not already enabled, and a phase for initializing logical decoding, which also covers obtaining the exported snapshot, and move the description of the waits into them. The decoding worker stores the phase to report in the existing shared memory and signals the existing condition variable when it changes, so the backend running REPACK reports it while waiting for the worker. Note that this changes the definition of pg_stat_progress_repack and therefore requires a catalog version bump, which is not included here. Author: Sami Imseih Reviewed-by: Masahiko Sawada Reviewed-by: Manuel Reyes Bravo Discussion: https://postgr.es/m/CAN12%2BYJOeOyPYB5JS28VE_%2Brzg%3DRbzQXfj58TGfMYwv-TX9mJA%40mail.gmail.com --- doc/src/sgml/monitoring.sgml | 38 ++++++++++++++++++-------- src/backend/catalog/system_views.sql | 2 ++ src/backend/commands/repack.c | 12 ++++++++ src/backend/commands/repack_worker.c | 27 ++++++++++++++++-- src/include/commands/progress.h | 2 ++ src/include/commands/repack_internal.h | 7 +++++ src/test/regress/expected/rules.out | 2 ++ 7 files changed, 75 insertions(+), 15 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 0d038de1a23..34ca1301790 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -7600,20 +7600,34 @@ FROM pg_stat_get_backend_idset() AS backendid; initializing - The command is preparing to begin scanning the heap. This phase is - expected to be very brief, except for - REPACK CONCURRENTLY, where it also covers enabling - logical decoding and then initializing it. Enabling has nothing to do - if is already - logical; otherwise it waits for all processes to - acknowledge that they started writing the additional WAL information - that logical decoding requires. Initializing then waits for the - transactions that have been assigned a transaction ID to end, including - ones that started while this wait was in progress, so this phase can - last longer than the longest transaction running when the command - started. This includes any other + The command is performing initial setup. For + REPACK CONCURRENTLY, this includes starting the + logical decoding worker. This phase is expected to be very brief. + + + + enabling logical decoding + + REPACK CONCURRENTLY is enabling logical decoding. + This has nothing to do if is + already logical; otherwise it waits for all processes + to acknowledge that they started writing the additional WAL information + that logical decoding requires. + This phase is skipped when not in concurrent mode. + + + + initializing logical decoding + + REPACK CONCURRENTLY is initializing logical + decoding and obtaining the initial snapshot used to copy the table. + This waits for the transactions that have been assigned a transaction ID + to end, including ones that started while this wait was in progress, so + this phase can last longer than the longest transaction running when the + command started. This includes any other REPACK CONCURRENTLY, which holds a transaction ID until it finishes. + This phase is skipped when not in concurrent mode. diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index ad340887f54..0bd29275644 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1376,6 +1376,8 @@ CREATE VIEW pg_stat_progress_repack AS WHEN 6 THEN 'swapping relation files' WHEN 7 THEN 'rebuilding index' WHEN 8 THEN 'performing final cleanup' + WHEN 9 THEN 'enabling logical decoding' + WHEN 10 THEN 'initializing logical decoding' END AS phase, CAST(S.param3 AS oid) AS repack_index_relid, S.param4 AS heap_tuples_scanned, diff --git a/src/backend/commands/repack.c b/src/backend/commands/repack.c index 759be53d6b8..6de84766814 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -3745,6 +3745,7 @@ start_repack_decoding_worker(Oid relid) DecodingWorkerShared *shared; shm_mq *mq; BackgroundWorker bgw; + int reported_phase = 0; decoding_worker = palloc0_object(DecodingWorker); @@ -3755,6 +3756,7 @@ start_repack_decoding_worker(Oid relid) shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg); shared->initialized = false; + shared->setup_phase = 0; shared->lsn_upto = InvalidXLogRecPtr; shared->done = false; SharedFileSetInit(&shared->sfs, decoding_worker->seg); @@ -3820,16 +3822,26 @@ start_repack_decoding_worker(Oid relid) * for any reason, otherwise the worker might end up in a deadlock, * waiting for the caller's transaction to end. Therefore wait here until * the worker indicates that it has the logical decoding initialized. + * While waiting, report the phase the worker asks us to report. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) { bool initialized; + int setup_phase; SpinLockAcquire(&shared->mutex); initialized = shared->initialized; + setup_phase = shared->setup_phase; SpinLockRelease(&shared->mutex); + /* Report each phase once. */ + if (setup_phase != reported_phase) + { + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, setup_phase); + reported_phase = setup_phase; + } + if (initialized) break; diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index 690863c6411..045333c06bf 100644 --- a/src/backend/commands/repack_worker.c +++ b/src/backend/commands/repack_worker.c @@ -19,6 +19,7 @@ #include "access/xlog_internal.h" #include "access/xlogutils.h" #include "access/xlogwait.h" +#include "commands/progress.h" #include "commands/repack.h" #include "commands/repack_internal.h" #include "libpq/libpq.h" @@ -33,7 +34,10 @@ #define PGREPACK_PLUGIN "pgrepack" static void RepackWorkerShutdown(int code, Datum arg); -static LogicalDecodingContext *repack_setup_logical_decoding(Oid relid); +static void set_repack_worker_setup_phase(DecodingWorkerShared *shared, + int setup_phase); +static LogicalDecodingContext *repack_setup_logical_decoding(Oid relid, + DecodingWorkerShared *shared); static void repack_cleanup_logical_decoding(LogicalDecodingContext *ctx); static void export_initial_snapshot(Snapshot snapshot, DecodingWorkerShared *shared); @@ -139,7 +143,7 @@ RepackWorkerMain(Datum main_arg) /* * Prepare to capture the concurrent data changes ourselves. */ - decoding_ctx = repack_setup_logical_decoding(shared->relid); + decoding_ctx = repack_setup_logical_decoding(shared->relid, shared); /* Announce that we're ready. */ SpinLockAcquire(&shared->mutex); @@ -212,6 +216,19 @@ AmRepackWorker(void) return am_repack_worker; } +/* + * Tell the backend running REPACK which progress phase to report while we + * initialize the decoding. + */ +static void +set_repack_worker_setup_phase(DecodingWorkerShared *shared, int setup_phase) +{ + SpinLockAcquire(&shared->mutex); + shared->setup_phase = setup_phase; + SpinLockRelease(&shared->mutex); + ConditionVariableSignal(&shared->cv); +} + /* * This function is much like pg_create_logical_replication_slot() except that * the new slot is neither released (if anyone else could read changes from @@ -220,7 +237,7 @@ AmRepackWorker(void) * crash by restarting all the work from scratch). */ static LogicalDecodingContext * -repack_setup_logical_decoding(Oid relid) +repack_setup_logical_decoding(Oid relid, DecodingWorkerShared *shared) { Relation rel; Oid toastrelid; @@ -249,6 +266,8 @@ repack_setup_logical_decoding(Oid relid) snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid); ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true, false, false); + set_repack_worker_setup_phase(shared, + PROGRESS_REPACK_PHASE_ENABLE_LOGICAL_DECODING); EnsureLogicalDecodingEnabled(); /* @@ -294,6 +313,8 @@ repack_setup_logical_decoding(Oid relid) Assert(!ctx->fast_forward); /* Find our decoding starting point. */ + set_repack_worker_setup_phase(shared, + PROGRESS_REPACK_PHASE_INIT_LOGICAL_DECODING); DecodingContextFindStartpoint(ctx); /* From this point on, we need non-blocking WAL reads */ diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index 2a12920c75f..711fadf81b3 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -104,6 +104,8 @@ #define PROGRESS_REPACK_PHASE_SWAP_REL_FILES 6 #define PROGRESS_REPACK_PHASE_REBUILD_INDEX 7 #define PROGRESS_REPACK_PHASE_FINAL_CLEANUP 8 +#define PROGRESS_REPACK_PHASE_ENABLE_LOGICAL_DECODING 9 +#define PROGRESS_REPACK_PHASE_INIT_LOGICAL_DECODING 10 /* Progress parameters for CREATE INDEX */ /* 3, 4 and 5 reserved for "waitfor" metrics */ diff --git a/src/include/commands/repack_internal.h b/src/include/commands/repack_internal.h index ec6e31d77f2..c14107b663f 100644 --- a/src/include/commands/repack_internal.h +++ b/src/include/commands/repack_internal.h @@ -67,6 +67,13 @@ typedef struct DecodingWorkerShared /* Is the decoding initialized? */ bool initialized; + /* + * The progress phase (a PROGRESS_REPACK_PHASE_* value) that the backend + * should report while the worker initializes the decoding. Zero means + * the initial phase, which the backend reports on its own. + */ + int setup_phase; + /* * Once the worker has reached this LSN, it should close the current * output file and either create a new one or exit, according to the field diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 4a8cc759d7b..b6e60dcd1bf 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2167,6 +2167,8 @@ pg_stat_progress_repack| SELECT s.pid, WHEN 6 THEN 'swapping relation files'::text WHEN 7 THEN 'rebuilding index'::text WHEN 8 THEN 'performing final cleanup'::text + WHEN 9 THEN 'enabling logical decoding'::text + WHEN 10 THEN 'initializing logical decoding'::text ELSE NULL::text END AS phase, (s.param3)::oid AS repack_index_relid, -- 2.50.1