agora inbox for [email protected]help / color / mirror / Atom feed
[PATCH v3 04/17] meson: prereq: output and depencency tracking work. 9+ messages / 3 participants [nested] [flat]
* [PATCH v3 04/17] meson: prereq: output and depencency tracking work. @ 2021-03-08 21:47 Andres Freund <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Andres Freund @ 2021-03-08 21:47 UTC (permalink / raw) --- src/backend/utils/misc/Makefile | 5 ++++- src/backend/utils/misc/guc.c | 2 +- src/bin/initdb/initdb.c | 5 +++-- src/bin/psql/Makefile | 4 ++-- src/bin/psql/create_help.pl | 16 ++++++++++++---- src/tools/msvc/MSBuildProject.pm | 9 +++++++-- src/tools/msvc/Mkvcbuild.pm | 3 +++ src/tools/msvc/Solution.pm | 2 +- src/tools/msvc/pgflex.pl | 4 ++-- 9 files changed, 35 insertions(+), 15 deletions(-) diff --git a/src/backend/utils/misc/Makefile b/src/backend/utils/misc/Makefile index 1d5327cf644..14861fd96b2 100644 --- a/src/backend/utils/misc/Makefile +++ b/src/backend/utils/misc/Makefile @@ -37,8 +37,11 @@ endif include $(top_srcdir)/src/backend/common.mk +guc-file.c.h: guc-file.l + flex -o $@ $< + # guc-file is compiled as part of guc -guc.o: guc-file.c +guc.o: guc-file.c.h # Note: guc-file.c is not deleted by 'make clean', # since we want to ship it in distribution tarballs. diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index d2ce4a84506..3786ae11a08 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -12559,4 +12559,4 @@ check_default_with_oids(bool *newval, void **extra, GucSource source) return true; } -#include "guc-file.c" +#include "guc-file.c.h" diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index 1ed4808d53f..9067a06e58a 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -1368,8 +1368,9 @@ bootstrap_template1(void) if (strcmp(headerline, *bki_lines) != 0) { - pg_log_error("input file \"%s\" does not belong to PostgreSQL %s", - bki_file, PG_VERSION); + pg_log_error("input file \"%s\" does not belong to PostgreSQL %s (expect %s, is %s)", + bki_file, PG_VERSION, + headerline, *bki_lines); fprintf(stderr, _("Check your installation or specify the correct path " "using the option -L.\n")); diff --git a/src/bin/psql/Makefile b/src/bin/psql/Makefile index d00881163c0..3851da1c8ef 100644 --- a/src/bin/psql/Makefile +++ b/src/bin/psql/Makefile @@ -56,7 +56,7 @@ sql_help.c: sql_help.h touch $@ sql_help.h: create_help.pl $(wildcard $(REFDOCDIR)/*.sgml) - $(PERL) $< $(REFDOCDIR) $* + $(PERL) $< $(REFDOCDIR) . $* psqlscanslash.c: FLEXFLAGS = -Cfe -p -p psqlscanslash.c: FLEX_NO_BACKUP=yes @@ -81,7 +81,7 @@ clean distclean: # files removed here are supposed to be in the distribution tarball, # so do not clean them in the clean/distclean rules maintainer-clean: distclean - rm -f sql_help.h sql_help.c psqlscanslash.c + rm -f sql_help.h sql_help.c sql_help.dep psqlscanslash.c check: $(prove_check) diff --git a/src/bin/psql/create_help.pl b/src/bin/psql/create_help.pl index 83324239740..40eb6ac2d3f 100644 --- a/src/bin/psql/create_help.pl +++ b/src/bin/psql/create_help.pl @@ -23,9 +23,12 @@ use strict; use warnings; my $docdir = $ARGV[0] or die "$0: missing required argument: docdir\n"; -my $hfile = $ARGV[1] . '.h' +my $outdir = $ARGV[1] or die "$0: missing required argument: outdir\n"; + +my $hfile = $ARGV[2] . '.h' or die "$0: missing required argument: output file\n"; -my $cfile = $ARGV[1] . '.c'; +my $cfile = $ARGV[2] . '.c'; +my $depfile = $ARGV[2] . '.dep'; my $hfilebasename; if ($hfile =~ m!.*/([^/]+)$!) @@ -43,10 +46,12 @@ $define =~ s/\W/_/g; opendir(DIR, $docdir) or die "$0: could not open documentation source dir '$docdir': $!\n"; -open(my $hfile_handle, '>', $hfile) +open(my $hfile_handle, '>', $outdir . '/' . $hfile) or die "$0: could not open output file '$hfile': $!\n"; -open(my $cfile_handle, '>', $cfile) +open(my $cfile_handle, '>', $outdir . '/' . $cfile) or die "$0: could not open output file '$cfile': $!\n"; +open(my $depfile_handle, '>', $outdir . '/' . $depfile) + or die "$0: could not open output file '$depfile': $!\n"; print $hfile_handle "/* * *** Do not change this file by hand. It is automatically @@ -98,6 +103,8 @@ foreach my $file (sort readdir DIR) my ($cmdid, @cmdnames, $cmddesc, $cmdsynopsis); $file =~ /\.sgml$/ or next; + print $depfile_handle "$cfile $hfile: $docdir/$file\n"; + open(my $fh, '<', "$docdir/$file") or next; my $filecontent = join('', <$fh>); close $fh; @@ -216,4 +223,5 @@ print $hfile_handle " close $cfile_handle; close $hfile_handle; +close $depfile_handle; closedir DIR; diff --git a/src/tools/msvc/MSBuildProject.pm b/src/tools/msvc/MSBuildProject.pm index fdd22e89eb2..036e44fcb83 100644 --- a/src/tools/msvc/MSBuildProject.pm +++ b/src/tools/msvc/MSBuildProject.pm @@ -211,14 +211,19 @@ EOF } else #if ($grammarFile =~ /\.l$/) { + if ($outputFile eq 'src/backend/utils/misc/guc-file.c') + { + $outputFile = 'src/backend/utils/misc/guc-file.c.h'; + } + print $f <<EOF; <CustomBuild Include="$grammarFile"> <Message Condition="'\$(Configuration)|\$(Platform)'=='Debug|$self->{platform}'">Running flex on $grammarFile</Message> - <Command Condition="'\$(Configuration)|\$(Platform)'=='Debug|$self->{platform}'">perl "src\\tools\\msvc\\pgflex.pl" "$grammarFile"</Command> + <Command Condition="'\$(Configuration)|\$(Platform)'=='Debug|$self->{platform}'">perl "src\\tools\\msvc\\pgflex.pl" "$grammarFile" "$outputFile"</Command> <AdditionalInputs Condition="'\$(Configuration)|\$(Platform)'=='Debug|$self->{platform}'">%(AdditionalInputs)</AdditionalInputs> <Outputs Condition="'\$(Configuration)|\$(Platform)'=='Debug|$self->{platform}'">$outputFile;%(Outputs)</Outputs> <Message Condition="'\$(Configuration)|\$(Platform)'=='Release|$self->{platform}'">Running flex on $grammarFile</Message> - <Command Condition="'\$(Configuration)|\$(Platform)'=='Release|$self->{platform}'">perl "src\\tools\\msvc\\pgflex.pl" "$grammarFile"</Command> + <Command Condition="'\$(Configuration)|\$(Platform)'=='Release|$self->{platform}'">perl "src\\tools\\msvc\\pgflex.pl" "$grammarFile" "$outputFile"</Command> <AdditionalInputs Condition="'\$(Configuration)|\$(Platform)'=='Release|$self->{platform}'">%(AdditionalInputs)</AdditionalInputs> <Outputs Condition="'\$(Configuration)|\$(Platform)'=='Release|$self->{platform}'">$outputFile;%(Outputs)</Outputs> </CustomBuild> diff --git a/src/tools/msvc/Mkvcbuild.pm b/src/tools/msvc/Mkvcbuild.pm index 4362bd44fd1..b8e62c6d3f7 100644 --- a/src/tools/msvc/Mkvcbuild.pm +++ b/src/tools/msvc/Mkvcbuild.pm @@ -330,6 +330,7 @@ sub mkvcbuild $pgregress_ecpg->AddFile('src/test/regress/pg_regress.c'); $pgregress_ecpg->AddIncludeDir('src/port'); $pgregress_ecpg->AddIncludeDir('src/test/regress'); + $pgregress_ecpg->AddDefine('DLSUFFIX=".dll"'); $pgregress_ecpg->AddDefine('HOST_TUPLE="i686-pc-win32vc"'); $pgregress_ecpg->AddLibrary('ws2_32.lib'); $pgregress_ecpg->AddDirResourceFile('src/interfaces/ecpg/test'); @@ -345,6 +346,7 @@ sub mkvcbuild $isolation_tester->AddIncludeDir('src/port'); $isolation_tester->AddIncludeDir('src/test/regress'); $isolation_tester->AddIncludeDir('src/interfaces/libpq'); + $isolation_tester->AddDefine('DLSUFFIX=".dll"'); $isolation_tester->AddDefine('HOST_TUPLE="i686-pc-win32vc"'); $isolation_tester->AddLibrary('ws2_32.lib'); $isolation_tester->AddDirResourceFile('src/test/isolation'); @@ -356,6 +358,7 @@ sub mkvcbuild $pgregress_isolation->AddFile('src/test/regress/pg_regress.c'); $pgregress_isolation->AddIncludeDir('src/port'); $pgregress_isolation->AddIncludeDir('src/test/regress'); + $pgregress_isolation->AddDefine('DLSUFFIX=".dll"'); $pgregress_isolation->AddDefine('HOST_TUPLE="i686-pc-win32vc"'); $pgregress_isolation->AddLibrary('ws2_32.lib'); $pgregress_isolation->AddDirResourceFile('src/test/isolation'); diff --git a/src/tools/msvc/Solution.pm b/src/tools/msvc/Solution.pm index 165a93987ac..721f690ae46 100644 --- a/src/tools/msvc/Solution.pm +++ b/src/tools/msvc/Solution.pm @@ -688,7 +688,7 @@ sub GenerateFiles { print "Generating sql_help.h...\n"; chdir('src/bin/psql'); - system("perl create_help.pl ../../../doc/src/sgml/ref sql_help"); + system("perl create_help.pl ../../../doc/src/sgml/ref . sql_help"); chdir('../../..'); } diff --git a/src/tools/msvc/pgflex.pl b/src/tools/msvc/pgflex.pl index 0728b85d4de..19f26ff213f 100644 --- a/src/tools/msvc/pgflex.pl +++ b/src/tools/msvc/pgflex.pl @@ -29,6 +29,8 @@ unless ($verparts[0] == 2 } my $input = shift; +my $output = shift; + if ($input !~ /\.l$/) { print "Input must be a .l file\n"; @@ -40,8 +42,6 @@ elsif (!-e $input) exit 1; } -(my $output = $input) =~ s/\.l$/.c/; - # get flex flags from make file my $makefile = dirname($input) . "/Makefile"; my ($mf, $make); -- 2.23.0.385.gbc12974a89 --qozqt6hocm2wibt2 Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v3-0005-meson-prereq-move-snowball_create.sql-creation-in.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-02-26 08:17 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 ++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 107 ++++++++++++++---- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - 12 files changed, 216 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 4ab5df9213..4e6e92c4db 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -97,6 +97,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -351,6 +352,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1546,6 +1548,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 0bfd329847..0e4f234440 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -137,6 +138,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -648,6 +655,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -662,6 +670,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -691,20 +725,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -729,6 +749,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2243,6 +2311,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 799fc739e1..4a8a43a6da 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index a5790d77b5..910ff9fa91 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1298,7 +1298,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, * * In the REPACK CONCURRENTLY case, the lock does not help because we need * to release it temporarily at some point. Instead, we expect VACUUM / - * CLUSTER to skip tables which are present in RepackedRelsHash. + * CLUSTER to skip tables which are present in repackedRels->hashtable. */ if (OldHeap->rd_rel->reltoastrelid && !concurrent) LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); @@ -2312,7 +2312,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* Maximum number of entries in the hashtable. */ static int maxRepackedRels = 0; @@ -2320,28 +2329,44 @@ static int maxRepackedRels = 0; Size RepackShmemSize(void) { + Size result; + + result = sizeof(RepackedRels); + /* * A replication slot is needed for the processing, so use this GUC to * allocate memory for the hashtable. */ maxRepackedRels = max_replication_slots; - return hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + result += hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - maxRepackedRels, - maxRepackedRels, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + maxRepackedRels, + maxRepackedRels, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2373,12 +2398,13 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, RelReopenInfo rri[2]; int nrel; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->Hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2393,7 +2419,7 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2411,6 +2437,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + /* * Even if the insertion of TOAST relid should fail below, the caller has * to do cleanup. @@ -2438,7 +2468,8 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, { key.relid = toastrelid; entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, + &found); if (found) /* * If we could enter the main fork the TOAST should succeed @@ -2452,6 +2483,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + Assert(!OidIsValid(repacked_rel_toast)); repacked_rel_toast = toastrelid; } @@ -2531,6 +2566,7 @@ end_concurrent_repack(bool error) RepackedRel *entry = NULL, *entry_toast = NULL; Oid relid = repacked_rel; Oid toastrelid = repacked_rel_toast; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2539,23 +2575,32 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); /* * By clearing this variable we also disable * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* Remove the TOAST relation if there is one. */ if (OidIsValid(repacked_rel_toast)) { key.relid = repacked_rel_toast; - entry_toast = hash_search(RepackedRelsHash, &key, HASH_REMOVE, + entry_toast = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, NULL); repacked_rel_toast = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } LWLockRelease(RepackedRelsLock); @@ -2621,7 +2666,7 @@ end_concurrent_repack(bool error) } /* - * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + * A wrapper to call end_concurrent_cluster() as a before_shmem_exit callback. */ static void cluster_before_shmem_exit_callback(int code, Datum arg) @@ -2632,24 +2677,48 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) { RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * Check if REPACK CONCURRENTLY is already running for given relation, and if * so, raise ERROR. The problem is that cluster_rel() needs to release its @@ -2944,8 +3013,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 8ea846bfc3..e5790d3fe8 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f8..413bcc1add 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c02..a325bb1d16 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index f5600bf4f6..3ed3066b36 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -173,6 +173,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void check_for_concurrent_repack(Oid relid, LOCKMODE lockmode); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 741b29226d..81f5348a4f 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -709,12 +709,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4..4f6c0ca3a8 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6c..0000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba99..716e5619aa 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v08-0009-Call-logical_rewrite_heap_tuple-when-applying-concur.patch ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v09 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-02-26 08:17 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-02-26 08:17 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 ++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 107 ++++++++++++++---- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - 12 files changed, 216 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index e766be7b81d..479fe62b1c7 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -138,6 +139,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -649,6 +656,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -663,6 +671,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -692,20 +726,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -730,6 +750,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2244,6 +2312,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 4b6c694a3f7..1b131e1436f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 0a2bacbb1df..fa6db0932c0 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1298,7 +1298,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, * * In the REPACK CONCURRENTLY case, the lock does not help because we need * to release it temporarily at some point. Instead, we expect VACUUM / - * CLUSTER to skip tables which are present in RepackedRelsHash. + * CLUSTER to skip tables which are present in repackedRels->hashtable. */ if (OldHeap->rd_rel->reltoastrelid && !concurrent) LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); @@ -2317,7 +2317,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* Maximum number of entries in the hashtable. */ static int maxRepackedRels = 0; @@ -2325,28 +2334,44 @@ static int maxRepackedRels = 0; Size RepackShmemSize(void) { + Size result; + + result = sizeof(RepackedRels); + /* * A replication slot is needed for the processing, so use this GUC to * allocate memory for the hashtable. */ maxRepackedRels = max_replication_slots; - return hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + result += hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - maxRepackedRels, - maxRepackedRels, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + maxRepackedRels, + maxRepackedRels, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2378,12 +2403,13 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, RelReopenInfo rri[2]; int nrel; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->Hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2398,7 +2424,7 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2416,6 +2442,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + /* * Even if the insertion of TOAST relid should fail below, the caller has * to do cleanup. @@ -2443,7 +2473,8 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, { key.relid = toastrelid; entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, + &found); if (found) /* * If we could enter the main fork the TOAST should succeed @@ -2457,6 +2488,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + Assert(!OidIsValid(repacked_rel_toast)); repacked_rel_toast = toastrelid; } @@ -2536,6 +2571,7 @@ end_concurrent_repack(bool error) RepackedRel *entry = NULL, *entry_toast = NULL; Oid relid = repacked_rel; Oid toastrelid = repacked_rel_toast; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2544,23 +2580,32 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); /* * By clearing this variable we also disable * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* Remove the TOAST relation if there is one. */ if (OidIsValid(repacked_rel_toast)) { key.relid = repacked_rel_toast; - entry_toast = hash_search(RepackedRelsHash, &key, HASH_REMOVE, + entry_toast = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, NULL); repacked_rel_toast = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } LWLockRelease(RepackedRelsLock); @@ -2626,7 +2671,7 @@ end_concurrent_repack(bool error) } /* - * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + * A wrapper to call end_concurrent_cluster() as a before_shmem_exit callback. */ static void cluster_before_shmem_exit_callback(int code, Datum arg) @@ -2637,24 +2682,48 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) { RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * Check if REPACK CONCURRENTLY is already running for given relation, and if * so, raise ERROR. The problem is that cluster_rel() needs to release its @@ -2949,8 +3018,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 8ea846bfc3b..e5790d3fe84 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f85..413bcc1addb 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index f5600bf4f62..3ed3066b364 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -173,6 +173,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void check_for_concurrent_repack(Oid relid, LOCKMODE lockmode); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index cb485d26f44..88bacc109ff 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -710,12 +710,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { -- 2.39.5 --v2l46qwn2nggemiy Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v09-0009-Call-logical_rewrite_heap_tuple-when-applying-co.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH v10 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-03-24 19:17 Álvaro Herrera <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Álvaro Herrera @ 2025-03-24 19:17 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 ++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 107 ++++++++++++++---- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 13 files changed, 217 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index aebad612ce8..1b02ef0bacb 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -138,6 +139,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -649,6 +656,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -663,6 +671,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -692,20 +726,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -730,6 +750,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2244,6 +2312,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 4b6c694a3f7..1b131e1436f 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index c272ed03cb9..2ab1756fa19 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -1304,7 +1304,7 @@ copy_table_data(Relation NewHeap, Relation OldHeap, Relation OldIndex, * * In the REPACK CONCURRENTLY case, the lock does not help because we need * to release it temporarily at some point. Instead, we expect VACUUM / - * CLUSTER to skip tables which are present in RepackedRelsHash. + * CLUSTER to skip tables which are present in repackedRels->hashtable. */ if (OldHeap->rd_rel->reltoastrelid && !concurrent) LockRelationOid(OldHeap->rd_rel->reltoastrelid, AccessExclusiveLock); @@ -2324,7 +2324,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* Maximum number of entries in the hashtable. */ static int maxRepackedRels = 0; @@ -2332,28 +2341,44 @@ static int maxRepackedRels = 0; Size RepackShmemSize(void) { + Size result; + + result = sizeof(RepackedRels); + /* * A replication slot is needed for the processing, so use this GUC to * allocate memory for the hashtable. */ maxRepackedRels = max_replication_slots; - return hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + result += hash_estimate_size(maxRepackedRels, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - maxRepackedRels, - maxRepackedRels, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + maxRepackedRels, + maxRepackedRels, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2387,12 +2412,13 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, RelReopenInfo rri[2]; int nrel; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->Hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2407,7 +2433,7 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2425,6 +2451,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + /* * Even if the insertion of TOAST relid should fail below, the caller has * to do cleanup. @@ -2452,7 +2482,8 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, { key.relid = toastrelid; entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, + &found); if (found) /* @@ -2467,6 +2498,10 @@ begin_concurrent_repack(Relation *rel_p, Relation *index_p, (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < maxRepackedRels); + Assert(!OidIsValid(repacked_rel_toast)); repacked_rel_toast = toastrelid; } @@ -2549,6 +2584,7 @@ end_concurrent_repack(bool error) *entry_toast = NULL; Oid relid = repacked_rel; Oid toastrelid = repacked_rel_toast; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2557,23 +2593,32 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); /* * By clearing this variable we also disable * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* Remove the TOAST relation if there is one. */ if (OidIsValid(repacked_rel_toast)) { key.relid = repacked_rel_toast; - entry_toast = hash_search(RepackedRelsHash, &key, HASH_REMOVE, + entry_toast = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, NULL); repacked_rel_toast = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } LWLockRelease(RepackedRelsLock); @@ -2639,7 +2684,7 @@ end_concurrent_repack(bool error) } /* - * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + * A wrapper to call end_concurrent_cluster() as a before_shmem_exit callback. */ static void cluster_before_shmem_exit_callback(int code, Datum arg) @@ -2650,6 +2695,8 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) @@ -2657,18 +2704,40 @@ is_concurrent_repack_in_progress(Oid relid) RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * Check if REPACK CONCURRENTLY is already running for given relation, and if * so, raise ERROR. The problem is that cluster_rel() needs to release its @@ -2967,8 +3036,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 8ea846bfc3b..e5790d3fe84 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f85..413bcc1addb 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 268c3098512..6f5566210a8 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -173,6 +173,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void check_for_concurrent_repack(Oid relid, LOCKMODE lockmode); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 372065fc570..fcbad5c1720 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -710,12 +710,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index ac52b8b0336..de1a178a4b9 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2508,6 +2508,7 @@ ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId RepackedRel +RepackedRels RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.39.5 --w2lc3gf2kw4enm6y Content-Type: text/x-diff; charset=utf-8 Content-Disposition: attachment; filename="v10-0009-Call-logical_rewrite_heap_tuple-when-applying-co.patch" ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-03-31 13:47 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-03-31 13:47 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 +++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 94 +++++++++++++--- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 13 files changed, 206 insertions(+), 44 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 3db4cac030e..608dc5c79bb 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -138,6 +139,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -650,6 +657,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -664,6 +672,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -693,20 +727,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -731,6 +751,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2245,6 +2313,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index fc30a52d496..ba758deefb4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 3f07b42d615..734e47eaba3 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -2203,7 +2203,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* * Maximum number of entries in the hashtable. @@ -2216,22 +2225,38 @@ static HTAB *RepackedRelsHash = NULL; Size RepackShmemSize(void) { - return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + Size result; + + result = sizeof(RepackedRels); + + result += hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - MAX_REPACKED_RELS, - MAX_REPACKED_RELS, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2266,13 +2291,14 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entry; bool found; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); index = index_p ? *index_p : NULL; /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2287,7 +2313,7 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2305,9 +2331,13 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < MAX_REPACKED_RELS); + /* - * Even if anything fails below, the caller has to do cleanup in the - * shared memory. + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. */ *entered_p = true; @@ -2389,6 +2419,7 @@ end_concurrent_repack(bool error) RepackedRel key; RepackedRel *entry = NULL; Oid relid = repacked_rel; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2397,7 +2428,8 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); LWLockRelease(RepackedRelsLock); /* @@ -2426,6 +2458,10 @@ end_concurrent_repack(bool error) * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* @@ -2478,6 +2514,8 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) @@ -2485,18 +2523,40 @@ is_concurrent_repack_in_progress(Oid relid) RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * 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 @@ -2524,8 +2584,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index a8d2e024d34..4909432d585 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f85..413bcc1addb 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index be283c70fce..0267357a261 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -172,6 +172,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 372065fc570..fcbad5c1720 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -710,12 +710,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index e1e3e619c4b..b3be8572132 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2511,6 +2511,7 @@ ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId RepackedRel +RepackedRels RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v11-0009-Call-logical_rewrite_heap_tuple-when-applying-concur.patch ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-04-01 11:48 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-04-01 11:48 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- src/backend/access/transam/parallel.c | 8 ++ src/backend/access/transam/xact.c | 106 +++++++++++++++--- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 94 +++++++++++++--- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/standby.c | 4 +- src/include/access/xlog.h | 15 ++- src/include/commands/cluster.h | 1 + src/include/utils/rel.h | 6 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 13 files changed, 206 insertions(+), 44 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 3db4cac030e..608dc5c79bb 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -138,6 +139,12 @@ static TransactionId *ParallelCurrentXids; static int nRepackCurrentXids = 0; static TransactionId *RepackCurrentXids = NULL; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -650,6 +657,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -664,6 +672,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -693,20 +727,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -731,6 +751,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2245,6 +2313,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index fc30a52d496..ba758deefb4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 085649716ac..c2201b046bc 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -2204,7 +2204,16 @@ typedef struct RepackedRel Oid dbid; } RepackedRel; -static HTAB *RepackedRelsHash = NULL; +typedef struct RepackedRels +{ + /* Hashtable of RepackedRel elements. */ + HTAB *hashtable; + + /* The number of elements in the hashtable.. */ + pg_atomic_uint32 nrels; +} RepackedRels; + +static RepackedRels *repackedRels = NULL; /* * Maximum number of entries in the hashtable. @@ -2217,22 +2226,38 @@ static HTAB *RepackedRelsHash = NULL; Size RepackShmemSize(void) { - return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + Size result; + + result = sizeof(RepackedRels); + + result += hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); + return result; } void RepackShmemInit(void) { + bool found; HASHCTL info; + repackedRels = ShmemInitStruct("Repacked Relations", + sizeof(RepackedRels), + &found); + if (!IsUnderPostmaster) + { + Assert(!found); + pg_atomic_init_u32(&repackedRels->nrels, 0); + } + else + Assert(found); + info.keysize = sizeof(RepackedRel); info.entrysize = info.keysize; - - RepackedRelsHash = ShmemInitHash("Repacked Relations", - MAX_REPACKED_RELS, - MAX_REPACKED_RELS, - &info, - HASH_ELEM | HASH_BLOBS); + repackedRels->hashtable = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS); } /* @@ -2267,13 +2292,14 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entry; bool found; static bool before_shmem_exit_callback_setup = false; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; relid = RelationGetRelid(rel); index = index_p ? *index_p : NULL; /* - * Make sure that we do not leave an entry in RepackedRelsHash if exiting - * due to FATAL. + * Make sure that we do not leave an entry in repackedRels->hashtable if + * exiting due to FATAL. */ if (!before_shmem_exit_callback_setup) { @@ -2288,7 +2314,7 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) *entered_p = false; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_ENTER_NULL, &found); + hash_search(repackedRels->hashtable, &key, HASH_ENTER_NULL, &found); if (found) { /* @@ -2306,9 +2332,13 @@ begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) (errmsg("too many requests for REPACK CONCURRENTLY at a time")), (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + /* Increment the number of relations. */ + nrels = pg_atomic_fetch_add_u32(&repackedRels->nrels, 1); + Assert(nrels < MAX_REPACKED_RELS); + /* - * Even if anything fails below, the caller has to do cleanup in the - * shared memory. + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. */ *entered_p = true; @@ -2390,6 +2420,7 @@ end_concurrent_repack(bool error) RepackedRel key; RepackedRel *entry = NULL; Oid relid = repacked_rel; + uint32 nrels PG_USED_FOR_ASSERTS_ONLY; /* Remove the relation from the hash if we managed to insert one. */ if (OidIsValid(repacked_rel)) @@ -2398,7 +2429,8 @@ end_concurrent_repack(bool error) key.relid = repacked_rel; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); - entry = hash_search(RepackedRelsHash, &key, HASH_REMOVE, NULL); + entry = hash_search(repackedRels->hashtable, &key, HASH_REMOVE, + NULL); LWLockRelease(RepackedRelsLock); /* @@ -2427,6 +2459,10 @@ end_concurrent_repack(bool error) * cluster_before_shmem_exit_callback(). */ repacked_rel = InvalidOid; + + /* Decrement the number of relations. */ + nrels = pg_atomic_fetch_sub_u32(&repackedRels->nrels, 1); + Assert(nrels > 0); } /* @@ -2479,6 +2515,8 @@ cluster_before_shmem_exit_callback(int code, Datum arg) /* * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. */ bool is_concurrent_repack_in_progress(Oid relid) @@ -2486,18 +2524,40 @@ is_concurrent_repack_in_progress(Oid relid) RepackedRel key, *entry; + /* + * If the caller is interested whether any relation is being repacked, + * just use the counter. + */ + if (!OidIsValid(relid)) + { + if (pg_atomic_read_u32(&repackedRels->nrels) > 0) + return true; + else + return false; + } + + /* For particular relation we need to search in the hashtable. */ memset(&key, 0, sizeof(key)); key.relid = relid; key.dbid = MyDatabaseId; LWLockAcquire(RepackedRelsLock, LW_SHARED); entry = (RepackedRel *) - hash_search(RepackedRelsHash, &key, HASH_FIND, NULL); + hash_search(repackedRels->hashtable, &key, HASH_FIND, NULL); LWLockRelease(RepackedRelsLock); return entry != NULL; } +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); +} + /* * 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 @@ -2525,8 +2585,8 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, - * regarding logical decoding, treated like a catalog. + * table we are processing is present in repackedRels->hashtable and + * therefore, regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, NIL, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index a8d2e024d34..4909432d585 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 5acb4508f85..413bcc1addb 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1313,13 +1313,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index be283c70fce..0267357a261 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -172,6 +172,7 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, extern Size RepackShmemSize(void); extern void RepackShmemInit(void); extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index 372065fc570..fcbad5c1720 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -710,12 +710,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index e1e3e619c4b..b3be8572132 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2511,6 +2511,7 @@ ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId RepackedRel +RepackedRels RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.43.5 --=-=-= Content-Type: text/x-diff Content-Disposition: attachment; filename=v12-0009-Call-logical_rewrite_heap_tuple-when-applying-concur.patch ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-04-11 09:13 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-04-11 09:13 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- doc/src/sgml/ref/repack.sgml | 7 - src/backend/access/transam/parallel.c | 8 + src/backend/access/transam/xact.c | 106 ++++- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 387 +++++++++++++++++- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/ipci.c | 2 + src/backend/storage/ipc/standby.c | 4 +- src/backend/utils/cache/inval.c | 21 + src/backend/utils/cache/relcache.c | 4 + src/include/access/xlog.h | 15 +- src/include/commands/cluster.h | 5 + src/include/utils/inval.h | 2 + src/include/utils/rel.h | 9 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 18 files changed, 540 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index 7e44fa636ac..28adb21738a 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -263,13 +263,6 @@ INDEX <replaceable class="parameter">index_name</replaceable> ] </para> </listitem> - <listitem> - <para> - The <link linkend="guc-wal-level"><varname>wal_level</varname></link> - configuration parameter is less than <literal>logical</literal>. - </para> - </listitem> - <listitem> <para> The <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link> diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 23f2de587a1..be568f70961 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -126,6 +127,12 @@ static FullTransactionId XactTopFullTransactionId = {InvalidTransactionId}; static int nParallelCurrentXids = 0; static TransactionId *ParallelCurrentXids; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -638,6 +645,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -652,6 +660,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -681,20 +715,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -719,6 +739,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2216,6 +2284,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index ec40c0b7c42..b4e07104083 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index cc765b88d52..6e1cdc7bca6 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -84,6 +84,14 @@ typedef struct * The following definitions are used for concurrent processing. */ +/* + * OID of the table being repacked by this backend. + */ +static Oid repacked_rel = InvalidOid; + +/* The same for its TOAST relation. */ +static Oid repacked_rel_toast = InvalidOid; + /* * The locators are used to avoid logical decoding of data that we do not need * for our table. @@ -135,8 +143,10 @@ static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context, ClusterCommand cmd); static bool cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd); -static void begin_concurrent_repack(Relation rel); -static void end_concurrent_repack(void); +static void begin_concurrent_repack(Relation rel, Relation *index_p, + bool *entered_p); +static void end_concurrent_repack(bool error); +static void cluster_before_shmem_exit_callback(int code, Datum arg); static LogicalDecodingContext *setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc); @@ -383,6 +393,8 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, Relation index; bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); LOCKMODE lmode; + bool entered, + success; /* * Check that the correct lock is held. The lock mode is @@ -558,23 +570,31 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, TransferPredicateLocksToHeapRelation(OldHeap); /* rebuild_relation does all the dirty work */ + entered = false; + success = false; PG_TRY(); { /* - * For concurrent processing, make sure that our logical decoding - * ignores data changes of other tables than the one we are - * processing. + * For concurrent processing, make sure that + * + * 1) our logical decoding ignores data changes of other tables than + * the one we are processing. + * + * 2) other transactions know that REPACK CONCURRENTLY is in progress + * for our table, so they write sufficient information to WAL even if + * wal_level is < LOGICAL. */ if (concurrent) - begin_concurrent_repack(OldHeap); + begin_concurrent_repack(OldHeap, &index, &entered); rebuild_relation(OldHeap, index, verbose, concurrent, save_userid, cmd); + success = true; } PG_FINALLY(); { - if (concurrent) - end_concurrent_repack(); + if (concurrent && entered) + end_concurrent_repack(!success); } PG_END_TRY(); @@ -2207,6 +2227,49 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) #define REPL_PLUGIN_NAME "pgoutput_repack" +/* + * Each relation being processed by REPACK CONCURRENTLY must be in the + * repackedRelsHash hashtable. + */ +typedef struct RepackedRel +{ + Oid relid; + Oid dbid; +} RepackedRel; + +/* Hashtable of RepackedRel elements. */ +static HTAB *repackedRelsHash = NULL;; + +/* + * Maximum number of entries in the hashtable. + * + * A replication slot is needed for the processing, so use this GUC to + * allocate memory for the hashtable. Multiply by two because TOAST relations + * also need to be added to the hashtable. + */ +#define MAX_REPACKED_RELS (max_replication_slots * 2) + +Size +RepackShmemSize(void) +{ + return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); +} + +void +RepackShmemInit(void) +{ + HASHCTL info; + + info.keysize = sizeof(RepackedRel); + info.entrysize = info.keysize; + repackedRelsHash = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS | + HASH_FIXED_SIZE); +} + /* * Call this function before REPACK CONCURRENTLY starts to setup logical * decoding. It makes sure that other users of the table put enough @@ -2221,11 +2284,150 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) * * Note that TOAST table needs no attention here as it's not scanned using * historic snapshot. + * + * 'index_p' is in/out argument because the function unlocks the index + * temporarily. + * + * 'enter_p' receives a bool value telling whether relation OID was entered + * into repackedRelsHash or not. */ static void -begin_concurrent_repack(Relation rel) +begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) { - Oid toastrelid; + Oid relid, + toastrelid; + Relation index = NULL; + Oid indexid = InvalidOid; + RepackedRel key, + *entry; + bool found; + static bool before_shmem_exit_callback_setup = false; + + relid = RelationGetRelid(rel); + index = index_p ? *index_p : NULL; + + /* + * Make sure that we do not leave an entry in repackedRelsHash if exiting + * due to FATAL. + */ + if (!before_shmem_exit_callback_setup) + { + before_shmem_exit(cluster_before_shmem_exit_callback, 0); + before_shmem_exit_callback_setup = true; + } + + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + *entered_p = false; + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + { + /* + * Since REPACK CONCURRENTLY takes ShareRowExclusiveLock, a conflict + * should occur much earlier. However that lock may be released + * temporarily, see below. Anyway, we should complain whatever the + * reason of the conflict might be. + */ + ereport(ERROR, + (errmsg("relation \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + } + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + /* + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. + */ + *entered_p = true; + + /* + * Enable the callback to remove the entry in case of exit. We should not + * do this earlier, otherwise an attempt to insert already existing entry + * could make us remove that entry (inserted by another backend) during + * ERROR handling. + */ + Assert(!OidIsValid(repacked_rel)); + repacked_rel = relid; + + /* + * TOAST relation is not accessed using historic snapshot, but we enter it + * here to protect it from being VACUUMed by another backend. (Lock does + * not help in the CONCURRENTLY case because cannot hold it continuously + * till the end of the transaction.) See the comments on locking TOAST + * relation in copy_table_data(). + */ + toastrelid = rel->rd_rel->reltoastrelid; + if (OidIsValid(toastrelid)) + { + key.relid = toastrelid; + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + + /* + * If we could enter the main fork the TOAST should succeed too. + * Nevertheless, check. + */ + ereport(ERROR, + (errmsg("TOAST relation of \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + Assert(!OidIsValid(repacked_rel_toast)); + repacked_rel_toast = toastrelid; + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make sure that other backends are aware of the new hash entry as soon + * as they open our table. + */ + CacheInvalidateRelcacheImmediate(relid); + + /* + * Also make sure that the existing users of the table update their + * relcache entry as soon as they try to run DML commands on it. + * + * ShareLock is the weakest lock that conflicts with DMLs. If any backend + * has a lower lock, we assume it'll accept our invalidation message when + * it changes the lock mode. + * + * Before upgrading the lock on the relation, close the index temporarily + * to avoid a deadlock if another backend running DML already has its lock + * (ShareLock) on the table and waits for the lock on the index. + */ + if (index) + { + indexid = RelationGetRelid(index); + index_close(index, ShareUpdateExclusiveLock); + } + LockRelationOid(relid, ShareLock); + UnlockRelationOid(relid, ShareLock); + if (OidIsValid(indexid)) + { + /* + * Re-open the index and check that it hasn't changed while unlocked. + */ + check_index_is_clusterable(rel, indexid, ShareUpdateExclusiveLock); + + /* + * Return the new relcache entry to the caller. (It's been locked by + * the call above.) + */ + index = index_open(indexid, NoLock); + *index_p = index; + } /* Avoid logical decoding of other relations by this backend. */ repacked_rel_locator = rel->rd_locator; @@ -2243,15 +2445,176 @@ begin_concurrent_repack(Relation rel) /* * Call this when done with REPACK CONCURRENTLY. + * + * 'error' tells whether the function is being called in order to handle + * error. */ static void -end_concurrent_repack(void) +end_concurrent_repack(bool error) { + RepackedRel key; + RepackedRel *entry = NULL; + RepackedRel *entry_toast = NULL; + Oid relid = repacked_rel; + Oid toastrelid = repacked_rel_toast; + + /* Remove the relation from the hash if we managed to insert one. */ + if (OidIsValid(repacked_rel)) + { + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + + memset(&key, 0, sizeof(key)); + key.relid = repacked_rel; + key.dbid = MyDatabaseId; + + entry = hash_search(repackedRelsHash, &key, HASH_REMOVE, NULL); + + /* Remove the TOAST relation if there is one. */ + if (OidIsValid(repacked_rel_toast)) + { + key.relid = repacked_rel_toast; + entry_toast = hash_search(repackedRelsHash, &key, HASH_REMOVE, + NULL); + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make others refresh their information whether they should still + * treat the table as catalog from the perspective of writing WAL. + * + * XXX Unlike entering the entry into the hashtable, we do not bother + * with locking and unlocking the table here: + * + * 1) On normal completion (and sometimes even on ERROR), the caller + * is already holding AccessExclusiveLock on the table, so there + * should be no relcache reference unaware of this change. + * + * 2) In the other cases, the worst scenario is that the other + * backends will write unnecessary information to WAL until they close + * the relation. + * + * Should we use ShareLock mode to fix 2) at least for the non-FATAL + * errors? (Our before_shmem_exit callback is in charge of FATAL, and + * that probably should not try to acquire any lock.) + */ + CacheInvalidateRelcacheImmediate(repacked_rel); + + /* + * By clearing repacked_rel we also disable + * cluster_before_shmem_exit_callback(). + */ + repacked_rel = InvalidOid; + repacked_rel_toast = InvalidOid; + } + /* * Restore normal function of (future) logical decoding for this backend. */ repacked_rel_locator.relNumber = InvalidOid; repacked_rel_toast_locator.relNumber = InvalidOid; + + /* + * On normal completion (!error), we should not really fail to remove the + * entry. But if it wasn't there for any reason, raise ERROR to make sure + * the transaction is aborted: if other transactions, while changing the + * contents of the relation, didn't know that REPACK CONCURRENTLY was in + * progress, they could have missed to WAL enough information, and thus we + * could have produced an inconsistent table contents. + * + * On the other hand, if we are already handling an error, there's no + * reason to worry about inconsistent contents of the new storage because + * the transaction is going to be rolled back anyway. Furthermore, by + * raising ERROR here we'd shadow the original error. + */ + if (!error) + { + char *relname; + + if (OidIsValid(relid) && entry == NULL) + { + relname = get_rel_name(relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + /* + * Likewise, the TOAST relation should not have disappeared. + */ + if (OidIsValid(toastrelid) && entry_toast == NULL) + { + relname = get_rel_name(key.relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + key.relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + } +} + +/* + * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + */ +static void +cluster_before_shmem_exit_callback(int code, Datum arg) +{ + if (OidIsValid(repacked_rel)) + end_concurrent_repack(true); +} + +/* + * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. + */ +bool +is_concurrent_repack_in_progress(Oid relid) +{ + RepackedRel key, + *entry; + + /* For particular relation we need to search in the hashtable. */ + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + LWLockAcquire(RepackedRelsLock, LW_SHARED); + /* + * If the caller is interested whether any relation is being repacked, + * just check the number of entries. + */ + if (!OidIsValid(relid)) + { + long n = hash_get_num_entries(repackedRelsHash); + + LWLockRelease(RepackedRelsLock); + return n > 0; + } + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_FIND, NULL); + LWLockRelease(RepackedRelsLock); + + return entry != NULL; +} + +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); } /* @@ -2281,7 +2644,7 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, + * table we are processing is present in repackedRelsHash and therefore, * regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index a8d2e024d34..4909432d585 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index f247e4e7c16..7b27068c338 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -153,6 +153,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, SlotSyncShmemSize()); size = add_size(size, AioShmemSize()); size = add_size(size, MemoryContextReportingShmemSize()); + size = add_size(size, RepackShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -347,6 +348,7 @@ CreateOrAttachShmemStructs(void) InjectionPointShmemInit(); AioShmemInit(); MemoryContextReportingShmemInit(); + RepackShmemInit(); } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 7fa8d9247e0..ab30d448d42 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1325,13 +1325,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c index 4eb67720737..14eda1c24ee 100644 --- a/src/backend/utils/cache/inval.c +++ b/src/backend/utils/cache/inval.c @@ -1633,6 +1633,27 @@ CacheInvalidateRelcache(Relation relation) databaseId, relationId); } +/* + * CacheInvalidateRelcacheImmediate + * Send invalidation message for the specified relation's relcache entry. + * + * Currently this is used in REPACK CONCURRENTLY, to make sure that other + * backends are aware that the command is being executed for the relation. + */ +void +CacheInvalidateRelcacheImmediate(Oid relid) +{ + SharedInvalidationMessage msg; + + msg.rc.id = SHAREDINVALRELCACHE_ID; + msg.rc.dbId = MyDatabaseId; + msg.rc.relId = relid; + /* check AddCatcacheInvalidationMessage() for an explanation */ + VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg)); + + SendSharedInvalidMessages(&msg, 1); +} + /* * CacheInvalidateRelcacheAll * Register invalidation of the whole relcache at the end of command. diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 75434e32198..058c95bc847 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -1279,6 +1279,10 @@ retry: /* make sure relation is marked as having no open file yet */ relation->rd_smgr = NULL; + /* Is REPACK CONCURRENTLY in progress? */ + relation->rd_repack_concurrent = + is_concurrent_repack_in_progress(targetRelId); + /* * now we can free the memory allocated for pg_class_tuple */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 4914f217267..9d5a30d0689 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -150,5 +150,10 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, MultiXactId cutoffMulti, char newrelpersistence); +extern Size RepackShmemSize(void); +extern void RepackShmemInit(void); +extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); + extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/inval.h b/src/include/utils/inval.h index 9b871caef62..ae9dee394dc 100644 --- a/src/include/utils/inval.h +++ b/src/include/utils/inval.h @@ -50,6 +50,8 @@ extern void CacheInvalidateCatalog(Oid catalogId); extern void CacheInvalidateRelcache(Relation relation); +extern void CacheInvalidateRelcacheImmediate(Oid relid); + extern void CacheInvalidateRelcacheAll(void); extern void CacheInvalidateRelcacheByTuple(HeapTuple classTuple); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index b552359915f..cc84592eb1f 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -253,6 +253,9 @@ typedef struct RelationData bool pgstat_enabled; /* should relation stats be counted */ /* use "struct" here to avoid needing to include pgstat.h: */ struct PgStat_TableStatus *pgstat_info; /* statistics collection area */ + + /* Is REPACK CONCURRENTLY being performed on this relation? */ + bool rd_repack_concurrent; } RelationData; @@ -708,12 +711,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 6bbc8b419f8..f885eb6d28b 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2529,6 +2529,7 @@ ReorderBufferTupleCidKey ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId +RepackedRel RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.43.5 --=-=-=-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-06-09 10:00 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-06-09 10:00 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- doc/src/sgml/ref/repack.sgml | 7 - src/backend/access/transam/parallel.c | 8 + src/backend/access/transam/xact.c | 106 ++++- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 387 +++++++++++++++++- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/ipci.c | 2 + src/backend/storage/ipc/standby.c | 4 +- src/backend/utils/cache/inval.c | 21 + src/backend/utils/cache/relcache.c | 4 + src/include/access/xlog.h | 15 +- src/include/commands/cluster.h | 5 + src/include/utils/inval.h | 2 + src/include/utils/rel.h | 9 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 18 files changed, 540 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index e1313f40599..0fd767eef98 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -260,13 +260,6 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU </para> </listitem> - <listitem> - <para> - The <link linkend="guc-wal-level"><varname>wal_level</varname></link> - configuration parameter is less than <literal>logical</literal>. - </para> - </listitem> - <listitem> <para> The <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link> diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 23f2de587a1..be568f70961 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -126,6 +127,12 @@ static FullTransactionId XactTopFullTransactionId = {InvalidTransactionId}; static int nParallelCurrentXids = 0; static TransactionId *ParallelCurrentXids; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -638,6 +645,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -652,6 +660,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -681,20 +715,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -719,6 +739,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2216,6 +2284,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 1914859b2ee..f9c0e947ba4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 0ebc7eacad9..be383a27712 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -84,6 +84,14 @@ typedef struct * The following definitions are used for concurrent processing. */ +/* + * OID of the table being repacked by this backend. + */ +static Oid repacked_rel = InvalidOid; + +/* The same for its TOAST relation. */ +static Oid repacked_rel_toast = InvalidOid; + /* * The locators are used to avoid logical decoding of data that we do not need * for our table. @@ -135,8 +143,10 @@ static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context, ClusterCommand cmd); static bool cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd); -static void begin_concurrent_repack(Relation rel); -static void end_concurrent_repack(void); +static void begin_concurrent_repack(Relation rel, Relation *index_p, + bool *entered_p); +static void end_concurrent_repack(bool error); +static void cluster_before_shmem_exit_callback(int code, Datum arg); static LogicalDecodingContext *setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc); @@ -383,6 +393,8 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, Relation index; bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); LOCKMODE lmode; + bool entered, + success; /* * Check that the correct lock is held. The lock mode is @@ -558,23 +570,31 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, TransferPredicateLocksToHeapRelation(OldHeap); /* rebuild_relation does all the dirty work */ + entered = false; + success = false; PG_TRY(); { /* - * For concurrent processing, make sure that our logical decoding - * ignores data changes of other tables than the one we are - * processing. + * For concurrent processing, make sure that + * + * 1) our logical decoding ignores data changes of other tables than + * the one we are processing. + * + * 2) other transactions know that REPACK CONCURRENTLY is in progress + * for our table, so they write sufficient information to WAL even if + * wal_level is < LOGICAL. */ if (concurrent) - begin_concurrent_repack(OldHeap); + begin_concurrent_repack(OldHeap, &index, &entered); rebuild_relation(OldHeap, index, verbose, concurrent, save_userid, cmd); + success = true; } PG_FINALLY(); { - if (concurrent) - end_concurrent_repack(); + if (concurrent && entered) + end_concurrent_repack(!success); } PG_END_TRY(); @@ -2208,6 +2228,49 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) #define REPL_PLUGIN_NAME "pgoutput_repack" +/* + * Each relation being processed by REPACK CONCURRENTLY must be in the + * repackedRelsHash hashtable. + */ +typedef struct RepackedRel +{ + Oid relid; + Oid dbid; +} RepackedRel; + +/* Hashtable of RepackedRel elements. */ +static HTAB *repackedRelsHash = NULL;; + +/* + * Maximum number of entries in the hashtable. + * + * A replication slot is needed for the processing, so use this GUC to + * allocate memory for the hashtable. Multiply by two because TOAST relations + * also need to be added to the hashtable. + */ +#define MAX_REPACKED_RELS (max_replication_slots * 2) + +Size +RepackShmemSize(void) +{ + return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); +} + +void +RepackShmemInit(void) +{ + HASHCTL info; + + info.keysize = sizeof(RepackedRel); + info.entrysize = info.keysize; + repackedRelsHash = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS | + HASH_FIXED_SIZE); +} + /* * Call this function before REPACK CONCURRENTLY starts to setup logical * decoding. It makes sure that other users of the table put enough @@ -2222,11 +2285,150 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) * * Note that TOAST table needs no attention here as it's not scanned using * historic snapshot. + * + * 'index_p' is in/out argument because the function unlocks the index + * temporarily. + * + * 'enter_p' receives a bool value telling whether relation OID was entered + * into repackedRelsHash or not. */ static void -begin_concurrent_repack(Relation rel) +begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) { - Oid toastrelid; + Oid relid, + toastrelid; + Relation index = NULL; + Oid indexid = InvalidOid; + RepackedRel key, + *entry; + bool found; + static bool before_shmem_exit_callback_setup = false; + + relid = RelationGetRelid(rel); + index = index_p ? *index_p : NULL; + + /* + * Make sure that we do not leave an entry in repackedRelsHash if exiting + * due to FATAL. + */ + if (!before_shmem_exit_callback_setup) + { + before_shmem_exit(cluster_before_shmem_exit_callback, 0); + before_shmem_exit_callback_setup = true; + } + + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + *entered_p = false; + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + { + /* + * Since REPACK CONCURRENTLY takes ShareRowExclusiveLock, a conflict + * should occur much earlier. However that lock may be released + * temporarily, see below. Anyway, we should complain whatever the + * reason of the conflict might be. + */ + ereport(ERROR, + (errmsg("relation \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + } + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + /* + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. + */ + *entered_p = true; + + /* + * Enable the callback to remove the entry in case of exit. We should not + * do this earlier, otherwise an attempt to insert already existing entry + * could make us remove that entry (inserted by another backend) during + * ERROR handling. + */ + Assert(!OidIsValid(repacked_rel)); + repacked_rel = relid; + + /* + * TOAST relation is not accessed using historic snapshot, but we enter it + * here to protect it from being VACUUMed by another backend. (Lock does + * not help in the CONCURRENTLY case because cannot hold it continuously + * till the end of the transaction.) See the comments on locking TOAST + * relation in copy_table_data(). + */ + toastrelid = rel->rd_rel->reltoastrelid; + if (OidIsValid(toastrelid)) + { + key.relid = toastrelid; + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + + /* + * If we could enter the main fork the TOAST should succeed too. + * Nevertheless, check. + */ + ereport(ERROR, + (errmsg("TOAST relation of \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + Assert(!OidIsValid(repacked_rel_toast)); + repacked_rel_toast = toastrelid; + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make sure that other backends are aware of the new hash entry as soon + * as they open our table. + */ + CacheInvalidateRelcacheImmediate(relid); + + /* + * Also make sure that the existing users of the table update their + * relcache entry as soon as they try to run DML commands on it. + * + * ShareLock is the weakest lock that conflicts with DMLs. If any backend + * has a lower lock, we assume it'll accept our invalidation message when + * it changes the lock mode. + * + * Before upgrading the lock on the relation, close the index temporarily + * to avoid a deadlock if another backend running DML already has its lock + * (ShareLock) on the table and waits for the lock on the index. + */ + if (index) + { + indexid = RelationGetRelid(index); + index_close(index, ShareUpdateExclusiveLock); + } + LockRelationOid(relid, ShareLock); + UnlockRelationOid(relid, ShareLock); + if (OidIsValid(indexid)) + { + /* + * Re-open the index and check that it hasn't changed while unlocked. + */ + check_index_is_clusterable(rel, indexid, ShareUpdateExclusiveLock); + + /* + * Return the new relcache entry to the caller. (It's been locked by + * the call above.) + */ + index = index_open(indexid, NoLock); + *index_p = index; + } /* Avoid logical decoding of other relations by this backend. */ repacked_rel_locator = rel->rd_locator; @@ -2244,15 +2446,176 @@ begin_concurrent_repack(Relation rel) /* * Call this when done with REPACK CONCURRENTLY. + * + * 'error' tells whether the function is being called in order to handle + * error. */ static void -end_concurrent_repack(void) +end_concurrent_repack(bool error) { + RepackedRel key; + RepackedRel *entry = NULL; + RepackedRel *entry_toast = NULL; + Oid relid = repacked_rel; + Oid toastrelid = repacked_rel_toast; + + /* Remove the relation from the hash if we managed to insert one. */ + if (OidIsValid(repacked_rel)) + { + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + + memset(&key, 0, sizeof(key)); + key.relid = repacked_rel; + key.dbid = MyDatabaseId; + + entry = hash_search(repackedRelsHash, &key, HASH_REMOVE, NULL); + + /* Remove the TOAST relation if there is one. */ + if (OidIsValid(repacked_rel_toast)) + { + key.relid = repacked_rel_toast; + entry_toast = hash_search(repackedRelsHash, &key, HASH_REMOVE, + NULL); + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make others refresh their information whether they should still + * treat the table as catalog from the perspective of writing WAL. + * + * XXX Unlike entering the entry into the hashtable, we do not bother + * with locking and unlocking the table here: + * + * 1) On normal completion (and sometimes even on ERROR), the caller + * is already holding AccessExclusiveLock on the table, so there + * should be no relcache reference unaware of this change. + * + * 2) In the other cases, the worst scenario is that the other + * backends will write unnecessary information to WAL until they close + * the relation. + * + * Should we use ShareLock mode to fix 2) at least for the non-FATAL + * errors? (Our before_shmem_exit callback is in charge of FATAL, and + * that probably should not try to acquire any lock.) + */ + CacheInvalidateRelcacheImmediate(repacked_rel); + + /* + * By clearing repacked_rel we also disable + * cluster_before_shmem_exit_callback(). + */ + repacked_rel = InvalidOid; + repacked_rel_toast = InvalidOid; + } + /* * Restore normal function of (future) logical decoding for this backend. */ repacked_rel_locator.relNumber = InvalidOid; repacked_rel_toast_locator.relNumber = InvalidOid; + + /* + * On normal completion (!error), we should not really fail to remove the + * entry. But if it wasn't there for any reason, raise ERROR to make sure + * the transaction is aborted: if other transactions, while changing the + * contents of the relation, didn't know that REPACK CONCURRENTLY was in + * progress, they could have missed to WAL enough information, and thus we + * could have produced an inconsistent table contents. + * + * On the other hand, if we are already handling an error, there's no + * reason to worry about inconsistent contents of the new storage because + * the transaction is going to be rolled back anyway. Furthermore, by + * raising ERROR here we'd shadow the original error. + */ + if (!error) + { + char *relname; + + if (OidIsValid(relid) && entry == NULL) + { + relname = get_rel_name(relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + /* + * Likewise, the TOAST relation should not have disappeared. + */ + if (OidIsValid(toastrelid) && entry_toast == NULL) + { + relname = get_rel_name(key.relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + key.relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + } +} + +/* + * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + */ +static void +cluster_before_shmem_exit_callback(int code, Datum arg) +{ + if (OidIsValid(repacked_rel)) + end_concurrent_repack(true); +} + +/* + * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. + */ +bool +is_concurrent_repack_in_progress(Oid relid) +{ + RepackedRel key, + *entry; + + /* For particular relation we need to search in the hashtable. */ + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + LWLockAcquire(RepackedRelsLock, LW_SHARED); + /* + * If the caller is interested whether any relation is being repacked, + * just check the number of entries. + */ + if (!OidIsValid(relid)) + { + long n = hash_get_num_entries(repackedRelsHash); + + LWLockRelease(RepackedRelsLock); + return n > 0; + } + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_FIND, NULL); + LWLockRelease(RepackedRelsLock); + + return entry != NULL; +} + +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); } /* @@ -2282,7 +2645,7 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, + * table we are processing is present in repackedRelsHash and therefore, * regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index 1d56d0c4ef3..2cde79c9ed4 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -30,6 +30,7 @@ #include "access/xact.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -112,10 +113,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index e9ddf39500c..e24e1795aa9 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -151,6 +151,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); size = add_size(size, AioShmemSize()); + size = add_size(size, RepackShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -344,6 +345,7 @@ CreateOrAttachShmemStructs(void) WaitEventCustomShmemInit(); InjectionPointShmemInit(); AioShmemInit(); + RepackShmemInit(); } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 7fa8d9247e0..ab30d448d42 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1325,13 +1325,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c index 02505c88b8e..ecaa2283c2a 100644 --- a/src/backend/utils/cache/inval.c +++ b/src/backend/utils/cache/inval.c @@ -1643,6 +1643,27 @@ CacheInvalidateRelcache(Relation relation) databaseId, relationId); } +/* + * CacheInvalidateRelcacheImmediate + * Send invalidation message for the specified relation's relcache entry. + * + * Currently this is used in REPACK CONCURRENTLY, to make sure that other + * backends are aware that the command is being executed for the relation. + */ +void +CacheInvalidateRelcacheImmediate(Oid relid) +{ + SharedInvalidationMessage msg; + + msg.rc.id = SHAREDINVALRELCACHE_ID; + msg.rc.dbId = MyDatabaseId; + msg.rc.relId = relid; + /* check AddCatcacheInvalidationMessage() for an explanation */ + VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg)); + + SendSharedInvalidMessages(&msg, 1); +} + /* * CacheInvalidateRelcacheAll * Register invalidation of the whole relcache at the end of command. diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 4911642fb3c..504cb8e56a8 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -1279,6 +1279,10 @@ retry: /* make sure relation is marked as having no open file yet */ relation->rd_smgr = NULL; + /* Is REPACK CONCURRENTLY in progress? */ + relation->rd_repack_concurrent = + is_concurrent_repack_in_progress(targetRelId); + /* * now we can free the memory allocated for pg_class_tuple */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 4914f217267..9d5a30d0689 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -150,5 +150,10 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, MultiXactId cutoffMulti, char newrelpersistence); +extern Size RepackShmemSize(void); +extern void RepackShmemInit(void); +extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); + extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/inval.h b/src/include/utils/inval.h index 9b871caef62..ae9dee394dc 100644 --- a/src/include/utils/inval.h +++ b/src/include/utils/inval.h @@ -50,6 +50,8 @@ extern void CacheInvalidateCatalog(Oid catalogId); extern void CacheInvalidateRelcache(Relation relation); +extern void CacheInvalidateRelcacheImmediate(Oid relid); + extern void CacheInvalidateRelcacheAll(void); extern void CacheInvalidateRelcacheByTuple(HeapTuple classTuple); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index b552359915f..cc84592eb1f 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -253,6 +253,9 @@ typedef struct RelationData bool pgstat_enabled; /* should relation stats be counted */ /* use "struct" here to avoid needing to include pgstat.h: */ struct PgStat_TableStatus *pgstat_info; /* statistics collection area */ + + /* Is REPACK CONCURRENTLY being performed on this relation? */ + bool rd_repack_concurrent; } RelationData; @@ -708,12 +711,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 405d0811b4f..4f6c0ca3a8a 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index 0e3c47ba999..716e5619aa7 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -50,9 +50,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 814a0ba7b69..e3be1f42ccf 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2527,6 +2527,7 @@ ReorderBufferTupleCidKey ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId +RepackedRel RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.43.5 --=-=-=-- ^ permalink raw reply [nested|flat] 9+ messages in thread
* [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. @ 2025-06-30 17:41 Antonin Houska <[email protected]> 0 siblings, 0 replies; 9+ messages in thread From: Antonin Houska @ 2025-06-30 17:41 UTC (permalink / raw) As REPACK CONCURRENTLY uses logical decoding, it requires wal_level to be set to 'logical', while 'replica' is the default value. If logical replication is not used, users will probably be reluctant to set the GUC to 'logical' because it can affect server performance (by writing additional information to WAL) and because it cannot be changed to 'logical' only for the time REPACK CONCURRENTLY is running: change of this GUC requires server restart to take effect. This patch teaches postgres backend to recognize whether it should consider wal_level='logical' "locally" for particular transaction, even if the wal_level GUC is actually set to 'replica'. Also it ensures that the logical decoding specific information is added to WAL only for the tables which are currently being processed by REPACK CONCURRENTLY. If the logical decoding is enabled this way, only temporary replication slots should be created. The problem of permanent slot is that it is restored during server restart, and the restore fails if wal_level is not "globally" 'logical'. There is an independent work in progres to enable logical decoding transiently [1]. ISTM that this is too "heavyweight" solution for our problem. And I think that these two approaches are not mutually exclusive: once [1] is committed, we only need to adjust the XLogLogicalInfoActive() macro. [1] https://www.postgresql.org/message-id/CAD21AoCVLeLYq09pQPaWs%2BJwdni5FuJ8v2jgq-u9_uFbcp6UbA%40mail.g... --- doc/src/sgml/ref/repack.sgml | 7 - src/backend/access/transam/parallel.c | 8 + src/backend/access/transam/xact.c | 106 ++++- src/backend/access/transam/xlog.c | 1 + src/backend/commands/cluster.c | 387 +++++++++++++++++- src/backend/replication/logical/logical.c | 9 +- src/backend/storage/ipc/ipci.c | 2 + src/backend/storage/ipc/standby.c | 4 +- src/backend/utils/cache/inval.c | 21 + src/backend/utils/cache/relcache.c | 4 + src/include/access/xlog.h | 15 +- src/include/commands/cluster.h | 5 + src/include/utils/inval.h | 2 + src/include/utils/rel.h | 9 +- src/test/modules/injection_points/Makefile | 1 - .../modules/injection_points/logical.conf | 1 - src/test/modules/injection_points/meson.build | 3 - src/tools/pgindent/typedefs.list | 1 + 18 files changed, 540 insertions(+), 46 deletions(-) delete mode 100644 src/test/modules/injection_points/logical.conf diff --git a/doc/src/sgml/ref/repack.sgml b/doc/src/sgml/ref/repack.sgml index e1313f40599..0fd767eef98 100644 --- a/doc/src/sgml/ref/repack.sgml +++ b/doc/src/sgml/ref/repack.sgml @@ -260,13 +260,6 @@ REPACK [ ( <replaceable class="parameter">option</replaceable> [, ...] ) ] CONCU </para> </listitem> - <listitem> - <para> - The <link linkend="guc-wal-level"><varname>wal_level</varname></link> - configuration parameter is less than <literal>logical</literal>. - </para> - </listitem> - <listitem> <para> The <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link> diff --git a/src/backend/access/transam/parallel.c b/src/backend/access/transam/parallel.c index 94db1ec3012..a33318ea7bd 100644 --- a/src/backend/access/transam/parallel.c +++ b/src/backend/access/transam/parallel.c @@ -98,6 +98,7 @@ typedef struct FixedParallelState TimestampTz xact_ts; TimestampTz stmt_ts; SerializableXactHandle serializable_xact_handle; + int wal_level_transient; /* Mutex protects remaining fields. */ slock_t mutex; @@ -355,6 +356,7 @@ InitializeParallelDSM(ParallelContext *pcxt) fps->xact_ts = GetCurrentTransactionStartTimestamp(); fps->stmt_ts = GetCurrentStatementStartTimestamp(); fps->serializable_xact_handle = ShareSerializableXact(); + fps->wal_level_transient = wal_level_transient; SpinLockInit(&fps->mutex); fps->last_xlog_end = 0; shm_toc_insert(pcxt->toc, PARALLEL_KEY_FIXED, fps); @@ -1550,6 +1552,12 @@ ParallelWorkerMain(Datum main_arg) /* Attach to the leader's serializable transaction, if SERIALIZABLE. */ AttachSerializableXact(fps->serializable_xact_handle); + /* + * Restore the information whether this worker should behave as if + * wal_level was WAL_LEVEL_LOGICAL.. + */ + wal_level_transient = fps->wal_level_transient; + /* * We've initialized all of our state now; nothing should change * hereafter. diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 23f2de587a1..be568f70961 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -36,6 +36,7 @@ #include "catalog/pg_enum.h" #include "catalog/storage.h" #include "commands/async.h" +#include "commands/cluster.h" #include "commands/tablecmds.h" #include "commands/trigger.h" #include "common/pg_prng.h" @@ -126,6 +127,12 @@ static FullTransactionId XactTopFullTransactionId = {InvalidTransactionId}; static int nParallelCurrentXids = 0; static TransactionId *ParallelCurrentXids; +/* + * Have we determined the value of wal_level_transient for the current + * transaction? + */ +static bool wal_level_transient_checked = false; + /* * Miscellaneous flag bits to record events which occur on the top level * transaction. These flags are only persisted in MyXactFlags and are intended @@ -638,6 +645,7 @@ AssignTransactionId(TransactionState s) bool isSubXact = (s->parent != NULL); ResourceOwner currentOwner; bool log_unknown_top = false; + bool set_wal_level_transient = false; /* Assert that caller didn't screw up */ Assert(!FullTransactionIdIsValid(s->fullTransactionId)); @@ -652,6 +660,32 @@ AssignTransactionId(TransactionState s) (errcode(ERRCODE_INVALID_TRANSACTION_STATE), errmsg("cannot assign transaction IDs during a parallel operation"))); + /* + * The first call (i.e. the first write) in the transaction tree + * determines whether the whole transaction assumes logical decoding or + * not. + */ + if (!wal_level_transient_checked) + { + Assert(wal_level_transient == WAL_LEVEL_MINIMAL); + + /* + * Do not repeat the check when calling this function for parent + * transactions. + */ + wal_level_transient_checked = true; + + /* + * Remember that the actual check is needed. We cannot do it until the + * top-level transaction has its XID assigned, see comments below. + * + * There is no use case for overriding MINIMAL, and LOGICAL cannot be + * overridden as such. + */ + if (wal_level == WAL_LEVEL_REPLICA) + set_wal_level_transient = true; + } + /* * Ensure parent(s) have XIDs, so that a child always has an XID later * than its parent. Mustn't recurse here, or we might get a stack @@ -681,20 +715,6 @@ AssignTransactionId(TransactionState s) pfree(parents); } - /* - * When wal_level=logical, guarantee that a subtransaction's xid can only - * be seen in the WAL stream if its toplevel xid has been logged before. - * If necessary we log an xact_assignment record with fewer than - * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set - * for a transaction even though it appears in a WAL record, we just might - * superfluously log something. That can happen when an xid is included - * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in - * xl_standby_locks. - */ - if (isSubXact && XLogLogicalInfoActive() && - !TopTransactionStateData.didLogXid) - log_unknown_top = true; - /* * Generate a new FullTransactionId and record its xid in PGPROC and * pg_subtrans. @@ -719,6 +739,54 @@ AssignTransactionId(TransactionState s) if (!isSubXact) RegisterPredicateLockingXid(XidFromFullTransactionId(s->fullTransactionId)); + /* + * Check if this transaction should consider wal_level=logical. + * + * Sometimes we need to turn on the logical decoding transiently although + * wal_level=WAL_LEVEL_REPLICA. Currently we do so when at least one table + * is being clustered concurrently, i.e. when we should assume that + * changes done by this transaction will be decoded. In such a case we + * adjust the value of XLogLogicalInfoActive() by setting + * wal_level_transient to LOGICAL. + * + * It's important not to do this check until the XID of the top-level + * transaction is in ProcGlobal: if the decoding becomes mandatory right + * after the check, our transaction will fail to write the necessary + * information to WAL. However, if the top-level transaction is already in + * ProcGlobal, its XID is guaranteed to appear in the xl_running_xacts + * record and therefore the snapshot builder will not try to decode the + * transaction (because it assumes it could have missed the initial part + * of the transaction). + * + * On the other hand, if the decoding became mandatory between the actual + * XID assignment and now, the transaction will WAL the decoding specific + * information unnecessarily. Let's assume that such race conditions do + * not happen too often. + */ + if (set_wal_level_transient) + { + /* + * Check for the operation that enables the logical decoding + * transiently. + */ + if (is_concurrent_repack_in_progress(InvalidOid)) + wal_level_transient = WAL_LEVEL_LOGICAL; + } + + /* + * When wal_level=logical, guarantee that a subtransaction's xid can only + * be seen in the WAL stream if its toplevel xid has been logged before. + * If necessary we log an xact_assignment record with fewer than + * PGPROC_MAX_CACHED_SUBXIDS. Note that it is fine if didLogXid isn't set + * for a transaction even though it appears in a WAL record, we just might + * superfluously log something. That can happen when an xid is included + * somewhere inside a wal record, but not in XLogRecord->xl_xid, like in + * xl_standby_locks. + */ + if (isSubXact && XLogLogicalInfoActive() && + !TopTransactionStateData.didLogXid) + log_unknown_top = true; + /* * Acquire lock on the transaction XID. (We assume this cannot block.) We * have to ensure that the lock is assigned to the transaction's own @@ -2216,6 +2284,16 @@ StartTransaction(void) if (TransactionTimeout > 0) enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout); + /* + * wal_level_transient can override wal_level for individual transactions, + * which effectively enables logical decoding for them. At the moment we + * don't know if this transaction will write any data changes to be + * decoded. Should it do, AssignTransactionId() will check if the decoding + * needs to be considered. + */ + wal_level_transient = WAL_LEVEL_MINIMAL; + wal_level_transient_checked = false; + ShowTransactionState("StartTransaction"); } diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 47ffc0a2307..dc222db6a5d 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -129,6 +129,7 @@ bool wal_recycle = true; bool log_checkpoints = true; int wal_sync_method = DEFAULT_WAL_SYNC_METHOD; int wal_level = WAL_LEVEL_REPLICA; +int wal_level_transient = WAL_LEVEL_MINIMAL; int CommitDelay = 0; /* precommit delay in microseconds */ int CommitSiblings = 5; /* # concurrent xacts needed to sleep */ int wal_retrieve_retry_interval = 5000; diff --git a/src/backend/commands/cluster.c b/src/backend/commands/cluster.c index 37f69f369eb..7ecef2b86fc 100644 --- a/src/backend/commands/cluster.c +++ b/src/backend/commands/cluster.c @@ -84,6 +84,14 @@ typedef struct * The following definitions are used for concurrent processing. */ +/* + * OID of the table being repacked by this backend. + */ +static Oid repacked_rel = InvalidOid; + +/* The same for its TOAST relation. */ +static Oid repacked_rel_toast = InvalidOid; + /* * The locators are used to avoid logical decoding of data that we do not need * for our table. @@ -135,8 +143,10 @@ static List *get_tables_to_cluster_partitioned(MemoryContext cluster_context, ClusterCommand cmd); static bool cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd); -static void begin_concurrent_repack(Relation rel); -static void end_concurrent_repack(void); +static void begin_concurrent_repack(Relation rel, Relation *index_p, + bool *entered_p); +static void end_concurrent_repack(bool error); +static void cluster_before_shmem_exit_callback(int code, Datum arg); static LogicalDecodingContext *setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc); @@ -383,6 +393,8 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, Relation index; bool concurrent = ((params->options & CLUOPT_CONCURRENT) != 0); LOCKMODE lmode; + bool entered, + success; /* * Check that the correct lock is held. The lock mode is @@ -558,23 +570,31 @@ cluster_rel(Relation OldHeap, Oid indexOid, ClusterParams *params, TransferPredicateLocksToHeapRelation(OldHeap); /* rebuild_relation does all the dirty work */ + entered = false; + success = false; PG_TRY(); { /* - * For concurrent processing, make sure that our logical decoding - * ignores data changes of other tables than the one we are - * processing. + * For concurrent processing, make sure that + * + * 1) our logical decoding ignores data changes of other tables than + * the one we are processing. + * + * 2) other transactions know that REPACK CONCURRENTLY is in progress + * for our table, so they write sufficient information to WAL even if + * wal_level is < LOGICAL. */ if (concurrent) - begin_concurrent_repack(OldHeap); + begin_concurrent_repack(OldHeap, &index, &entered); rebuild_relation(OldHeap, index, verbose, concurrent, save_userid, cmd); + success = true; } PG_FINALLY(); { - if (concurrent) - end_concurrent_repack(); + if (concurrent && entered) + end_concurrent_repack(!success); } PG_END_TRY(); @@ -2208,6 +2228,49 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) #define REPL_PLUGIN_NAME "pgoutput_repack" +/* + * Each relation being processed by REPACK CONCURRENTLY must be in the + * repackedRelsHash hashtable. + */ +typedef struct RepackedRel +{ + Oid relid; + Oid dbid; +} RepackedRel; + +/* Hashtable of RepackedRel elements. */ +static HTAB *repackedRelsHash = NULL;; + +/* + * Maximum number of entries in the hashtable. + * + * A replication slot is needed for the processing, so use this GUC to + * allocate memory for the hashtable. Multiply by two because TOAST relations + * also need to be added to the hashtable. + */ +#define MAX_REPACKED_RELS (max_replication_slots * 2) + +Size +RepackShmemSize(void) +{ + return hash_estimate_size(MAX_REPACKED_RELS, sizeof(RepackedRel)); +} + +void +RepackShmemInit(void) +{ + HASHCTL info; + + info.keysize = sizeof(RepackedRel); + info.entrysize = info.keysize; + repackedRelsHash = ShmemInitHash("Repacked Relations Hash", + MAX_REPACKED_RELS, + MAX_REPACKED_RELS, + &info, + HASH_ELEM | HASH_BLOBS | + HASH_FIXED_SIZE); +} + /* * Call this function before REPACK CONCURRENTLY starts to setup logical * decoding. It makes sure that other users of the table put enough @@ -2222,11 +2285,150 @@ cluster_is_permitted_for_relation(Oid relid, Oid userid, ClusterCommand cmd) * * Note that TOAST table needs no attention here as it's not scanned using * historic snapshot. + * + * 'index_p' is in/out argument because the function unlocks the index + * temporarily. + * + * 'enter_p' receives a bool value telling whether relation OID was entered + * into repackedRelsHash or not. */ static void -begin_concurrent_repack(Relation rel) +begin_concurrent_repack(Relation rel, Relation *index_p, bool *entered_p) { - Oid toastrelid; + Oid relid, + toastrelid; + Relation index = NULL; + Oid indexid = InvalidOid; + RepackedRel key, + *entry; + bool found; + static bool before_shmem_exit_callback_setup = false; + + relid = RelationGetRelid(rel); + index = index_p ? *index_p : NULL; + + /* + * Make sure that we do not leave an entry in repackedRelsHash if exiting + * due to FATAL. + */ + if (!before_shmem_exit_callback_setup) + { + before_shmem_exit(cluster_before_shmem_exit_callback, 0); + before_shmem_exit_callback_setup = true; + } + + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + *entered_p = false; + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + { + /* + * Since REPACK CONCURRENTLY takes ShareRowExclusiveLock, a conflict + * should occur much earlier. However that lock may be released + * temporarily, see below. Anyway, we should complain whatever the + * reason of the conflict might be. + */ + ereport(ERROR, + (errmsg("relation \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + } + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + /* + * Even if the insertion of TOAST relid should fail below, the caller has + * to do cleanup. + */ + *entered_p = true; + + /* + * Enable the callback to remove the entry in case of exit. We should not + * do this earlier, otherwise an attempt to insert already existing entry + * could make us remove that entry (inserted by another backend) during + * ERROR handling. + */ + Assert(!OidIsValid(repacked_rel)); + repacked_rel = relid; + + /* + * TOAST relation is not accessed using historic snapshot, but we enter it + * here to protect it from being VACUUMed by another backend. (Lock does + * not help in the CONCURRENTLY case because cannot hold it continuously + * till the end of the transaction.) See the comments on locking TOAST + * relation in copy_table_data(). + */ + toastrelid = rel->rd_rel->reltoastrelid; + if (OidIsValid(toastrelid)) + { + key.relid = toastrelid; + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_ENTER_NULL, &found); + if (found) + + /* + * If we could enter the main fork the TOAST should succeed too. + * Nevertheless, check. + */ + ereport(ERROR, + (errmsg("TOAST relation of \"%s\" is already being processed by REPACK CONCURRENTLY", + RelationGetRelationName(rel)))); + if (entry == NULL) + ereport(ERROR, + (errmsg("too many requests for REPACK CONCURRENTLY at a time")), + (errhint("Please consider increasing the \"max_replication_slots\" configuration parameter."))); + + Assert(!OidIsValid(repacked_rel_toast)); + repacked_rel_toast = toastrelid; + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make sure that other backends are aware of the new hash entry as soon + * as they open our table. + */ + CacheInvalidateRelcacheImmediate(relid); + + /* + * Also make sure that the existing users of the table update their + * relcache entry as soon as they try to run DML commands on it. + * + * ShareLock is the weakest lock that conflicts with DMLs. If any backend + * has a lower lock, we assume it'll accept our invalidation message when + * it changes the lock mode. + * + * Before upgrading the lock on the relation, close the index temporarily + * to avoid a deadlock if another backend running DML already has its lock + * (ShareLock) on the table and waits for the lock on the index. + */ + if (index) + { + indexid = RelationGetRelid(index); + index_close(index, ShareUpdateExclusiveLock); + } + LockRelationOid(relid, ShareLock); + UnlockRelationOid(relid, ShareLock); + if (OidIsValid(indexid)) + { + /* + * Re-open the index and check that it hasn't changed while unlocked. + */ + check_index_is_clusterable(rel, indexid, ShareUpdateExclusiveLock); + + /* + * Return the new relcache entry to the caller. (It's been locked by + * the call above.) + */ + index = index_open(indexid, NoLock); + *index_p = index; + } /* Avoid logical decoding of other relations by this backend. */ repacked_rel_locator = rel->rd_locator; @@ -2244,15 +2446,176 @@ begin_concurrent_repack(Relation rel) /* * Call this when done with REPACK CONCURRENTLY. + * + * 'error' tells whether the function is being called in order to handle + * error. */ static void -end_concurrent_repack(void) +end_concurrent_repack(bool error) { + RepackedRel key; + RepackedRel *entry = NULL; + RepackedRel *entry_toast = NULL; + Oid relid = repacked_rel; + Oid toastrelid = repacked_rel_toast; + + /* Remove the relation from the hash if we managed to insert one. */ + if (OidIsValid(repacked_rel)) + { + LWLockAcquire(RepackedRelsLock, LW_EXCLUSIVE); + + memset(&key, 0, sizeof(key)); + key.relid = repacked_rel; + key.dbid = MyDatabaseId; + + entry = hash_search(repackedRelsHash, &key, HASH_REMOVE, NULL); + + /* Remove the TOAST relation if there is one. */ + if (OidIsValid(repacked_rel_toast)) + { + key.relid = repacked_rel_toast; + entry_toast = hash_search(repackedRelsHash, &key, HASH_REMOVE, + NULL); + } + + LWLockRelease(RepackedRelsLock); + + /* + * Make others refresh their information whether they should still + * treat the table as catalog from the perspective of writing WAL. + * + * XXX Unlike entering the entry into the hashtable, we do not bother + * with locking and unlocking the table here: + * + * 1) On normal completion (and sometimes even on ERROR), the caller + * is already holding AccessExclusiveLock on the table, so there + * should be no relcache reference unaware of this change. + * + * 2) In the other cases, the worst scenario is that the other + * backends will write unnecessary information to WAL until they close + * the relation. + * + * Should we use ShareLock mode to fix 2) at least for the non-FATAL + * errors? (Our before_shmem_exit callback is in charge of FATAL, and + * that probably should not try to acquire any lock.) + */ + CacheInvalidateRelcacheImmediate(repacked_rel); + + /* + * By clearing repacked_rel we also disable + * cluster_before_shmem_exit_callback(). + */ + repacked_rel = InvalidOid; + repacked_rel_toast = InvalidOid; + } + /* * Restore normal function of (future) logical decoding for this backend. */ repacked_rel_locator.relNumber = InvalidOid; repacked_rel_toast_locator.relNumber = InvalidOid; + + /* + * On normal completion (!error), we should not really fail to remove the + * entry. But if it wasn't there for any reason, raise ERROR to make sure + * the transaction is aborted: if other transactions, while changing the + * contents of the relation, didn't know that REPACK CONCURRENTLY was in + * progress, they could have missed to WAL enough information, and thus we + * could have produced an inconsistent table contents. + * + * On the other hand, if we are already handling an error, there's no + * reason to worry about inconsistent contents of the new storage because + * the transaction is going to be rolled back anyway. Furthermore, by + * raising ERROR here we'd shadow the original error. + */ + if (!error) + { + char *relname; + + if (OidIsValid(relid) && entry == NULL) + { + relname = get_rel_name(relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + /* + * Likewise, the TOAST relation should not have disappeared. + */ + if (OidIsValid(toastrelid) && entry_toast == NULL) + { + relname = get_rel_name(key.relid); + if (!relname) + ereport(ERROR, + (errmsg("cache lookup failed for relation %u", + key.relid))); + + ereport(ERROR, + (errmsg("relation \"%s\" not found among repacked relations", + relname))); + } + + } +} + +/* + * A wrapper to call end_concurrent_repack() as a before_shmem_exit callback. + */ +static void +cluster_before_shmem_exit_callback(int code, Datum arg) +{ + if (OidIsValid(repacked_rel)) + end_concurrent_repack(true); +} + +/* + * Check if relation is currently being processed by REPACK CONCURRENTLY. + * + * If relid is InvalidOid, check if any relation is being processed. + */ +bool +is_concurrent_repack_in_progress(Oid relid) +{ + RepackedRel key, + *entry; + + /* For particular relation we need to search in the hashtable. */ + memset(&key, 0, sizeof(key)); + key.relid = relid; + key.dbid = MyDatabaseId; + + LWLockAcquire(RepackedRelsLock, LW_SHARED); + /* + * If the caller is interested whether any relation is being repacked, + * just check the number of entries. + */ + if (!OidIsValid(relid)) + { + long n = hash_get_num_entries(repackedRelsHash); + + LWLockRelease(RepackedRelsLock); + return n > 0; + } + entry = (RepackedRel *) + hash_search(repackedRelsHash, &key, HASH_FIND, NULL); + LWLockRelease(RepackedRelsLock); + + return entry != NULL; +} + +/* + * Is this backend performing REPACK CONCURRENTLY? + */ +bool +is_concurrent_repack_run_by_me(void) +{ + return OidIsValid(repacked_rel); } /* @@ -2282,7 +2645,7 @@ setup_logical_decoding(Oid relid, const char *slotname, TupleDesc tupdesc) * useful for us. * * Regarding the value of need_full_snapshot, we pass false because the - * table we are processing is present in RepackedRelsHash and therefore, + * table we are processing is present in repackedRelsHash and therefore, * regarding logical decoding, treated like a catalog. */ ctx = CreateInitDecodingContext(REPL_PLUGIN_NAME, diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c index f1eb798f3e9..5e6000db086 100644 --- a/src/backend/replication/logical/logical.c +++ b/src/backend/replication/logical/logical.c @@ -31,6 +31,7 @@ #include "access/xact.h" #include "access/xlog_internal.h" #include "access/xlogutils.h" +#include "commands/cluster.h" #include "fmgr.h" #include "miscadmin.h" #include "pgstat.h" @@ -114,10 +115,12 @@ CheckLogicalDecodingRequirements(void) /* * NB: Adding a new requirement likely means that RestoreSlotFromDisk() - * needs the same check. + * needs the same check. (Except that only temporary slots should be + * created for REPACK CONCURRENTLY, which effectively raises wal_level to + * LOGICAL.) */ - - if (wal_level < WAL_LEVEL_LOGICAL) + if ((wal_level < WAL_LEVEL_LOGICAL && !is_concurrent_repack_run_by_me()) + || wal_level < WAL_LEVEL_REPLICA) ereport(ERROR, (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), errmsg("logical decoding requires \"wal_level\" >= \"logical\""))); diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index e9ddf39500c..e24e1795aa9 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -151,6 +151,7 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, InjectionPointShmemSize()); size = add_size(size, SlotSyncShmemSize()); size = add_size(size, AioShmemSize()); + size = add_size(size, RepackShmemSize()); /* include additional requested shmem from preload libraries */ size = add_size(size, total_addin_request); @@ -344,6 +345,7 @@ CreateOrAttachShmemStructs(void) WaitEventCustomShmemInit(); InjectionPointShmemInit(); AioShmemInit(); + RepackShmemInit(); } /* diff --git a/src/backend/storage/ipc/standby.c b/src/backend/storage/ipc/standby.c index 7fa8d9247e0..ab30d448d42 100644 --- a/src/backend/storage/ipc/standby.c +++ b/src/backend/storage/ipc/standby.c @@ -1325,13 +1325,13 @@ LogStandbySnapshot(void) * record. Fortunately this routine isn't executed frequently, and it's * only a shared lock. */ - if (wal_level < WAL_LEVEL_LOGICAL) + if (!XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); recptr = LogCurrentRunningXacts(running); /* Release lock if we kept it longer ... */ - if (wal_level >= WAL_LEVEL_LOGICAL) + if (XLogLogicalInfoActive()) LWLockRelease(ProcArrayLock); /* GetRunningTransactionData() acquired XidGenLock, we must release it */ diff --git a/src/backend/utils/cache/inval.c b/src/backend/utils/cache/inval.c index 02505c88b8e..ecaa2283c2a 100644 --- a/src/backend/utils/cache/inval.c +++ b/src/backend/utils/cache/inval.c @@ -1643,6 +1643,27 @@ CacheInvalidateRelcache(Relation relation) databaseId, relationId); } +/* + * CacheInvalidateRelcacheImmediate + * Send invalidation message for the specified relation's relcache entry. + * + * Currently this is used in REPACK CONCURRENTLY, to make sure that other + * backends are aware that the command is being executed for the relation. + */ +void +CacheInvalidateRelcacheImmediate(Oid relid) +{ + SharedInvalidationMessage msg; + + msg.rc.id = SHAREDINVALRELCACHE_ID; + msg.rc.dbId = MyDatabaseId; + msg.rc.relId = relid; + /* check AddCatcacheInvalidationMessage() for an explanation */ + VALGRIND_MAKE_MEM_DEFINED(&msg, sizeof(msg)); + + SendSharedInvalidMessages(&msg, 1); +} + /* * CacheInvalidateRelcacheAll * Register invalidation of the whole relcache at the end of command. diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c index 4911642fb3c..504cb8e56a8 100644 --- a/src/backend/utils/cache/relcache.c +++ b/src/backend/utils/cache/relcache.c @@ -1279,6 +1279,10 @@ retry: /* make sure relation is marked as having no open file yet */ relation->rd_smgr = NULL; + /* Is REPACK CONCURRENTLY in progress? */ + relation->rd_repack_concurrent = + is_concurrent_repack_in_progress(targetRelId); + /* * now we can free the memory allocated for pg_class_tuple */ diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h index d313099c027..a325bb1d16b 100644 --- a/src/include/access/xlog.h +++ b/src/include/access/xlog.h @@ -95,6 +95,12 @@ typedef enum RecoveryState extern PGDLLIMPORT int wal_level; +/* + * wal_level_transient overrides wal_level if logical decoding needs to be + * enabled transiently. + */ +extern PGDLLIMPORT int wal_level_transient; + /* Is WAL archiving enabled (always or only while server is running normally)? */ #define XLogArchivingActive() \ (AssertMacro(XLogArchiveMode == ARCHIVE_MODE_OFF || wal_level >= WAL_LEVEL_REPLICA), XLogArchiveMode > ARCHIVE_MODE_OFF) @@ -122,8 +128,13 @@ extern PGDLLIMPORT int wal_level; /* Do we need to WAL-log information required only for Hot Standby and logical replication? */ #define XLogStandbyInfoActive() (wal_level >= WAL_LEVEL_REPLICA) -/* Do we need to WAL-log information required only for logical replication? */ -#define XLogLogicalInfoActive() (wal_level >= WAL_LEVEL_LOGICAL) +/* + * Do we need to WAL-log information required only for logical replication? + * + * wal_level_transient overrides wal_level if logical decoding needs to be + * active transiently. + */ +#define XLogLogicalInfoActive() (Max(wal_level, wal_level_transient) == WAL_LEVEL_LOGICAL) #ifdef WAL_DEBUG extern PGDLLIMPORT bool XLOG_DEBUG; diff --git a/src/include/commands/cluster.h b/src/include/commands/cluster.h index 4914f217267..9d5a30d0689 100644 --- a/src/include/commands/cluster.h +++ b/src/include/commands/cluster.h @@ -150,5 +150,10 @@ extern void finish_heap_swap(Oid OIDOldHeap, Oid OIDNewHeap, MultiXactId cutoffMulti, char newrelpersistence); +extern Size RepackShmemSize(void); +extern void RepackShmemInit(void); +extern bool is_concurrent_repack_in_progress(Oid relid); +extern bool is_concurrent_repack_run_by_me(void); + extern void repack(ParseState *pstate, RepackStmt *stmt, bool isTopLevel); #endif /* CLUSTER_H */ diff --git a/src/include/utils/inval.h b/src/include/utils/inval.h index 9b871caef62..ae9dee394dc 100644 --- a/src/include/utils/inval.h +++ b/src/include/utils/inval.h @@ -50,6 +50,8 @@ extern void CacheInvalidateCatalog(Oid catalogId); extern void CacheInvalidateRelcache(Relation relation); +extern void CacheInvalidateRelcacheImmediate(Oid relid); + extern void CacheInvalidateRelcacheAll(void); extern void CacheInvalidateRelcacheByTuple(HeapTuple classTuple); diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h index b552359915f..cc84592eb1f 100644 --- a/src/include/utils/rel.h +++ b/src/include/utils/rel.h @@ -253,6 +253,9 @@ typedef struct RelationData bool pgstat_enabled; /* should relation stats be counted */ /* use "struct" here to avoid needing to include pgstat.h: */ struct PgStat_TableStatus *pgstat_info; /* statistics collection area */ + + /* Is REPACK CONCURRENTLY being performed on this relation? */ + bool rd_repack_concurrent; } RelationData; @@ -708,12 +711,16 @@ RelationCloseSmgr(Relation relation) * it would complicate decoding slightly for little gain). Note that we *do* * log information for user defined catalog tables since they presumably are * interesting to the user... + * + * If particular relations require that, the logical decoding can be active + * even if wal_level is REPLICA. Do not log other relations in that case. */ #define RelationIsLogicallyLogged(relation) \ (XLogLogicalInfoActive() && \ RelationNeedsWAL(relation) && \ (relation)->rd_rel->relkind != RELKIND_FOREIGN_TABLE && \ - !IsCatalogRelation(relation)) + !IsCatalogRelation(relation) && \ + (wal_level == WAL_LEVEL_LOGICAL || (relation)->rd_repack_concurrent)) /* routines in utils/cache/relcache.c */ extern void RelationIncrementReferenceCount(Relation rel); diff --git a/src/test/modules/injection_points/Makefile b/src/test/modules/injection_points/Makefile index 30ffe509239..e71b8a19116 100644 --- a/src/test/modules/injection_points/Makefile +++ b/src/test/modules/injection_points/Makefile @@ -15,7 +15,6 @@ REGRESS = injection_points hashagg reindex_conc vacuum REGRESS_OPTS = --dlpath=$(top_builddir)/src/test/regress ISOLATION = basic inplace syscache-update-pruned repack -ISOLATION_OPTS = --temp-config $(top_srcdir)/src/test/modules/injection_points/logical.conf TAP_TESTS = 1 diff --git a/src/test/modules/injection_points/logical.conf b/src/test/modules/injection_points/logical.conf deleted file mode 100644 index c8f264bc6cb..00000000000 --- a/src/test/modules/injection_points/logical.conf +++ /dev/null @@ -1 +0,0 @@ -wal_level = logical \ No newline at end of file diff --git a/src/test/modules/injection_points/meson.build b/src/test/modules/injection_points/meson.build index c7daa669548..13c2b627a0b 100644 --- a/src/test/modules/injection_points/meson.build +++ b/src/test/modules/injection_points/meson.build @@ -51,9 +51,6 @@ tests += { 'syscache-update-pruned', ], 'runningcheck': false, # see syscache-update-pruned - # 'repack' requires wal_level = 'logical'. - 'regress_args': ['--temp-config', files('logical.conf')], - }, 'tap': { 'env': { diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 879977ea41f..add58883124 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -2532,6 +2532,7 @@ ReorderBufferTupleCidKey ReorderBufferUpdateProgressTxnCB ReorderTuple RepOriginId +RepackedRel RepackDecodingState RepackStmt ReparameterizeForeignPathByChild_function -- 2.47.1 --=-=-=-- ^ permalink raw reply [nested|flat] 9+ messages in thread
end of thread, other threads:[~2025-06-30 17:41 UTC | newest] Thread overview: 9+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2021-03-08 21:47 [PATCH v3 04/17] meson: prereq: output and depencency tracking work. Andres Freund <[email protected]> 2025-02-26 08:17 [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-02-26 08:17 [PATCH v09 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-03-24 19:17 [PATCH v10 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Álvaro Herrera <[email protected]> 2025-03-31 13:47 [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-04-01 11:48 [PATCH 8/9] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-04-11 09:13 [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-06-09 10:00 [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]> 2025-06-30 17:41 [PATCH 7/7] Enable logical decoding transiently, only for REPACK CONCURRENTLY. Antonin Houska <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox