From 2b60e10220f15e9ad47d21a1f76332ddab52d431 Mon Sep 17 00:00:00 2001 From: Sami Imseih Date: Fri, 18 Sep 2026 21:12:32 +0000 Subject: [PATCH v3 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 already had a transaction ID assigned to end. That wait can last as long as the longest of those transactions, which 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. --- doc/src/sgml/monitoring.sgml | 35 ++++++++++++++++++-------- src/backend/catalog/system_views.sql | 2 ++ src/backend/commands/repack.c | 7 ++++++ src/backend/commands/repack_worker.c | 29 ++++++++++++++++++--- src/include/commands/progress.h | 2 ++ src/include/commands/repack_internal.h | 7 ++++++ src/test/regress/expected/rules.out | 2 ++ 7 files changed, 71 insertions(+), 13 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index b7e3741a190..63a65a62d66 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -7600,16 +7600,31 @@ 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 backends to - acknowledge that they started writing the additional WAL information - that logical decoding requires. Initializing then waits for the - transactions that had already been assigned a transaction ID to end, so - this phase can last as long as the longest of those transactions. + 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 backends + 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 to copy the table with. + This waits for the transactions that had already been assigned a + transaction ID to end, so this phase can last as long as the longest of + those transactions. + 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..80213dfdb50 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -3755,6 +3755,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 +3821,22 @@ 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 the phase the worker has reached. */ + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, setup_phase); + if (initialized) break; diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index 690863c6411..79d622bf40d 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,21 @@ AmRepackWorker(void) return am_repack_worker; } +/* + * Tell the backend running REPACK which progress phase to report while we + * initialize the decoding. The phases must only advance, so that the backend + * never reports one twice. + */ +static void +set_repack_worker_setup_phase(DecodingWorkerShared *shared, int setup_phase) +{ + SpinLockAcquire(&shared->mutex); + Assert(setup_phase > shared->setup_phase); + 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 +239,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; @@ -247,6 +266,8 @@ repack_setup_logical_decoding(Oid relid) * to make the slot name unique. */ snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid); + set_repack_worker_setup_phase(shared, + PROGRESS_REPACK_PHASE_ENABLE_LOGICAL_DECODING); ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true, false, false); EnsureLogicalDecodingEnabled(); @@ -294,6 +315,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