public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v24 1/2] In-place table persistence change 8+ messages / 3 participants [nested] [flat]
* [PATCH v24 1/2] In-place table persistence change @ 2022-07-19 04:23 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 8+ messages in thread From: Kyotaro Horiguchi @ 2022-07-19 04:23 UTC (permalink / raw) Even though ALTER TABLE SET LOGGED/UNLOGGED does not require data rewriting, currently it runs heap rewrite which causes large amount of file I/O. This patch makes the command run without heap rewrite. Addition to that, SET LOGGED while wal_level > minimal emits WAL using XLOG_FPI instead of massive number of HEAP_INSERT's, which should be smaller. Also this allows for the cleanup of files left behind in the crash of the transaction that created it. --- src/backend/access/rmgrdesc/smgrdesc.c | 49 ++ src/backend/access/transam/README | 10 + src/backend/access/transam/xact.c | 7 + src/backend/access/transam/xlogrecovery.c | 18 + src/backend/backup/basebackup.c | 9 +- src/backend/catalog/storage.c | 561 +++++++++++++++++++++- src/backend/commands/tablecmds.c | 267 ++++++++-- src/backend/storage/buffer/bufmgr.c | 85 ++++ src/backend/storage/file/fd.c | 4 +- src/backend/storage/file/reinit.c | 313 ++++++++---- src/backend/storage/smgr/md.c | 95 +++- src/backend/storage/smgr/smgr.c | 32 ++ src/backend/storage/sync/sync.c | 21 +- src/bin/pg_rewind/parsexlog.c | 22 + src/bin/pg_rewind/pg_rewind.c | 1 - src/common/relpath.c | 47 +- src/include/catalog/storage.h | 3 + src/include/catalog/storage_xlog.h | 43 +- src/include/common/relpath.h | 9 +- src/include/storage/bufmgr.h | 2 + src/include/storage/fd.h | 1 + src/include/storage/md.h | 8 +- src/include/storage/reinit.h | 8 +- src/include/storage/smgr.h | 17 + src/tools/pgindent/typedefs.list | 7 + 25 files changed, 1471 insertions(+), 168 deletions(-) diff --git a/src/backend/access/rmgrdesc/smgrdesc.c b/src/backend/access/rmgrdesc/smgrdesc.c index e0ee8a078a..2f92c06f70 100644 --- a/src/backend/access/rmgrdesc/smgrdesc.c +++ b/src/backend/access/rmgrdesc/smgrdesc.c @@ -40,6 +40,46 @@ smgr_desc(StringInfo buf, XLogReaderState *record) xlrec->blkno, xlrec->flags); pfree(path); } + else if (info == XLOG_SMGR_UNLINK) + { + xl_smgr_unlink *xlrec = (xl_smgr_unlink *) rec; + char *path = relpathperm(xlrec->rlocator, xlrec->forkNum); + + appendStringInfoString(buf, path); + pfree(path); + } + else if (info == XLOG_SMGR_MARK) + { + xl_smgr_mark *xlrec = (xl_smgr_mark *) rec; + char *path = GetRelationPath(xlrec->rlocator.dbOid, + xlrec->rlocator.spcOid, + xlrec->rlocator.relNumber, + InvalidBackendId, + xlrec->forkNum, xlrec->mark); + char *action = "<none>"; + + switch (xlrec->action) + { + case XLOG_SMGR_MARK_CREATE: + action = "CREATE"; + break; + case XLOG_SMGR_MARK_UNLINK: + action = "DELETE"; + break; + } + + appendStringInfo(buf, "%s %s", action, path); + pfree(path); + } + else if (info == XLOG_SMGR_BUFPERSISTENCE) + { + xl_smgr_bufpersistence *xlrec = (xl_smgr_bufpersistence *) rec; + char *path = relpathperm(xlrec->rlocator, MAIN_FORKNUM); + + appendStringInfoString(buf, path); + appendStringInfo(buf, " persistence %d", xlrec->persistence); + pfree(path); + } } const char * @@ -55,6 +95,15 @@ smgr_identify(uint8 info) case XLOG_SMGR_TRUNCATE: id = "TRUNCATE"; break; + case XLOG_SMGR_UNLINK: + id = "UNLINK"; + break; + case XLOG_SMGR_MARK: + id = "MARK"; + break; + case XLOG_SMGR_BUFPERSISTENCE: + id = "BUFPERSISTENCE"; + break; } return id; diff --git a/src/backend/access/transam/README b/src/backend/access/transam/README index 91c2578f7a..37527e16ca 100644 --- a/src/backend/access/transam/README +++ b/src/backend/access/transam/README @@ -725,6 +725,16 @@ we must panic and abort recovery. The DBA will have to manually clean up then restart recovery. This is part of the reason for not writing a WAL entry until we've successfully done the original action. +================================ +The Smgr MARK files +-------------------------------- + +An smgr mark file is an empty file that is created when a new relation +storage file is created to signal that the storage file needs to be +cleaned up at recovery time. In contrast to the four actions above, +failure to remove smgr mark files will lead to data loss, in which +case the server will shut down. + Skipping WAL for New RelFileLocator -------------------------------- diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 2bb975943c..03e4bcec34 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -2213,6 +2213,9 @@ CommitTransaction(void) */ smgrDoPendingSyncs(true, is_parallel_worker); + /* Likewise delete mark files for files created during this transaction. */ + smgrDoPendingCleanups(true); + /* close large objects before lower-level cleanup */ AtEOXact_LargeObject(true); @@ -2463,6 +2466,9 @@ PrepareTransaction(void) */ smgrDoPendingSyncs(true, false); + /* Likewise delete mark files for files created during this transaction. */ + smgrDoPendingCleanups(true); + /* close large objects before lower-level cleanup */ AtEOXact_LargeObject(true); @@ -2788,6 +2794,7 @@ AbortTransaction(void) AfterTriggerEndXact(false); /* 'false' means it's abort */ AtAbort_Portals(); smgrDoPendingSyncs(false, is_parallel_worker); + smgrDoPendingCleanups(false); AtEOXact_LargeObject(false); AtAbort_Notify(); AtEOXact_RelationMap(false, is_parallel_worker); diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c index 1026ce5dcf..aac303934c 100644 --- a/src/backend/access/transam/xlogrecovery.c +++ b/src/backend/access/transam/xlogrecovery.c @@ -42,6 +42,7 @@ #include "access/xlogutils.h" #include "backup/basebackup.h" #include "catalog/pg_control.h" +#include "catalog/storage.h" #include "commands/tablespace.h" #include "common/file_utils.h" #include "miscadmin.h" @@ -56,6 +57,7 @@ #include "storage/pmsignal.h" #include "storage/proc.h" #include "storage/procarray.h" +#include "storage/reinit.h" #include "storage/spin.h" #include "utils/builtins.h" #include "utils/datetime.h" @@ -1777,6 +1779,14 @@ PerformWalRecovery(void) RmgrCleanup(); + /* cleanup garbage files left during crash recovery */ + if (!InArchiveRecovery) + ResetUnloggedRelations(UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_CLEANUP); + + /* run rollback cleanup if any */ + smgrDoPendingDeletes(false); + ereport(LOG, (errmsg("redo done at %X/%X system usage: %s", LSN_FORMAT_ARGS(xlogreader->ReadRecPtr), @@ -3113,6 +3123,14 @@ ReadRecord(XLogPrefetcher *xlogprefetcher, int emode, { ereport(DEBUG1, (errmsg_internal("reached end of WAL in pg_wal, entering archive recovery"))); + + /* cleanup garbage files left during crash recovery */ + ResetUnloggedRelations(UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_CLEANUP); + + /* run rollback cleanup if any */ + smgrDoPendingDeletes(false); + InArchiveRecovery = true; if (StandbyModeRequested) StandbyMode = true; diff --git a/src/backend/backup/basebackup.c b/src/backend/backup/basebackup.c index 1434bcdd85..4e0c6c4e98 100644 --- a/src/backend/backup/basebackup.c +++ b/src/backend/backup/basebackup.c @@ -1190,6 +1190,7 @@ sendDir(bbsink *sink, const char *path, int basepathlen, bool sizeonly, ForkNumber relForkNum; /* Type of fork if file is a relation */ int relnumchars; /* Chars in filename that are the * relnumber */ + StorageMarks mark; /* marker file sign */ /* Skip special stuff */ if (strcmp(de->d_name, ".") == 0 || strcmp(de->d_name, "..") == 0) @@ -1240,7 +1241,7 @@ sendDir(bbsink *sink, const char *path, int basepathlen, bool sizeonly, /* Exclude all forks for unlogged tables except the init fork */ if (isDbDir && parse_filename_for_nontemp_relation(de->d_name, &relnumchars, - &relForkNum)) + &relForkNum, &mark)) { /* Never exclude init forks */ if (relForkNum != INIT_FORKNUM) @@ -1447,6 +1448,7 @@ is_checksummed_file(const char *fullpath, const char *filename) strncmp(fullpath, "/", 1) == 0) { int excludeIdx; + char *p; /* Compare file against noChecksumFiles skip list */ for (excludeIdx = 0; noChecksumFiles[excludeIdx].name != NULL; excludeIdx++) @@ -1460,6 +1462,11 @@ is_checksummed_file(const char *fullpath, const char *filename) return false; } + /* exclude mark files */ + p = strchr(filename, '.'); + if (p && isalpha(p[1]) && p[2] == 0) + return false; + return true; } else diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c index 38bbe32550..2c6472cfd5 100644 --- a/src/backend/catalog/storage.c +++ b/src/backend/catalog/storage.c @@ -19,6 +19,7 @@ #include "postgres.h" +#include "access/amapi.h" #include "access/parallel.h" #include "access/visibilitymap.h" #include "access/xact.h" @@ -66,6 +67,23 @@ typedef struct PendingRelDelete struct PendingRelDelete *next; /* linked-list link */ } PendingRelDelete; +#define PCOP_UNLINK_FORK (1 << 0) +#define PCOP_UNLINK_MARK (1 << 1) +#define PCOP_SET_PERSISTENCE (1 << 2) + +typedef struct PendingCleanup +{ + RelFileLocator rlocator; /* relation that need a cleanup */ + int op; /* operation mask */ + bool bufpersistence; /* buffer persistence to set */ + ForkNumber unlink_forknum; /* forknum to unlink */ + StorageMarks unlink_mark; /* mark to unlink */ + BackendId backend; /* InvalidBackendId if not a temp rel */ + bool atCommit; /* T=delete at commit; F=delete at abort */ + int nestLevel; /* xact nesting level of request */ + struct PendingCleanup *next; /* linked-list link */ +} PendingCleanup; + typedef struct PendingRelSync { RelFileLocator rlocator; @@ -73,6 +91,7 @@ typedef struct PendingRelSync } PendingRelSync; static PendingRelDelete *pendingDeletes = NULL; /* head of linked list */ +static PendingCleanup * pendingCleanups = NULL; /* head of linked list */ static HTAB *pendingSyncHash = NULL; @@ -123,6 +142,7 @@ RelationCreateStorage(RelFileLocator rlocator, char relpersistence, SMgrRelation srel; BackendId backend; bool needs_wal; + PendingCleanup *pendingclean; Assert(!IsInParallelMode()); /* couldn't update pendingSyncHash */ @@ -145,9 +165,23 @@ RelationCreateStorage(RelFileLocator rlocator, char relpersistence, return NULL; /* placate compiler */ } + /* + * We are going to create a new storage file. If server crashes before the + * current transaction ends the file needs to be cleaned up. The + * SMGR_MARK_UNCOMMITED mark file prompts that work at the next startup. + * We don't need this during WAL-loggged CREATE DATABASE. See + * CreateAndCopyRelationData for detail. + */ srel = smgropen(rlocator, backend); + + if (register_delete) + { + log_smgrcreatemark(&rlocator, MAIN_FORKNUM, SMGR_MARK_UNCOMMITTED); + smgrcreatemark(srel, MAIN_FORKNUM, SMGR_MARK_UNCOMMITTED, false); + } + smgrcreate(srel, MAIN_FORKNUM, false); - + if (needs_wal) log_smgrcreate(&srel->smgr_rlocator.locator, MAIN_FORKNUM); @@ -157,16 +191,29 @@ RelationCreateStorage(RelFileLocator rlocator, char relpersistence, */ if (register_delete) { - PendingRelDelete *pending; + PendingRelDelete *pendingdel; - pending = (PendingRelDelete *) + pendingdel = (PendingRelDelete *) MemoryContextAlloc(TopMemoryContext, sizeof(PendingRelDelete)); - pending->rlocator = rlocator; - pending->backend = backend; - pending->atCommit = false; /* delete if abort */ - pending->nestLevel = GetCurrentTransactionNestLevel(); - pending->next = pendingDeletes; - pendingDeletes = pending; + pendingdel->rlocator = rlocator; + pendingdel->backend = backend; + pendingdel->atCommit = false; /* delete if abort */ + pendingdel->nestLevel = GetCurrentTransactionNestLevel(); + pendingdel->next = pendingDeletes; + pendingDeletes = pendingdel; + + /* drop mark files at commit */ + pendingclean = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pendingclean->rlocator = rlocator; + pendingclean->op = PCOP_UNLINK_MARK; + pendingclean->unlink_forknum = MAIN_FORKNUM; + pendingclean->unlink_mark = SMGR_MARK_UNCOMMITTED; + pendingclean->backend = backend; + pendingclean->atCommit = true; + pendingclean->nestLevel = GetCurrentTransactionNestLevel(); + pendingclean->next = pendingCleanups; + pendingCleanups = pendingclean; } if (relpersistence == RELPERSISTENCE_PERMANENT && !XLogIsNeeded()) @@ -178,6 +225,204 @@ RelationCreateStorage(RelFileLocator rlocator, char relpersistence, return srel; } +/* + * RelationCreateInitFork + * Create physical storage for the init fork of a relation. + * + * Create the init fork for the relation. + * + * This function is transactional. The creation is WAL-logged, and if the + * transaction aborts later on, the init fork will be removed. + */ +void +RelationCreateInitFork(Relation rel) +{ + RelFileLocator rlocator = rel->rd_locator; + PendingCleanup *pending; + PendingCleanup *prev; + PendingCleanup *next; + SMgrRelation srel; + bool create = true; + + /* switch buffer persistence */ + SetRelationBuffersPersistence(RelationGetSmgr(rel), false, false); + + /* + * If we have a pending-unlink for the init-fork of this relation, that + * means the init-fork exists since before the current transaction + * started. This function reverts that change just by removing the entry. + * See RelationDropInitFork. + */ + prev = NULL; + for (pending = pendingCleanups; pending != NULL; pending = next) + { + next = pending->next; + + if (RelFileLocatorEquals(rlocator, pending->rlocator) && + pending->unlink_forknum == INIT_FORKNUM) + { + if (prev) + prev->next = next; + else + pendingCleanups = next; + + pfree(pending); + /* prev does not change */ + + create = false; + } + else + prev = pending; + } + + if (!create) + return; + + /* create the init fork, along with the commit-sentinel file */ + srel = smgropen(rlocator, InvalidBackendId); + log_smgrcreatemark(&rlocator, INIT_FORKNUM, SMGR_MARK_UNCOMMITTED); + smgrcreatemark(srel, INIT_FORKNUM, SMGR_MARK_UNCOMMITTED, false); + + /* We don't have existing init fork, create it. */ + smgrcreate(srel, INIT_FORKNUM, false); + + /* + * init fork for indexes needs further initialization. ambuildempty should + * do WAL-log and file sync by itself but otherwise we do that by + * ourselves. + */ + if (rel->rd_rel->relkind == RELKIND_INDEX) + rel->rd_indam->ambuildempty(rel); + else + { + log_smgrcreate(&rlocator, INIT_FORKNUM); + smgrimmedsync(srel, INIT_FORKNUM); + } + + /* drop the init fork, mark file and revert persistence at abort */ + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = rlocator; + pending->op = PCOP_UNLINK_FORK | PCOP_UNLINK_MARK | PCOP_SET_PERSISTENCE; + pending->unlink_forknum = INIT_FORKNUM; + pending->unlink_mark = SMGR_MARK_UNCOMMITTED; + pending->bufpersistence = true; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; + + /* drop mark file at commit */ + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = rlocator; + pending->op = PCOP_UNLINK_MARK; + pending->unlink_forknum = INIT_FORKNUM; + pending->unlink_mark = SMGR_MARK_UNCOMMITTED; + pending->backend = InvalidBackendId; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; +} + +/* + * RelationDropInitFork + * Delete physical storage for the init fork of a relation. + * + * Register pending-delete of the init fork. The real deletion is performed by + * smgrDoPendingDeletes at commit. + * + * This function is transactional. If the transaction aborts later on, the + * deletion is canceled. + */ +void +RelationDropInitFork(Relation rel) +{ + RelFileLocator rlocator = rel->rd_locator; + PendingCleanup *pending; + PendingCleanup *prev; + PendingCleanup *next; + bool inxact_created = false; + + /* switch buffer persistence */ + SetRelationBuffersPersistence(RelationGetSmgr(rel), true, false); + + /* + * If we have a pending-unlink for the init-fork of this relation, that + * means the init fork is created in the current transaction. We remove + * both the init fork and mark file immediately in that case. Otherwise + * register an at-commit pending-unlink for the existing init fork. See + * RelationCreateInitFork. + */ + prev = NULL; + for (pending = pendingCleanups; pending != NULL; pending = next) + { + next = pending->next; + + if (RelFileLocatorEquals(rlocator, pending->rlocator) && + pending->unlink_forknum != INIT_FORKNUM) + { + /* unlink list entry */ + if (prev) + prev->next = next; + else + pendingCleanups = next; + + pfree(pending); + /* prev does not change */ + + inxact_created = true; + } + else + prev = pending; + } + + if (inxact_created) + { + SMgrRelation srel = smgropen(rlocator, InvalidBackendId); + ForkNumber forknum = INIT_FORKNUM; + BlockNumber firstblock = 0; + + /* + * Some AMs initializes INIT fork via buffer manager. Drop all buffers + * for the INIT fork then unlink the INIT fork along with the mark + * file. + */ + DropRelationBuffers(srel, &forknum, 1, &firstblock); + log_smgrunlinkmark(&rlocator, INIT_FORKNUM, SMGR_MARK_UNCOMMITTED); + smgrunlinkmark(srel, INIT_FORKNUM, SMGR_MARK_UNCOMMITTED, false); + log_smgrunlink(&rlocator, INIT_FORKNUM); + smgrunlink(srel, INIT_FORKNUM, false); + return; + } + + /* register drop of this init fork file at commit */ + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = rlocator; + pending->op = PCOP_UNLINK_FORK; + pending->unlink_forknum = INIT_FORKNUM; + pending->backend = InvalidBackendId; + pending->atCommit = true; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; + + /* revert buffer-persistence changes at abort */ + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = rlocator; + pending->op = PCOP_SET_PERSISTENCE; + pending->bufpersistence = false; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; +} + /* * Perform XLogInsert of an XLOG_SMGR_CREATE record to WAL. */ @@ -197,6 +442,88 @@ log_smgrcreate(const RelFileLocator *rlocator, ForkNumber forkNum) XLogInsert(RM_SMGR_ID, XLOG_SMGR_CREATE | XLR_SPECIAL_REL_UPDATE); } +/* + * Perform XLogInsert of an XLOG_SMGR_UNLINK record to WAL. + */ +void +log_smgrunlink(const RelFileLocator *rlocator, ForkNumber forkNum) +{ + xl_smgr_unlink xlrec; + + /* + * Make an XLOG entry reporting the file unlink. + */ + xlrec.rlocator = *rlocator; + xlrec.forkNum = forkNum; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_UNLINK | XLR_SPECIAL_REL_UPDATE); +} + +/* + * Perform XLogInsert of an XLOG_SMGR_CREATEMARK record to WAL. + */ +void +log_smgrcreatemark(const RelFileLocator *rlocator, ForkNumber forkNum, + StorageMarks mark) +{ + xl_smgr_mark xlrec; + + /* + * Make an XLOG entry reporting the file creation. + */ + xlrec.rlocator = *rlocator; + xlrec.forkNum = forkNum; + xlrec.mark = mark; + xlrec.action = XLOG_SMGR_MARK_CREATE; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_MARK | XLR_SPECIAL_REL_UPDATE); +} + +/* + * Perform XLogInsert of an XLOG_SMGR_UNLINKMARK record to WAL. + */ +void +log_smgrunlinkmark(const RelFileLocator *rlocator, ForkNumber forkNum, + StorageMarks mark) +{ + xl_smgr_mark xlrec; + + /* + * Make an XLOG entry reporting the file creation. + */ + xlrec.rlocator = *rlocator; + xlrec.forkNum = forkNum; + xlrec.mark = mark; + xlrec.action = XLOG_SMGR_MARK_UNLINK; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_MARK | XLR_SPECIAL_REL_UPDATE); +} + +/* + * Perform XLogInsert of an XLOG_SMGR_BUFPERSISTENCE record to WAL. + */ +void +log_smgrbufpersistence(const RelFileLocator *rlocator, bool persistence) +{ + xl_smgr_bufpersistence xlrec; + + /* + * Make an XLOG entry reporting the change of buffer persistence. + */ + xlrec.rlocator = *rlocator; + xlrec.persistence = persistence; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + XLogInsert(RM_SMGR_ID, XLOG_SMGR_BUFPERSISTENCE | XLR_SPECIAL_REL_UPDATE); +} + /* * RelationDropStorage * Schedule unlinking of physical storage at transaction commit. @@ -711,6 +1038,95 @@ smgrDoPendingDeletes(bool isCommit) } } +/* + * smgrDoPendingUnmark() -- Clean up work that emits WAL records + * + * The operations handled in the function emits WAL records, which must be + * part of the current transaction. + */ +void +smgrDoPendingCleanups(bool isCommit) +{ + int nestLevel = GetCurrentTransactionNestLevel(); + PendingCleanup *pending; + PendingCleanup *prev; + PendingCleanup *next; + + prev = NULL; + for (pending = pendingCleanups; pending != NULL; pending = next) + { + next = pending->next; + if (pending->nestLevel < nestLevel) + { + /* outer-level entries should not be processed yet */ + prev = pending; + } + else + { + /* unlink list entry first, so we don't retry on failure */ + if (prev) + prev->next = next; + else + pendingCleanups = next; + + /* do cleanup if called for */ + if (pending->atCommit == isCommit) + { + SMgrRelation srel; + + srel = smgropen(pending->rlocator, pending->backend); + + Assert((pending->op & + ~(PCOP_UNLINK_FORK | PCOP_UNLINK_MARK | + PCOP_SET_PERSISTENCE)) == 0); + + if (pending->op & PCOP_SET_PERSISTENCE) + { + SetRelationBuffersPersistence(srel, pending->bufpersistence, + InRecovery); + } + + if (pending->op & PCOP_UNLINK_FORK) + { + /* + * Unlink the fork file. Currently we use this only for + * init forks and we're sure that the init fork is not + * loaded on shared buffers. For RelationDropInitFork + * case, the function dropped that buffers. For + * RelationCreateInitFork case, PCOP_SET_PERSISTENCE(true) + * is set and the buffers have been dropped just before. + */ + Assert(pending->unlink_forknum == INIT_FORKNUM); + + /* Don't emit wal while recovery. */ + if (!InRecovery) + log_smgrunlink(&pending->rlocator, + pending->unlink_forknum); + smgrunlink(srel, pending->unlink_forknum, false); + } + + if (pending->op & PCOP_UNLINK_MARK) + { + SMgrRelation srel; + + if (!InRecovery) + log_smgrunlinkmark(&pending->rlocator, + pending->unlink_forknum, + pending->unlink_mark); + srel = smgropen(pending->rlocator, pending->backend); + smgrunlinkmark(srel, pending->unlink_forknum, + pending->unlink_mark, InRecovery); + smgrclose(srel); + } + } + + /* must explicitly free the list entry */ + pfree(pending); + /* prev does not change */ + } + } +} + /* * smgrDoPendingSyncs() -- Take care of relation syncs at end of xact. */ @@ -975,6 +1391,15 @@ smgr_redo(XLogReaderState *record) reln = smgropen(xlrec->rlocator, InvalidBackendId); smgrcreate(reln, xlrec->forkNum, true); } + else if (info == XLOG_SMGR_UNLINK) + { + xl_smgr_unlink *xlrec = (xl_smgr_unlink *) XLogRecGetData(record); + SMgrRelation reln; + + reln = smgropen(xlrec->rlocator, InvalidBackendId); + smgrunlink(reln, xlrec->forkNum, true); + smgrclose(reln); + } else if (info == XLOG_SMGR_TRUNCATE) { xl_smgr_truncate *xlrec = (xl_smgr_truncate *) XLogRecGetData(record); @@ -1067,6 +1492,124 @@ smgr_redo(XLogReaderState *record) FreeFakeRelcacheEntry(rel); } + else if (info == XLOG_SMGR_MARK) + { + xl_smgr_mark *xlrec = (xl_smgr_mark *) XLogRecGetData(record); + SMgrRelation reln; + PendingCleanup *pending; + bool created = false; + + reln = smgropen(xlrec->rlocator, InvalidBackendId); + + switch (xlrec->action) + { + case XLOG_SMGR_MARK_CREATE: + smgrcreatemark(reln, xlrec->forkNum, xlrec->mark, true); + created = true; + break; + case XLOG_SMGR_MARK_UNLINK: + smgrunlinkmark(reln, xlrec->forkNum, xlrec->mark, true); + break; + default: + elog(ERROR, "unknown smgr_mark action \"%c\"", xlrec->mark); + } + + if (created) + { + /* revert mark file operation at abort */ + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = xlrec->rlocator; + pending->op = PCOP_UNLINK_MARK; + pending->unlink_forknum = xlrec->forkNum; + pending->unlink_mark = xlrec->mark; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; + } + else + { + /* + * Delete pending action for this mark file if any. We should have + * at most one entry for this action. + */ + PendingCleanup *prev = NULL; + + for (pending = pendingCleanups; pending != NULL; + pending = pending->next) + { + if (RelFileLocatorEquals(xlrec->rlocator, pending->rlocator) && + pending->unlink_forknum == xlrec->forkNum && + (pending->op & PCOP_UNLINK_MARK) != 0) + { + if (prev) + prev->next = pending->next; + else + pendingCleanups = pending->next; + + pfree(pending); + break; + } + + prev = pending; + } + } + } + else if (info == XLOG_SMGR_BUFPERSISTENCE) + { + xl_smgr_bufpersistence *xlrec = + (xl_smgr_bufpersistence *) XLogRecGetData(record); + SMgrRelation reln; + PendingCleanup *pending; + PendingCleanup *prev = NULL; + + reln = smgropen(xlrec->rlocator, InvalidBackendId); + SetRelationBuffersPersistence(reln, xlrec->persistence, true); + + /* + * Delete pending action for persistence change if any. We should have + * at most one entry for this action. + */ + for (pending = pendingCleanups; pending != NULL; + pending = pending->next) + { + if (RelFileLocatorEquals(xlrec->rlocator, pending->rlocator) && + (pending->op & PCOP_SET_PERSISTENCE) != 0) + { + Assert(pending->bufpersistence == xlrec->persistence); + + if (prev) + prev->next = pending->next; + else + pendingCleanups = pending->next; + + pfree(pending); + break; + } + + prev = pending; + } + + /* + * Revert buffer-persistence changes at abort if the relation is going + * to different persistence from before this transaction. + */ + if (!pending) + { + pending = (PendingCleanup *) + MemoryContextAlloc(TopMemoryContext, sizeof(PendingCleanup)); + pending->rlocator = xlrec->rlocator; + pending->op = PCOP_SET_PERSISTENCE; + pending->bufpersistence = !xlrec->persistence; + pending->backend = InvalidBackendId; + pending->atCommit = false; + pending->nestLevel = GetCurrentTransactionNestLevel(); + pending->next = pendingCleanups; + pendingCleanups = pending; + } + } else elog(PANIC, "smgr_redo: unknown op code %u", info); } diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c index 1b8e6d5729..1779959410 100644 --- a/src/backend/commands/tablecmds.c +++ b/src/backend/commands/tablecmds.c @@ -55,6 +55,7 @@ #include "commands/defrem.h" #include "commands/event_trigger.h" #include "commands/policy.h" +#include "commands/progress.h" #include "commands/sequence.h" #include "commands/tablecmds.h" #include "commands/tablespace.h" @@ -5407,6 +5408,187 @@ ATParseTransformCmd(List **wqueue, AlteredTableInfo *tab, Relation rel, return newcmd; } +/* + * RelationChangePersistence: do in-place persistence change of a relation + */ +static void +RelationChangePersistence(AlteredTableInfo *tab, char persistence, + LOCKMODE lockmode) +{ + Relation rel; + Relation classRel; + HeapTuple tuple, + newtuple; + Datum new_val[Natts_pg_class]; + bool new_null[Natts_pg_class], + new_repl[Natts_pg_class]; + int i; + List *relids; + ListCell *lc_oid; + + Assert(tab->rewrite == AT_REWRITE_ALTER_PERSISTENCE); + Assert(lockmode == AccessExclusiveLock); + + /* + * Under the following condition, we need to call ATRewriteTable, which + * cannot be false in the AT_REWRITE_ALTER_PERSISTENCE case. + */ + Assert(tab->constraints == NULL && tab->partition_constraint == NULL && + tab->newvals == NULL && !tab->verify_new_notnull); + + rel = table_open(tab->relid, lockmode); + + Assert(rel->rd_rel->relpersistence != persistence); + + elog(DEBUG1, "perform in-place persistence change"); + + /* + * First we collect all relations that we need to change persistence. + */ + + /* Collect OIDs of indexes and toast relations */ + relids = RelationGetIndexList(rel); + relids = lcons_oid(rel->rd_id, relids); + + /* Add toast relation if any */ + if (OidIsValid(rel->rd_rel->reltoastrelid)) + { + List *toastidx; + Relation toastrel = table_open(rel->rd_rel->reltoastrelid, lockmode); + + relids = lappend_oid(relids, rel->rd_rel->reltoastrelid); + toastidx = RelationGetIndexList(toastrel); + relids = list_concat(relids, toastidx); + pfree(toastidx); + table_close(toastrel, NoLock); + } + + table_close(rel, NoLock); + + /* Make changes in storage */ + classRel = table_open(RelationRelationId, RowExclusiveLock); + + foreach(lc_oid, relids) + { + Oid reloid = lfirst_oid(lc_oid); + Relation r = relation_open(reloid, lockmode); + + /* + * XXXX: Some access methods do not bear up an in-place persistence + * change. Specifically, GiST uses page LSNs to figure out whether a + * block has changed, where UNLOGGED GiST indexes use fake LSNs that + * are incompatible with real LSNs used for LOGGED ones. + * + * Maybe if gistGetFakeLSN behaved the same way for permanent and + * unlogged indexes, we could skip index rebuild in exchange of some + * extra WAL records emitted while it is unlogged. + * + * Check relam against a positive list so that we take the hard way for + * unknown AMs. + */ + if (r->rd_rel->relkind == RELKIND_INDEX && + /* GiST is excluded */ + r->rd_rel->relam != BTREE_AM_OID && + r->rd_rel->relam != HASH_AM_OID && + r->rd_rel->relam != GIN_AM_OID && + r->rd_rel->relam != SPGIST_AM_OID && + r->rd_rel->relam != BRIN_AM_OID) + { + int reindex_flags; + ReindexParams params = {0}; + + /* reindex doesn't allow concurrent use of the index */ + table_close(r, NoLock); + + reindex_flags = + REINDEX_REL_SUPPRESS_INDEX_USE | + REINDEX_REL_CHECK_CONSTRAINTS; + + /* Set the same persistence with the parent relation. */ + if (persistence == RELPERSISTENCE_UNLOGGED) + reindex_flags |= REINDEX_REL_FORCE_INDEXES_UNLOGGED; + else + reindex_flags |= REINDEX_REL_FORCE_INDEXES_PERMANENT; + + reindex_index(reloid, reindex_flags, persistence, ¶ms); + + continue; + } + + /* Create or drop init fork */ + if (persistence == RELPERSISTENCE_UNLOGGED) + RelationCreateInitFork(r); + else + RelationDropInitFork(r); + + /* + * When this relation gets WAL-logged, immediately sync all files but + * initfork to establish the initial state on storage. Buffers have + * already flushed out by RelationCreate(Drop)InitFork called just + * above. Initfork should have been synced as needed. + */ + if (persistence == RELPERSISTENCE_PERMANENT) + { + for (i = 0; i < INIT_FORKNUM; i++) + { + if (smgrexists(RelationGetSmgr(r), i)) + smgrimmedsync(RelationGetSmgr(r), i); + } + } + + /* Update catalog */ + tuple = SearchSysCacheCopy1(RELOID, ObjectIdGetDatum(reloid)); + if (!HeapTupleIsValid(tuple)) + elog(ERROR, "cache lookup failed for relation %u", reloid); + + memset(new_val, 0, sizeof(new_val)); + memset(new_null, false, sizeof(new_null)); + memset(new_repl, false, sizeof(new_repl)); + + new_val[Anum_pg_class_relpersistence - 1] = CharGetDatum(persistence); + new_null[Anum_pg_class_relpersistence - 1] = false; + new_repl[Anum_pg_class_relpersistence - 1] = true; + + newtuple = heap_modify_tuple(tuple, RelationGetDescr(classRel), + new_val, new_null, new_repl); + + CatalogTupleUpdate(classRel, &newtuple->t_self, newtuple); + heap_freetuple(newtuple); + + /* + * While wal_level >= replica, switching to LOGGED requires the + * relation content to be WAL-logged to recover the table. We don't + * emit this if wal_level = minimal. + */ + if (persistence == RELPERSISTENCE_PERMANENT && XLogIsNeeded()) + { + ForkNumber fork; + xl_smgr_truncate xlrec; + + xlrec.blkno = 0; + xlrec.rlocator = r->rd_locator; + xlrec.flags = SMGR_TRUNCATE_ALL; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, sizeof(xlrec)); + + XLogInsert(RM_SMGR_ID, XLOG_SMGR_TRUNCATE | XLR_SPECIAL_REL_UPDATE); + + for (fork = 0; fork < INIT_FORKNUM; fork++) + { + if (smgrexists(RelationGetSmgr(r), fork)) + log_newpage_range(r, fork, 0, + smgrnblocks(RelationGetSmgr(r), fork), + false); + } + } + + table_close(r, NoLock); + } + + table_close(classRel, NoLock); +} + /* * ATRewriteTables: ALTER TABLE phase 3 */ @@ -5537,48 +5719,55 @@ ATRewriteTables(AlterTableStmt *parsetree, List **wqueue, LOCKMODE lockmode, tab->relid, tab->rewrite); - /* - * Create transient table that will receive the modified data. - * - * Ensure it is marked correctly as logged or unlogged. We have - * to do this here so that buffers for the new relfilenumber will - * have the right persistence set, and at the same time ensure - * that the original filenumbers's buffers will get read in with - * the correct setting (i.e. the original one). Otherwise a - * rollback after the rewrite would possibly result with buffers - * for the original filenumbers having the wrong persistence - * setting. - * - * NB: This relies on swap_relation_files() also swapping the - * persistence. That wouldn't work for pg_class, but that can't be - * unlogged anyway. - */ - OIDNewHeap = make_new_heap(tab->relid, NewTableSpace, NewAccessMethod, - persistence, lockmode); + if (tab->rewrite == AT_REWRITE_ALTER_PERSISTENCE) + RelationChangePersistence(tab, persistence, lockmode); + else + { + /* + * Create transient table that will receive the modified data. + * + * Ensure it is marked correctly as logged or unlogged. We + * have to do this here so that buffers for the new + * relfilenumber will have the right persistence set, and at + * the same time ensure that the original filenumbers's buffers + * will get read in with the correct setting (i.e. the original + * one). Otherwise a rollback after the rewrite would possibly + * result with buffers for the original filenumbers having the + * wrong persistence setting. + * + * NB: This relies on swap_relation_files() also swapping the + * persistence. That wouldn't work for pg_class, but that + * can't be unlogged anyway. + */ + OIDNewHeap = make_new_heap(tab->relid, NewTableSpace, + NewAccessMethod, + persistence, lockmode); - /* - * Copy the heap data into the new table with the desired - * modifications, and test the current data within the table - * against new constraints generated by ALTER TABLE commands. - */ - ATRewriteTable(tab, OIDNewHeap, lockmode); + /* + * Copy the heap data into the new table with the desired + * modifications, and test the current data within the table + * against new constraints generated by ALTER TABLE commands. + */ + ATRewriteTable(tab, OIDNewHeap, lockmode); - /* - * Swap the physical files of the old and new heaps, then rebuild - * indexes and discard the old heap. We can use RecentXmin for - * the table's new relfrozenxid because we rewrote all the tuples - * in ATRewriteTable, so no older Xid remains in the table. Also, - * we never try to swap toast tables by content, since we have no - * interest in letting this code work on system catalogs. - */ - finish_heap_swap(tab->relid, OIDNewHeap, - false, false, true, - !OidIsValid(tab->newTableSpace), - RecentXmin, - ReadNextMultiXactId(), - persistence); + /* + * Swap the physical files of the old and new heaps, then + * rebuild indexes and discard the old heap. We can use + * RecentXmin for the table's new relfrozenxid because we + * rewrote all the tuples in ATRewriteTable, so no older Xid + * remains in the table. Also, we never try to swap toast + * tables by content, since we have no interest in letting + * this code work on system catalogs. + */ + finish_heap_swap(tab->relid, OIDNewHeap, + false, false, true, + !OidIsValid(tab->newTableSpace), + RecentXmin, + ReadNextMultiXactId(), + persistence); - InvokeObjectPostAlterHook(RelationRelationId, tab->relid, 0); + InvokeObjectPostAlterHook(RelationRelationId, tab->relid, 0); + } } else if (tab->rewrite > 0 && tab->relkind == RELKIND_SEQUENCE) { diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 5b0e531f97..796cb139c3 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -3172,6 +3172,91 @@ DropRelationBuffers(SMgrRelation smgr_reln, ForkNumber *forkNum, } } +/* --------------------------------------------------------------------- + * SetRelationBuffersPersistence + * + * This function changes the persistence of all buffer pages of a relation + * then writes all dirty pages of the relation out to disk when switching + * to PERMANENT. (or more accurately, out to kernel disk buffers), + * ensuring that the kernel has an up-to-date view of the relation. + * + * Generally, the caller should be holding AccessExclusiveLock on the + * target relation to ensure that no other backend is busy dirtying + * more blocks of the relation; the effects can't be expected to last + * after the lock is released. + * + * XXX currently it sequentially searches the buffer pool, should be + * changed to more clever ways of searching. This routine is not + * used in any performance-critical code paths, so it's not worth + * adding additional overhead to normal paths to make it go faster; + * but see also DropRelFileLocatorBuffers. + * -------------------------------------------------------------------- + */ +void +SetRelationBuffersPersistence(SMgrRelation srel, bool permanent, bool isRedo) +{ + int i; + RelFileLocatorBackend rlocator = srel->smgr_rlocator; + + Assert(!RelFileLocatorBackendIsTemp(rlocator)); + + if (!isRedo) + log_smgrbufpersistence(&srel->smgr_rlocator.locator, permanent); + + ResourceOwnerEnlargeBuffers(CurrentResourceOwner); + + for (i = 0; i < NBuffers; i++) + { + BufferDesc *bufHdr = GetBufferDescriptor(i); + uint32 buf_state; + + if (!RelFileLocatorEquals(bufHdr->tag.rlocator, rlocator.locator)) + continue; + + ReservePrivateRefCountEntry(); + + buf_state = LockBufHdr(bufHdr); + + if (!RelFileLocatorEquals(bufHdr->tag.rlocator, rlocator.locator)) + { + UnlockBufHdr(bufHdr, buf_state); + continue; + } + + if (permanent) + { + /* Init fork is being dropped, drop buffers for it. */ + if (bufHdr->tag.forkNum == INIT_FORKNUM) + { + InvalidateBuffer(bufHdr); + continue; + } + + buf_state |= BM_PERMANENT; + pg_atomic_write_u32(&bufHdr->state, buf_state); + + /* we flush this buffer when switching to PERMANENT */ + if ((buf_state & (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY)) + { + PinBuffer_Locked(bufHdr); + LWLockAcquire(BufferDescriptorGetContentLock(bufHdr), + LW_SHARED); + FlushBuffer(bufHdr, srel); + LWLockRelease(BufferDescriptorGetContentLock(bufHdr)); + UnpinBuffer(bufHdr, true); + } + else + UnlockBufHdr(bufHdr, buf_state); + } + else + { + /* There shouldn't be an init fork */ + Assert(bufHdr->tag.forkNum != INIT_FORKNUM); + UnlockBufHdr(bufHdr, buf_state); + } + } +} + /* --------------------------------------------------------------------- * DropRelationsAllBuffers * diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c index 073dab2be5..c5f577a694 100644 --- a/src/backend/storage/file/fd.c +++ b/src/backend/storage/file/fd.c @@ -347,8 +347,6 @@ static void pre_sync_fname(const char *fname, bool isdir, int elevel); static void datadir_fsync_fname(const char *fname, bool isdir, int elevel); static void unlink_if_exists_fname(const char *fname, bool isdir, int elevel); -static int fsync_parent_path(const char *fname, int elevel); - /* * pg_fsync --- do fsync with or without writethrough @@ -3670,7 +3668,7 @@ fsync_fname_ext(const char *fname, bool isdir, bool ignore_perm, int elevel) * This is aimed at making file operations persistent on disk in case of * an OS crash or power failure. */ -static int +int fsync_parent_path(const char *fname, int elevel) { char parentpath[MAXPGPATH]; diff --git a/src/backend/storage/file/reinit.c b/src/backend/storage/file/reinit.c index c3faa68126..ea5d6bbba1 100644 --- a/src/backend/storage/file/reinit.c +++ b/src/backend/storage/file/reinit.c @@ -16,29 +16,49 @@ #include <unistd.h> +#include "access/xlogrecovery.h" +#include "catalog/pg_tablespace_d.h" #include "common/relpath.h" #include "postmaster/startup.h" +#include "storage/bufmgr.h" #include "storage/copydir.h" #include "storage/fd.h" +#include "storage/md.h" #include "storage/reinit.h" +#include "storage/smgr.h" #include "utils/hsearch.h" #include "utils/memutils.h" static void ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, - int op); + Oid tspid, int op); static void ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, - int op); + Oid tspid, Oid dbid, int op); typedef struct { RelFileNumber relnumber; /* hash key */ -} unlogged_relation_entry; + bool has_init; /* has INIT fork */ + bool dirty_init; /* needs to remove INIT fork */ + bool dirty_all; /* needs to remove all forks */ +} relfile_entry; /* - * Reset unlogged relations from before the last restart. + * Clean up and reset relation files from before the last restart. * - * If op includes UNLOGGED_RELATION_CLEANUP, we remove all forks of any - * relation with an "init" fork, except for the "init" fork itself. + * If op includes UNLOGGED_RELATION_CLEANUP, we perform different operations + * depending on the existence of the "cleanup" forks. + * + * If SMGR_MARK_UNCOMMITTED mark file for init fork is present, we remove the + * init fork along with the mark file. + * + * If SMGR_MARK_UNCOMMITTED mark file for main fork is present we remove the + * whole relation along with the mark file. + * + * Otherwise, if the "init" fork is found. we remove all forks of any relation + * with the "init" fork, except for the "init" fork itself. + * + * If op includes UNLOGGED_RELATION_DROP_BUFFER, we drop all buffers for all + * relations that have the "cleanup" and/or the "init" forks. * * If op includes UNLOGGED_RELATION_INIT, we copy the "init" fork to the main * fork. @@ -72,7 +92,7 @@ ResetUnloggedRelations(int op) /* * First process unlogged files in pg_default ($PGDATA/base) */ - ResetUnloggedRelationsInTablespaceDir("base", op); + ResetUnloggedRelationsInTablespaceDir("base", DEFAULTTABLESPACE_OID, op); /* * Cycle through directories for all non-default tablespaces. @@ -81,13 +101,19 @@ ResetUnloggedRelations(int op) while ((spc_de = ReadDir(spc_dir, "pg_tblspc")) != NULL) { + Oid tspid; + if (strcmp(spc_de->d_name, ".") == 0 || strcmp(spc_de->d_name, "..") == 0) continue; snprintf(temp_path, sizeof(temp_path), "pg_tblspc/%s/%s", spc_de->d_name, TABLESPACE_VERSION_DIRECTORY); - ResetUnloggedRelationsInTablespaceDir(temp_path, op); + + tspid = atooid(spc_de->d_name); + + Assert(tspid != 0); + ResetUnloggedRelationsInTablespaceDir(temp_path, tspid, op); } FreeDir(spc_dir); @@ -103,7 +129,8 @@ ResetUnloggedRelations(int op) * Process one tablespace directory for ResetUnloggedRelations */ static void -ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) +ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, + Oid tspid, int op) { DIR *ts_dir; struct dirent *de; @@ -130,6 +157,8 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) while ((de = ReadDir(ts_dir, tsdirname)) != NULL) { + Oid dbid; + /* * We're only interested in the per-database directories, which have * numeric names. Note that this code will also (properly) ignore "." @@ -148,7 +177,10 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) ereport_startup_progress("resetting unlogged relations (cleanup), elapsed time: %ld.%02d s, current path: %s", dbspace_path); - ResetUnloggedRelationsInDbspaceDir(dbspace_path, op); + dbid = atooid(de->d_name); + Assert(dbid != 0); + + ResetUnloggedRelationsInDbspaceDir(dbspace_path, tspid, dbid, op); } FreeDir(ts_dir); @@ -158,125 +190,228 @@ ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op) * Process one per-dbspace directory for ResetUnloggedRelations */ static void -ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) +ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, + Oid tspid, Oid dbid, int op) { DIR *dbspace_dir; struct dirent *de; char rm_path[MAXPGPATH * 2]; + HTAB *hash; + HASHCTL ctl; /* Caller must specify at least one operation. */ - Assert((op & (UNLOGGED_RELATION_CLEANUP | UNLOGGED_RELATION_INIT)) != 0); + Assert((op & (UNLOGGED_RELATION_CLEANUP | + UNLOGGED_RELATION_DROP_BUFFER | + UNLOGGED_RELATION_INIT)) != 0); /* * Cleanup is a two-pass operation. First, we go through and identify all * the files with init forks. Then, we go through again and nuke * everything with the same OID except the init fork. */ - if ((op & UNLOGGED_RELATION_CLEANUP) != 0) + + /* + * It's possible that someone could create tons of unlogged relations in + * the same database & tablespace, so we'd better use a hash table rather + * than an array or linked list to keep track of which files need to be + * reset. Otherwise, this cleanup operation would be O(n^2). + */ + memset(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(RelFileNumber); + ctl.entrysize = sizeof(relfile_entry); + hash = hash_create("unlogged relation RelFileNumbers", + 32, &ctl, HASH_ELEM | HASH_BLOBS); + + /* Collect INIT fork and mark files in the directory. */ + dbspace_dir = AllocateDir(dbspacedirname); + while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) { - HTAB *hash; - HASHCTL ctl; + ForkNumber forkNum; + int relnumchars; + StorageMarks mark; - /* - * It's possible that someone could create a ton of unlogged relations - * in the same database & tablespace, so we'd better use a hash table - * rather than an array or linked list to keep track of which files - * need to be reset. Otherwise, this cleanup operation would be - * O(n^2). - */ - ctl.keysize = sizeof(RelFileNumber); - ctl.entrysize = sizeof(unlogged_relation_entry); - ctl.hcxt = CurrentMemoryContext; - hash = hash_create("unlogged relation RelFileNumbers", 32, &ctl, - HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); + /* Skip anything that doesn't look like a relation data file. */ + if (!parse_filename_for_nontemp_relation(de->d_name, &relnumchars, + &forkNum, &mark)) + continue; - /* Scan the directory. */ - dbspace_dir = AllocateDir(dbspacedirname); - while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) + if (forkNum == INIT_FORKNUM || mark == SMGR_MARK_UNCOMMITTED) { - ForkNumber forkNum; - int relnumchars; - unlogged_relation_entry ent; - - /* Skip anything that doesn't look like a relation data file. */ - if (!parse_filename_for_nontemp_relation(de->d_name, &relnumchars, - &forkNum)) - continue; - - /* Also skip it unless this is the init fork. */ - if (forkNum != INIT_FORKNUM) - continue; + RelFileNumbers key; + relfile_entry *ent; + bool found; /* * Put the RELFILENUMBER portion of the name into the hash table, - * if it isn't already. + * if it isn't already. If it has SMGR_MARK_UNCOMMITTED mark + * files, the storage file is in dirty state, where clean up is + * needed. */ - ent.relnumber = atorelnumber(de->d_name); - (void) hash_search(hash, &ent, HASH_ENTER, NULL); + key = atorelnumber(de->d_name); + ent = hash_search(hash, &key, HASH_ENTER, &found); + + if (!found) + { + ent->has_init = false; + ent->dirty_init = false; + ent->dirty_all = false; + } + + if (forkNum == INIT_FORKNUM && mark == SMGR_MARK_UNCOMMITTED) + ent->dirty_init = true; + else if (forkNum == MAIN_FORKNUM && mark == SMGR_MARK_UNCOMMITTED) + ent->dirty_all = true; + else + { + Assert(forkNum == INIT_FORKNUM); + ent->has_init = true; + } } + } - /* Done with the first pass. */ - FreeDir(dbspace_dir); + /* Done with the first pass. */ + FreeDir(dbspace_dir); + /* nothing to do if we don't have init nor cleanup forks */ + if (hash_get_num_entries(hash) < 1) + { + hash_destroy(hash); + return; + } + + if ((op & UNLOGGED_RELATION_DROP_BUFFER) != 0) + { /* - * If we didn't find any init forks, there's no point in continuing; - * we can bail out now. + * When we come here after recovery, smgr object for this file might + * have been created. In that case we need to drop all buffers then + * the smgr object before initializing the unlogged relation. This is + * safe as far as no other backends have accessed the relation before + * starting archive recovery. */ - if (hash_get_num_entries(hash) == 0) + HASH_SEQ_STATUS status; + relfile_entry *ent; + SMgrRelation *srels = palloc(sizeof(SMgrRelation) * 8); + int maxrels = 8; + int nrels = 0; + int i; + + Assert(!HotStandbyActive()); + + hash_seq_init(&status, hash); + while ((ent = (relfile_entry *) hash_seq_search(&status)) != NULL) { - hash_destroy(hash); - return; + RelFileLocatorBackend rel; + + /* + * The relation is persistent and stays persistent. Don't drop the + * buffers for this relation. + */ + if (ent->has_init && ent->dirty_init) + continue; + + if (maxrels <= nrels) + { + maxrels *= 2; + srels = repalloc(srels, sizeof(SMgrRelation) * maxrels); + } + + rel.backend = InvalidBackendId; + rel.locator.spcOid = tspid; + rel.locator.dbOid = dbid; + rel.locator.relNumber = ent->reloid; + + srels[nrels++] = smgropen(rel.locator, InvalidBackendId); } - /* - * Now, make a second pass and remove anything that matches. - */ + DropRelationsAllBuffers(srels, nrels); + + for (i = 0; i < nrels; i++) + smgrclose(srels[i]); + } + + /* + * Now, make a second pass and remove anything that matches. + */ + if ((op & UNLOGGED_RELATION_CLEANUP) != 0) + { dbspace_dir = AllocateDir(dbspacedirname); while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) { ForkNumber forkNum; + StorageMarks mark; int relnumchars; - unlogged_relation_entry ent; + RelFileNumber key; + relfile_entry *ent; + RelFileLocatorBackend rel; /* Skip anything that doesn't look like a relation data file. */ if (!parse_filename_for_nontemp_relation(de->d_name, &relnumchars, - &forkNum)) - continue; - - /* We never remove the init fork. */ - if (forkNum == INIT_FORKNUM) + &forkNum, &mark)) continue; /* * See whether the RELFILENUMBER portion of the name shows up in * the hash table. If so, nuke it! */ - ent.relnumber = atorelnumber(de->d_name); - if (hash_search(hash, &ent, HASH_FIND, NULL)) + key = atorelnumber(de->d_name); + ent = hash_search(hash, &key, HASH_FIND, NULL); + + if (!ent) + continue; + + if (!ent->dirty_all) { - snprintf(rm_path, sizeof(rm_path), "%s/%s", - dbspacedirname, de->d_name); - if (unlink(rm_path) < 0) - ereport(ERROR, - (errcode_for_file_access(), - errmsg("could not remove file \"%s\": %m", - rm_path))); + /* clean permanent relations don't need cleanup */ + if (!ent->has_init) + continue; + + if (ent->dirty_init) + { + /* + * The crashed transaction did SET UNLOGGED. This relation + * is restored to a LOGGED relation. + */ + if (forkNum != INIT_FORKNUM) + continue; + } else - elog(DEBUG2, "unlinked file \"%s\"", rm_path); + { + /* + * we don't remove the INIT fork of a non-dirty + * relation files. + */ + if (forkNum == INIT_FORKNUM && mark == SMGR_MARK_NONE) + continue; + } } + + /* so, nuke it! */ + snprintf(rm_path, sizeof(rm_path), "%s/%s", + dbspacedirname, de->d_name); + if (unlink(rm_path) < 0) + ereport(ERROR, + errcode_for_file_access(), + errmsg("could not remove file \"%s\": %m", + rm_path)); + + rel.backend = InvalidBackendId; + rel.locator.spcOid = tspid; + rel.locator.dbOid = dbid; + rel.locator.relNumber = atooid(de->d_name); + + ForgetRelationForkSyncRequests(rel, forkNum); } /* Cleanup is complete. */ FreeDir(dbspace_dir); - hash_destroy(hash); } + hash_destroy(hash); + hash = NULL; + /* * Initialization happens after cleanup is complete: we copy each init - * fork file to the corresponding main fork file. Note that if we are - * asked to do both cleanup and init, we may never get here: if the - * cleanup code determines that there are no init forks in this dbspace, - * it will return before we get to this point. + * fork file to the corresponding main fork file. */ if ((op & UNLOGGED_RELATION_INIT) != 0) { @@ -285,6 +420,7 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) { ForkNumber forkNum; + StorageMarks mark; int relnumchars; char relnumbuf[RELNUMBERCHARS + 1]; char srcpath[MAXPGPATH * 2]; @@ -292,9 +428,11 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) /* Skip anything that doesn't look like a relation data file. */ if (!parse_filename_for_nontemp_relation(de->d_name, &relnumchars, - &forkNum)) + &forkNum, &mark)) continue; + Assert(mark == SMGR_MARK_NONE); + /* Also skip it unless this is the init fork. */ if (forkNum != INIT_FORKNUM) continue; @@ -328,15 +466,18 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL) { ForkNumber forkNum; + StorageMarks mark; int relnumchars; char relnumbuf[RELNUMBERCHARS + 1]; char mainpath[MAXPGPATH]; /* Skip anything that doesn't look like a relation data file. */ if (!parse_filename_for_nontemp_relation(de->d_name, &relnumchars, - &forkNum)) + &forkNum, &mark)) continue; + Assert(mark == SMGR_MARK_NONE); + /* Also skip it unless this is the init fork. */ if (forkNum != INIT_FORKNUM) continue; @@ -379,7 +520,7 @@ ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op) */ bool parse_filename_for_nontemp_relation(const char *name, int *relnumchars, - ForkNumber *fork) + ForkNumber *fork, StorageMarks *mark) { int pos; @@ -410,11 +551,19 @@ parse_filename_for_nontemp_relation(const char *name, int *relnumchars, for (segchar = 1; isdigit((unsigned char) name[pos + segchar]); ++segchar) ; - if (segchar <= 1) - return false; - pos += segchar; + if (segchar > 1) + pos += segchar; } + /* mark file? */ + if (name[pos] == '.' && name[pos + 1] != 0) + { + *mark = name[pos + 1]; + pos += 2; + } + else + *mark = SMGR_MARK_NONE; + /* Now we should be at the end. */ if (name[pos] != '\0') return false; diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c index bed47f07d7..d628461970 100644 --- a/src/backend/storage/smgr/md.c +++ b/src/backend/storage/smgr/md.c @@ -141,7 +141,8 @@ static MdfdVec *_mdfd_getseg(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno, bool skipFsync, int behavior); static BlockNumber _mdnblocks(SMgrRelation reln, ForkNumber forknum, MdfdVec *seg); - +static bool mdmarkexists(SMgrRelation reln, ForkNumber forkNum, + StorageMarks mark); /* * mdinit() -- Initialize private state for magnetic disk storage manager. @@ -173,6 +174,82 @@ mdexists(SMgrRelation reln, ForkNumber forknum) return (mdopenfork(reln, forknum, EXTENSION_RETURN_NULL) != NULL); } +/* + * mdcreatemark() -- Create a mark file. + * + * If isRedo is true, it's okay for the file to exist already. + */ +void +mdcreatemark(SMgrRelation reln, ForkNumber forkNum, StorageMarks mark, + bool isRedo) +{ + char *path = markpath(reln->smgr_rlocator, forkNum, mark); + int fd; + + /* See mdcreate for details.. */ + TablespaceCreateDbspace(reln->smgr_rlocator.locator.spcOid, + reln->smgr_rlocator.locator.dbOid, + isRedo); + + fd = BasicOpenFile(path, O_WRONLY | O_CREAT | O_EXCL); + if (fd < 0 && (!isRedo || errno != EEXIST)) + ereport(ERROR, + errcode_for_file_access(), + errmsg("could not create mark file \"%s\": %m", path)); + + pg_fsync(fd); + close(fd); + + /* + * To guarantee that the creation of the file is persistent, fsync its + * parent directory. + */ + fsync_parent_path(path, ERROR); + + pfree(path); +} + + +/* + * mdunlinkmark() -- Delete the mark file + * + * If isRedo is true, it's okay for the file being not found. + */ +void +mdunlinkmark(SMgrRelation reln, ForkNumber forkNum, StorageMarks mark, + bool isRedo) +{ + char *path = markpath(reln->smgr_rlocator, forkNum, mark); + + if (!isRedo || mdmarkexists(reln, forkNum, mark)) + durable_unlink(path, ERROR); + + pfree(path); +} + +/* + * mdmarkexists() -- Check if the file exists. + */ +static bool +mdmarkexists(SMgrRelation reln, ForkNumber forkNum, StorageMarks mark) +{ + char *path = markpath(reln->smgr_rlocator, forkNum, mark); + int fd; + + fd = BasicOpenFile(path, O_RDONLY); + if (fd < 0 && errno != ENOENT) + ereport(ERROR, + errcode_for_file_access(), + errmsg("could not access mark file \"%s\": %m", path)); + pfree(path); + + if (fd < 0) + return false; + + close(fd); + return true; +} + /* * mdcreate() -- Create a new relation on magnetic disk. * @@ -1056,6 +1133,16 @@ register_forget_request(RelFileLocatorBackend rlocator, ForkNumber forknum, RegisterSyncRequest(&tag, SYNC_FORGET_REQUEST, true /* retryOnError */ ); } +/* + * ForgetRelationForkSyncRequests -- forget any fsyncs and unlinks for a fork + */ +void +ForgetRelationForkSyncRequests(RelFileLocatorBackend rlocator, + ForkNumber forknum) +{ + register_forget_request(rlocator, forknum, 0); +} + /* * ForgetDatabaseSyncRequests -- forget any fsyncs and unlinks for a DB */ @@ -1414,12 +1501,14 @@ mdsyncfiletag(const FileTag *ftag, char *path) * Return 0 on success, -1 on failure, with errno set. */ int -mdunlinkfiletag(const FileTag *ftag, char *path) +mdunlinkfiletag(const FileTag *ftag, char *path, StorageMarks mark) { char *p; /* Compute the path. */ - p = relpathperm(ftag->rlocator, MAIN_FORKNUM); + p = GetRelationPath(ftag->rlocator.dbOid, ftag->rlocator.spcOid, + ftag->rlocator.relNumber,InvalidBackendId, + MAIN_FORKNUM, mark); strlcpy(path, p, MAXPGPATH); pfree(p); diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c index ed46ac3f44..c7353d3dcf 100644 --- a/src/backend/storage/smgr/smgr.c +++ b/src/backend/storage/smgr/smgr.c @@ -62,6 +62,10 @@ typedef struct f_smgr void (*smgr_truncate) (SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks); void (*smgr_immedsync) (SMgrRelation reln, ForkNumber forknum); + void (*smgr_createmark) (SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); + void (*smgr_unlinkmark) (SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); } f_smgr; static const f_smgr smgrsw[] = { @@ -82,6 +86,8 @@ static const f_smgr smgrsw[] = { .smgr_nblocks = mdnblocks, .smgr_truncate = mdtruncate, .smgr_immedsync = mdimmedsync, + .smgr_createmark = mdcreatemark, + .smgr_unlinkmark = mdunlinkmark, } }; @@ -371,6 +377,26 @@ smgrcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo) smgrsw[reln->smgr_which].smgr_create(reln, forknum, isRedo); } +/* + * smgrcreatemark() -- Create a mark file + */ +void +smgrcreatemark(SMgrRelation reln, ForkNumber forknum, StorageMarks mark, + bool isRedo) +{ + smgrsw[reln->smgr_which].smgr_createmark(reln, forknum, mark, isRedo); +} + +/* + * smgrunlinkmark() -- Delete a mark file + */ +void +smgrunlinkmark(SMgrRelation reln, ForkNumber forknum, StorageMarks mark, + bool isRedo) +{ + smgrsw[reln->smgr_which].smgr_unlinkmark(reln, forknum, mark, isRedo); +} + /* * smgrdosyncall() -- Immediately sync all forks of all given relations * @@ -693,6 +719,12 @@ smgrimmedsync(SMgrRelation reln, ForkNumber forknum) smgrsw[reln->smgr_which].smgr_immedsync(reln, forknum); } +void +smgrunlink(SMgrRelation reln, ForkNumber forknum, bool isRedo) +{ + smgrsw[reln->smgr_which].smgr_unlink(reln->smgr_rlocator, forknum, isRedo); +} + /* * AtEOXact_SMgr * diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c index 9d6a9e9109..74357efb1c 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -91,7 +91,8 @@ static CycleCtr checkpoint_cycle_ctr = 0; typedef struct SyncOps { int (*sync_syncfiletag) (const FileTag *ftag, char *path); - int (*sync_unlinkfiletag) (const FileTag *ftag, char *path); + int (*sync_unlinkfiletag) (const FileTag *ftag, char *path, + StorageMarks mark); bool (*sync_filetagmatches) (const FileTag *ftag, const FileTag *candidate); } SyncOps; @@ -235,7 +236,8 @@ SyncPostCheckpoint(void) /* Unlink the file */ if (syncsw[entry->tag.handler].sync_unlinkfiletag(&entry->tag, - path) < 0) + path, + SMGR_MARK_NONE) < 0) { /* * There's a race condition, when the database is dropped at the @@ -244,6 +246,21 @@ SyncPostCheckpoint(void) * here. rmtree() also has to ignore ENOENT errors, to deal with * the possibility that we delete the file first. */ + if (errno != ENOENT) + ereport(WARNING, + errcode_for_file_access(), + errmsg("could not remove file \"%s\": %m", path)); + } + else if (syncsw[entry->tag.handler].sync_unlinkfiletag(&entry->tag, + path, + SMGR_MARK_UNCOMMITTED) + < 0) + { + /* + * And we may have SMGR_MARK_UNCOMMITTED file. Remove it if the + * fork file has been successfully removed. It's ok if the file + * does not exist. + */ if (errno != ENOENT) ereport(WARNING, (errcode_for_file_access(), diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c index 53f011a2fe..4da62f71ce 100644 --- a/src/bin/pg_rewind/parsexlog.c +++ b/src/bin/pg_rewind/parsexlog.c @@ -418,6 +418,28 @@ extractPageInfo(XLogReaderState *record) * source system. */ } + else if (rmid == RM_SMGR_ID && rminfo == XLOG_SMGR_UNLINK) + { + /* + * We can safely ignore there. We'll see that the file don't exist in + * the target data dir, and copy them in from the source system. No + * need to do anything special here. + */ + } + else if (rmid == RM_SMGR_ID && rminfo == XLOG_SMGR_MARK) + { + /* + * We can safely ignore these, The file will be removed from the + * target, if it doesn't exist in the source system. The files are + * empty so we don't need to bother the content. + */ + } + else if (rmid == RM_SMGR_ID && rminfo == XLOG_SMGR_BUFPERSISTENCE) + { + /* + * We can safely ignore these. These don't make any on-disk changes. + */ + } else if (rmid == RM_XACT_ID && ((rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_COMMIT || (rminfo & XLOG_XACT_OPMASK) == XLOG_XACT_COMMIT_PREPARED || diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c index 3cd77c09b1..f87dfda8fe 100644 --- a/src/bin/pg_rewind/pg_rewind.c +++ b/src/bin/pg_rewind/pg_rewind.c @@ -419,7 +419,6 @@ main(int argc, char **argv) if (showprogress) pg_log_info("reading source file list"); source->traverse_files(source, &process_source_file); - if (showprogress) pg_log_info("reading target file list"); traverse_datadir(datadir_target, &process_target_file); diff --git a/src/common/relpath.c b/src/common/relpath.c index d0d83e593b..f4f7435cf9 100644 --- a/src/common/relpath.c +++ b/src/common/relpath.c @@ -139,9 +139,15 @@ GetDatabasePath(Oid dbOid, Oid spcOid) */ char * GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, - int backendId, ForkNumber forkNumber) + int backendId, ForkNumber forkNumber, char mark) { char *path; + char markstr[4]; + + if (mark == 0) + markstr[0] = 0; + else + snprintf(markstr, sizeof(markstr), ".%c", mark); if (spcOid == GLOBALTABLESPACE_OID) { @@ -149,10 +155,10 @@ GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, Assert(dbOid == 0); Assert(backendId == InvalidBackendId); if (forkNumber != MAIN_FORKNUM) - path = psprintf("global/" UINT64_FORMAT "_%s", - relNumber, forkNames[forkNumber]); + path = psprintf("global/" UINT64_FORMAT "_%s%s", + relNumber, forkNames[forkNumber], markstr); else - path = psprintf("global/" UINT64_FORMAT, relNumber); + path = psprintf("global/" UINT64_FORMAT "%s", relNumber, markstr); } else if (spcOid == DEFAULTTABLESPACE_OID) { @@ -160,22 +166,22 @@ GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, if (backendId == InvalidBackendId) { if (forkNumber != MAIN_FORKNUM) - path = psprintf("base/%u/" UINT64_FORMAT "_%s", + path = psprintf("base/%u/" UINT64_FORMAT "_%s%s", dbOid, relNumber, - forkNames[forkNumber]); + forkNames[forkNumber], markstr); else - path = psprintf("base/%u/" UINT64_FORMAT, - dbOid, relNumber); + path = psprintf("base/%u/" UINT64_FORMAT "%s", + dbOid, relNumber, markstr); } else { if (forkNumber != MAIN_FORKNUM) - path = psprintf("base/%u/t%d_" UINT64_FORMAT "_%s", + path = psprintf("base/%u/t%d_" UINT64_FORMAT "_%s%s", dbOid, backendId, relNumber, - forkNames[forkNumber]); + forkNames[forkNumber], markstr); else - path = psprintf("base/%u/t%d_" UINT64_FORMAT, - dbOid, backendId, relNumber); + path = psprintf("base/%u/t%d_" UINT64_FORMAT "%s", + dbOid, backendId, relNumber, markstr); } } else @@ -184,27 +190,28 @@ GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, if (backendId == InvalidBackendId) { if (forkNumber != MAIN_FORKNUM) - path = psprintf("pg_tblspc/%u/%s/%u/" UINT64_FORMAT "_%s", + path = psprintf("pg_tblspc/%u/%s/%u/" UINT64_FORMAT "_%s%s", spcOid, TABLESPACE_VERSION_DIRECTORY, dbOid, relNumber, - forkNames[forkNumber]); + forkNames[forkNumber], markstr); else - path = psprintf("pg_tblspc/%u/%s/%u/" UINT64_FORMAT, + path = psprintf("pg_tblspc/%u/%s/%u/" UINT64_FORMAT "%s", spcOid, TABLESPACE_VERSION_DIRECTORY, - dbOid, relNumber); + dbOid, relNumber, markstr); } else { if (forkNumber != MAIN_FORKNUM) - path = psprintf("pg_tblspc/%u/%s/%u/t%d_" UINT64_FORMAT "_%s", + path = psprintf("pg_tblspc/%u/%s/%u/t%d_" UINT64_FORMAT "_%s%s", spcOid, TABLESPACE_VERSION_DIRECTORY, dbOid, backendId, relNumber, - forkNames[forkNumber]); + forkNames[forkNumber], markstr); else - path = psprintf("pg_tblspc/%u/%s/%u/t%d_" UINT64_FORMAT, + path = psprintf("pg_tblspc/%u/%s/%u/t%d_" UINT64_FORMAT "%s", spcOid, TABLESPACE_VERSION_DIRECTORY, - dbOid, backendId, relNumber); + dbOid, backendId, relNumber, markstr); } } + return path; } diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h index 9964c312aa..ee4179699a 100644 --- a/src/include/catalog/storage.h +++ b/src/include/catalog/storage.h @@ -25,6 +25,8 @@ extern PGDLLIMPORT int wal_skip_threshold; extern SMgrRelation RelationCreateStorage(RelFileLocator rlocator, char relpersistence, bool register_delete); +extern void RelationCreateInitFork(Relation rel); +extern void RelationDropInitFork(Relation rel); extern void RelationDropStorage(Relation rel); extern void RelationPreserveStorage(RelFileLocator rlocator, bool atCommit); extern void RelationPreTruncate(Relation rel); @@ -43,6 +45,7 @@ extern void RestorePendingSyncs(char *startAddress); extern void smgrDoPendingDeletes(bool isCommit); extern void smgrDoPendingSyncs(bool isCommit, bool isParallelWorker); extern int smgrGetPendingDeletes(bool forCommit, RelFileLocator **ptr); +extern void smgrDoPendingCleanups(bool isCommit); extern void AtSubCommit_smgr(void); extern void AtSubAbort_smgr(void); extern void PostPrepare_smgr(void); diff --git a/src/include/catalog/storage_xlog.h b/src/include/catalog/storage_xlog.h index 44a5e2043b..9f48fb5e6f 100644 --- a/src/include/catalog/storage_xlog.h +++ b/src/include/catalog/storage_xlog.h @@ -18,17 +18,23 @@ #include "lib/stringinfo.h" #include "storage/block.h" #include "storage/relfilelocator.h" +#include "storage/smgr.h" /* * Declarations for smgr-related XLOG records * - * Note: we log file creation and truncation here, but logging of deletion - * actions is handled by xact.c, because it is part of transaction commit. + * Note: we log file creation, truncation and buffer persistence change here, + * but logging of deletion actions is handled mainly by xact.c, because it is + * part of transaction commit in most cases. However, there's a case where + * init forks are deleted outside control of transaction. */ /* XLOG gives us high 4 bits */ #define XLOG_SMGR_CREATE 0x10 #define XLOG_SMGR_TRUNCATE 0x20 +#define XLOG_SMGR_UNLINK 0x30 +#define XLOG_SMGR_MARK 0x40 +#define XLOG_SMGR_BUFPERSISTENCE 0x50 typedef struct xl_smgr_create { @@ -36,6 +42,32 @@ typedef struct xl_smgr_create ForkNumber forkNum; } xl_smgr_create; +typedef struct xl_smgr_unlink +{ + RelFileLocator rlocator; + ForkNumber forkNum; +} xl_smgr_unlink; + +typedef enum smgr_mark_action +{ + XLOG_SMGR_MARK_CREATE = 'c', + XLOG_SMGR_MARK_UNLINK = 'u' +} smgr_mark_action; + +typedef struct xl_smgr_mark +{ + RelFileLocator rlocator; + ForkNumber forkNum; + StorageMarks mark; + smgr_mark_action action; +} xl_smgr_mark; + +typedef struct xl_smgr_bufpersistence +{ + RelFileLocator rlocator; + bool persistence; +} xl_smgr_bufpersistence; + /* flags for xl_smgr_truncate */ #define SMGR_TRUNCATE_HEAP 0x0001 #define SMGR_TRUNCATE_VM 0x0002 @@ -51,6 +83,13 @@ typedef struct xl_smgr_truncate } xl_smgr_truncate; extern void log_smgrcreate(const RelFileLocator *rlocator, ForkNumber forkNum); +extern void log_smgrunlink(const RelFileLocator *rlocator, ForkNumber forkNum); +extern void log_smgrcreatemark(const RelFileLocator *rlocator, + ForkNumber forkNum, StorageMarks mark); +extern void log_smgrunlinkmark(const RelFileLocator *rlocator, + ForkNumber forkNum, StorageMarks mark); +extern void log_smgrbufpersistence(const RelFileLocator *rlocator, + bool persistence); extern void smgr_redo(XLogReaderState *record); extern void smgr_desc(StringInfo buf, XLogReaderState *record); diff --git a/src/include/common/relpath.h b/src/include/common/relpath.h index 2d3b52fe0b..bcbd66ead3 100644 --- a/src/include/common/relpath.h +++ b/src/include/common/relpath.h @@ -77,7 +77,7 @@ extern int forkname_chars(const char *str, ForkNumber *fork); extern char *GetDatabasePath(Oid dbOid, Oid spcOid); extern char *GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, - int backendId, ForkNumber forkNumber); + int backendId, ForkNumber forkNumber, char mark); /* * Wrapper macros for GetRelationPath. Beware of multiple @@ -87,7 +87,7 @@ extern char *GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, /* First argument is a RelFileLocator */ #define relpathbackend(rlocator, backend, forknum) \ GetRelationPath((rlocator).dbOid, (rlocator).spcOid, (rlocator).relNumber, \ - backend, forknum) + backend, forknum, 0) /* First argument is a RelFileLocator */ #define relpathperm(rlocator, forknum) \ @@ -97,4 +97,9 @@ extern char *GetRelationPath(Oid dbOid, Oid spcOid, RelFileNumber relNumber, #define relpath(rlocator, forknum) \ relpathbackend((rlocator).locator, (rlocator).backend, forknum) +/* First argument is a RelFileLocatorBackend */ +#define markpath(rlocator, forknum, mark) \ + GetRelationPath((rlocator).locator.dbOid, (rlocator).locator.spcOid, \ + (rlocator).locator.relNumber, \ + (rlocator).backend, forknum, mark) #endif /* RELPATH_H */ diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h index 6f4dfa0960..5b753b768b 100644 --- a/src/include/storage/bufmgr.h +++ b/src/include/storage/bufmgr.h @@ -142,6 +142,8 @@ extern void DropRelationBuffers(struct SMgrRelationData *smgr_reln, int nforks, BlockNumber *firstDelBlock); extern void DropRelationsAllBuffers(struct SMgrRelationData **smgr_reln, int nlocators); +extern void SetRelationBuffersPersistence(struct SMgrRelationData *srel, + bool permanent, bool isRedo); extern void DropDatabaseBuffers(Oid dbid); #define RelationGetNumberOfBlocks(reln) \ diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h index 5a48fccd9c..b2773ae8bb 100644 --- a/src/include/storage/fd.h +++ b/src/include/storage/fd.h @@ -185,6 +185,7 @@ extern ssize_t pg_pwritev_with_retry(int fd, extern int pg_truncate(const char *path, off_t length); extern void fsync_fname(const char *fname, bool isdir); extern int fsync_fname_ext(const char *fname, bool isdir, bool ignore_perm, int elevel); +extern int fsync_parent_path(const char *fname, int elevel); extern int durable_rename(const char *oldfile, const char *newfile, int elevel); extern int durable_unlink(const char *fname, int elevel); extern void SyncDataDirectory(void); diff --git a/src/include/storage/md.h b/src/include/storage/md.h index 10aa1b0109..294a09444c 100644 --- a/src/include/storage/md.h +++ b/src/include/storage/md.h @@ -23,6 +23,10 @@ extern void mdinit(void); extern void mdopen(SMgrRelation reln); extern void mdclose(SMgrRelation reln, ForkNumber forknum); +extern void mdcreatemark(SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); +extern void mdunlinkmark(SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); extern void mdcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo); extern bool mdexists(SMgrRelation reln, ForkNumber forknum); extern void mdunlink(RelFileLocatorBackend rlocator, ForkNumber forknum, bool isRedo); @@ -41,12 +45,14 @@ extern void mdtruncate(SMgrRelation reln, ForkNumber forknum, BlockNumber nblocks); extern void mdimmedsync(SMgrRelation reln, ForkNumber forknum); +extern void ForgetRelationForkSyncRequests(RelFileLocatorBackend rlocator, + ForkNumber forknum); extern void ForgetDatabaseSyncRequests(Oid dbid); extern void DropRelationFiles(RelFileLocator *delrels, int ndelrels, bool isRedo); /* md sync callbacks */ extern int mdsyncfiletag(const FileTag *ftag, char *path); -extern int mdunlinkfiletag(const FileTag *ftag, char *path); +extern int mdunlinkfiletag(const FileTag *ftag, char *path, StorageMarks mark); extern bool mdfiletagmatches(const FileTag *ftag, const FileTag *candidate); #endif /* MD_H */ diff --git a/src/include/storage/reinit.h b/src/include/storage/reinit.h index b990d28d38..cacd5e7d2c 100644 --- a/src/include/storage/reinit.h +++ b/src/include/storage/reinit.h @@ -16,14 +16,16 @@ #define REINIT_H #include "common/relpath.h" - +#include "storage/smgr.h" extern void ResetUnloggedRelations(int op); extern bool parse_filename_for_nontemp_relation(const char *name, int *relnumchars, - ForkNumber *fork); + ForkNumber *fork, + StorageMarks *mark); #define UNLOGGED_RELATION_CLEANUP 0x0001 -#define UNLOGGED_RELATION_INIT 0x0002 +#define UNLOGGED_RELATION_DROP_BUFFER 0x0002 +#define UNLOGGED_RELATION_INIT 0x0004 #endif /* REINIT_H */ diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h index a07715356b..75b9d41e4b 100644 --- a/src/include/storage/smgr.h +++ b/src/include/storage/smgr.h @@ -18,6 +18,18 @@ #include "storage/block.h" #include "storage/relfilelocator.h" +/* + * Storage marks is a file of which existence suggests something about a + * file. The name of such files is "<filename>.<mark>", where the mark is one + * of the values of StorageMarks. Since ".<digit>" means segment files so don't + * use digits for the mark character. + */ +typedef enum StorageMarks +{ + SMGR_MARK_NONE = 0, + SMGR_MARK_UNCOMMITTED = 'u' /* the file is not committed yet */ +} StorageMarks; + /* * smgr.c maintains a table of SMgrRelation objects, which are essentially * cached file handles. An SMgrRelation is created (if not already present) @@ -87,7 +99,12 @@ extern void smgrcloseall(void); extern void smgrcloserellocator(RelFileLocatorBackend rlocator); extern void smgrrelease(SMgrRelation reln); extern void smgrreleaseall(void); +extern void smgrcreatemark(SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); +extern void smgrunlinkmark(SMgrRelation reln, ForkNumber forknum, + StorageMarks mark, bool isRedo); extern void smgrcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo); +extern void smgrunlink(SMgrRelation reln, ForkNumber forknum, bool isRedo); extern void smgrdosyncall(SMgrRelation *rels, int nrels); extern void smgrdounlinkall(SMgrRelation *rels, int nrels, bool isRedo); extern void smgrextend(SMgrRelation reln, ForkNumber forknum, diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 97c9bc1861..2fdd221408 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -1970,6 +1970,7 @@ PatternInfoArray Pattern_Prefix_Status Pattern_Type PendingFsyncEntry +PendingMarkCleanup PendingRelDelete PendingRelSync PendingUnlinkEntry @@ -2600,6 +2601,7 @@ StdRdOptIndexCleanup StdRdOptions Step StopList +StorageMarks StrategyNumber StreamCtl String @@ -3607,6 +3609,7 @@ registered_buffer regmatch_t regoff_t regproc +relfile_entry relopt_bool relopt_enum relopt_enum_elt_def @@ -3660,6 +3663,7 @@ slist_iter slist_mutable_iter slist_node slock_t +smgr_mark_action socket_set socklen_t spgBulkDeleteState @@ -3858,8 +3862,11 @@ xl_replorigin_set xl_restore_point xl_running_xacts xl_seq_rec +xl_smgr_bufpersistence xl_smgr_create +xl_smgr_mark xl_smgr_truncate +xl_smgr_unlink xl_standby_lock xl_standby_locks xl_tblspc_create_rec -- 2.31.1 ----Next_Part(Wed_Sep_28_17_21_19_2022_527)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: attachment; filename="v24-0002-New-command-ALTER-TABLE-ALL-IN-TABLESPACE-SET-LO.patch" ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication @ 2024-07-08 19:12 John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: John H @ 2024-07-08 19:12 UTC (permalink / raw) To: Amit Kapila <[email protected]>; [email protected] <[email protected]>; +Cc: pgsql-hackers Hi Amit, Thanks for taking a look. On Tue, Jun 18, 2024 at 10:34 PM Amit Kapila <[email protected]> wrote: > > > The reading indicates when you set 'standby_slot_names_from_syncrep', > the TPS reduces as compared to when it is not set. It would be better > to see the data comparing 'standby_slot_names_from_syncrep' and the > existing parameter 'standby_slot_names'. I added new benchmark numbers in the reply to Bertrand, but I'll include in this thread for posterity. Writer with 5 synchronous replicas, 10 pg_recvlogical clients and pgbench all running from the same server. Command: pgbench -c 4 -j 4 -T 600 -U "ec2-user" -d postgres -r -P 5 Result with: standby_slot_names = 'replica_1,replica_2,replica_3,replica_4,replica_5' latency average = 5.600 ms latency stddev = 2.854 ms initial connection time = 5.503 ms tps = 714.148263 (without initial connection time) Result with: standby_slot_names_from_syncrep = 'true', synchronous_standby_names = 'ANY 3 (A,B,C,D,E)' latency average = 5.740 ms latency stddev = 2.543 ms initial connection time = 4.093 ms tps = 696.776249 (without initial connection time) Result with nothing set: latency average = 5.090 ms latency stddev = 3.467 ms initial connection time = 4.989 ms tps = 785.665963 (without initial connection time) > Can we make it a default > behavior that logical slots marked with a failover option will wait > for 'synchronous_standby_names' as per your patch's idea unless > 'standby_slot_names' is specified? I don't know if there is any value > in setting the 'failover' option for a slot without specifying > 'standby_slot_names', so was wondering if we can additionally tie it > to 'synchronous_standby_names'. Any better ideas? > No, I think that works pretty cleanly actually. Reserving some special keyword isn't great which is the only other thing I can think of. I've updated the patch and tests to reflect that. Attached the patch that addresses these changes. -- John Hsu - Amazon Web Services Attachments: [application/octet-stream] 0002-Wait-on-synchronous-replication-by-default-for-logic.patch (21.8K, ../../CA+-JvFsn_FMKVcbztr2LiVKzwjz6uQTNKtakR47mXg+78Nt2iA@mail.gmail.com/2-0002-Wait-on-synchronous-replication-by-default-for-logic.patch) download | inline diff: From 805e5b7e35913f53b9caac996aaf0b62e937f274 Mon Sep 17 00:00:00 2001 From: John Hsu <[email protected]> Date: Tue, 4 Jun 2024 20:27:47 +0000 Subject: [PATCH] Wait on synchronous replication by default for logical failover slots If synchronous replication is enabled, this patch allows logical subscribers to wait for changes to be replicated to synchronous replicas before consuming the changes. In the event that 'synchronized_standby_slots' is set, it will take precedence. --- src/backend/replication/slot.c | 271 ++++++++++-------- src/backend/replication/syncrep.c | 27 +- src/include/replication/syncrep.h | 3 + .../t/040_standby_failover_slots_sync.pl | 135 +++++++++ 4 files changed, 321 insertions(+), 115 deletions(-) diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index baf9b89dc4..fafb6c7873 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -49,6 +49,7 @@ #include "postmaster/interrupt.h" #include "replication/slotsync.h" #include "replication/slot.h" +#include "replication/syncrep.h" #include "replication/walsender_private.h" #include "storage/fd.h" #include "storage/ipc.h" @@ -2570,8 +2571,8 @@ SlotExistsInSyncStandbySlots(const char *slot_name) } /* - * Return true if the slots specified in synchronized_standby_slots have caught up to - * the given WAL location, false otherwise. + * Return true if the slots specified in synchronized_standby_slots or synchronous + * replication have caught up to the given WAL location, false otherwise. * * The elevel parameter specifies the error level used for logging messages * related to slots that do not exist, are invalidated, or are inactive. @@ -2585,9 +2586,9 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel) /* * Don't need to wait for the standbys to catch up if there is no value in - * synchronized_standby_slots. + * synchronized_standby_slots or synchronous replication is not configured. */ - if (synchronized_standby_slots_config == NULL) + if (synchronized_standby_slots_config == NULL && !SyncRepConfigured()) return true; /* @@ -2598,144 +2599,190 @@ StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel) return true; /* - * Don't need to wait for the standbys to catch up if they are already - * beyond the specified WAL location. - */ - if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) && - ss_oldest_flush_lsn >= wait_for_lsn) - return true; - - /* - * To prevent concurrent slot dropping and creation while filtering the - * slots, take the ReplicationSlotControlLock outside of the loop. + * In the event that synchronized_standby_slots and synchronous replication is + * configured, have the former take precedence. */ - LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); - - name = synchronized_standby_slots_config->slot_names; - for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++) + if (synchronized_standby_slots_config != NULL) { - XLogRecPtr restart_lsn; - bool invalidated; - bool inactive; - ReplicationSlot *slot; + /* + * Don't need to wait for the standbys to catch up if they are already + * beyond the specified WAL location. + */ + if (!XLogRecPtrIsInvalid(ss_oldest_flush_lsn) && + ss_oldest_flush_lsn >= wait_for_lsn) + return true; - slot = SearchNamedReplicationSlot(name, false); + /* + * To prevent concurrent slot dropping and creation while filtering the + * slots, take the ReplicationSlotControlLock outside of the loop. + */ + LWLockAcquire(ReplicationSlotControlLock, LW_SHARED); - if (!slot) + name = synchronized_standby_slots_config->slot_names; + for (int i = 0; i < synchronized_standby_slots_config->nslotnames; i++) { - /* - * If a slot name provided in synchronized_standby_slots does not - * exist, report a message and exit the loop. A user can specify a - * slot name that does not exist just before the server startup. - * The GUC check_hook(validate_sync_standby_slots) cannot validate - * such a slot during startup as the ReplicationSlotCtl shared - * memory is not initialized at that time. It is also possible for - * a user to drop the slot in synchronized_standby_slots - * afterwards. - */ - ereport(elevel, - errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("replication slot \"%s\" specified in parameter %s does not exist", - name, "synchronized_standby_slots"), - errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider creating the slot \"%s\" or amend parameter %s.", - name, "synchronized_standby_slots")); - break; - } + XLogRecPtr restart_lsn; + bool invalidated; + bool inactive; + ReplicationSlot *slot; - if (SlotIsLogical(slot)) - { - /* - * If a logical slot name is provided in - * synchronized_standby_slots, report a message and exit the loop. - * Similar to the non-existent case, a user can specify a logical - * slot name in synchronized_standby_slots before the server - * startup, or drop an existing physical slot and recreate a - * logical slot with the same name. - */ - ereport(elevel, - errcode(ERRCODE_INVALID_PARAMETER_VALUE), - errmsg("cannot have logical replication slot \"%s\" in parameter %s", - name, "synchronized_standby_slots"), - errdetail("Logical replication is waiting for correction on \"%s\".", - name), - errhint("Consider removing logical slot \"%s\" from parameter %s.", - name, "synchronized_standby_slots")); - break; - } + slot = SearchNamedReplicationSlot(name, false); - SpinLockAcquire(&slot->mutex); - restart_lsn = slot->data.restart_lsn; - invalidated = slot->data.invalidated != RS_INVAL_NONE; - inactive = slot->active_pid == 0; - SpinLockRelease(&slot->mutex); + if (!slot) + { + /* + * If a slot name provided in synchronized_standby_slots does not + * exist, report a message and exit the loop. A user can specify a + * slot name that does not exist just before the server startup. + * The GUC check_hook(validate_sync_standby_slots) cannot validate + * such a slot during startup as the ReplicationSlotCtl shared + * memory is not initialized at that time. It is also possible for + * a user to drop the slot in synchronized_standby_slots + * afterwards. + */ + ereport(elevel, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("replication slot \"%s\" specified in parameter %s does not exist", + name, "synchronized_standby_slots"), + errdetail("Logical replication is waiting on the standby associated with \"%s\".", + name), + errhint("Consider creating the slot \"%s\" or amend parameter %s.", + name, "synchronized_standby_slots")); + break; + } - if (invalidated) - { - /* Specified physical slot has been invalidated */ - ereport(elevel, - errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("physical slot \"%s\" specified in parameter %s has been invalidated", - name, "synchronized_standby_slots"), - errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.", - name, "synchronized_standby_slots")); - break; - } + if (SlotIsLogical(slot)) + { + /* + * If a logical slot name is provided in + * synchronized_standby_slots, report a message and exit the loop. + * Similar to the non-existent case, a user can specify a logical + * slot name in synchronized_standby_slots before the server + * startup, or drop an existing physical slot and recreate a + * logical slot with the same name. + */ + ereport(elevel, + errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("cannot have logical replication slot \"%s\" in parameter %s", + name, "synchronized_standby_slots"), + errdetail("Logical replication is waiting for correction on \"%s\".", + name), + errhint("Consider removing logical slot \"%s\" from parameter %s.", + name, "synchronized_standby_slots")); + break; + } - if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn) - { - /* Log a message if no active_pid for this physical slot */ - if (inactive) + SpinLockAcquire(&slot->mutex); + restart_lsn = slot->data.restart_lsn; + invalidated = slot->data.invalidated != RS_INVAL_NONE; + inactive = slot->active_pid == 0; + SpinLockRelease(&slot->mutex); + + if (invalidated) + { + /* Specified physical slot has been invalidated */ ereport(elevel, errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), - errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid", - name, "synchronized_standby_slots"), + errmsg("physical slot \"%s\" specified in parameter %s has been invalidated", + name, "synchronized_standby_slots"), errdetail("Logical replication is waiting on the standby associated with \"%s\".", - name), - errhint("Consider starting standby associated with \"%s\" or amend parameter %s.", + name), + errhint("Consider dropping and recreating the slot \"%s\" or amend parameter %s.", name, "synchronized_standby_slots")); + break; + } - /* Continue if the current slot hasn't caught up. */ - break; + if (XLogRecPtrIsInvalid(restart_lsn) || restart_lsn < wait_for_lsn) + { + /* Log a message if no active_pid for this physical slot */ + if (inactive) + ereport(elevel, + errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("replication slot \"%s\" specified in parameter %s does not have active_pid", + name, "synchronized_standby_slots"), + errdetail("Logical replication is waiting on the standby associated with \"%s\".", + name), + errhint("Consider starting standby associated with \"%s\" or amend parameter %s.", + name, "synchronized_standby_slots")); + + /* Continue if the current slot hasn't caught up. */ + break; + } + + Assert(restart_lsn >= wait_for_lsn); + + if (XLogRecPtrIsInvalid(min_restart_lsn) || + min_restart_lsn > restart_lsn) + min_restart_lsn = restart_lsn; + + caught_up_slot_num++; + + name += strlen(name) + 1; } - Assert(restart_lsn >= wait_for_lsn); + LWLockRelease(ReplicationSlotControlLock); + + /* + * Return false if not all the standbys have caught up to the specified + * WAL location. + */ + if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames) + return false; - if (XLogRecPtrIsInvalid(min_restart_lsn) || - min_restart_lsn > restart_lsn) - min_restart_lsn = restart_lsn; + /* The ss_oldest_flush_lsn must not retreat. */ + Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) || + min_restart_lsn >= ss_oldest_flush_lsn); - caught_up_slot_num++; + ss_oldest_flush_lsn = min_restart_lsn; - name += strlen(name) + 1; + return true; } + else + { + volatile WalSndCtlData *walsndctl = WalSndCtl; + static XLogRecPtr lsn[NUM_SYNC_REP_WAIT_MODE]; + static bool initialized = false; + /* Initialize value in case SIGHUP changing to SYNC_REP_NO_WAIT */ + int mode = SyncRepWaitMode; + int i; + + if (!initialized) + { + for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) + { + lsn[i] = InvalidXLogRecPtr; + } + } - LWLockRelease(ReplicationSlotControlLock); + if (mode == SYNC_REP_NO_WAIT) + return true; - /* - * Return false if not all the standbys have caught up to the specified - * WAL location. - */ - if (caught_up_slot_num != synchronized_standby_slots_config->nslotnames) - return false; + if (lsn[mode] >= wait_for_lsn) + return true; + + LWLockAcquire(SyncRepLock, LW_SHARED); - /* The ss_oldest_flush_lsn must not retreat. */ - Assert(XLogRecPtrIsInvalid(ss_oldest_flush_lsn) || - min_restart_lsn >= ss_oldest_flush_lsn); + /* Cache values to reduce contention on lock */ + for (i = 0; i < NUM_SYNC_REP_WAIT_MODE; i++) + { + lsn[i] = walsndctl->lsn[i]; + } - ss_oldest_flush_lsn = min_restart_lsn; + LWLockRelease(SyncRepLock); - return true; + if (lsn[mode] >= wait_for_lsn) + return true; + + return false; + } } /* * Wait for physical standbys to confirm receiving the given lsn. * * Used by logical decoding SQL functions. It waits for physical standbys - * corresponding to the physical slots specified in the synchronized_standby_slots GUC. + * corresponding to the physical slots specified in the synchronized_standby_slots GUC, + * or synchronous replication. */ void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn) @@ -2745,7 +2792,7 @@ WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn) * slot is not a logical failover slot, or there is no value in * synchronized_standby_slots. */ - if (!MyReplicationSlot->data.failover || !synchronized_standby_slots_config) + if (!MyReplicationSlot->data.failover || !(synchronized_standby_slots_config || SyncRepConfigured())) return; ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv); diff --git a/src/backend/replication/syncrep.c b/src/backend/replication/syncrep.c index fa5988c824..69c061d8dc 100644 --- a/src/backend/replication/syncrep.c +++ b/src/backend/replication/syncrep.c @@ -78,6 +78,7 @@ #include "common/int.h" #include "miscadmin.h" #include "pgstat.h" +#include "replication/slot.h" #include "replication/syncrep.h" #include "replication/walsender.h" #include "replication/walsender_private.h" @@ -95,7 +96,7 @@ char *SyncRepStandbyNames; static bool announce_next_takeover = true; SyncRepConfigData *SyncRepConfig = NULL; -static int SyncRepWaitMode = SYNC_REP_NO_WAIT; +int SyncRepWaitMode = SYNC_REP_NO_WAIT; static void SyncRepQueueInsert(int mode); static void SyncRepCancelWait(void); @@ -124,6 +125,8 @@ static int cmp_lsn(const void *a, const void *b); static bool SyncRepQueueIsOrderedByLSN(int mode); #endif +bool SyncRepConfigured(void); + /* * =========================================================== * Synchronous Replication functions for normal user backends @@ -169,8 +172,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit) * described in SyncRepUpdateSyncStandbysDefined(). On the other hand, if * it's false, the lock is not necessary because we don't touch the queue. */ - if (!SyncRepRequested() || - !((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined) + if (!SyncRepConfigured()) return; /* Cap the level for anything other than commit to remote flush only. */ @@ -523,6 +525,15 @@ SyncRepReleaseWaiters(void) LWLockRelease(SyncRepLock); + /* + * If synchronized_standby_slots is set, the respective walsender's + * will be responsible for broadcasting the value. + */ + if (strcmp(synchronized_standby_slots, "") == 0) + { + ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv); + } + elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X", numwrite, LSN_FORMAT_ARGS(writePtr), numflush, LSN_FORMAT_ARGS(flushPtr), @@ -1069,3 +1080,13 @@ assign_synchronous_commit(int newval, void *extra) break; } } + +bool +SyncRepConfigured() +{ + if (!SyncRepRequested() || + !((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined) + return false; + + return true; +} diff --git a/src/include/replication/syncrep.h b/src/include/replication/syncrep.h index ea439e6da6..ec22bc72df 100644 --- a/src/include/replication/syncrep.h +++ b/src/include/replication/syncrep.h @@ -82,6 +82,7 @@ extern PGDLLIMPORT char *SyncRepStandbyNames; /* called by user backend */ extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit); +extern bool SyncRepConfigured(); /* called at backend exit */ extern void SyncRepCleanupAtProcExit(void); @@ -96,6 +97,8 @@ extern int SyncRepGetCandidateStandbys(SyncRepStandbyData **standbys); /* called by checkpointer */ extern void SyncRepUpdateSyncStandbysDefined(void); +extern int SyncRepWaitMode; + /* * Internal functions for parsing synchronous_standby_names grammar, * in syncrep_gram.y and syncrep_scanner.l diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl index 2c51cfc3c8..00130d92dc 100644 --- a/src/test/recovery/t/040_standby_failover_slots_sync.pl +++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl @@ -710,6 +710,141 @@ $result = $subscriber1->safe_psql('postgres', is($result, 't', "subscriber1 gets data from primary after standby1 acknowledges changes"); +################################################## +# Test that logical failover replication slots wait for the specified +# synchronous replicas to receive the changes first. It uses the +# following set up: +# +# (synchronous physical standbys) +# | ----> standby1 (application_name = standby1) +# | ----> standby2 (application_name = standby2) +# primary ----- | +# (logical replication) +# | ----> subscriber1 (failover = true, slot_name = lsub1_slot) +# | ----> subscriber2 (failover = false, slot_name = lsub2_slot) +# +# synchronous_commit = 'on' +# synchronous_standby_names = 'ANY 2 (standby1, standby2)' +# +# The setup is configured in such a way that the logical slot of subscriber1 is +# enabled for failover, and thus the subscriber1 will wait for the changes to have +# been synchronously replicated before receiving the decoded changes. +################################################## + +$primary->safe_psql('postgres', "TRUNCATE tab_int;"); +# Setup synchronous replication +$primary->append_conf( + 'postgresql.conf', qq( + synchronous_commit = 'on' + synchronous_standby_names = 'ANY 2 (standby1, standby2)' +)); + +$primary->reload; + +$standby1->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby1'"); +$standby1->reload; + +$standby2->append_conf('postgresql.conf', "primary_conninfo = '$connstr_1 dbname=postgres application_name=standby2'"); +$standby2->reload; + +# Check that synchronous replication is setup properly +$standby2->stop; + +# Create some data on the primary +$primary_row_count = 10; + +my $sync_back_q = $primary->background_psql( + 'postgres', + on_error_stop => 0, + timeout => $PostgreSQL::Test::Utils::timeout_default); + +$sync_back_q->query_until(qr/insert_blocked_on_sync_rep/, q( + \echo insert_blocked_on_sync_rep + INSERT INTO tab_int SELECT generate_series(1, 10); +)); + +$result = $primary->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); + +is($result, 'f', "primary row count is not updated due to synchronous replication"); + +# Verify the standby specified in synchronized_standby_slots (sb1_slot aka standby1) +# catches up with the primary. +$primary->wait_for_replay_catchup($standby1); +$result = $standby1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', "standby1 gets data from primary"); + +# Validate that synchronized_standby_slots takes precedence over waiting for +# changes to have been synchronous replicated. +# Since the slot specified (sb1_slot) has received the changes, primary can send +# the decoded changes to the subscription enabled for failover +# (i.e. regress_mysub1). +$primary->wait_for_catchup('regress_mysub1'); +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber1 gets data from primary after standby1 acknowledges changes"); + +$standby2->start; +$primary->wait_for_replay_catchup($standby2); + +# Unset synchronized_standby_slots to test synchronous replication +# blocks primary from sending logical decoded changes to failover slots until +# changes have been synchronously replicated. +$primary->append_conf( + 'postgresql.conf', qq( + synchronized_standby_slots = '')); +$primary->reload; + +$primary_row_count = 20; +$standby2->stop; + +$sync_back_q->query_until( + qr/insert_blocked_on_additional_sync_rep/, q( + \echo insert_blocked_on_additional_sync_rep + INSERT INTO tab_int SELECT generate_series(11, 20); +)); + +# Since $standby2 has not received the changes, validate that subscriber1 (failover = true) +# has not received the decoded changes, but subscriber2 (failover = false) has. +$primary->wait_for_catchup('regress_mysub2'); + +$result = $subscriber2->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber2 gets data from primary even if the changes have not been synchronously acknowledged."); + +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 'f', + "subscriber1 does not get data from primary since changes have not been synchronously acknowledged."); + +# Start standby2 to allow the changes to be acknowledged by all the synchronous standbys. +$standby2->start; +$primary->wait_for_replay_catchup($standby2); +$primary->wait_for_catchup('regress_mysub1'); + +# Now that the changes have been replicated to all synchronous nodes, +# primary can send the decoded changes to the subscription enabled for failover +# (i.e. regress_mysub1). While the standby was down, regress_mysub1 didn't +# receive any data from the primary. i.e. the primary didn't allow it to go +# ahead of requirements of synchronous commit. +$result = $subscriber1->safe_psql('postgres', + "SELECT count(*) = $primary_row_count FROM tab_int;"); +is($result, 't', + "subscriber1 gets data from primary since changes have been syhcronously acknowledged."); +# No longer need to run the background session. +$sync_back_q->quit; + +# Reset synchronized_standby_slots and synchronous commit for below test cases +$primary->append_conf( + 'postgresql.conf', qq( +synchronized_standby_slots = 'sb1_slot' +synchronous_standby_names = '' +)); +$primary->reload; + ################################################## # Verify that when using pg_logical_slot_get_changes to consume changes from a # logical failover slot, it will also wait for the slots specified in -- 2.40.1 ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> @ 2024-07-18 03:55 ` shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: shveta malik @ 2024-07-18 03:55 UTC (permalink / raw) To: John H <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; shveta malik <[email protected]> On Tue, Jul 9, 2024 at 12:42 AM John H <[email protected]> wrote: > > > > Can we make it a default > > behavior that logical slots marked with a failover option will wait > > for 'synchronous_standby_names' as per your patch's idea unless > > 'standby_slot_names' is specified? I don't know if there is any value > > in setting the 'failover' option for a slot without specifying > > 'standby_slot_names', so was wondering if we can additionally tie it > > to 'synchronous_standby_names'. Any better ideas? > > > > No, I think that works pretty cleanly actually. Reserving some special > keyword isn't great > which is the only other thing I can think of. I've updated the patch > and tests to reflect that. > > Attached the patch that addresses these changes. Thank You for the patch. I like the overall idea, it is a useful one and will make user experience better. Please find few comments. 1) I agree with the idea that instead of introducing new GUC, we can make failover logical slots wait on 'synchronous_standby_names', but this will leave user no option to unlink failover-enabled logical subscribers from the wait on synchronous standbys. We provide user a way to switch off automatic slot-sync by disabling 'sync_replication_slots' and we also provide user a way to manually sync the slots using function 'pg_sync_replication_slots()' and nowhere we make it mandatory to set 'synchronized_standby_slots'; in fact in docs, it is a recommended setting and not a mandatory one. User can always create failover slots, switch off automatic slot sync and disable wait on standbys by not specifying 'synchronized_standby_slots' and do the slot-sync and consistency checks manually when needed. I feel, for worst case scenarios, we should provide user an option to delink failover-enabled logical subscriptions from 'synchronous_standby_names'. We can have 'synchronized_standby_slots' (standby_slot_names) to accept one such option as in 'SAME_AS_SYNCREP_STANDBYS' or say 'DEFAULT'. So when 'synchronous_standby_names' is comma separated list, we pick those slots; if it is empty, then no wait on standbys, and if its value is 'DEFAULT' as configured by user, then go with 'synchronous_standby_names'. Thoughts? 2) When 'synchronized_standby_slots' is configured but standby named in it is down blocking logical replication, then we get a WARNING in subscriber's log file: WARNING: replication slot "standby_2" specified in parameter synchronized_standby_slots does not have active_pid. DETAIL: Logical replication is waiting on the standby associated with "standby_2". HINT: Consider starting standby associated with "standby_2" or amend parameter synchronized_standby_slots. But OTOH, when 'synchronous_standby_names' is configured instead of 'synchronized_standby_slots' and any of the standbys listed is down blocking logical replication, we do not get any sort of warning. It is inconsistent behavior. Also user might be left clueless on why subscribers are not getting changes. thanks Shveta ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> @ 2024-07-18 04:27 ` shveta malik <[email protected]> 2024-07-18 21:21 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: shveta malik @ 2024-07-18 04:27 UTC (permalink / raw) To: John H <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; shveta malik <[email protected]> On Thu, Jul 18, 2024 at 9:25 AM shveta malik <[email protected]> wrote: > > On Tue, Jul 9, 2024 at 12:42 AM John H <[email protected]> wrote: > > > > > > > Can we make it a default > > > behavior that logical slots marked with a failover option will wait > > > for 'synchronous_standby_names' as per your patch's idea unless > > > 'standby_slot_names' is specified? I don't know if there is any value > > > in setting the 'failover' option for a slot without specifying > > > 'standby_slot_names', so was wondering if we can additionally tie it > > > to 'synchronous_standby_names'. Any better ideas? > > > > > > > No, I think that works pretty cleanly actually. Reserving some special > > keyword isn't great > > which is the only other thing I can think of. I've updated the patch > > and tests to reflect that. > > > > Attached the patch that addresses these changes. > > Thank You for the patch. I like the overall idea, it is a useful one > and will make user experience better. Please find few comments. > > 1) > I agree with the idea that instead of introducing new GUC, we can make > failover logical slots wait on 'synchronous_standby_names', but this > will leave user no option to unlink failover-enabled logical > subscribers from the wait on synchronous standbys. We provide user a > way to switch off automatic slot-sync by disabling > 'sync_replication_slots' and we also provide user a way to manually > sync the slots using function 'pg_sync_replication_slots()' and > nowhere we make it mandatory to set 'synchronized_standby_slots'; in > fact in docs, it is a recommended setting and not a mandatory one. > User can always create failover slots, switch off automatic slot sync > and disable wait on standbys by not specifying > 'synchronized_standby_slots' and do the slot-sync and consistency > checks manually when needed. I feel, for worst case scenarios, we > should provide user an option to delink failover-enabled logical > subscriptions from 'synchronous_standby_names'. > > We can have 'synchronized_standby_slots' (standby_slot_names) to > accept one such option as in 'SAME_AS_SYNCREP_STANDBYS' or say > 'DEFAULT'. So when 'synchronous_standby_names' is comma separated > list, we pick those slots; if it is empty, then no wait on standbys, > and if its value is 'DEFAULT' as configured by the user, then go with > 'synchronous_standby_names'. Thoughts? One correction here ('synchronous_standby_names-->synchronized_standby_slots). Corrected version: So when 'synchronized_standby_slots' is comma separated list, we pick those slots; if it is empty, then no wait on standbys, and if its value is 'DEFAULT' as configured by user, then go with 'synchronous_standby_names'. Thoughts? > > 2) > When 'synchronized_standby_slots' is configured but standby named in > it is down blocking logical replication, then we get a WARNING in > subscriber's log file: > > WARNING: replication slot "standby_2" specified in parameter > synchronized_standby_slots does not have active_pid. > DETAIL: Logical replication is waiting on the standby associated with > "standby_2". > HINT: Consider starting standby associated with "standby_2" or amend > parameter synchronized_standby_slots. > > But OTOH, when 'synchronous_standby_names' is configured instead of > 'synchronized_standby_slots' and any of the standbys listed is down > blocking logical replication, we do not get any sort of warning. It is > inconsistent behavior. Also user might be left clueless on why > subscribers are not getting changes. > > thanks > Shveta ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> @ 2024-07-18 21:21 ` John H <[email protected]> 2024-07-22 03:41 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: John H @ 2024-07-18 21:21 UTC (permalink / raw) To: shveta malik <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers Hi Shveta, Thanks for taking a look at the patch. > > will leave user no option to unlink failover-enabled logical > > subscribers from the wait on synchronous standbys. That's a good point. I'm a bit biased in that I don't think there's a great reason why someone would want to replicate logical changes out of the synchronous cluster without it having been synchronously replicated but yes this would be different behavior compared to strictly the slot one. > ... > So when 'synchronized_standby_slots' is comma separated list, we pick > those slots; if it is empty, then no wait on standbys, and if its > value is 'DEFAULT' as configured by user, then go with > 'synchronous_standby_names'. Thoughts? I think I'd prefer having a separate GUC if the alternative is to reserve special keywords in 'synchronized_standby_slots' but I'm not sure if I feel strongly about that. > > 2) > > When 'synchronized_standby_slots' is configured but standby named in > > it is down blocking logical replication, then we get a WARNING in > > subscriber's log file: > > > > WARNING: replication slot "standby_2" specified in parameter > > synchronized_standby_slots does not have active_pid. > > DETAIL: Logical replication is waiting on the standby associated with > > "standby_2". > > HINT: Consider starting standby associated with "standby_2" or amend > > parameter synchronized_standby_slots. > > > > But OTOH, when 'synchronous_standby_names' is configured instead of > > 'synchronized_standby_slots' and any of the standbys listed is down > > blocking logical replication, we do not get any sort of warning. It is > > inconsistent behavior. Also user might be left clueless on why > > subscribers are not getting changes. Ah that's a gap. Let me add some logging/warning in a similar fashion. Although I think I'd have the warning be relatively generic (e.g. changes are blocked because they're not synchronously committed) Thanks, -- John Hsu - Amazon Web Services ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 21:21 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> @ 2024-07-22 03:41 ` shveta malik <[email protected]> 2024-08-26 19:25 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: shveta malik @ 2024-07-22 03:41 UTC (permalink / raw) To: John H <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; shveta malik <[email protected]> On Fri, Jul 19, 2024 at 2:52 AM John H <[email protected]> wrote: > > Hi Shveta, > > Thanks for taking a look at the patch. > > > > will leave user no option to unlink failover-enabled logical > > > subscribers from the wait on synchronous standbys. > > That's a good point. I'm a bit biased in that I don't think there's a > great reason why someone would > want to replicate logical changes out of the synchronous cluster > without it having been synchronously replicated > but yes this would be different behavior compared to strictly the slot one. > > > ... > > So when 'synchronized_standby_slots' is comma separated list, we pick > > those slots; if it is empty, then no wait on standbys, and if its > > value is 'DEFAULT' as configured by user, then go with > > 'synchronous_standby_names'. Thoughts? > > I think I'd prefer having a separate GUC if the alternative is to reserve > special keywords in 'synchronized_standby_slots' but I'm not sure if I > feel strongly about that. My only concern is, earlier we provided a way to set the failover property of slots even without mandatorily wait on physical standbys. But now we will be changing this behaviour. Okay, we can see what other comments. If we plan to go this way, we can change docs to clearly mention this. > > > 2) > > > When 'synchronized_standby_slots' is configured but standby named in > > > it is down blocking logical replication, then we get a WARNING in > > > subscriber's log file: > > > > > > WARNING: replication slot "standby_2" specified in parameter > > > synchronized_standby_slots does not have active_pid. > > > DETAIL: Logical replication is waiting on the standby associated with > > > "standby_2". > > > HINT: Consider starting standby associated with "standby_2" or amend > > > parameter synchronized_standby_slots. > > > > > > But OTOH, when 'synchronous_standby_names' is configured instead of > > > 'synchronized_standby_slots' and any of the standbys listed is down > > > blocking logical replication, we do not get any sort of warning. It is > > > inconsistent behavior. Also user might be left clueless on why > > > subscribers are not getting changes. > > Ah that's a gap. Let me add some logging/warning in a similar fashion. > Although I think I'd have the warning be relatively generic (e.g. > changes are blocked because > they're not synchronously committed) > okay, sounds good. thanks Shveta ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 21:21 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-22 03:41 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> @ 2024-08-26 19:25 ` John H <[email protected]> 2024-08-28 03:16 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 0 siblings, 1 reply; 8+ messages in thread From: John H @ 2024-08-26 19:25 UTC (permalink / raw) To: shveta malik <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers Hi Shveta, On Sun, Jul 21, 2024 at 8:42 PM shveta malik <[email protected]> wrote: > > Ah that's a gap. Let me add some logging/warning in a similar fashion. > > Although I think I'd have the warning be relatively generic (e.g. > > changes are blocked because > > they're not synchronously committed) > > > > okay, sounds good. > > thanks > Shveta I took a look at having similar warnings the existing 'synchronized_standby_slots' feature has and I don't think it's particularly feasible. The checks/warnings for 'synchronized_standby_slots' are intended to protect against misconfiguration. They consist of slot validation (valid slot_name, not logical slot, slot has not been invalidated), and whether or not the slot is active. I don't think there's a "misconfiguration" equivalent for waiting on synchronous_commit. With the current proposal, once you have (synchronous_commit enabled && failover_slots), logical decoding is dependent on whether or not the writes have been replicated to a synchronous replica. If there is no data being replicated out of the logical slot, it is because from the perspective of the database no writes have been committed yet. I don't think it would make sense to add logging/warning as to why a transaction is still not committed (e.g. which potential replica is the one lagging). There isn't a nice way to determine why synchronous commit is waiting without being particularly invasive, and even then it could be particularly noisy (e.g. provide all the application_names that we are potentially waiting on). Thanks, -- John Hsu - Amazon Web Services ^ permalink raw reply [nested|flat] 8+ messages in thread
* Re: Allow logical failover slots to wait on synchronous replication 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 21:21 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-22 03:41 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-08-26 19:25 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> @ 2024-08-28 03:16 ` shveta malik <[email protected]> 0 siblings, 0 replies; 8+ messages in thread From: shveta malik @ 2024-08-28 03:16 UTC (permalink / raw) To: John H <[email protected]>; +Cc: Amit Kapila <[email protected]>; [email protected] <[email protected]>; pgsql-hackers; shveta malik <[email protected]> On Tue, Aug 27, 2024 at 12:56 AM John H <[email protected]> wrote: > > Hi Shveta, > > On Sun, Jul 21, 2024 at 8:42 PM shveta malik <[email protected]> wrote: > > > > Ah that's a gap. Let me add some logging/warning in a similar fashion. > > > Although I think I'd have the warning be relatively generic (e.g. > > > changes are blocked because > > > they're not synchronously committed) > > > > > > > okay, sounds good. > > > > thanks > > Shveta > > I took a look at having similar warnings the existing > 'synchronized_standby_slots' feature has > and I don't think it's particularly feasible. > > The checks/warnings for 'synchronized_standby_slots' are intended to > protect against misconfiguration. > They consist of slot validation (valid slot_name, not logical slot, > slot has not been invalidated), and > whether or not the slot is active. > > I don't think there's a "misconfiguration" equivalent for waiting on > synchronous_commit. > With the current proposal, once you have (synchronous_commit enabled > && failover_slots), logical > decoding is dependent on whether or not the writes have been > replicated to a synchronous replica. > If there is no data being replicated out of the logical slot, it is > because from the perspective of the > database no writes have been committed yet. I don't think it would > make sense to add logging/warning as to > why a transaction is still not committed (e.g. which potential replica > is the one lagging). There isn't a > nice way to determine why synchronous commit is waiting without being > particularly invasive, and even then > it could be particularly noisy (e.g. provide all the application_names > that we are potentially waiting on). > Okay. Thanks for the details. I see your point. I will review to see if anything comes to my mind for a simpler way to do this. thanks Shveta ^ permalink raw reply [nested|flat] 8+ messages in thread
end of thread, other threads:[~2024-08-28 03:16 UTC | newest] Thread overview: 8+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2022-07-19 04:23 [PATCH v24 1/2] In-place table persistence change Kyotaro Horiguchi <[email protected]> 2024-07-08 19:12 Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-18 03:55 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 04:27 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-07-18 21:21 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-07-22 03:41 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[email protected]> 2024-08-26 19:25 ` Re: Allow logical failover slots to wait on synchronous replication John H <[email protected]> 2024-08-28 03:16 ` Re: Allow logical failover slots to wait on synchronous replication shveta malik <[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