From a08efb4bb2b01bbaf1bef7d5420ee71c902299d1 Mon Sep 17 00:00:00 2001 From: Sami Imseih Date: Fri, 18 Sep 2026 21:12:32 +0000 Subject: [PATCH v1 1/1] Add REPACK progress phases for logical decoding setup REPACK (CONCURRENTLY) reported all work before the heap scan as "initializing". This includes enabling logical decoding, waiting for old transactions, and building the initial snapshot. The documentation describes the whole phase as expected to be brief, but the first two operations can wait indefinitely. Add separate phases for enabling logical decoding, waiting for old transactions, and building the initial snapshot. Store the decoding worker's setup state in the existing shared memory and signal the existing condition variable when it changes. --- doc/src/sgml/monitoring.sgml | 36 +++++++++++++++++- src/backend/catalog/system_views.sql | 3 ++ src/backend/commands/repack.c | 51 ++++++++++++++++++++------ src/backend/commands/repack_worker.c | 40 +++++++++++++++----- src/include/catalog/catversion.h | 2 +- src/include/commands/progress.h | 3 ++ src/include/commands/repack_internal.h | 16 +++++++- src/test/regress/expected/rules.out | 3 ++ src/tools/pgindent/typedefs.list | 1 + 9 files changed, 129 insertions(+), 26 deletions(-) diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 62dadf3e86c..c49cda43d39 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -7600,8 +7600,39 @@ 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. + 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. + If logical decoding is not already enabled, this phase waits for all + backends to process a request to start writing the additional WAL + information required for logical decoding. + This phase is skipped when not in concurrent mode. + + + + waiting for old transactions + + REPACK (CONCURRENTLY) is waiting for old + transactions with assigned transaction IDs to end so that the initial + logical decoding snapshot can be built. This can include prepared + transactions and other REPACK (CONCURRENTLY) + operations that have already acquired a transaction ID. + This phase is skipped when not in concurrent mode. + + + + building initial snapshot + + REPACK (CONCURRENTLY) is obtaining the initial + snapshot from its logical decoding worker and preparing to copy the + table. + This phase is skipped when not in concurrent mode. @@ -7634,6 +7665,7 @@ FROM pg_stat_get_backend_idset() AS backendid; REPACK CONCURRENTLY is currently processing the DML commands that other transactions executed during any of the preceding phases. + 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..454ff32d67c 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1376,6 +1376,9 @@ 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 'waiting for old transactions' + WHEN 11 THEN 'building initial snapshot' 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..2dccedea22f 100644 --- a/src/backend/commands/repack.c +++ b/src/backend/commands/repack.c @@ -519,6 +519,10 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, }; const int64 progress_values[] = {cmd, indexOid}; + pgstat_progress_start_command(PROGRESS_COMMAND_REPACK, tableOid); + /* Report the ordering index even when using a sequential scan and sort. */ + pgstat_progress_update_multi_param(2, progress_index, progress_values); + /* Determine the lock mode to use. */ lmode = RepackLockLevel(concurrent); @@ -540,10 +544,6 @@ cluster_rel(RepackCommand cmd, Relation OldHeap, Oid indexOid, /* Check for user-requested abort. */ CHECK_FOR_INTERRUPTS(); - pgstat_progress_start_command(PROGRESS_COMMAND_REPACK, tableOid); - /* Report the ordering index even when using a sequential scan and sort. */ - pgstat_progress_update_multi_param(2, progress_index, progress_values); - /* * Switch to the table owner's userid, so that any index functions are run * as that user. Also lock down security-restricted operations and @@ -3745,6 +3745,8 @@ start_repack_decoding_worker(Oid relid) DecodingWorkerShared *shared; shm_mq *mq; BackgroundWorker bgw; + RepackWorkerSetupState last_reported_state = + REPACK_WORKER_SETUP_STARTING; decoding_worker = palloc0_object(DecodingWorker); @@ -3754,7 +3756,7 @@ start_repack_decoding_worker(Oid relid) decoding_worker->seg = dsm_create(size, 0); shared = (DecodingWorkerShared *) dsm_segment_address(decoding_worker->seg); - shared->initialized = false; + shared->setup_state = REPACK_WORKER_SETUP_STARTING; shared->lsn_upto = InvalidXLogRecPtr; shared->done = false; SharedFileSetInit(&shared->sfs, decoding_worker->seg); @@ -3820,17 +3822,45 @@ 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 setup state as the worker advances. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) { - bool initialized; + RepackWorkerSetupState setup_state; SpinLockAcquire(&shared->mutex); - initialized = shared->initialized; + setup_state = shared->setup_state; SpinLockRelease(&shared->mutex); - if (initialized) + if (setup_state > last_reported_state) + { + int progress_phase; + + switch (setup_state) + { + case REPACK_WORKER_SETUP_ENABLING_LOGICAL_DECODING: + progress_phase = + PROGRESS_REPACK_PHASE_ENABLE_LOGICAL_DECODING; + break; + case REPACK_WORKER_SETUP_WAITING_FOR_OLD_TRANSACTIONS: + progress_phase = PROGRESS_REPACK_PHASE_WAIT_XACTS; + break; + case REPACK_WORKER_SETUP_BUILDING_INITIAL_SNAPSHOT: + progress_phase = + PROGRESS_REPACK_PHASE_BUILD_INITIAL_SNAPSHOT; + break; + default: + elog(ERROR, "unexpected REPACK worker setup state: %d", + setup_state); + } + + pgstat_progress_update_param(PROGRESS_REPACK_PHASE, + progress_phase); + last_reported_state = setup_state; + } + + if (setup_state == REPACK_WORKER_SETUP_BUILDING_INITIAL_SNAPSHOT) break; ConditionVariableSleep(&shared->cv, WAIT_EVENT_REPACK_WORKER_EXPORT); @@ -3998,9 +4028,8 @@ get_initial_snapshot(DecodingWorker *worker) shared = (DecodingWorkerShared *) dsm_segment_address(worker->seg); /* - * The worker needs to initialize the logical decoding, which usually - * takes some time. Therefore it makes sense to prepare for the sleep - * first. + * Building and serializing the initial snapshot can take some time. + * Therefore it makes sense to prepare for the sleep first. */ ConditionVariablePrepareToSleep(&shared->cv); for (;;) diff --git a/src/backend/commands/repack_worker.c b/src/backend/commands/repack_worker.c index 690863c6411..64ea01cb08e 100644 --- a/src/backend/commands/repack_worker.c +++ b/src/backend/commands/repack_worker.c @@ -33,7 +33,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_state(DecodingWorkerShared *shared, + RepackWorkerSetupState setup_state); +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,19 +142,15 @@ RepackWorkerMain(Datum main_arg) /* * Prepare to capture the concurrent data changes ourselves. */ - decoding_ctx = repack_setup_logical_decoding(shared->relid); - - /* Announce that we're ready. */ - SpinLockAcquire(&shared->mutex); - shared->initialized = true; - SpinLockRelease(&shared->mutex); - ConditionVariableSignal(&shared->cv); + decoding_ctx = repack_setup_logical_decoding(shared->relid, shared); /* There doesn't seem to a nice API to set these */ XactIsoLevel = XACT_REPEATABLE_READ; XactReadOnly = true; /* Build the initial snapshot and export it. */ + set_repack_worker_setup_state(shared, + REPACK_WORKER_SETUP_BUILDING_INITIAL_SNAPSHOT); snapshot = SnapBuildInitialSnapshot(decoding_ctx->snapshot_builder); export_initial_snapshot(snapshot, shared); @@ -212,6 +211,20 @@ AmRepackWorker(void) return am_repack_worker; } +/* + * Report the current setup state to the backend running REPACK. + */ +static void +set_repack_worker_setup_state(DecodingWorkerShared *shared, + RepackWorkerSetupState setup_state) +{ + SpinLockAcquire(&shared->mutex); + Assert(setup_state > shared->setup_state); + shared->setup_state = setup_state; + 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 +233,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 +260,8 @@ repack_setup_logical_decoding(Oid relid) * to make the slot name unique. */ snprintf(slotname, NAMEDATALEN, "pg_repack_%d", MyProcPid); + set_repack_worker_setup_state(shared, + REPACK_WORKER_SETUP_ENABLING_LOGICAL_DECODING); ReplicationSlotCreate(slotname, true, RS_TEMPORARY, false, true, false, false); EnsureLogicalDecodingEnabled(); @@ -293,7 +308,12 @@ repack_setup_logical_decoding(Oid relid) /* We don't have control on fast_forward, but verify it's sane */ Assert(!ctx->fast_forward); - /* Find our decoding starting point. */ + /* + * Find our decoding starting point, waiting for old transactions as + * needed. + */ + set_repack_worker_setup_state(shared, + REPACK_WORKER_SETUP_WAITING_FOR_OLD_TRANSACTIONS); DecodingContextFindStartpoint(ctx); /* From this point on, we need non-blocking WAL reads */ diff --git a/src/include/catalog/catversion.h b/src/include/catalog/catversion.h index 6f3e526de96..fa5ccdf79eb 100644 --- a/src/include/catalog/catversion.h +++ b/src/include/catalog/catversion.h @@ -57,6 +57,6 @@ */ /* yyyymmddN */ -#define CATALOG_VERSION_NO 202609152 +#define CATALOG_VERSION_NO 202609180 #endif diff --git a/src/include/commands/progress.h b/src/include/commands/progress.h index 2a12920c75f..7d48444aac3 100644 --- a/src/include/commands/progress.h +++ b/src/include/commands/progress.h @@ -104,6 +104,9 @@ #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_WAIT_XACTS 10 +#define PROGRESS_REPACK_PHASE_BUILD_INITIAL_SNAPSHOT 11 /* 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..020f3ce59e2 100644 --- a/src/include/commands/repack_internal.h +++ b/src/include/commands/repack_internal.h @@ -58,14 +58,26 @@ typedef struct RepackDecodingState BufFile *file; } RepackDecodingState; +/* + * Setup states reported by the decoding worker to the backend running REPACK, + * in transition order. + */ +typedef enum RepackWorkerSetupState +{ + REPACK_WORKER_SETUP_STARTING, + REPACK_WORKER_SETUP_ENABLING_LOGICAL_DECODING, + REPACK_WORKER_SETUP_WAITING_FOR_OLD_TRANSACTIONS, + REPACK_WORKER_SETUP_BUILDING_INITIAL_SNAPSHOT +} RepackWorkerSetupState; + /* * Shared memory used for communication between the backend running REPACK and * the worker that performs logical decoding of data changes. */ typedef struct DecodingWorkerShared { - /* Is the decoding initialized? */ - bool initialized; + /* Current setup state. */ + RepackWorkerSetupState setup_state; /* * Once the worker has reached this LSN, it should close the current diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 4a8cc759d7b..9eb26d34239 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -2167,6 +2167,9 @@ 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 'waiting for old transactions'::text + WHEN 11 THEN 'building initial snapshot'::text ELSE NULL::text END AS phase, (s.param3)::oid AS repack_index_relid, diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 5d432074c2c..cdaef34bf46 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2644,6 +2644,7 @@ ReorderTuple RepackCommand RepackDecodingState RepackStmt +RepackWorkerSetupState ReparameterizeForeignPathByChild_function ReplOriginId ReplOriginXactState -- 2.50.1