From: Andrey Lepikhov Date: Wed, 17 Nov 2021 11:13:37 +0500 Subject: [PATCH] Add Commit Sequence Number (CSN) machinery into MVCC implementation for a timestamp-based resolving of visibility conflicts. It allows to achieve proper snapshot isolation semantics in the case of distributed transactions involving more than one Postgres instance. Authors: K.Knizhnik, S.Kelvich, A.Sher, A.Lepikhov, M.Usama. Discussion: (2020/05/21 -) https://www.postgresql.org/message-id/flat/CA%2Bfd4k6HE8xLGEvqWzABEg8kkju5MxU%2Bif7bf-md0_2pjzXp9Q%40mail.gmail.com#ed1359340871688bed2e643921f73365 (2018/05/01 - 2019/04/21) https://www.postgresql.org/message-id/flat/21BC916B-80A1-43BF-8650-3363CCDAE09C%40postgrespro.ru --- doc/src/sgml/config.sgml | 50 +- src/backend/access/rmgrdesc/Makefile | 1 + src/backend/access/rmgrdesc/csnlogdesc.c | 95 +++ src/backend/access/rmgrdesc/xlogdesc.c | 6 +- src/backend/access/transam/Makefile | 2 + src/backend/access/transam/csn_log.c | 748 ++++++++++++++++++ src/backend/access/transam/csn_snapshot.c | 687 ++++++++++++++++ src/backend/access/transam/rmgr.c | 1 + src/backend/access/transam/twophase.c | 154 ++++ src/backend/access/transam/varsup.c | 2 + src/backend/access/transam/xact.c | 32 + src/backend/access/transam/xlog.c | 23 +- src/backend/access/transam/xloginsert.c | 2 + src/backend/commands/vacuum.c | 3 +- src/backend/storage/ipc/ipci.c | 6 + src/backend/storage/ipc/procarray.c | 85 ++ src/backend/storage/lmgr/lwlock.c | 2 + src/backend/storage/lmgr/lwlocknames.txt | 2 + src/backend/storage/lmgr/proc.c | 6 + src/backend/storage/sync/sync.c | 5 + src/backend/utils/misc/guc.c | 37 + src/backend/utils/probes.d | 2 + src/backend/utils/time/snapmgr.c | 149 +++- src/bin/initdb/initdb.c | 3 +- src/bin/pg_controldata/pg_controldata.c | 2 + src/bin/pg_upgrade/pg_upgrade.c | 5 + src/bin/pg_upgrade/pg_upgrade.h | 2 + src/bin/pg_waldump/rmgrdesc.c | 1 + src/include/access/csn_log.h | 98 +++ src/include/access/csn_snapshot.h | 54 ++ src/include/access/rmgrlist.h | 1 + src/include/access/xlog_internal.h | 2 + src/include/catalog/pg_control.h | 1 + src/include/catalog/pg_proc.dat | 17 + src/include/datatype/timestamp.h | 3 + src/include/fmgr.h | 1 + src/include/portability/instr_time.h | 10 + src/include/storage/lwlock.h | 1 + src/include/storage/proc.h | 14 + src/include/storage/procarray.h | 7 + src/include/storage/sync.h | 1 + src/include/utils/snapmgr.h | 7 +- src/include/utils/snapshot.h | 11 + src/test/modules/Makefile | 1 + src/test/modules/csnsnapshot/Makefile | 25 + .../modules/csnsnapshot/csn_snapshot.conf | 1 + .../csnsnapshot/expected/csnsnapshot.out | 1 + src/test/modules/csnsnapshot/t/001_base.pl | 103 +++ src/test/modules/csnsnapshot/t/002_standby.pl | 66 ++ .../modules/csnsnapshot/t/003_time_skew.pl | 214 +++++ .../csnsnapshot/t/004_read_committed.pl | 97 +++ .../csnsnapshot/t/005_basic_visibility.pl | 181 +++++ src/test/modules/snapshot_too_old/sto.conf | 1 + src/test/regress/expected/sysviews.out | 4 +- 54 files changed, 3024 insertions(+), 11 deletions(-) create mode 100644 src/backend/access/rmgrdesc/csnlogdesc.c create mode 100644 src/backend/access/transam/csn_log.c create mode 100644 src/backend/access/transam/csn_snapshot.c create mode 100644 src/include/access/csn_log.h create mode 100644 src/include/access/csn_snapshot.h create mode 100644 src/test/modules/csnsnapshot/Makefile create mode 100644 src/test/modules/csnsnapshot/csn_snapshot.conf create mode 100644 src/test/modules/csnsnapshot/expected/csnsnapshot.out create mode 100644 src/test/modules/csnsnapshot/t/001_base.pl create mode 100644 src/test/modules/csnsnapshot/t/002_standby.pl create mode 100644 src/test/modules/csnsnapshot/t/003_time_skew.pl create mode 100644 src/test/modules/csnsnapshot/t/004_read_committed.pl create mode 100644 src/test/modules/csnsnapshot/t/005_basic_visibility.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 3f806740d5..f4f6c83fd0 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -9682,8 +9682,56 @@ dynamic_library_path = 'C:\tools\postgresql;H:\my_project\lib;$libdir' - + + CSN Based Snapshot + + By default, snapshots in PostgreSQL contains a + XID (TransactionID) that allows to identify the status of a transaction + and make arbitrary visibility calculations. + + + + PostgreSQL also provides a CSN (Commit + Sequence Number) based machinery as an additional tool for visibility + calculations. It may be used within distributed transactions when a xid of + a local transaction can't correctly identify order of the distributed one. + + + + + enable_csn_snapshot (boolean) + + enable_csn_snapshot configuration parameter + + + + + + Enable/disable the CSN tracking for the snapshot. + + + + PostgreSQL uses a physical clock timestamp as + a CSN, so enabling the CSN based snapshots can be useful for implementing + cross-instance snapshots and visibility of distributed transaction. + + + + when enabled PostgreSQL creates + pg_csn directory under PGDATA to keep + the track of CSN and XID mappings. + + + + The default value is on. + + + + + + + Version and Platform Compatibility diff --git a/src/backend/access/rmgrdesc/Makefile b/src/backend/access/rmgrdesc/Makefile index f88d72fd86..15fc36f7b4 100644 --- a/src/backend/access/rmgrdesc/Makefile +++ b/src/backend/access/rmgrdesc/Makefile @@ -11,6 +11,7 @@ include $(top_builddir)/src/Makefile.global OBJS = \ brindesc.o \ clogdesc.o \ + csnlogdesc.o \ committsdesc.o \ dbasedesc.o \ genericdesc.o \ diff --git a/src/backend/access/rmgrdesc/csnlogdesc.c b/src/backend/access/rmgrdesc/csnlogdesc.c new file mode 100644 index 0000000000..f8c644e906 --- /dev/null +++ b/src/backend/access/rmgrdesc/csnlogdesc.c @@ -0,0 +1,95 @@ +/*------------------------------------------------------------------------- + * + * clogdesc.c + * rmgr descriptor routines for access/transam/csn_log.c + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * src/backend/access/rmgrdesc/csnlogdesc.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/csn_log.h" + + +void +csnlog_desc(StringInfo buf, XLogReaderState *record) +{ + char *rec = XLogRecGetData(record); + uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK; + + if (info == XLOG_CSN_ZEROPAGE) + { + int pageno; + + memcpy(&pageno, XLogRecGetData(record), sizeof(int)); + appendStringInfo(buf, "pageno %d", pageno); + } + else if (info == XLOG_CSN_TRUNCATE) + { + int pageno; + + memcpy(&pageno, XLogRecGetData(record), sizeof(int)); + appendStringInfo(buf, "pageno %d", pageno); + } + else if (info == XLOG_CSN_ASSIGNMENT) + { + CSN csn; + + memcpy(&csn, XLogRecGetData(record), sizeof(CSN)); + appendStringInfo(buf, "assign "INT64_FORMAT"", csn); + } + else if (info == XLOG_CSN_SETCSN) + { + xl_csn_set *xlrec = (xl_csn_set *) rec; + int nsubxids; + + appendStringInfo(buf, "set "INT64_FORMAT" for: %u", + xlrec->csn, + xlrec->xtop); + nsubxids = ((XLogRecGetDataLen(record) - MinSizeOfCSNSet) / + sizeof(TransactionId)); + if (nsubxids > 0) + { + int i; + TransactionId *subxids; + + subxids = palloc(sizeof(TransactionId) * nsubxids); + memcpy(subxids, + XLogRecGetData(record) + MinSizeOfCSNSet, + sizeof(TransactionId) * nsubxids); + for (i = 0; i < nsubxids; i++) + appendStringInfo(buf, ", %u", subxids[i]); + pfree(subxids); + } + } +} + +const char * +csnlog_identify(uint8 info) +{ + const char *id = NULL; + + switch (info & ~XLR_INFO_MASK) + { + case XLOG_CSN_ASSIGNMENT: + id = "ASSIGNMENT"; + break; + case XLOG_CSN_SETCSN: + id = "SETCSN"; + break; + case XLOG_CSN_ZEROPAGE: + id = "ZEROPAGE"; + break; + case XLOG_CSN_TRUNCATE: + id = "TRUNCATE"; + break; + } + + return id; +} diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.c index 5bf2346dd9..ea433046cf 100644 --- a/src/backend/access/rmgrdesc/xlogdesc.c +++ b/src/backend/access/rmgrdesc/xlogdesc.c @@ -113,7 +113,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record) appendStringInfo(buf, "max_connections=%d max_worker_processes=%d " "max_wal_senders=%d max_prepared_xacts=%d " "max_locks_per_xact=%d wal_level=%s " - "wal_log_hints=%s track_commit_timestamp=%s", + "wal_log_hints=%s track_commit_timestamp=%s " + "enable_csn_snapshot=%s", xlrec.MaxConnections, xlrec.max_worker_processes, xlrec.max_wal_senders, @@ -121,7 +122,8 @@ xlog_desc(StringInfo buf, XLogReaderState *record) xlrec.max_locks_per_xact, wal_level_str, xlrec.wal_log_hints ? "on" : "off", - xlrec.track_commit_timestamp ? "on" : "off"); + xlrec.track_commit_timestamp ? "on" : "off", + xlrec.enable_csn_snapshot ? "on" : "off"); } else if (info == XLOG_FPW_CHANGE) { diff --git a/src/backend/access/transam/Makefile b/src/backend/access/transam/Makefile index 595e02de72..fc0321ee6b 100644 --- a/src/backend/access/transam/Makefile +++ b/src/backend/access/transam/Makefile @@ -15,6 +15,8 @@ include $(top_builddir)/src/Makefile.global OBJS = \ clog.o \ commit_ts.o \ + csn_log.o \ + csn_snapshot.o \ generic_xlog.o \ multixact.o \ parallel.o \ diff --git a/src/backend/access/transam/csn_log.c b/src/backend/access/transam/csn_log.c new file mode 100644 index 0000000000..33517271ed --- /dev/null +++ b/src/backend/access/transam/csn_log.c @@ -0,0 +1,748 @@ +/*----------------------------------------------------------------------------- + * + * csn_log.c + * Track commit sequence numbers of finished transactions + * + * This module provides SLRU to store CSN for each transaction. This + * mapping need to be kept only for xid's greater then oldestXid, but + * that can require arbitrary large amounts of memory in case of long-lived + * transactions. Because of same lifetime and persistancy requirements + * this module is quite similar to subtrans.c + * + * If we switch database from CSN-base snapshot to xid-base snapshot then, + * nothing wrong. But if we switch xid-base snapshot to CSN-base snapshot + * it should decide a new xid which begin csn-base check. It can not be + * oldestActiveXID because of prepared transaction. + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/backend/access/transam/csn_log.c + * + *----------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/csn_log.h" +#include "access/slru.h" +#include "access/csn_snapshot.h" +#include "access/subtrans.h" +#include "access/transam.h" +#include "access/xlogutils.h" +#include "miscadmin.h" +#include "pg_trace.h" +#include "portability/instr_time.h" +#include "storage/shmem.h" +#include "storage/spin.h" +#include "utils/snapmgr.h" +#include "access/xlog_internal.h" + +/* + * We use csnSnapshotActive to judge if csn snapshot enabled instead of by + * enable_csn_snapshot, this design is similar to 'track_commit_timestamp'. + * + * Because in process of replication if master changes 'enable_csn_snapshot' + * in a database restart, standby should apply wal record for GUC changed, + * then it's difficult to notice all backends about that. So they can get + * the message by 'csnSnapshotActive' which in shared buffer. It will not + * acquire a lock, so without performance issue. + * last_max_csn - Record the max csn till now. + * last_csn_log_wal - for interval we log the assign csn to wal + * oldestXmin - first sensible Xmin on the first existed page in the CSN Log + */ +typedef struct CSNShared +{ + bool csnSnapshotActive; + pg_atomic_uint32 oldestXmin; + CSN last_max_csn; + CSN last_csn_log_wal; + volatile slock_t lock; +} CSNShared; + +CSNShared *csnShared; + +/* + * Defines for CSNLog page sizes. A page is the same BLCKSZ as is used + * everywhere else in Postgres. + * + * Note: because TransactionIds are 32 bits and wrap around at 0xFFFFFFFF, + * CSNLog page numbering also wraps around at + * 0xFFFFFFFF/CSN_LOG_XACTS_PER_PAGE, and CSNLog segment numbering at + * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT. We need take no + * explicit notice of that fact in this module, except when comparing segment + * and page numbers in TruncateCSNLog (see CSNLogPagePrecedes). + */ + +/* We store the commit CSN for each xid */ +#define CSN_LOG_XACTS_PER_PAGE (BLCKSZ / sizeof(CSN)) + +#define TransactionIdToPage(xid) ((xid) / (TransactionId) CSN_LOG_XACTS_PER_PAGE) +#define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CSN_LOG_XACTS_PER_PAGE) + +/* + * Link to shared-memory data structures for CLOG control + */ +static SlruCtlData CSNLogCtlData; +#define CsnlogCtl (&CSNLogCtlData) + +static int ZeroCSNLogPage(int pageno, bool write_xlog); +static void ZeroTruncateCSNLogPage(int pageno, bool write_xlog); +static bool CSNLogPagePrecedes(int page1, int page2); +static void CSNLogSetPageStatus(TransactionId xid, int nsubxids, + TransactionId *subxids, + CSN csn, int pageno); +static void CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno); + +static void WriteCSNXlogRec(TransactionId xid, int nsubxids, + TransactionId *subxids, CSN csn); +static void WriteZeroCSNPageXlogRec(int pageno); +static void WriteTruncateCSNXlogRec(int pageno); +static void set_oldest_xmin(TransactionId xid); + + +/* + * Number of shared CSNLog buffers. + */ +static Size +CSNLogShmemBuffers(void) +{ + return Min(32, Max(4, NBuffers / 512)); +} + +/* + * Reserve shared memory for CsnlogCtl. + */ +Size +CSNLogShmemSize(void) +{ + return SimpleLruShmemSize(CSNLogShmemBuffers(), 0); +} + +/* + * Initialization of shared memory for CSNLog. + */ +void +CSNLogShmemInit(void) +{ + bool found; + + CsnlogCtl->PagePrecedes = CSNLogPagePrecedes; + SimpleLruInit(CsnlogCtl, "CSNLog Ctl", CSNLogShmemBuffers(), 0, + CSNLogSLRULock, "pg_csn", LWTRANCHE_CSN_LOG_BUFFERS, + SYNC_HANDLER_CSN); + + csnShared = ShmemInitStruct("CSNlog shared", + sizeof(CSNShared), + &found); + if (!found) + { + csnShared->csnSnapshotActive = false; + pg_atomic_init_u32(&csnShared->oldestXmin, InvalidTransactionId); + csnShared->last_max_csn = InvalidCSN; + csnShared->last_csn_log_wal = InvalidCSN; + SpinLockInit(&csnShared->lock); + } +} + +/* + * CSNLogSetCSN + * + * Record CSN of transaction and its subtransaction tree. + * + * xid is a single xid to set status for. This will typically be the top level + * transactionid for a top level commit or abort. It can also be a + * subtransaction when we record transaction aborts. + * + * subxids is an array of xids of length nsubxids, representing subtransactions + * in the tree of xid. In various cases nsubxids may be zero. + * + * csn is the commit sequence number of the transaction. It should be + * AbortedCSN for abort cases. + */ +void +CSNLogSetCSN(TransactionId xid, int nsubxids, TransactionId *subxids, CSN csn, + bool write_xlog) +{ + int pageno; + int i = 0; + int offset = 0; + + Assert(TransactionIdIsValid(xid)); + + pageno = TransactionIdToPage(xid); /* get page of parent */ + + if(write_xlog) + WriteCSNXlogRec(xid, nsubxids, subxids, csn); + + for (;;) + { + int num_on_page = 0; + + /* Form subtransactions bucket that can be written on the same page */ + while (i < nsubxids && TransactionIdToPage(subxids[i]) == pageno) + { + num_on_page++; + i++; + } + + CSNLogSetPageStatus(xid, + num_on_page, subxids + offset, + csn, pageno); + if (i >= nsubxids) + break; + + offset = i; + pageno = TransactionIdToPage(subxids[offset]); + xid = InvalidTransactionId; + } +} + +/* + * Record the final state of transaction entries in the csn log for + * all entries on a single page. Atomic only on this page. + * + * Otherwise API is same as TransactionIdSetTreeStatus() + */ +static void +CSNLogSetPageStatus(TransactionId xid, int nsubxids, TransactionId *subxids, + CSN csn, int pageno) +{ + int slotno; + int i; + + LWLockAcquire(CSNLogSLRULock, LW_EXCLUSIVE); + + slotno = SimpleLruReadPage(CsnlogCtl, pageno, true, xid); + + /* Subtransactions first, if needed ... */ + for (i = 0; i < nsubxids; i++) + { + Assert(CsnlogCtl->shared->page_number[slotno] == TransactionIdToPage(subxids[i])); + CSNLogSetCSNInSlot(subxids[i], csn, slotno); + } + + /* ... then the main transaction */ + if (TransactionIdIsValid(xid)) + CSNLogSetCSNInSlot(xid, csn, slotno); + + CsnlogCtl->shared->page_dirty[slotno] = true; + + LWLockRelease(CSNLogSLRULock); +} + +/* + * Sets the commit status of a single transaction. + */ +static void +CSNLogSetCSNInSlot(TransactionId xid, CSN csn, int slotno) +{ + int entryno = TransactionIdToPgIndex(xid); + CSN *ptr; + + Assert(LWLockHeldByMe(CSNLogSLRULock)); + + ptr = (CSN *) (CsnlogCtl->shared->page_buffer[slotno] + + entryno * sizeof(CSN)); + *ptr = csn; +} + +/* + * Interrogate the state of a transaction in the log. + * + * NB: this is a low-level routine and is NOT the preferred entry point + * for most uses; TransactionIdGetCSN() in csn_snapshot.c is the + * intended caller. + */ +CSN +CSNLogGetCSNByXid(TransactionId xid) +{ + int pageno = TransactionIdToPage(xid); + int entryno = TransactionIdToPgIndex(xid); + int slotno; + CSN csn; + + /* lock is acquired by SimpleLruReadPage_ReadOnly */ + slotno = SimpleLruReadPage_ReadOnly(CsnlogCtl, pageno, xid); + csn = *(CSN *) (CsnlogCtl->shared->page_buffer[slotno] + + entryno * sizeof(CSN)); + LWLockRelease(CSNLogSLRULock); + + return csn; +} + +/* + * Initialize (or reinitialize) a page of CSNLog to zeroes. + * + * The page is not actually written, just set up in shared memory. + * The slot number of the new page is returned. + * + * Control lock must be held at entry, and will be held at exit. + */ +static int +ZeroCSNLogPage(int pageno, bool write_xlog) +{ + Assert(LWLockHeldByMe(CSNLogSLRULock)); + if(write_xlog) + WriteZeroCSNPageXlogRec(pageno); + return SimpleLruZeroPage(CsnlogCtl, pageno); +} + +static void +ZeroTruncateCSNLogPage(int pageno, bool write_xlog) +{ + if(write_xlog) + WriteTruncateCSNXlogRec(pageno); + SimpleLruTruncate(CsnlogCtl, pageno); +} + +void +ActivateCSNlog(void) +{ + int pageno; + TransactionId nextXid = InvalidTransactionId; + TransactionId oldest_xid = InvalidTransactionId; + + if (csnShared->csnSnapshotActive) + return; + + nextXid = XidFromFullTransactionId(ShmemVariableCache->nextXid); + pageno = TransactionIdToPage(nextXid); + + LWLockAcquire(CSNLogSLRULock, LW_EXCLUSIVE); + + /* + * Create the current segment file, if necessary. + * This means that + */ + if (!SimpleLruDoesPhysicalPageExist(CsnlogCtl, pageno)) + { + int slotno; + TransactionId curxid = nextXid; + + slotno = ZeroCSNLogPage(pageno, false); + SimpleLruWritePage(CsnlogCtl, slotno); + + elog(LOG, "Create SLRU page=%d, slotno=%d for xid %u on a CSN log activation", + pageno, slotno, nextXid); + + /* + * nextXid isn't first xid on the page. It is the first page in the CSN + * log. Set UnclearCSN value into all previous slots on this page. + * This xid value can be used as an oldest xid in the CSN log. + */ + if (TransactionIdToPgIndex(nextXid) > 0) + { + /* Cleaning procedure. Can be optimized. */ + do + { + curxid--; + CSNLogSetCSNInSlot(curxid, UnclearCSN, slotno); + } while (TransactionIdToPgIndex(curxid) > 0); + + elog(LOG, + "Set UnclearCSN values for %d xids in the range [%u,%u]", + nextXid - curxid, curxid, nextXid-1); + + /* Oldest XID found on this page */ + oldest_xid = nextXid; + } + } + LWLockRelease(CSNLogSLRULock); + + if (!TransactionIdIsValid(oldest_xid)) + { + TransactionId curxid; + + elog(LOG, "Search for the oldest xid across previous pages"); + + /* Need to scan previous pages for an oldest xid. */ + while (pageno > 0 && SimpleLruDoesPhysicalPageExist(CsnlogCtl, pageno - 1)) + pageno--; + + /* look up for the first clear xid value. */ + curxid = pageno * (TransactionId) CSN_LOG_XACTS_PER_PAGE; + while(CSNLogGetCSNByXid(curxid) == UnclearCSN) + curxid++; + oldest_xid = curxid; + } + + set_oldest_xmin(oldest_xid); + csnShared->csnSnapshotActive = true; +} + +bool +get_csnlog_status(void) +{ + return csnShared->csnSnapshotActive; +} + +void +DeactivateCSNlog(void) +{ + csnShared->csnSnapshotActive = false; + set_oldest_xmin(InvalidTransactionId); + LWLockAcquire(CSNLogSLRULock, LW_EXCLUSIVE); + (void) SlruScanDirectory(CsnlogCtl, SlruScanDirCbDeleteAll, NULL); + LWLockRelease(CSNLogSLRULock); + elog(LOG, "CSN log has deactivated"); +} + +void +StartupCSN(void) +{ + ActivateCSNlog(); +} + +void +CompleteCSNInitialization(void) +{ + /* + * If the feature is not enabled, turn it off for good. This also removes + * any leftover data. + * + * Conversely, we activate the module if the feature is enabled. This is + * necessary for primary and standby as the activation depends on the + * control file contents at the beginning of recovery or when a + * XLOG_PARAMETER_CHANGE is replayed. + */ + if (!enable_csn_snapshot) + DeactivateCSNlog(); + else + ActivateCSNlog(); +} + +void +CSNlogParameterChange(bool newvalue, bool oldvalue) +{ + if (newvalue) + { + if (!csnShared->csnSnapshotActive) + ActivateCSNlog(); + } + else if (csnShared->csnSnapshotActive) + DeactivateCSNlog(); +} + +/* + * Perform a checkpoint --- either during shutdown, or on-the-fly + */ +void +CheckPointCSNLog(void) +{ + if (!get_csnlog_status()) + return; + + /* + * Flush dirty CSNLog pages to disk. + * + * This is not actually necessary from a correctness point of view. We do + * it merely to improve the odds that writing of dirty pages is done by + * the checkpoint process and not by backends. + */ + TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_START(true); + SimpleLruWriteAll(CsnlogCtl, true); + TRACE_POSTGRESQL_CSNLOG_CHECKPOINT_DONE(true); +} + +/* + * Make sure that CSNLog has room for a newly-allocated XID. + * + * NB: this is called while holding XidGenLock. We want it to be very fast + * most of the time; even when it's not so fast, no actual I/O need happen + * unless we're forced to write out a dirty clog or xlog page to make room + * in shared memory. + */ +void +ExtendCSNLog(TransactionId newestXact) +{ + int pageno; + + if (!get_csnlog_status()) + return; + + /* + * No work except at first XID of a page. But beware: just after + * wraparound, the first XID of page zero is FirstNormalTransactionId. + */ + if (TransactionIdToPgIndex(newestXact) != 0 && + !TransactionIdEquals(newestXact, FirstNormalTransactionId)) + return; + + pageno = TransactionIdToPage(newestXact); + + LWLockAcquire(CSNLogSLRULock, LW_EXCLUSIVE); + + /* Zero the page and make an XLOG entry about it */ + ZeroCSNLogPage(pageno, !InRecovery); + + LWLockRelease(CSNLogSLRULock); +} + +/* + * Remove all CSNLog segments before the one holding the passed + * transaction ID. + * + * This is normally called during checkpoint, with oldestXact being the + * oldest TransactionXmin of any running transaction. + */ +void +TruncateCSNLog(TransactionId oldestXact) +{ + int cutoffPage; + TransactionId oldestXmin; + + /* Can't do truncation because WAL messages isn't allowed during recovery */ + if (RecoveryInProgress() || !get_csnlog_status()) + return; + + /* + * The cutoff point is the start of the segment containing oldestXact. We + * pass the *page* containing oldestXact to SimpleLruTruncate. We step + * back one transaction to avoid passing a cutoff page that hasn't been + * created yet in the rare case that oldestXact would be the first item on + * a page and oldestXact == next XID. In that case, if we didn't subtract + * one, we'd trigger SimpleLruTruncate's wraparound detection. + */ + TransactionIdRetreat(oldestXact); + cutoffPage = TransactionIdToPage(oldestXact); + + /* Detect, that we really need to cut CSN log. */ + oldestXmin = pg_atomic_read_u32(&csnShared->oldestXmin); + + if (TransactionIdToPage(oldestXmin) < cutoffPage) + { + /* OldestXact is located in the same page as oldestXmin. No actions needed. */ + return; + } + + /* + * Shift oldestXmin to the start of new first page. Use first position + * on the page because all transactions on this page is created with enabled + * CSN snapshot machinery. + */ + pg_atomic_write_u32(&csnShared->oldestXmin, + oldestXact - TransactionIdToPgIndex(oldestXact)); + + SpinLockRelease(&csnShared->lock); + ZeroTruncateCSNLogPage(cutoffPage, true); +} + +/* + * Decide which of two CSNLog page numbers is "older" for truncation + * purposes. + * + * We need to use comparison of TransactionIds here in order to do the right + * thing with wraparound XID arithmetic. However, if we are asked about + * page number zero, we don't want to hand InvalidTransactionId to + * TransactionIdPrecedes: it'll get weird about permanent xact IDs. So, + * offset both xids by FirstNormalTransactionId to avoid that. + */ +static bool +CSNLogPagePrecedes(int page1, int page2) +{ + TransactionId xid1; + TransactionId xid2; + + xid1 = ((TransactionId) page1) * CSN_LOG_XACTS_PER_PAGE; + xid1 += FirstNormalTransactionId; + xid2 = ((TransactionId) page2) * CSN_LOG_XACTS_PER_PAGE; + xid2 += FirstNormalTransactionId; + + return TransactionIdPrecedes(xid1, xid2); +} + +void +WriteAssignCSNXlogRec(CSN csn) +{ + Assert(enable_csn_wal && csn <= csnShared->last_csn_log_wal); + + XLogBeginInsert(); + XLogRegisterData((char *) (&csn), sizeof(CSN)); + XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ASSIGNMENT); +} + +static void +WriteCSNXlogRec(TransactionId xid, int nsubxids, + TransactionId *subxids, CSN csn) +{ + xl_csn_set xlrec; + + if(!enable_csn_wal) + return; + + xlrec.xtop = xid; + xlrec.nsubxacts = nsubxids; + xlrec.csn = csn; + + XLogBeginInsert(); + XLogRegisterData((char *) &xlrec, MinSizeOfCSNSet); + XLogRegisterData((char *) subxids, nsubxids * sizeof(TransactionId)); + XLogInsert(RM_CSNLOG_ID, XLOG_CSN_SETCSN); +} + +/* + * Write a ZEROPAGE xlog record + */ +static void +WriteZeroCSNPageXlogRec(int pageno) +{ + if(!enable_csn_wal) + { + return; + } + XLogBeginInsert(); + XLogRegisterData((char *) (&pageno), sizeof(int)); + (void) XLogInsert(RM_CSNLOG_ID, XLOG_CSN_ZEROPAGE); +} + +/* + * Write a TRUNCATE xlog record + */ +static void +WriteTruncateCSNXlogRec(int pageno) +{ + if(!enable_csn_wal) + { + return; + } + XLogBeginInsert(); + XLogRegisterData((char *) (&pageno), sizeof(int)); + XLogInsert(RM_CSNLOG_ID, XLOG_CSN_TRUNCATE); +} + + +void +csnlog_redo(XLogReaderState *record) +{ + uint8 info = XLogRecGetInfo(record) & ~XLR_INFO_MASK; + + /* Backup blocks are not used in csnlog records */ + Assert(!XLogRecHasAnyBlockRefs(record)); + + if (info == XLOG_CSN_ASSIGNMENT) + { + CSN csn; + + memcpy(&csn, XLogRecGetData(record), sizeof(CSN)); + /* XXX: Do we really not needed to acquire the lock here? */ + csnShared->last_max_csn = csn; + } + else if (info == XLOG_CSN_SETCSN) + { + xl_csn_set *xlrec = (xl_csn_set *) XLogRecGetData(record); + CSNLogSetCSN(xlrec->xtop, xlrec->nsubxacts, xlrec->xsub, xlrec->csn, false); + } + else if (info == XLOG_CSN_ZEROPAGE) + { + int pageno; + int slotno; + + memcpy(&pageno, XLogRecGetData(record), sizeof(int)); + LWLockAcquire(CSNLogSLRULock, LW_EXCLUSIVE); + slotno = ZeroCSNLogPage(pageno, false); + SimpleLruWritePage(CsnlogCtl, slotno); + LWLockRelease(CSNLogSLRULock); + Assert(!CsnlogCtl->shared->page_dirty[slotno]); + + } + else if (info == XLOG_CSN_TRUNCATE) + { + int pageno; + + memcpy(&pageno, XLogRecGetData(record), sizeof(int)); + CsnlogCtl->shared->latest_page_number = pageno; + ZeroTruncateCSNLogPage(pageno, false); + } + else + elog(PANIC, "csnlog_redo: unknown op code %u", info); +} + +/* + * Entrypoint for sync.c to sync members files. + */ +int +csnsyncfiletag(const FileTag *ftag, char *path) +{ + return SlruSyncFileTag(&CSNLogCtlData, ftag, path); +} + +/* + * GenerateCSN + * + * Generate CSN which is actually a local time. Also we are forcing + * this time to be always increasing. Since now it is not uncommon to have + * millions of read transactions per second we are trying to use nanoseconds + * if such time resolution is available. + */ +CSN +GenerateCSN(bool locked, CSN assign) +{ + instr_time current_time; + CSN csn; + CSN log_csn = InvalidCSN; + + Assert(get_csnlog_status() || csn_snapshot_defer_time > 0); + + /* TODO: create some macro that add small random shift to current time. */ + INSTR_TIME_SET_CURRENT(current_time); + csn = (CSN) INSTR_TIME_GET_NANOSEC(current_time) + (int64) (csn_time_shift * 1E9); + + if(assign != InvalidCSN && csn < assign) + csn = assign; + + /* TODO: change to atomics? */ + if (!locked) + SpinLockAcquire(&csnShared->lock); + + if (csn <= csnShared->last_max_csn) + csn = csnShared->last_max_csn + 1; + csnShared->last_max_csn = csn; + + if (enable_csn_wal && csn > csnShared->last_csn_log_wal) + { + /* + * We log the CSN 5s greater than generated, you can see comments on + * the CSN_ASSIGN_TIME_INTERVAL. + */ + log_csn = CSNAddByNanosec(csn, CSN_ASSIGN_TIME_INTERVAL); + csnShared->last_csn_log_wal = log_csn; + } + + if (!locked) + SpinLockRelease(&csnShared->lock); + + if (log_csn != InvalidCSN) + WriteAssignCSNXlogRec(csn); + + return csn; +} + +CSN +GetLastGeneratedCSN(void) +{ + CSN csn; + + SpinLockAcquire(&csnShared->lock); + csn = csnShared->last_max_csn; + SpinLockRelease(&csnShared->lock); + return csn; +} + +/* + * Mostly for debug purposes. + */ +static void +set_oldest_xmin(TransactionId xid) +{ + elog(LOG, "Oldest Xmin for CSN will be changed from %u to %u", + pg_atomic_read_u32(&csnShared->oldestXmin), xid); + + pg_atomic_write_u32(&csnShared->oldestXmin, xid); +} + +TransactionId +GetOldestXmin(void) +{ + Assert(get_csnlog_status()); + return pg_atomic_read_u32(&csnShared->oldestXmin); +} diff --git a/src/backend/access/transam/csn_snapshot.c b/src/backend/access/transam/csn_snapshot.c new file mode 100644 index 0000000000..a381d219ea --- /dev/null +++ b/src/backend/access/transam/csn_snapshot.c @@ -0,0 +1,687 @@ +/*------------------------------------------------------------------------- + * + * csn_snapshot.c + * Support for cross-node snapshot isolation. + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/backend/access/transam/csn_snapshot.c + * + *------------------------------------------------------------------------- + */ +#include "postgres.h" + +#include "access/csn_snapshot.h" +#include "access/subtrans.h" +#include "access/transam.h" +#include "access/twophase.h" +#include "access/xact.h" +#include "portability/instr_time.h" +#include "storage/lmgr.h" +#include "storage/proc.h" +#include "storage/procarray.h" +#include "storage/shmem.h" +#include "storage/spin.h" +#include "utils/builtins.h" +#include "utils/guc.h" +#include "utils/snapmgr.h" +#include "miscadmin.h" + +/* Raise a warning if imported snapshot_csn exceeds ours by this value. */ +#define SNAP_DESYNC_COMPLAIN (1*NSECS_PER_SEC) /* 1 second */ + +static TransactionId xmin_for_csn = InvalidTransactionId; + + +/* + * GUC to delay advance of oldestXid for this amount of time. Also determines + * the size CSNSnapshotXidMap circular buffer. + */ +int csn_snapshot_defer_time; + +int csn_time_shift; + +/* + * CSNSnapshotXidMap + * + * To be able to install csn snapshot that points to past we need to keep + * old versions of tuples and therefore delay advance of oldestXid. Here we + * keep track of correspondence between snapshot's snapshot_csn and oldestXid + * that was set at the time when the snapshot was taken. Much like the + * snapshot too old's OldSnapshotControlData does, but with finer granularity + * to seconds. + * + * Different strategies can be employed to hold oldestXid (e.g. we can track + * oldest csn-based snapshot among cluster nodes and map it oldestXid + * on each node). + * + * On each snapshot acquisition CSNSnapshotMapXmin() is called and stores + * correspondence between current snapshot_csn and oldestXmin in a sparse way: + * snapshot_csn is rounded to seconds (and here we use the fact that snapshot_csn + * is just a timestamp) and oldestXmin is stored in the circular buffer where + * rounded snapshot_csn acts as an offset from current circular buffer head. + * Size of the circular buffer is controlled by csn_snapshot_defer_time GUC. + * + * When csn snapshot arrives we check that its + * snapshot_csn is still in our map, otherwise we'll error out with "snapshot too + * old" message. If snapshot_csn is successfully mapped to oldestXid we move + * backend's pgxact->xmin to proc->originalXmin and fill pgxact->xmin to + * mapped oldestXid. That way GetOldestXmin() can take into account backends + * with imported csn snapshot and old tuple versions will be preserved. + * + * Also while calculating oldestXmin for our map in presence of imported + * csn snapshots we should use proc->originalXmin instead of pgxact->xmin + * that was set during import. Otherwise, we can create a feedback loop: + * xmin's of imported csn snapshots were calculated using our map and new + * entries in map going to be calculated based on that xmin's, and there is + * a risk to stuck forever with one non-increasing oldestXmin. All other + * callers of GetOldestXmin() are using pgxact->xmin so the old tuple versions + * are preserved. + */ +typedef struct CSNSnapshotXidMap +{ + int head; /* offset of current freshest value */ + int size; /* total size of circular buffer */ + CSN_atomic last_csn_seconds; /* last rounded csn that changed + * xmin_by_second[] */ + TransactionId *xmin_by_second; /* circular buffer of oldestXmin's */ +} +CSNSnapshotXidMap; + +static CSNSnapshotXidMap *csnXidMap; + + +/* Estimate shared memory space needed */ +Size +CSNSnapshotShmemSize(void) +{ + Size size = 0; + + if (csn_snapshot_defer_time > 0) + { + size += sizeof(CSNSnapshotXidMap); + size += csn_snapshot_defer_time*sizeof(TransactionId); + size = MAXALIGN(size); + } + + return size; +} + +/* Init shared memory structures */ +void +CSNSnapshotShmemInit() +{ + bool found; + + if (csn_snapshot_defer_time > 0) + { + csnXidMap = ShmemInitStruct("csnXidMap", + sizeof(CSNSnapshotXidMap), + &found); + if (!found) + { + int i; + + pg_atomic_init_u64(&csnXidMap->last_csn_seconds, 0); + csnXidMap->head = 0; + csnXidMap->size = csn_snapshot_defer_time; + csnXidMap->xmin_by_second = + ShmemAlloc(sizeof(TransactionId)*csnXidMap->size); + + for (i = 0; i < csnXidMap->size; i++) + csnXidMap->xmin_by_second[i] = InvalidTransactionId; + } + } +} + +/* + * CSNSnapshotStartup + * + * Set csnXidMap entries to oldestActiveXID during startup. + */ +void +CSNSnapshotStartup(TransactionId oldestActiveXID) +{ + /* + * Run only if we have initialized shared memory and csnXidMap + * is enabled. + */ + if (IsNormalProcessingMode() && + enable_csn_snapshot && csn_snapshot_defer_time > 0) + { + int i; + + Assert(TransactionIdIsValid(oldestActiveXID)); + for (i = 0; i < csnXidMap->size; i++) + csnXidMap->xmin_by_second[i] = oldestActiveXID; + ProcArraySetCSNSnapshotXmin(oldestActiveXID); + + elog(LOG, "CSN map initialized with oldest active xid %u", oldestActiveXID); + } +} + +/* + * CSNSnapshotMapXmin + * + * Maintain circular buffer of oldestXmins for several seconds in past. This + * buffer allows to shift oldestXmin in the past when backend is importing + * CSN snapshot. Otherwise old versions of tuples that were needed for + * this transaction can be recycled by other processes (vacuum, HOT, etc). + * + * Locking here is not trivial. Called upon each snapshot creation after + * ProcArrayLock is released. Such usage creates several race conditions. It + * is possible that backend who got csn called CSNSnapshotMapXmin() + * only after other backends managed to get snapshot and complete + * CSNSnapshotMapXmin() call, or even committed. This is safe because + * + * * We already hold our xmin in MyPgXact, so our snapshot will not be + * harmed even though ProcArrayLock is released. + * + * * snapshot_csn is always pessmistically rounded up to the next + * second. + * + * * For performance reasons, xmin value for particular second is filled + * only once. Because of that instead of writing to buffer just our + * xmin (which is enough for our snapshot), we bump oldestXmin there -- + * it mitigates the possibility of damaging someone else's snapshot by + * writing to the buffer too advanced value in case of slowness of + * another backend who generated csn earlier, but didn't manage to + * insert it before us. + * + * * if CSNSnapshotMapXmin() founds a gap in several seconds between + * current call and latest completed call then it should fill that gap + * with latest known values instead of new one. Otherwise it is + * possible (however highly unlikely) that this gap also happend + * between taking snapshot and call to CSNSnapshotMapXmin() for some + * backend. And we are at risk to fill circullar buffer with + * oldestXmin's that are bigger then they actually were. + */ +void +CSNSnapshotMapXmin(SnapshotCSN snapshot_csn) +{ + int offset, gap, i; + SnapshotCSN csn_seconds; + SnapshotCSN last_csn_seconds; + volatile TransactionId oldest_deferred_xmin; + TransactionId current_oldest_xmin, previous_oldest_xmin; + TransactionId ImportedXmin; + + /* Callers should check config values */ + Assert(csn_snapshot_defer_time > 0); + Assert(csnXidMap != NULL); + /* + * Round up snapshot_csn to the next second -- pessimistically and safely. + */ + csn_seconds = (snapshot_csn / NSECS_PER_SEC + 1); + + /* + * Fast-path check. Avoid taking exclusive CSNSnapshotXidMapLock lock + * if oldestXid was already written to xmin_by_second[] for this rounded + * snapshot_csn. + */ + if (pg_atomic_read_u64(&csnXidMap->last_csn_seconds) >= csn_seconds) + return; + + /* Ok, we have new entry (or entries) */ + LWLockAcquire(CSNSnapshotXidMapLock, LW_EXCLUSIVE); + + /* Re-check last_csn_seconds under lock */ + last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds); + if (last_csn_seconds >= csn_seconds) + { + LWLockRelease(CSNSnapshotXidMapLock); + return; + } + pg_atomic_write_u64(&csnXidMap->last_csn_seconds, csn_seconds); + + /* + * Count oldest_xmin. + * + * It was possible to calculate oldest_xmin during corresponding snapshot + * creation, but GetSnapshotData() intentionally reads only PgXact, but not + * PgProc. And we need info about originalXmin (see comment to csnXidMap) + * which is stored in PgProc because of threats in comments around PgXact + * about extending it with new fields. So just calculate oldest_xmin again, + * that anyway happens quite rarely. + */ + + /* + * Don't afraid here because csn_snapshot_xmin will hold border of + * minimal non-removable from vacuuming. + */ + ImportedXmin = MyProc->xmin; + MyProc->xmin = MyProc->originalXmin; + current_oldest_xmin = GetOldestNonRemovableTransactionId(NULL); + MyProc->xmin = ImportedXmin; + Assert(TransactionIdIsNormal(current_oldest_xmin)); + + previous_oldest_xmin = csnXidMap->xmin_by_second[csnXidMap->head]; + Assert(TransactionIdIsNormal(previous_oldest_xmin) || !enable_csn_snapshot); + + gap = csn_seconds - last_csn_seconds; + offset = csn_seconds % csnXidMap->size; + + /* Sanity check before we update head and gap */ + Assert( gap >= 1 ); + Assert( (csnXidMap->head + gap) % csnXidMap->size == offset ); + + gap = gap > csnXidMap->size ? csnXidMap->size : gap; + csnXidMap->head = offset; + + /* Fill new entry with current_oldest_xmin */ + csnXidMap->xmin_by_second[offset] = current_oldest_xmin; + + /* + * If we have gap then fill it with previous_oldest_xmin for reasons + * outlined in comment above this function. + */ + for (i = 1; i < gap; i++) + { + offset = (offset + csnXidMap->size - 1) % csnXidMap->size; + csnXidMap->xmin_by_second[offset] = previous_oldest_xmin; + } + + oldest_deferred_xmin = + csnXidMap->xmin_by_second[ (csnXidMap->head + 1) % csnXidMap->size ]; + + LWLockRelease(CSNSnapshotXidMapLock); + + elog(DEBUG5, "Advance xmin for CSN. Oldest deferred xmin = %u", + oldest_deferred_xmin); + + /* + * Advance procArray->csn_snapshot_xmin after we released + * CSNSnapshotXidMapLock. Since we gather not xmin but oldestXmin, it + * never goes backwards regardless of how slow we can do that. + */ + /*Assert(TransactionIdFollowsOrEquals(oldest_deferred_xmin, + ProcArrayGetCSNSnapshotXmin()));*/ + ProcArraySetCSNSnapshotXmin(oldest_deferred_xmin); +} + + +/* + * CSNSnapshotToXmin + * + * Get oldestXmin that took place when snapshot_csn was taken. + */ +TransactionId +CSNSnapshotToXmin(SnapshotCSN snapshot_csn) +{ + TransactionId xmin; + SnapshotCSN csn_seconds; + volatile SnapshotCSN last_csn_seconds; + + /* Callers should check config values */ + Assert(csn_snapshot_defer_time > 0); + Assert(csnXidMap != NULL); + + /* Round down to get conservative estimates */ + csn_seconds = (snapshot_csn / NSECS_PER_SEC); + + LWLockAcquire(CSNSnapshotXidMapLock, LW_SHARED); + last_csn_seconds = pg_atomic_read_u64(&csnXidMap->last_csn_seconds); + if (csn_seconds > last_csn_seconds) + { + /* we don't have entry for this snapshot_csn yet, return latest known */ + xmin = csnXidMap->xmin_by_second[csnXidMap->head]; + } + else if (last_csn_seconds - csn_seconds < csnXidMap->size) + { + /* we are good, retrieve value from our map */ + Assert(last_csn_seconds % csnXidMap->size == csnXidMap->head); + xmin = csnXidMap->xmin_by_second[csn_seconds % csnXidMap->size]; + } + else + { + /* requested snapshot_csn is too old, let caller know */ + xmin = InvalidTransactionId; + } + LWLockRelease(CSNSnapshotXidMapLock); + + return xmin; +} + +/* + * CSNSnapshotPrepareCurrent + * + * Set InDoubt state for currently active transaction and return commit's + * global snapshot. + */ +SnapshotCSN +CSNSnapshotPrepareCurrent(void) +{ + TransactionId xid = GetCurrentTransactionIdIfAny(); + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (TransactionIdIsValid(xid)) + { + TransactionId *subxids; + int nsubxids = xactGetCommittedChildren(&subxids); + CSNLogSetCSN(xid, nsubxids, subxids, InDoubtCSN, true); + } + + /* Nothing to write if we don't have xid */ + + return GenerateCSN(false, InvalidCSN); +} + + +/* + * CSNSnapshotAssignCurrent + * + * Assign SnapshotCSN to the currently active transaction. SnapshotCSN is supposedly + * maximal among of values returned by CSNSnapshotPrepareCurrent and + * pg_csn_snapshot_prepare. + */ +void +CSNSnapshotAssignCurrent(SnapshotCSN snapshot_csn) +{ + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (!CSNIsNormal(snapshot_csn)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("pg_csn_snapshot_assign expects normal snapshot_csn"))); + + Assert(snapshot_csn != InvalidCSN); + /* We do not care the Generate result, we just want to make sure max + * csnShared->last_max_csn value. + */ + GenerateCSN(false, snapshot_csn); + + /* Set csn and defuse ProcArrayEndTransaction from assigning one */ + pg_atomic_write_u64(&MyProc->assignedCSN, snapshot_csn); +} + +/* + * CSNSnapshotSync + * + * Due to time desynchronization on different nodes we can receive snapshot_csn + * which is greater than snapshot_csn on this node. To preserve proper isolation + * this node needs to wait when such snapshot_csn comes on local clock. + * + * This should happend relatively rare if nodes have running NTP/PTP/etc. + * Complain if wait time is more than SNAP_SYNC_COMPLAIN. + */ +void +CSNSnapshotSync(SnapshotCSN remote_csn) +{ + SnapshotCSN local_csn; + SnapshotCSN delta; + + Assert(enable_csn_snapshot); + + for(;;) + { + if (GetLastGeneratedCSN() > remote_csn) + return; + + local_csn = GenerateCSN(true, InvalidCSN); + + if (local_csn >= remote_csn) + /* + * Everything is fine too, but last_max_csn wasn't updated for + * some time. + */ + return; + + /* Okay we need to sleep now */ + delta = remote_csn - local_csn; + if (delta > SNAP_DESYNC_COMPLAIN) + ereport(WARNING, + (errmsg("remote global snapshot exceeds ours by more than a second"), + errhint("Consider running NTPd on servers participating in global transaction"))); + + /* TODO: report this sleeptime somewhere? */ + pg_usleep((long) (delta/NSECS_PER_USEC)); + + /* + * Loop that checks to ensure that we actually slept for specified + * amount of time. + */ + } + + Assert(false); /* Should not happend */ + return; +} + +/* + * TransactionIdGetCSN + * + * Get CSN for specified TransactionId taking care about special xids, + * xids beyond TransactionXmin and InDoubt states. + */ +CSN +TransactionIdGetCSN(TransactionId xid) +{ + CSN csn; + + /* Handle permanent TransactionId's for which we don't have mapping */ + if (!TransactionIdIsNormal(xid)) + { + if (xid == InvalidTransactionId) + return AbortedCSN; + if (xid == FrozenTransactionId || xid == BootstrapTransactionId) + return FrozenCSN; + Assert(false); /* Should not happend */ + } + + /* + * If we just switch a xid-snapsot to a csn_snapshot, we should handle a start + * xid for csn base check. Just in case we have prepared transaction which + * hold the TransactionXmin but without CSN. + */ + xmin_for_csn = GetOldestXmin(); + + /* + * For the xid with 'xid >= TransactionXmin and xid < xmin_for_csn', + * it defined as unclear csn which follow xid-snapshot result. + */ + if(!TransactionIdPrecedes(xid, TransactionXmin) && + TransactionIdPrecedes(xid, xmin_for_csn)) + { + elog(LOG, "UnclearCSN was returned. xid=%u, TransactionXmin=%u, xmin_for_csn=%u", + xid, TransactionXmin, xmin_for_csn); + return UnclearCSN; + } + /* + * For xids which less then TransactionXmin CSNLog can be already + * trimmed but we know that such transaction is definitely not concurrently + * running according to any snapshot including timetravel ones. Callers + * should check TransactionDidCommit after. + */ + if (TransactionIdPrecedes(xid, TransactionXmin)) + return FrozenCSN; + + /* Read CSN from SLRU */ + csn = CSNLogGetCSNByXid(xid); + + /* + * If we faced InDoubt state then transaction is being committed and we + * should wait until CSN will be assigned so that visibility check + * could decide whether tuple is in snapshot. See also comments in + * CSNSnapshotPrecommit(). + */ + if (CSNIsInDoubt(csn)) + { + XactLockTableWait(SubTransGetTopmostTransaction(xid), NULL, NULL, XLTW_None); + csn = CSNLogGetCSNByXid(xid); + Assert(CSNIsNormal(csn) || CSNIsAborted(csn)); + } + + Assert(CSNIsNormal(csn) || CSNIsInProgress(csn) || CSNIsAborted(csn)); + return csn; +} + +/* + * XidInCSNSnapshot + * + * Version of XidInMVCCSnapshot for transactions. For non-imported + * csn snapshots this should give same results as XidInLocalMVCCSnapshot + * (except that aborts will be shown as invisible without going to clog) and to + * ensure such behaviour XidInMVCCSnapshot is coated with asserts that checks + * identicalness of XidInCSNSnapshot/XidInLocalMVCCSnapshot in + * case of ordinary snapshot. + */ +bool +XidInCSNSnapshot(TransactionId xid, Snapshot snapshot) +{ + CSN csn; + + csn = TransactionIdGetCSN(xid); + + if (CSNIsNormal(csn)) + return (csn >= snapshot->snapshot_csn); + else if (CSNIsFrozen(csn)) + { + /* It is bootstrap or frozen transaction */ + return false; + } + else if(CSNIsUnclear(csn)) + { + /* + * Some xid can not figure out csn because of snapshot switch, + * and we can follow xid-base result. + */ + return true; + } + else + { + /* It is aborted or in-progress */ + Assert(CSNIsAborted(csn) || CSNIsInProgress(csn)); + if (CSNIsAborted(csn)) + Assert(TransactionIdDidAbort(xid)); + return true; + } +} + + +/***************************************************************************** + * Functions to handle transactions commit. + * + * For local transactions CSNSnapshotPrecommit sets InDoubt state before + * ProcArrayEndTransaction is called and transaction data potetntially becomes + * visible to other backends. ProcArrayEndTransaction (or ProcArrayRemove in + * twophase case) then acquires csn under ProcArray lock and stores it + * in proc->assignedCSN. It's important that csn for commit is + * generated under ProcArray lock, otherwise snapshots won't + * be equivalent. Consequent call to CSNSnapshotCommit will write + * proc->assignedCSN to CSNLog. + * + * + * CSNSnapshotAbort is slightly different comparing to commit because abort + * can skip InDoubt phase and can be called for transaction subtree. + *****************************************************************************/ + + +/* + * CSNSnapshotAbort + * + * Abort transaction in CsnLog. We can skip InDoubt state for aborts + * since no concurrent transactions allowed to see aborted data anyway. + */ +void +CSNSnapshotAbort(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + if (!get_csnlog_status()) + return; + + CSNLogSetCSN(xid, nsubxids, subxids, AbortedCSN, true); + + /* + * Clean assignedCSN anyway, as it was possibly set in + * XidSnapshotAssignCsnCurrent. + */ + pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN); +} + +/* + * CSNSnapshotPrecommit + * + * Set InDoubt status for local transaction that we are going to commit. + * This step is needed to achieve consistency between local snapshots and + * csn-based snapshots. We don't hold ProcArray lock while writing + * csn for transaction in SLRU but instead we set InDoubt status before + * transaction is deleted from ProcArray so the readers who will read csn + * in the gap between ProcArray removal and CSN assignment can wait + * until CSN is finally assigned. See also TransactionIdGetCSN(). + * + * This should be called only from parallel group leader before backend is + * deleted from ProcArray. + */ +void +CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + CSN oldassignedCSN = InProgressCSN; + bool in_progress; + + if (!get_csnlog_status()) + return; + + /* Set InDoubt status if it is local transaction */ + in_progress = pg_atomic_compare_exchange_u64(&proc->assignedCSN, + &oldassignedCSN, + InDoubtCSN); + if (in_progress) + { + Assert(CSNIsInProgress(oldassignedCSN)); + CSNLogSetCSN(xid, nsubxids, subxids, InDoubtCSN, true); + } + else + { + /* Otherwise we should have valid CSN by this time */ + Assert(CSNIsNormal(oldassignedCSN)); + Assert(CSNIsInDoubt(CSNLogGetCSNByXid(xid))); + } +} + +/* + * CSNSnapshotCommit + * + * Write CSN that were acquired earlier to CsnLog. Should be + * preceded by CSNSnapshotPrecommit() so readers can wait until we finally + * finished writing to SLRU. + * + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks, so that TransactionIdGetCSN can wait on this + * lock for CSN. + */ +void +CSNSnapshotCommit(PGPROC *proc, TransactionId xid, + int nsubxids, TransactionId *subxids) +{ + volatile CSN assignedCSN; + + if (!get_csnlog_status()) + return; + + if (!TransactionIdIsValid(xid)) + { + assignedCSN = pg_atomic_read_u64(&proc->assignedCSN); + Assert(CSNIsInProgress(assignedCSN)); + return; + } + + /* Finally write resulting CSN in SLRU */ + assignedCSN = pg_atomic_read_u64(&proc->assignedCSN); + Assert(CSNIsNormal(assignedCSN)); + CSNLogSetCSN(xid, nsubxids, subxids, assignedCSN, true); + + /* Reset for next transaction */ + pg_atomic_write_u64(&proc->assignedCSN, InProgressCSN); +} diff --git a/src/backend/access/transam/rmgr.c b/src/backend/access/transam/rmgr.c index 58091f6b52..b86c172e46 100644 --- a/src/backend/access/transam/rmgr.c +++ b/src/backend/access/transam/rmgr.c @@ -10,6 +10,7 @@ #include "access/brin_xlog.h" #include "access/clog.h" #include "access/commit_ts.h" +#include "access/csn_log.h" #include "access/generic_xlog.h" #include "access/ginxlog.h" #include "access/gistxlog.h" diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c index 28b153abc3..7bc6aae9a4 100644 --- a/src/backend/access/transam/twophase.c +++ b/src/backend/access/transam/twophase.c @@ -77,6 +77,7 @@ #include #include "access/commit_ts.h" +#include "access/csn_snapshot.h" #include "access/htup_details.h" #include "access/subtrans.h" #include "access/transam.h" @@ -1536,8 +1537,34 @@ FinishPreparedTransaction(const char *gid, bool isCommit) hdr->nabortrels, abortrels, gid); + /* + * CSNSnapshot callbacks that should be called right before we are + * going to become visible. Details in comments to this functions. + */ + if (isCommit) + CSNSnapshotPrecommit(proc, xid, hdr->nsubxacts, children); + else + CSNSnapshotAbort(proc, xid, hdr->nsubxacts, children); + + ProcArrayRemove(proc, latestXid); + /* + * Stamp our transaction with CSN in CSNLog. + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks, since TransactionIdGetCSN relies on + * XactLockTableWait to await csn. + */ + if (isCommit) + { + CSNSnapshotCommit(proc, xid, hdr->nsubxacts, children); + } + else + { + Assert(CSNIsInProgress( + pg_atomic_read_u64(&proc->assignedCSN))); + } + /* * In case we fail while running the callbacks, mark the gxact invalid so * no one else will try to commit/rollback, and so it will be recycled if @@ -2583,3 +2610,130 @@ LookupGXact(const char *gid, XLogRecPtr prepare_end_lsn, LWLockRelease(TwoPhaseStateLock); return found; } + +/* + * CSNSnapshotPrepareTwophase + * + * Set InDoubt state for currently active transaction and return commit's + * global snapshot. + */ +static SnapshotCSN +CSNSnapshotPrepareTwophase(const char *gid) +{ + GlobalTransaction gxact; + PGPROC *proc; + char *buf; + TransactionId xid; + xl_xact_parsed_prepare parsed; + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + /* + * Validate the GID, and lock the GXACT to ensure that two backends do not + * try to access the same GID at once. + */ + gxact = LockGXact(gid, GetUserId()); + proc = &ProcGlobal->allProcs[gxact->pgprocno]; + xid = proc->xid; + + if (gxact->ondisk) + buf = ReadTwoPhaseFile(xid, true); + else + XlogReadTwoPhaseData(gxact->prepare_start_lsn, &buf, NULL); + + ParsePrepareRecord(0, (xl_xact_prepare *)buf, &parsed); + + CSNLogSetCSN(xid, parsed.nsubxacts, + parsed.subxacts, InDoubtCSN, true); + + /* Unlock our GXACT */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + gxact->locking_backend = InvalidBackendId; + LWLockRelease(TwoPhaseStateLock); + + pfree(buf); + return GenerateCSN(false, InvalidCSN); +} + +/* + * CSNSnapshotAssignTwoPhase + * + * Asign SnapshotCSN for currently active transaction. SnapshotCSN is supposedly + * maximal among of values returned by CSNSnapshotPrepareCurrent and + * pg_csn_snapshot_prepare. + * + * This function is a counterpart of CSNSnapshotAssignCurrent() for + * twophase transactions. + */ +static void +CSNSnapshotAssignTwoPhase(const char *gid, SnapshotCSN csn) +{ + GlobalTransaction gxact; + PGPROC *proc; + + if (!enable_csn_snapshot) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not prepare transaction for global commit"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (!CSNIsNormal(csn)) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("pg_csn_snapshot_assign expects normal snapshot_csn"))); + + /* + * Validate the GID, and lock the GXACT to ensure that two backends do not + * try to access the same GID at once. + */ + gxact = LockGXact(gid, GetUserId()); + proc = &ProcGlobal->allProcs[gxact->pgprocno]; + + Assert(csn != InvalidCSN); + /* We do not care the Generate result, we just want to make sure max + * csnShared->last_max_csn value. + */ + GenerateCSN(false, csn); + /* Set snapshot_csn and defuse ProcArrayRemove from assigning one. */ + pg_atomic_write_u64(&proc->assignedCSN, csn); + + /* Unlock our GXACT */ + LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE); + gxact->locking_backend = InvalidBackendId; + LWLockRelease(TwoPhaseStateLock); +} + +/* + * SQL interface to CSNSnapshotPrepareTwophase() + * + * TODO: Rewrite this as PREPARE TRANSACTION 'gid' RETURNING SNAPSHOT + */ +Datum +pg_csn_snapshot_prepare(PG_FUNCTION_ARGS) +{ + const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0)); + SnapshotCSN csn = CSNSnapshotPrepareTwophase(gid); + + PG_RETURN_INT64(csn); +} + +/* + * SQL interface to CSNSnapshotAssignTwoPhase() + * + * TODO: Rewrite this as COMMIT PREPARED 'gid' SNAPSHOT 'csn' + */ +Datum +pg_csn_snapshot_assign(PG_FUNCTION_ARGS) +{ + const char *gid = text_to_cstring(PG_GETARG_TEXT_PP(0)); + SnapshotCSN csn = PG_GETARG_INT64(1); + + CSNSnapshotAssignTwoPhase(gid, csn); + PG_RETURN_VOID(); +} diff --git a/src/backend/access/transam/varsup.c b/src/backend/access/transam/varsup.c index a6e98e71bd..8e1d074806 100644 --- a/src/backend/access/transam/varsup.c +++ b/src/backend/access/transam/varsup.c @@ -15,6 +15,7 @@ #include "access/clog.h" #include "access/commit_ts.h" +#include "access/csn_log.h" #include "access/subtrans.h" #include "access/transam.h" #include "access/xact.h" @@ -175,6 +176,7 @@ GetNewTransactionId(bool isSubXact) * Extend pg_subtrans and pg_commit_ts too. */ ExtendCLOG(xid); + ExtendCSNLog(xid); ExtendCommitTs(xid); ExtendSUBTRANS(xid); diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c index 8e35c432f5..e6baf880d9 100644 --- a/src/backend/access/transam/xact.c +++ b/src/backend/access/transam/xact.c @@ -21,6 +21,7 @@ #include #include "access/commit_ts.h" +#include "access/csn_snapshot.h" #include "access/multixact.h" #include "access/parallel.h" #include "access/subtrans.h" @@ -1418,6 +1419,12 @@ RecordTransactionCommit(void) TransactionTreeSetCommitTsData(xid, nchildren, children, replorigin_session_origin_timestamp, replorigin_session_origin); + + /* + * Mark our transaction as InDoubt in CsnLog and get ready for + * commit. + */ + CSNSnapshotPrecommit(MyProc, xid, nchildren, children); } /* @@ -1772,6 +1779,9 @@ RecordTransactionAbort(bool isSubXact) */ TransactionIdAbortTree(xid, nchildren, children); + /* Mark our transaction as Aborted in CSN Log. */ + CSNSnapshotAbort(MyProc, xid, nchildren, children); + END_CRIT_SECTION(); /* Compute latestXid while we have the child XIDs handy */ @@ -2114,6 +2124,13 @@ StartTransaction(void) ShowTransactionState("StartTransaction"); } +Datum +pg_current_csn(PG_FUNCTION_ARGS) +{ + SnapshotCSN csn = GenerateCSN(false, InvalidCSN); + + PG_RETURN_INT64(csn); +} /* * CommitTransaction @@ -2262,6 +2279,21 @@ CommitTransaction(void) */ ProcArrayEndTransaction(MyProc, latestXid); + /* + * Stamp our transaction with CSN in CsnLog. + * Should be called after ProcArrayEndTransaction, but before releasing + * transaction locks. + */ + if (!is_parallel_worker) + { + TransactionId xid = GetTopTransactionIdIfAny(); + TransactionId *subxids; + int nsubxids; + + nsubxids = xactGetCommittedChildren(&subxids); + CSNSnapshotCommit(MyProc, xid, nsubxids, subxids); + } + /* * This is all post-commit cleanup. Note that if an error is raised here, * it's too late to abort the transaction. This should be just diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 1616448368..2a8de10038 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -24,6 +24,7 @@ #include "access/clog.h" #include "access/commit_ts.h" +#include "access/csn_log.h" #include "access/heaptoast.h" #include "access/multixact.h" #include "access/rewriteheap.h" @@ -4747,6 +4748,7 @@ InitControlFile(uint64 sysidentifier) ControlFile->wal_level = wal_level; ControlFile->wal_log_hints = wal_log_hints; ControlFile->track_commit_timestamp = track_commit_timestamp; + ControlFile->enable_csn_snapshot = enable_csn_snapshot; ControlFile->data_checksum_version = bootstrap_data_checksum_version; } @@ -7181,6 +7183,9 @@ StartupXLOG(void) if (ControlFile->track_commit_timestamp) StartupCommitTs(); + if(ControlFile->enable_csn_snapshot) + StartupCSN(); + /* * Recover knowledge about replay progress of known replication partners. */ @@ -7448,6 +7453,8 @@ StartupXLOG(void) */ StartupSUBTRANS(oldestActiveXID); + CSNSnapshotStartup(oldestActiveXID); + /* * If we're beginning at a shutdown checkpoint, we know that * nothing was running on the primary at this point. So fake-up an @@ -8117,7 +8124,10 @@ StartupXLOG(void) * timestamps are started below, if necessary.) */ if (standbyState == STANDBY_DISABLED) + { StartupSUBTRANS(oldestActiveXID); + CSNSnapshotStartup(oldestActiveXID); + } /* * Perform end of recovery actions for any SLRUs that need it. @@ -8183,6 +8193,7 @@ StartupXLOG(void) * commit timestamp. */ CompleteCommitTsInitialization(); + CompleteCSNInitialization(); /* * All done with end-of-recovery actions. @@ -9616,6 +9627,7 @@ CheckPointGuts(XLogRecPtr checkPointRedo, int flags) TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags); CheckpointStats.ckpt_write_t = GetCurrentTimestamp(); CheckPointCLOG(); + CheckPointCSNLog(); CheckPointCommitTs(); CheckPointSUBTRANS(); CheckPointMultiXact(); @@ -9894,7 +9906,10 @@ CreateRestartPoint(int flags) * this because StartupSUBTRANS hasn't been called yet. */ if (EnableHotStandby) + { TruncateSUBTRANS(GetOldestTransactionIdConsideredRunning()); + TruncateCSNLog(GetOldestTransactionIdConsideredRunning()); + } /* Real work is done; log and update stats. */ LogCheckpointEnd(true); @@ -10172,7 +10187,8 @@ XLogReportParameters(void) max_wal_senders != ControlFile->max_wal_senders || max_prepared_xacts != ControlFile->max_prepared_xacts || max_locks_per_xact != ControlFile->max_locks_per_xact || - track_commit_timestamp != ControlFile->track_commit_timestamp) + track_commit_timestamp != ControlFile->track_commit_timestamp || + enable_csn_snapshot != ControlFile->enable_csn_snapshot) { /* * The change in number of backend slots doesn't need to be WAL-logged @@ -10194,6 +10210,7 @@ XLogReportParameters(void) xlrec.wal_level = wal_level; xlrec.wal_log_hints = wal_log_hints; xlrec.track_commit_timestamp = track_commit_timestamp; + xlrec.enable_csn_snapshot = enable_csn_snapshot; XLogBeginInsert(); XLogRegisterData((char *) &xlrec, sizeof(xlrec)); @@ -10212,6 +10229,7 @@ XLogReportParameters(void) ControlFile->wal_level = wal_level; ControlFile->wal_log_hints = wal_log_hints; ControlFile->track_commit_timestamp = track_commit_timestamp; + ControlFile->enable_csn_snapshot = enable_csn_snapshot; UpdateControlFile(); LWLockRelease(ControlFileLock); @@ -10665,6 +10683,9 @@ xlog_redo(XLogReaderState *record) CommitTsParameterChange(xlrec.track_commit_timestamp, ControlFile->track_commit_timestamp); ControlFile->track_commit_timestamp = xlrec.track_commit_timestamp; + CSNlogParameterChange(xlrec.enable_csn_snapshot, + ControlFile->enable_csn_snapshot); + ControlFile->enable_csn_snapshot = xlrec.enable_csn_snapshot; UpdateControlFile(); LWLockRelease(ControlFileLock); diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.c index 689384a411..e6585a94ba 100644 --- a/src/backend/access/transam/xloginsert.c +++ b/src/backend/access/transam/xloginsert.c @@ -73,6 +73,8 @@ typedef struct char compressed_page[COMPRESS_BUFSIZE]; } registered_buffer; +bool enable_csn_wal = true; + static registered_buffer *registered_buffers; static int max_registered_buffers; /* allocated size */ static int max_registered_block_id = 0; /* highest block_id + 1 currently diff --git a/src/backend/commands/vacuum.c b/src/backend/commands/vacuum.c index 5c4bc15b44..e64ada86c7 100644 --- a/src/backend/commands/vacuum.c +++ b/src/backend/commands/vacuum.c @@ -53,7 +53,7 @@ #include "utils/memutils.h" #include "utils/snapmgr.h" #include "utils/syscache.h" - +#include "access/csn_log.h" /* * GUC parameters @@ -1760,6 +1760,7 @@ vac_truncate_clog(TransactionId frozenXID, */ TruncateCLOG(frozenXID, oldestxid_datoid); TruncateCommitTs(frozenXID); + TruncateCSNLog(frozenXID); TruncateMultiXact(minMulti, minmulti_datoid); /* diff --git a/src/backend/storage/ipc/ipci.c b/src/backend/storage/ipc/ipci.c index 9fa3e0631e..2a7e184da9 100644 --- a/src/backend/storage/ipc/ipci.c +++ b/src/backend/storage/ipc/ipci.c @@ -16,6 +16,8 @@ #include "access/clog.h" #include "access/commit_ts.h" +#include "access/csn_log.h" +#include "access/csn_snapshot.h" #include "access/heapam.h" #include "access/multixact.h" #include "access/nbtree.h" @@ -120,6 +122,8 @@ CalculateShmemSize(int *num_semaphores) size = add_size(size, ProcGlobalShmemSize()); size = add_size(size, XLOGShmemSize()); size = add_size(size, CLOGShmemSize()); + size = add_size(size, CSNLogShmemSize()); + size = add_size(size, CSNSnapshotShmemSize()); size = add_size(size, CommitTsShmemSize()); size = add_size(size, SUBTRANSShmemSize()); size = add_size(size, TwoPhaseShmemSize()); @@ -242,6 +246,8 @@ CreateSharedMemoryAndSemaphores(void) */ XLOGShmemInit(); CLOGShmemInit(); + CSNLogShmemInit(); + CSNSnapshotShmemInit(); CommitTsShmemInit(); SUBTRANSShmemInit(); MultiXactShmemInit(); diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index 892f0f6799..5bc7370c73 100644 --- a/src/backend/storage/ipc/procarray.c +++ b/src/backend/storage/ipc/procarray.c @@ -48,6 +48,7 @@ #include #include "access/clog.h" +#include "access/csn_snapshot.h" #include "access/subtrans.h" #include "access/transam.h" #include "access/twophase.h" @@ -96,6 +97,8 @@ typedef struct ProcArrayStruct TransactionId replication_slot_xmin; /* oldest catalog xmin of any replication slot */ TransactionId replication_slot_catalog_xmin; + /* xmin of oldest active csn snapshot */ + TransactionId csn_snapshot_xmin; /* indexes into allProcs[], has PROCARRAY_MAXPROCS entries */ int pgprocnos[FLEXIBLE_ARRAY_MEMBER]; @@ -429,6 +432,7 @@ CreateSharedProcArray(void) procArray->lastOverflowedXid = InvalidTransactionId; procArray->replication_slot_xmin = InvalidTransactionId; procArray->replication_slot_catalog_xmin = InvalidTransactionId; + procArray->csn_snapshot_xmin = InvalidTransactionId; ShmemVariableCache->xactCompletionCount = 1; } @@ -577,6 +581,14 @@ ProcArrayRemove(PGPROC *proc, TransactionId latestXid) /* Advance global latestCompletedXid while holding the lock */ MaintainLatestCompletedXid(latestXid); + /* + * Assign xid csn while holding ProcArrayLock for non-distributed + * COMMIT PREPARED. After lock is released consequent + * CSNSnapshotCommit() will write this value to CsnLog. + */ + if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN))) + pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(false, InvalidCSN)); + /* Same with xactCompletionCount */ ShmemVariableCache->xactCompletionCount++; @@ -691,6 +703,7 @@ ProcArrayEndTransaction(PGPROC *proc, TransactionId latestXid) proc->xmin = InvalidTransactionId; proc->delayChkpt = false; /* be sure this is cleared in abort */ proc->recoveryConflictPending = false; + proc->originalXmin = InvalidTransactionId; /* must be cleared with xid/xmin: */ /* avoid unnecessarily dirtying shared cachelines */ @@ -730,6 +743,7 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid) proc->xmin = InvalidTransactionId; proc->delayChkpt = false; /* be sure this is cleared in abort */ proc->recoveryConflictPending = false; + proc->originalXmin = InvalidTransactionId; /* must be cleared with xid/xmin: */ /* avoid unnecessarily dirtying shared cachelines */ @@ -753,6 +767,16 @@ ProcArrayEndTransactionInternal(PGPROC *proc, TransactionId latestXid) /* Also advance global latestCompletedXid while holding the lock */ MaintainLatestCompletedXid(latestXid); + /* + * Assign xid csn while holding ProcArrayLock for + * COMMIT. + * + * TODO: in case of group commit we can generate one CSNSnapshot for + * whole group to save time on timestamp aquisition. + */ + if (CSNIsInDoubt(pg_atomic_read_u64(&proc->assignedCSN))) + pg_atomic_write_u64(&proc->assignedCSN, GenerateCSN(false, InvalidCSN)); + /* Same with xactCompletionCount */ ShmemVariableCache->xactCompletionCount++; } @@ -912,6 +936,7 @@ ProcArrayClearTransaction(PGPROC *proc) proc->lxid = InvalidLocalTransactionId; proc->xmin = InvalidTransactionId; proc->recoveryConflictPending = false; + proc->originalXmin = InvalidTransactionId; Assert(!(proc->statusFlags & PROC_VACUUM_STATE_MASK)); Assert(!proc->delayChkpt); @@ -1204,6 +1229,7 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running) while (TransactionIdPrecedes(latestObservedXid, running->nextXid)) { ExtendSUBTRANS(latestObservedXid); + ExtendCSNLog(latestObservedXid); TransactionIdAdvance(latestObservedXid); } TransactionIdRetreat(latestObservedXid); /* = running->nextXid - 1 */ @@ -1704,6 +1730,7 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) TransactionId kaxmin; bool in_recovery = RecoveryInProgress(); TransactionId *other_xids = ProcGlobal->xids; + TransactionId csn_snapshot_xmin = InvalidTransactionId; LWLockAcquire(ProcArrayLock, LW_SHARED); @@ -1843,6 +1870,10 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) if (in_recovery) kaxmin = KnownAssignedXidsGetOldestXmin(); + /* Get value of xmin, delayed by a CSN snapshot settings. */ + if (get_csnlog_status() && csn_snapshot_defer_time > 0 && IsUnderPostmaster) + csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin(); + /* * No other information from shared state is needed, release the lock * immediately. The rest of the computations can be done without a lock. @@ -1899,6 +1930,15 @@ ComputeXidHorizons(ComputeXidHorizonsResult *h) h->data_oldest_nonremovable = TransactionIdOlder(h->data_oldest_nonremovable, h->slot_xmin); + /* + * Hold non-removable border because distributed transactions + * can wish to see old data. + */ + h->shared_oldest_nonremovable = + TransactionIdOlder(h->shared_oldest_nonremovable, csn_snapshot_xmin); + h->data_oldest_nonremovable = + TransactionIdOlder(h->data_oldest_nonremovable, csn_snapshot_xmin); + /* * The only difference between catalog / data horizons is that the slot's * catalog xmin is applied to the catalog one (so catalogs can be accessed @@ -2133,6 +2173,9 @@ GetSnapshotDataReuse(Snapshot snapshot) if (curXactCompletionCount != snapshot->snapXactCompletionCount) return false; + if (get_csnlog_status()) + return false; + /* * If the current xactCompletionCount is still the same as it was at the * time the snapshot was built, we can be sure that rebuilding the @@ -2212,6 +2255,8 @@ GetSnapshotData(Snapshot snapshot) int count = 0; int subcount = 0; bool suboverflowed = false; + CSN csn = FrozenCSN; + TransactionId csn_snapshot_xmin = InvalidTransactionId; FullTransactionId latest_completed; TransactionId oldestxid; int mypgxactoff; @@ -2444,6 +2489,20 @@ GetSnapshotData(Snapshot snapshot) if (!TransactionIdIsValid(MyProc->xmin)) MyProc->xmin = TransactionXmin = xmin; + /* Take CSN under ProcArrayLock so the snapshot stays synchronized. */ + if (!snapshot->takenDuringRecovery && get_csnlog_status()) + csn = GenerateCSN(false, InvalidCSN); + + if (get_csnlog_status() && csn_snapshot_defer_time > 0 && IsUnderPostmaster) + { + CSNSnapshotMapXmin(snapshot->snapshot_csn); + + /* Get value of xmin, delayed by a CSN snapshot settings. */ + csn_snapshot_xmin = ProcArrayGetCSNSnapshotXmin(); + /* Adjust an oldest xid value with a xmin, delayed by CSN options. */ + oldestxid = TransactionIdOlder(oldestxid, csn_snapshot_xmin); + } + LWLockRelease(ProcArrayLock); /* maintain state for GlobalVis* */ @@ -2469,6 +2528,10 @@ GetSnapshotData(Snapshot snapshot) def_vis_xid_data = TransactionIdOlder(def_vis_xid_data, replication_slot_xmin); + /* The csn-related settings can require an older xmin. */ + def_vis_xid_data = + TransactionIdOlder(def_vis_xid_data, csn_snapshot_xmin); + /* * Rows in non-shared, non-catalog tables possibly could be vacuumed * if older than this xid. @@ -2549,6 +2612,8 @@ GetSnapshotData(Snapshot snapshot) snapshot->active_count = 0; snapshot->regd_count = 0; snapshot->copied = false; + snapshot->imported_csn = false; + snapshot->snapshot_csn = csn; GetSnapshotDataInitOldSnapshot(snapshot); @@ -3901,6 +3966,25 @@ ProcArrayGetReplicationSlotXmin(TransactionId *xmin, LWLockRelease(ProcArrayLock); } +/* + * ProcArraySetCSNSnapshotXmin + */ +void +ProcArraySetCSNSnapshotXmin(TransactionId xmin) +{ + /* We rely on atomic fetch/store of xid */ + procArray->csn_snapshot_xmin = xmin; +} + +/* + * ProcArrayGetCSNSnapshotXmin + */ +TransactionId +ProcArrayGetCSNSnapshotXmin(void) +{ + return procArray->csn_snapshot_xmin; +} + /* * XidCacheRemoveRunningXids * @@ -4383,6 +4467,7 @@ RecordKnownAssignedTransactionIds(TransactionId xid) while (TransactionIdPrecedes(next_expected_xid, xid)) { TransactionIdAdvance(next_expected_xid); + ExtendCSNLog(next_expected_xid); ExtendSUBTRANS(next_expected_xid); } Assert(next_expected_xid == xid); diff --git a/src/backend/storage/lmgr/lwlock.c b/src/backend/storage/lmgr/lwlock.c index 862097352b..1f78161d9a 100644 --- a/src/backend/storage/lmgr/lwlock.c +++ b/src/backend/storage/lmgr/lwlock.c @@ -135,6 +135,8 @@ static const char *const BuiltinTrancheNames[] = { "CommitTSBuffer", /* LWTRANCHE_SUBTRANS_BUFFER: */ "SubtransBuffer", + /* LWTRANCHE_CSN_LOG_BUFFERS */ + "CSNLogBuffer", /* LWTRANCHE_MULTIXACTOFFSET_BUFFER: */ "MultiXactOffsetBuffer", /* LWTRANCHE_MULTIXACTMEMBER_BUFFER: */ diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt index 6c7cf6c295..e8ca393611 100644 --- a/src/backend/storage/lmgr/lwlocknames.txt +++ b/src/backend/storage/lmgr/lwlocknames.txt @@ -53,3 +53,5 @@ XactTruncationLock 44 # 45 was XactTruncationLock until removal of BackendRandomLock WrapLimitsVacuumLock 46 NotifyQueueTailLock 47 +CSNLogSLRULock 48 +CSNSnapshotXidMapLock 49 diff --git a/src/backend/storage/lmgr/proc.c b/src/backend/storage/lmgr/proc.c index b7d9da0aa9..88f4f42456 100644 --- a/src/backend/storage/lmgr/proc.c +++ b/src/backend/storage/lmgr/proc.c @@ -35,9 +35,11 @@ #include #include +#include "access/csn_snapshot.h" #include "access/transam.h" #include "access/twophase.h" #include "access/xlogutils.h" +#include "access/xact.h" #include "miscadmin.h" #include "pgstat.h" #include "postmaster/autovacuum.h" @@ -440,6 +442,9 @@ InitProcess(void) MyProc->clogGroupMemberLsn = InvalidXLogRecPtr; Assert(pg_atomic_read_u32(&MyProc->clogGroupNext) == INVALID_PGPROCNO); + MyProc->originalXmin = InvalidTransactionId; + pg_atomic_init_u64(&MyProc->assignedCSN, InProgressCSN); + /* * Acquire ownership of the PGPROC's latch, so that we can use WaitLatch * on it. That allows us to repoint the process latch, which so far @@ -585,6 +590,7 @@ InitAuxiliaryProcess(void) MyProc->lwWaitMode = 0; MyProc->waitLock = NULL; MyProc->waitProcLock = NULL; + MyProc->originalXmin = InvalidTransactionId; pg_atomic_write_u64(&MyProc->waitStart, 0); #ifdef USE_ASSERT_CHECKING { diff --git a/src/backend/storage/sync/sync.c b/src/backend/storage/sync/sync.c index d4083e8a56..383f1e4566 100644 --- a/src/backend/storage/sync/sync.c +++ b/src/backend/storage/sync/sync.c @@ -20,6 +20,7 @@ #include "access/commit_ts.h" #include "access/clog.h" +#include "access/csn_log.h" #include "access/multixact.h" #include "access/xlog.h" #include "access/xlogutils.h" @@ -119,6 +120,10 @@ static const SyncOps syncsw[] = { /* pg_multixact/members */ [SYNC_HANDLER_MULTIXACT_MEMBER] = { .sync_syncfiletag = multixactmemberssyncfiletag + }, + /* pg_multixact/members */ + [SYNC_HANDLER_CSN] = { + .sync_syncfiletag = csnsyncfiletag } }; diff --git a/src/backend/utils/misc/guc.c b/src/backend/utils/misc/guc.c index e91d5a3cfd..4d9833fb5f 100644 --- a/src/backend/utils/misc/guc.c +++ b/src/backend/utils/misc/guc.c @@ -34,6 +34,7 @@ #include "access/commit_ts.h" #include "access/gin.h" +#include "access/csn_snapshot.h" #include "access/rmgr.h" #include "access/tableam.h" #include "access/toast_compression.h" @@ -1212,6 +1213,24 @@ static struct config_bool ConfigureNamesBool[] = false, NULL, NULL, NULL }, + { + {"enable_csn_snapshot", PGC_POSTMASTER, RESOURCES_MEM, + gettext_noop("Enable csn-base snapshot."), + gettext_noop("Used to achieve REPEATABLE READ isolation level for postgres_fdw transactions.") + }, + &enable_csn_snapshot, + true, + NULL, NULL, NULL + }, + { + {"enable_csn_wal", PGC_POSTMASTER, RESOURCES_MEM, + gettext_noop("Enable csn-wal record."), + gettext_noop("Used to enable csn-wal record") + }, + &enable_csn_wal, + true, + NULL, NULL, NULL + }, { {"ssl", PGC_SIGHUP, CONN_AUTH_SSL, gettext_noop("Enables SSL connections."), @@ -3195,6 +3214,24 @@ static struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"csn_snapshot_defer_time", PGC_POSTMASTER, REPLICATION_PRIMARY, + gettext_noop("Minimal age of records which allowed to be vacuumed, in seconds."), + NULL + }, + &csn_snapshot_defer_time, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, + { + {"csn_time_shift", PGC_USERSET, RESOURCES_MEM, + gettext_noop("Do the time shift in the CSN generator."), + gettext_noop("Used for debug purposes.") + }, + &csn_time_shift, + 0, INT_MIN, INT_MAX, + NULL, NULL, NULL + }, { {"block_size", PGC_INTERNAL, PRESET_OPTIONS, gettext_noop("Shows the size of a disk block."), diff --git a/src/backend/utils/probes.d b/src/backend/utils/probes.d index b0c50a3c7f..3fcd0f4ccf 100644 --- a/src/backend/utils/probes.d +++ b/src/backend/utils/probes.d @@ -77,6 +77,8 @@ provider postgresql { probe clog__checkpoint__done(bool); probe subtrans__checkpoint__start(bool); probe subtrans__checkpoint__done(bool); + probe csnlog__checkpoint__start(bool); + probe csnlog__checkpoint__done(bool); probe multixact__checkpoint__start(bool); probe multixact__checkpoint__done(bool); probe twophase__checkpoint__start(); diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 5001efdf7a..eaf2082e41 100644 --- a/src/backend/utils/time/snapmgr.c +++ b/src/backend/utils/time/snapmgr.c @@ -48,6 +48,7 @@ #include #include +#include "access/csn_log.h" #include "access/subtrans.h" #include "access/transam.h" #include "access/xact.h" @@ -77,6 +78,8 @@ */ int old_snapshot_threshold; /* number of minutes, -1 disables */ +bool enable_csn_snapshot; + volatile OldSnapshotControlData *oldSnapshotControl; @@ -173,6 +176,7 @@ static TimestampTz AlignTimestampToMinuteBoundary(TimestampTz ts); static Snapshot CopySnapshot(Snapshot snapshot); static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); +static bool XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot); /* * Snapshot fields to be serialized. @@ -191,6 +195,8 @@ typedef struct SerializedSnapshotData CommandId curcid; TimestampTz whenTaken; XLogRecPtr lsn; + CSN csn; + bool imported_csn; } SerializedSnapshotData; Size @@ -2130,6 +2136,8 @@ SerializeSnapshot(Snapshot snapshot, char *start_address) serialized_snapshot.curcid = snapshot->curcid; serialized_snapshot.whenTaken = snapshot->whenTaken; serialized_snapshot.lsn = snapshot->lsn; + serialized_snapshot.csn = snapshot->snapshot_csn; + serialized_snapshot.imported_csn = snapshot->imported_csn; /* * Ignore the SubXID array if it has overflowed, unless the snapshot was @@ -2204,6 +2212,8 @@ RestoreSnapshot(char *start_address) snapshot->curcid = serialized_snapshot.curcid; snapshot->whenTaken = serialized_snapshot.whenTaken; snapshot->lsn = serialized_snapshot.lsn; + snapshot->snapshot_csn = serialized_snapshot.csn; + snapshot->imported_csn = serialized_snapshot.imported_csn; snapshot->snapXactCompletionCount = 0; /* Copy XIDs, if present. */ @@ -2245,6 +2255,44 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc) /* * XidInMVCCSnapshot + * + * Check whether this xid is in snapshot. When enable_csn_snapshot is + * switched off just call XidInLocalMVCCSnapshot(). + */ +bool +XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +{ + bool in_snapshot; + + if (snapshot->imported_csn) + { + Assert(enable_csn_snapshot); + /* No point to using snapshot info except CSN */ + return XidInCSNSnapshot(xid, snapshot); + } + + in_snapshot = XidInLocalMVCCSnapshot(xid, snapshot); + + if (!get_csnlog_status()) + { + Assert(CSNIsFrozen(snapshot->snapshot_csn)); + return in_snapshot; + } + + if (in_snapshot) + { + /* + * This xid may be already in unknown state and in that case + * we must wait and recheck. + */ + return XidInCSNSnapshot(xid, snapshot); + } + else + return false; +} + +/* + * XidInLocalMVCCSnapshot * Is the given XID still-in-progress according to the snapshot? * * Note: GetSnapshotData never stores either top xid or subxids of our own @@ -2253,8 +2301,8 @@ RestoreTransactionSnapshot(Snapshot snapshot, void *source_pgproc) * TransactionIdIsCurrentTransactionId first, except when it's known the * XID could not be ours anyway. */ -bool -XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) +static bool +XidInLocalMVCCSnapshot(TransactionId xid, Snapshot snapshot) { uint32 i; @@ -2364,3 +2412,100 @@ XidInMVCCSnapshot(TransactionId xid, Snapshot snapshot) return false; } + + +/* + * ExportCSNSnapshot + * + * Export snapshot_csn so that caller can expand this transaction to other + * nodes. + * + * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and + * add some additional checks that transaction did not yet acquired xid, but + * for current iteration of this patch I don't want to hack on parser. + */ +SnapshotCSN +ExportCSNSnapshot() +{ + if (!get_csnlog_status()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not export csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + elog(DEBUG5, "Export CSN Snapshot: csn = %lu", + CurrentSnapshot->snapshot_csn); + return CurrentSnapshot->snapshot_csn; +} + +/* SQL accessor to ExportCSNSnapshot() */ +Datum +pg_csn_snapshot_export(PG_FUNCTION_ARGS) +{ + SnapshotCSN csn = ExportCSNSnapshot(); + + PG_RETURN_UINT64(csn); +} + +/* + * ImportCSNSnapshot + * + * Import csn and retract this backends xmin to the value that was + * actual when we had such csn. + * + * TODO: it's better to do this through EXPORT/IMPORT SNAPSHOT syntax and + * add some additional checks that transaction did not yet acquired xid, but + * for current iteration of this patch I don't want to hack on parser. + */ +void +ImportCSNSnapshot(SnapshotCSN snapshot_csn) +{ + volatile TransactionId xmin; + + if (!get_csnlog_status()) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not import csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is enabled.", + "enable_csn_snapshot"))); + + if (csn_snapshot_defer_time <= 0) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not import csn snapshot"), + errhint("Make sure the configuration parameter \"%s\" is positive.", + "csn_snapshot_defer_time"))); + + /* + * Call CSNSnapshotToXmin under ProcArrayLock to avoid situation that + * resulting xmin will be evicted from map before we will set it into our + * backend's xmin. + */ + LWLockAcquire(ProcArrayLock, LW_SHARED); + xmin = CSNSnapshotToXmin(snapshot_csn); + if (!TransactionIdIsValid(xmin)) + { + LWLockRelease(ProcArrayLock); + elog(ERROR, "CSNSnapshotToXmin: csn snapshot too old"); + } + + MyProc->originalXmin = MyProc->xmin; + MyProc->xmin = TransactionXmin = xmin; + LWLockRelease(ProcArrayLock); + + CurrentSnapshot->xmin = xmin; /* defuse SnapshotResetXmin() */ + CurrentSnapshot->snapshot_csn = snapshot_csn; + CurrentSnapshot->imported_csn = true; + CSNSnapshotSync(snapshot_csn); +} + +/* SQL accessor to ImportCSNSnapshot() */ +Datum +pg_csn_snapshot_import(PG_FUNCTION_ARGS) +{ + SnapshotCSN csn = PG_GETARG_UINT64(0); + + ImportCSNSnapshot(csn); + PG_RETURN_VOID(); +} diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index 31839c1a19..1864952bd2 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -227,7 +227,8 @@ static const char *const subdirs[] = { "pg_xact", "pg_logical", "pg_logical/snapshots", - "pg_logical/mappings" + "pg_logical/mappings", + "pg_csn" }; diff --git a/src/bin/pg_controldata/pg_controldata.c b/src/bin/pg_controldata/pg_controldata.c index f911f98d94..325e6a0e2b 100644 --- a/src/bin/pg_controldata/pg_controldata.c +++ b/src/bin/pg_controldata/pg_controldata.c @@ -300,6 +300,8 @@ main(int argc, char *argv[]) ControlFile->max_locks_per_xact); printf(_("track_commit_timestamp setting: %s\n"), ControlFile->track_commit_timestamp ? _("on") : _("off")); + printf(_("enable_csn_snapshot setting: %s\n"), + ControlFile->enable_csn_snapshot ? _("on") : _("off")); printf(_("Maximum data alignment: %u\n"), ControlFile->maxAlign); /* we don't print floatFormat since can't say much useful about it */ diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c index 3628bd74a7..18cf9197cc 100644 --- a/src/bin/pg_upgrade/pg_upgrade.c +++ b/src/bin/pg_upgrade/pg_upgrade.c @@ -548,6 +548,11 @@ copy_xact_xlog_xid(void) check_ok(); } + if(old_cluster.controldata.cat_ver > CSN_BASE_SNAPSHOT_ADD_VER) + { + copy_subdir_files("pg_csn", "pg_csn"); + } + /* now reset the wal archives in the new cluster */ prep_status("Resetting WAL archives"); exec_prog(UTILITY_LOG_FILE, NULL, true, true, diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h index ca0795f68f..54f2984387 100644 --- a/src/bin/pg_upgrade/pg_upgrade.h +++ b/src/bin/pg_upgrade/pg_upgrade.h @@ -124,6 +124,8 @@ extern char *output_files[]; */ #define JSONB_FORMAT_CHANGE_CAT_VER 201409291 +#define CSN_BASE_SNAPSHOT_ADD_VER 202002010 + /* * Each relation is represented by a relinfo structure. diff --git a/src/bin/pg_waldump/rmgrdesc.c b/src/bin/pg_waldump/rmgrdesc.c index 852d8ca4b1..2d280ce940 100644 --- a/src/bin/pg_waldump/rmgrdesc.c +++ b/src/bin/pg_waldump/rmgrdesc.c @@ -11,6 +11,7 @@ #include "access/brin_xlog.h" #include "access/clog.h" #include "access/commit_ts.h" +#include "access/csn_log.h" #include "access/generic_xlog.h" #include "access/ginxlog.h" #include "access/gistxlog.h" diff --git a/src/include/access/csn_log.h b/src/include/access/csn_log.h new file mode 100644 index 0000000000..12df028bf4 --- /dev/null +++ b/src/include/access/csn_log.h @@ -0,0 +1,98 @@ +/* + * csn_log.h + * + * Commit-Sequence-Number log. + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/access/csn_log.h + */ +#ifndef CSNLOG_H +#define CSNLOG_H + +#include "access/xlog.h" +#include "utils/snapshot.h" +#include "storage/sync.h" + + +#define InProgressCSN UINT64CONST(0x0) +#define AbortedCSN UINT64CONST(0x1) +#define FrozenCSN UINT64CONST(0x2) +#define InDoubtCSN UINT64CONST(0x3) +#define UnclearCSN UINT64CONST(0x4) +#define FirstNormalCSN UINT64CONST(0x5) + +#define CSNIsInProgress(csn) ((csn) == InProgressCSN) +#define CSNIsAborted(csn) ((csn) == AbortedCSN) +#define CSNIsFrozen(csn) ((csn) == FrozenCSN) +#define CSNIsInDoubt(csn) ((csn) == InDoubtCSN) +#define CSNIsUnclear(csn) ((csn) == UnclearCSN) +#define CSNIsNormal(csn) ((csn) >= FirstNormalCSN) + +/* XLOG stuff */ +#define XLOG_CSN_ASSIGNMENT 0x00 +#define XLOG_CSN_SETCSN 0x10 +#define XLOG_CSN_ZEROPAGE 0x20 +#define XLOG_CSN_TRUNCATE 0x30 + +/* + * We should log MAX generated CSN to wal, so that database will not generate + * a historical CSN after database restart. This may appear when system time + * turned back. + * + * However we can not log the MAX CSN every time it generated, if so it will + * cause too many wal expend, so we log it 5s more in the future. + * + * As a trade off, when this database restart, there will be 5s bad performance + * for time synchronization among sharding nodes. + * + * It looks like we can redefine this as a configure parameter, and the user + * can decide which way they prefer. + * + */ +#define CSN_ASSIGN_TIME_INTERVAL 5 + +typedef struct xl_csn_set +{ + CSN csn; + TransactionId xtop; /* XID's top-level XID */ + int nsubxacts; /* number of subtransaction XIDs */ + TransactionId xsub[FLEXIBLE_ARRAY_MEMBER]; /* assigned subxids */ +} xl_csn_set; + +#define MinSizeOfCSNSet offsetof(xl_csn_set, xsub) +#define CSNAddByNanosec(csn,second) (csn + second * 1000000000L) + +/* Main functions */ +extern void CSNLogSetCSN(TransactionId xid, int nsubxids, + TransactionId *subxids, CSN csn, bool write_xlog); +extern CSN CSNLogGetCSNByXid(TransactionId xid); + +/* Infrastructure functions */ +extern Size CSNLogShmemSize(void); +extern void CSNLogShmemInit(void); +extern void ActivateCSNlog(void); +extern void ExtendCSNLog(TransactionId newestXact); +extern void DeactivateCSNlog(void); + +extern void CheckPointCSNLog(void); +extern void TruncateCSNLog(TransactionId oldestXact); + +extern void csnlog_redo(XLogReaderState *record); +extern void csnlog_desc(StringInfo buf, XLogReaderState *record); +extern const char *csnlog_identify(uint8 info); +extern void WriteAssignCSNXlogRec(CSN csn); +extern void CatchCSNLog(void); +extern void StartupCSN(void); +extern void CompleteCSNInitialization(void); +extern void CSNlogParameterChange(bool newvalue, bool oldvalue); +extern bool get_csnlog_status(void); +extern int csnsyncfiletag(const FileTag *ftag, char *path); + +extern CSN GenerateCSN(bool locked, CSN assign); +extern CSN GetLastGeneratedCSN(void); + +extern TransactionId GetOldestXmin(void); + +#endif /* CSNLOG_H */ \ No newline at end of file diff --git a/src/include/access/csn_snapshot.h b/src/include/access/csn_snapshot.h new file mode 100644 index 0000000000..916603af0c --- /dev/null +++ b/src/include/access/csn_snapshot.h @@ -0,0 +1,54 @@ +/*------------------------------------------------------------------------- + * + * csn_snapshot.h + * Support for cross-node snapshot isolation. + * + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * src/include/access/csn_snapshot.h + * + *------------------------------------------------------------------------- + */ +#ifndef CSN_SNAPSHOT_H +#define CSN_SNAPSHOT_H + +#include "access/csn_log.h" +#include "port/atomics.h" +#include "storage/lock.h" +#include "utils/snapshot.h" +#include "utils/guc.h" + +/* + * snapshot.h is used in frontend code so atomic variant of SnapshotCSN type + * is defined here. + */ +typedef pg_atomic_uint64 CSN_atomic; + + +extern int csn_snapshot_defer_time; +extern int csn_time_shift; + + +extern Size CSNSnapshotShmemSize(void); +extern void CSNSnapshotShmemInit(void); +extern void CSNSnapshotStartup(TransactionId oldestActiveXID); + +extern void CSNSnapshotMapXmin(SnapshotCSN snapshot_csn); +extern TransactionId CSNSnapshotToXmin(SnapshotCSN snapshot_csn); + +extern bool XidInCSNSnapshot(TransactionId xid, Snapshot snapshot); + +extern CSN TransactionIdGetCSN(TransactionId xid); + +extern void CSNSnapshotAbort(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotPrecommit(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotCommit(PGPROC *proc, TransactionId xid, int nsubxids, + TransactionId *subxids); +extern void CSNSnapshotAssignCurrent(SnapshotCSN snapshot_csn); +extern SnapshotCSN CSNSnapshotPrepareCurrent(void); +extern void CSNSnapshotSync(SnapshotCSN remote_csn); + +#endif /* CSN_SNAPSHOT_H */ diff --git a/src/include/access/rmgrlist.h b/src/include/access/rmgrlist.h index f582cf535f..3cf0775176 100644 --- a/src/include/access/rmgrlist.h +++ b/src/include/access/rmgrlist.h @@ -47,3 +47,4 @@ PG_RMGR(RM_COMMIT_TS_ID, "CommitTs", commit_ts_redo, commit_ts_desc, commit_ts_i PG_RMGR(RM_REPLORIGIN_ID, "ReplicationOrigin", replorigin_redo, replorigin_desc, replorigin_identify, NULL, NULL, NULL) PG_RMGR(RM_GENERIC_ID, "Generic", generic_redo, generic_desc, generic_identify, NULL, NULL, generic_mask) PG_RMGR(RM_LOGICALMSG_ID, "LogicalMessage", logicalmsg_redo, logicalmsg_desc, logicalmsg_identify, NULL, NULL, NULL) +PG_RMGR(RM_CSNLOG_ID, "CSN", csnlog_redo, csnlog_desc, csnlog_identify, NULL, NULL, NULL) diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.h index c0da76cab4..2ee489dcad 100644 --- a/src/include/access/xlog_internal.h +++ b/src/include/access/xlog_internal.h @@ -242,6 +242,7 @@ typedef struct xl_parameter_change int wal_level; bool wal_log_hints; bool track_commit_timestamp; + bool enable_csn_snapshot; } xl_parameter_change; /* logs restore point */ @@ -332,5 +333,6 @@ extern bool ArchiveRecoveryRequested; extern bool InArchiveRecovery; extern bool StandbyMode; extern char *recoveryRestoreCommand; +extern bool enable_csn_wal; #endif /* XLOG_INTERNAL_H */ diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.h index 749bce0cc6..a7da532f3a 100644 --- a/src/include/catalog/pg_control.h +++ b/src/include/catalog/pg_control.h @@ -183,6 +183,7 @@ typedef struct ControlFileData int max_prepared_xacts; int max_locks_per_xact; bool track_commit_timestamp; + bool enable_csn_snapshot; /* * This data is used to check for hardware-architecture compatibility of diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index d068d6532e..d578aceb40 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11689,4 +11689,21 @@ prorettype => 'bytea', proargtypes => 'pg_brin_minmax_multi_summary', prosrc => 'brin_minmax_multi_summary_send' }, +# csn shnapshot handling +{ oid => '10001', descr => 'export csn snapshot', + proname => 'pg_csn_snapshot_export', provolatile => 'v', proparallel => 'u', + prorettype => 'int8', proargtypes => '', prosrc => 'pg_csn_snapshot_export' }, +{ oid => '10002', descr => 'import csn snapshot', + proname => 'pg_csn_snapshot_import', provolatile => 'v', proparallel => 'u', + prorettype => 'void', proargtypes => 'int8', prosrc => 'pg_csn_snapshot_import' }, +{ oid => '10003', descr => 'prepare distributed transaction for commit, get csn', + proname => 'pg_csn_snapshot_prepare', provolatile => 'v', proparallel => 'u', + prorettype => 'int8', proargtypes => 'text', prosrc => 'pg_csn_snapshot_prepare' }, +{ oid => '10004', descr => 'assign csn to distributed transaction', + proname => 'pg_csn_snapshot_assign', provolatile => 'v', proparallel => 'u', + prorettype => 'void', proargtypes => 'text int8', prosrc => 'pg_csn_snapshot_assign' }, +{ oid => '10005', descr => 'get current CSN', + proname => 'pg_current_csn', provolatile => 'v', proparallel => 'u', + prorettype => 'int8', proargtypes => '', prosrc => 'pg_current_csn' }, + ] diff --git a/src/include/datatype/timestamp.h b/src/include/datatype/timestamp.h index 99873497a6..8d1ced7430 100644 --- a/src/include/datatype/timestamp.h +++ b/src/include/datatype/timestamp.h @@ -93,6 +93,9 @@ typedef struct #define USECS_PER_MINUTE INT64CONST(60000000) #define USECS_PER_SEC INT64CONST(1000000) +#define NSECS_PER_SEC INT64CONST(1000000000) +#define NSECS_PER_USEC INT64CONST(1000) + /* * We allow numeric timezone offsets up to 15:59:59 either way from Greenwich. * Currently, the record holders for wackiest offsets in actual use are zones diff --git a/src/include/fmgr.h b/src/include/fmgr.h index ab7b85c86e..f08999740b 100644 --- a/src/include/fmgr.h +++ b/src/include/fmgr.h @@ -281,6 +281,7 @@ extern struct varlena *pg_detoast_datum_packed(struct varlena *datum); #define PG_GETARG_FLOAT4(n) DatumGetFloat4(PG_GETARG_DATUM(n)) #define PG_GETARG_FLOAT8(n) DatumGetFloat8(PG_GETARG_DATUM(n)) #define PG_GETARG_INT64(n) DatumGetInt64(PG_GETARG_DATUM(n)) +#define PG_GETARG_UINT64(n) DatumGetUInt64(PG_GETARG_DATUM(n)) /* use this if you want the raw, possibly-toasted input datum: */ #define PG_GETARG_RAW_VARLENA_P(n) ((struct varlena *) PG_GETARG_POINTER(n)) /* use this if you want the input datum de-toasted: */ diff --git a/src/include/portability/instr_time.h b/src/include/portability/instr_time.h index 39a4f0600e..a78f0d284b 100644 --- a/src/include/portability/instr_time.h +++ b/src/include/portability/instr_time.h @@ -141,6 +141,9 @@ typedef struct timespec instr_time; #define INSTR_TIME_GET_MICROSEC(t) \ (((uint64) (t).tv_sec * (uint64) 1000000) + (uint64) ((t).tv_nsec / 1000)) +#define INSTR_TIME_GET_NANOSEC(t) \ + (((uint64) (t).tv_sec * (uint64) 1000000000) + (uint64) ((t).tv_nsec)) + #else /* !HAVE_CLOCK_GETTIME */ /* Use gettimeofday() */ @@ -205,6 +208,10 @@ typedef struct timeval instr_time; #define INSTR_TIME_GET_MICROSEC(t) \ (((uint64) (t).tv_sec * (uint64) 1000000) + (uint64) (t).tv_usec) +#define INSTR_TIME_GET_NANOSEC(t) \ + (((uint64) (t).tv_sec * (uint64) 1000000000) + \ + (uint64) (t).tv_usec * (uint64) 1000) + #endif /* HAVE_CLOCK_GETTIME */ #else /* WIN32 */ @@ -237,6 +244,9 @@ typedef LARGE_INTEGER instr_time; #define INSTR_TIME_GET_MICROSEC(t) \ ((uint64) (((double) (t).QuadPart * 1000000.0) / GetTimerFrequency())) +#define INSTR_TIME_GET_NANOSEC(t) \ + ((uint64) (((double) (t).QuadPart * 1000000000.0) / GetTimerFrequency())) + static inline double GetTimerFrequency(void) { diff --git a/src/include/storage/lwlock.h b/src/include/storage/lwlock.h index a8f052e484..65d1e49fb2 100644 --- a/src/include/storage/lwlock.h +++ b/src/include/storage/lwlock.h @@ -168,6 +168,7 @@ typedef enum BuiltinTrancheIds LWTRANCHE_XACT_BUFFER = NUM_INDIVIDUAL_LWLOCKS, LWTRANCHE_COMMITTS_BUFFER, LWTRANCHE_SUBTRANS_BUFFER, + LWTRANCHE_CSN_LOG_BUFFERS, LWTRANCHE_MULTIXACTOFFSET_BUFFER, LWTRANCHE_MULTIXACTMEMBER_BUFFER, LWTRANCHE_NOTIFY_BUFFER, diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h index be67d8a861..ade5d8e169 100644 --- a/src/include/storage/proc.h +++ b/src/include/storage/proc.h @@ -15,12 +15,14 @@ #define _PROC_H_ #include "access/clog.h" +#include "access/csn_snapshot.h" #include "access/xlogdefs.h" #include "lib/ilist.h" #include "storage/latch.h" #include "storage/lock.h" #include "storage/pg_sema.h" #include "storage/proclist_types.h" +#include "utils/snapshot.h" /* * Each backend advertises up to PGPROC_MAX_CACHED_SUBXIDS TransactionIds @@ -251,6 +253,18 @@ struct PGPROC PGPROC *lockGroupLeader; /* lock group leader, if I'm a member */ dlist_head lockGroupMembers; /* list of members, if I'm a leader */ dlist_node lockGroupLink; /* my member link, if I'm a member */ + + /* + * assignedCSN holds CSN for this transaction. It is generated + * under a ProcArray lock and later is written to a CSNLog. This + * variable defined as atomic only for case of group commit, in all other + * scenarios only backend responsible for this proc entry is working with + * this variable. + */ + CSN_atomic assignedCSN; + + /* Original xmin of this backend before csn snapshot was imported */ + TransactionId originalXmin; }; /* NOTE: "typedef struct PGPROC PGPROC" appears in storage/lock.h. */ diff --git a/src/include/storage/procarray.h b/src/include/storage/procarray.h index b01fa52139..ba580435f9 100644 --- a/src/include/storage/procarray.h +++ b/src/include/storage/procarray.h @@ -20,6 +20,10 @@ #include "utils/snapshot.h" +#define PROCARRAY_NON_IMPORTED_XMIN 0x80 /* use originalXmin instead + * of xmin to properly + * maintain csnXidMap */ + extern Size ProcArrayShmemSize(void); extern void CreateSharedProcArray(void); extern void ProcArrayAdd(PGPROC *proc); @@ -94,4 +98,7 @@ extern void ProcArraySetReplicationSlotXmin(TransactionId xmin, extern void ProcArrayGetReplicationSlotXmin(TransactionId *xmin, TransactionId *catalog_xmin); +extern void ProcArraySetCSNSnapshotXmin(TransactionId xmin); + +extern TransactionId ProcArrayGetCSNSnapshotXmin(void); #endif /* PROCARRAY_H */ diff --git a/src/include/storage/sync.h b/src/include/storage/sync.h index 6fd50cfa7b..eb1d52673a 100644 --- a/src/include/storage/sync.h +++ b/src/include/storage/sync.h @@ -39,6 +39,7 @@ typedef enum SyncRequestHandler SYNC_HANDLER_COMMIT_TS, SYNC_HANDLER_MULTIXACT_OFFSET, SYNC_HANDLER_MULTIXACT_MEMBER, + SYNC_HANDLER_CSN, SYNC_HANDLER_NONE } SyncRequestHandler; diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index c6a176cc95..122eea20ba 100644 --- a/src/include/utils/snapmgr.h +++ b/src/include/utils/snapmgr.h @@ -41,10 +41,11 @@ && !RelationIsAccessibleInLogicalDecoding(rel) \ ) -#define EarlyPruningEnabled(rel) (old_snapshot_threshold >= 0 && RelationAllowsEarlyPruning(rel)) +#define EarlyPruningEnabled(rel) (old_snapshot_threshold >= 0 && !enable_csn_snapshot && RelationAllowsEarlyPruning(rel)) /* GUC variables */ extern PGDLLIMPORT int old_snapshot_threshold; +extern PGDLLIMPORT bool enable_csn_snapshot; extern Size SnapMgrShmemSize(void); @@ -100,7 +101,7 @@ extern PGDLLIMPORT SnapshotData CatalogSnapshotData; static inline bool OldSnapshotThresholdActive(void) { - return old_snapshot_threshold >= 0; + return (old_snapshot_threshold >= 0) && (!enable_csn_snapshot); } extern Snapshot GetTransactionSnapshot(void); @@ -130,6 +131,8 @@ extern void AtSubCommit_Snapshot(int level); extern void AtSubAbort_Snapshot(int level); extern void AtEOXact_Snapshot(bool isCommit, bool resetXmin); +extern SnapshotCSN ExportCSNSnapshot(void); +extern void ImportCSNSnapshot(SnapshotCSN snapshot_csn); extern void ImportSnapshot(const char *idstr); extern bool XactHasExportedSnapshots(void); extern void DeleteAllExportedSnapshotFiles(void); diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h index 6b60755c53..3580a94c43 100644 --- a/src/include/utils/snapshot.h +++ b/src/include/utils/snapshot.h @@ -121,6 +121,9 @@ typedef enum SnapshotType typedef struct SnapshotData *Snapshot; #define InvalidSnapshot ((Snapshot) NULL) +#define InvalidCSN ((CSN) 0) +typedef uint64 CSN; +typedef uint64 SnapshotCSN; /* * Struct representing all kind of possible snapshots. @@ -214,6 +217,14 @@ typedef struct SnapshotData * transactions completed since the last GetSnapshotData(). */ uint64 snapXactCompletionCount; + + /* + * SnapshotCSN for snapshot isolation support. + * Will be used only if enable_csn_snapshot is enabled. + */ + SnapshotCSN snapshot_csn; + /* Did we have our own snapshot_csn or imported one from different node */ + bool imported_csn; } SnapshotData; #endif /* SNAPSHOT_H */ diff --git a/src/test/modules/Makefile b/src/test/modules/Makefile index dffc79b2d9..16bb65e7e1 100644 --- a/src/test/modules/Makefile +++ b/src/test/modules/Makefile @@ -7,6 +7,7 @@ include $(top_builddir)/src/Makefile.global SUBDIRS = \ brin \ commit_ts \ + csnsnapshot \ delay_execution \ dummy_index_am \ dummy_seclabel \ diff --git a/src/test/modules/csnsnapshot/Makefile b/src/test/modules/csnsnapshot/Makefile new file mode 100644 index 0000000000..fa4245752b --- /dev/null +++ b/src/test/modules/csnsnapshot/Makefile @@ -0,0 +1,25 @@ +# src/test/modules/csnsnapshot/Makefile + +REGRESS_OPTS = --temp-config=$(top_srcdir)/src/test/modules/csnsnapshot/csn_snapshot.conf +NO_INSTALLCHECK = 1 + +TAP_TESTS = 1 +EXTRA_INSTALL=contrib/postgres_fdw + +# Don't support full consistency of distributed commit in READ COMMITTED +# transactions. +#PROVE_TESTS = t/001_base.pl \ +# t/002_standby.pl \ +# t/003_time_skew.pl +PROVE_TESTS = t/005_basic_visibility.pl + +ifdef USE_PGXS +PG_CONFIG = pg_config +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) +else +subdir = src/test/modules/csnsnapshot +top_builddir = ../../../.. +include $(top_builddir)/src/Makefile.global +include $(top_srcdir)/contrib/contrib-global.mk +endif diff --git a/src/test/modules/csnsnapshot/csn_snapshot.conf b/src/test/modules/csnsnapshot/csn_snapshot.conf new file mode 100644 index 0000000000..e9d3c35756 --- /dev/null +++ b/src/test/modules/csnsnapshot/csn_snapshot.conf @@ -0,0 +1 @@ +track_commit_timestamp = on diff --git a/src/test/modules/csnsnapshot/expected/csnsnapshot.out b/src/test/modules/csnsnapshot/expected/csnsnapshot.out new file mode 100644 index 0000000000..ac28e417b6 --- /dev/null +++ b/src/test/modules/csnsnapshot/expected/csnsnapshot.out @@ -0,0 +1 @@ +create table t1(i int, j int, k varchar); diff --git a/src/test/modules/csnsnapshot/t/001_base.pl b/src/test/modules/csnsnapshot/t/001_base.pl new file mode 100644 index 0000000000..3b5a09000b --- /dev/null +++ b/src/test/modules/csnsnapshot/t/001_base.pl @@ -0,0 +1,103 @@ +# Single-node test: value can be set, and is still present after recovery + +use strict; +use warnings; + +use TestLib; +use Test::More tests => 5; +use PostgresNode; + +my $node = PostgresNode->new('csntest'); +$node->init; +$node->append_conf('postgresql.conf', qq{ + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + max_prepared_transactions = 10 + }); +$node->start; + +my $test_1 = 1; + +# Create a table +$node->safe_psql('postgres', 'create table t1(i int, j int)'); + +# insert test record +$node->safe_psql('postgres', 'insert into t1 values(1,1)'); +# export csn snapshot +my $test_snapshot = $node->safe_psql('postgres', 'select pg_csn_snapshot_export()'); +# insert test record +$node->safe_psql('postgres', 'insert into t1 values(2,1)'); + +my $count1 = $node->safe_psql('postgres', "select count(*) from t1"); +is($count1, '2', 'Get right number in normal query'); +my $count2 = $node->safe_psql('postgres', " + begin transaction isolation level repeatable read; + select pg_csn_snapshot_import($test_snapshot); + select count(*) from t1; + commit;" + ); + +is($count2, ' +1', 'Get right number in csn import query'); + +#prepare transaction test +$node->safe_psql('postgres', " + begin; + insert into t1 values(3,1); + insert into t1 values(3,2); + prepare transaction 'pt3'; + "); +$node->safe_psql('postgres', " + begin; + insert into t1 values(4,1); + insert into t1 values(4,2); + prepare transaction 'pt4'; + "); +$node->safe_psql('postgres', " + begin; + insert into t1 values(5,1); + insert into t1 values(5,2); + prepare transaction 'pt5'; + "); +$node->safe_psql('postgres', " + begin; + insert into t1 values(6,1); + insert into t1 values(6,2); + prepare transaction 'pt6'; + "); +$node->safe_psql('postgres', "commit prepared 'pt4';"); + +# restart with enable_csn_snapshot off +$node->append_conf('postgresql.conf', "enable_csn_snapshot = off"); +$node->restart; +$node->safe_psql('postgres', " + insert into t1 values(7,1); + insert into t1 values(7,2); + "); +$node->safe_psql('postgres', "commit prepared 'pt3';"); +$count1 = $node->safe_psql('postgres', "select count(*) from t1"); +is($count1, '8', 'Get right number in normal query'); + + +# restart with enable_csn_snapshot on +$node->append_conf('postgresql.conf', "enable_csn_snapshot = on"); +$node->restart; +$node->safe_psql('postgres', " + insert into t1 values(8,1); + insert into t1 values(8,2); + "); +$node->safe_psql('postgres', "commit prepared 'pt5';"); +$count1 = $node->safe_psql('postgres', "select count(*) from t1"); +is($count1, '12', 'Get right number in normal query'); + +# restart with enable_csn_snapshot off +$node->append_conf('postgresql.conf', "enable_csn_snapshot = on"); +$node->restart; +$node->safe_psql('postgres', " + insert into t1 values(9,1); + insert into t1 values(9,2); + "); +$node->safe_psql('postgres', "commit prepared 'pt6';"); + +$count1 = $node->safe_psql('postgres', "select count(*) from t1"); +is($count1, '16', 'Get right number in normal query'); diff --git a/src/test/modules/csnsnapshot/t/002_standby.pl b/src/test/modules/csnsnapshot/t/002_standby.pl new file mode 100644 index 0000000000..2b09712141 --- /dev/null +++ b/src/test/modules/csnsnapshot/t/002_standby.pl @@ -0,0 +1,66 @@ +# Test simple scenario involving a standby + +use strict; +use warnings; + +use TestLib; +use Test::More tests => 6; +use PostgresNode; + +my $bkplabel = 'backup'; +my $master = PostgresNode->new('master'); +$master->init(allows_streaming => 1); + +$master->append_conf( + 'postgresql.conf', qq{ + enable_csn_snapshot = on + max_wal_senders = 5 + }); +$master->start; +$master->backup($bkplabel); + +my $standby = PostgresNode->new('standby'); +$standby->init_from_backup($master, $bkplabel, has_streaming => 1); +$standby->start; + +$master->safe_psql('postgres', "create table t1(i int, j int)"); + +my $guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot'); +is($guc_on_master, 'on', "GUC on master"); + +my $guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot'); +is($guc_on_standby, 'on', "GUC on standby"); + +$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off'); +$master->restart; + +$guc_on_master = $master->safe_psql('postgres', 'show enable_csn_snapshot'); +is($guc_on_master, 'off', "GUC off master"); + +$guc_on_standby = $standby->safe_psql('postgres', 'show enable_csn_snapshot'); +is($guc_on_standby, 'on', "GUC on standby"); + +# We consume a large number of transaction,for skip page +for my $i (1 .. 4096) #4096 +{ + $master->safe_psql('postgres', "insert into t1 values(1,$i)"); +} +$master->safe_psql('postgres', "select pg_sleep(2)"); +$master->append_conf('postgresql.conf', 'enable_csn_snapshot = on'); +$master->restart; + +my $count_standby = $standby->safe_psql('postgres', 'select count(*) from t1'); +is($count_standby, '4096', "Ok for siwtch xid-base > csn-base"); #4096 + +# We consume a large number of transaction,for skip page +for my $i (1 .. 4096) #4096 +{ + $master->safe_psql('postgres', "insert into t1 values(1,$i)"); +} +$master->safe_psql('postgres', "select pg_sleep(2)"); + +$master->append_conf('postgresql.conf', 'enable_csn_snapshot = off'); +$master->restart; + +$count_standby = $standby->safe_psql('postgres', 'select count(*) from t1'); +is($count_standby, '8192', "Ok for switch csn-base > xid-base"); #8192 \ No newline at end of file diff --git a/src/test/modules/csnsnapshot/t/003_time_skew.pl b/src/test/modules/csnsnapshot/t/003_time_skew.pl new file mode 100644 index 0000000000..f2496ea883 --- /dev/null +++ b/src/test/modules/csnsnapshot/t/003_time_skew.pl @@ -0,0 +1,214 @@ +use strict; +use warnings; + +use TestLib; +use Test::More tests => 13; +use PostgresNode; + +my $node1 = PostgresNode->new('csn1'); +$node1->init; +$node1->append_conf('postgresql.conf', qq{ + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + max_prepared_transactions = 10 + csn_time_shift = 0 + shared_preload_libraries = 'postgres_fdw' + postgres_fdw.use_csn_snapshots = true + }); +$node1->start; +my $node2 = PostgresNode->new('csn2'); +$node2->init; +$node2->append_conf('postgresql.conf', qq{ + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + max_prepared_transactions = 10 + csn_time_shift = 0 + shared_preload_libraries = 'postgres_fdw' + postgres_fdw.use_csn_snapshots = true + }); +$node2->start; + +$node1->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node2->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; + CREATE TABLE summary(value int, ntrans int); + INSERT INTO summary (value, ntrans) VALUES (0, 0); +"); +$node2->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node1->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; + CREATE FOREIGN TABLE summary(value int, ntrans int) SERVER remote; +"); + +$node1->safe_psql('postgres', " + CREATE TABLE t (id int, payload int) PARTITION BY HASH(id); + CREATE TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0); + CREATE FOREIGN TABLE t_2 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1) SERVER remote; +"); +$node2->safe_psql('postgres', " + CREATE TABLE t (id serial, payload int) PARTITION BY HASH(id); + CREATE FOREIGN TABLE t_1 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 0) SERVER remote; + CREATE TABLE t_2 PARTITION OF t FOR VALUES WITH (modulus 2, remainder 1); +"); + +$node1->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 1 FROM generate_series(1,100) AS gs)"); +$node2->safe_psql('postgres', "INSERT INTO t(id, payload) (SELECT gs.*, 2 FROM generate_series(101,200) AS gs)"); +my $count1 = $node1->safe_psql('postgres', "SELECT SUM(payload) FROM t"); +my $count2 = $node2->safe_psql('postgres', "SELECT SUM(payload) FROM t"); +is( (($count1 == 300) and ($count1 == $count2)), 1, 'Correct insert'); + +# ############################################################################## +# +# Basic test. Check REPEATABLE READ anomaly. +# ntrans is needed to control that some transactions were committed. +# +# ############################################################################## + +my $q1 = File::Temp->new(); +append_to_file($q1, q{ + START TRANSACTION ISOLATION LEVEL REPEATABLE READ; + UPDATE summary SET value = value + (SELECT SUM(payload) FROM t); + UPDATE summary SET value = value - (SELECT SUM(payload) FROM t); + UPDATE summary SET ntrans = ntrans + 1; + COMMIT; +}); +my $q2 = File::Temp->new(); +append_to_file($q2, q{ + BEGIN; + \set pl random(-100, 100) + \set id random(1, 200) + UPDATE t SET payload = :pl WHERE id = :id; + COMMIT; +}); + +my $seconds = 5; +my $pgb_handle1; +my $pgb_handle2; + +$pgb_handle1 = $node1->pgbench_async(-n, -c => 5, -T => $seconds, -f => $q1, 'postgres' ); +$pgb_handle2 = $node2->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q2, 'postgres' ); +$node1->pgbench_await($pgb_handle1); +$node2->pgbench_await($pgb_handle2); + +$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary"); +$count2 = $node2->safe_psql('postgres', "SELECT SUM(value) FROM summary"); +my $ntrans = $node2->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary"); +note("$count1, $count2, $ntrans"); +is( ( ($ntrans > 0) and ($count1 == 0) and ($count1 == $count2)), 1, 'Correct update'); + +# ############################################################################## +# +# Test on 'snapshot too old' +# +# ############################################################################## +$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0;"); +$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = -100"); +$node2->restart(); + +# READ COMMITTED transactions ignores the time skew. +$node2->psql('postgres', "UPDATE summary SET ntrans = 1"); +$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary"); +note("$ntrans"); +is( $ntrans, 1, 'Read committed behavior if snapshot turn sour'); + +# But REPEATABLE READ transactions isn't +$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = +100"); +$node1->restart(); +my $err = ''; +$node2->psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 2; COMMIT;", stderr => \$err); +$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary"); +note("$ntrans"); +is( (($ntrans == 1) and (index($err, 'csn snapshot too old') != -1)), 1, 'Read committed can\'t update if snapshot turn sour'); + +# ############################################################################## +# +# Test on issue #1: +# 'xact confirmed as committed, so any following xact must see its effects'. +# +# ############################################################################## +$node1->safe_psql('postgres', "delete from t"); +$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 5"); +$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 0"); +$node1->restart(); +$node2->restart(); + +my $st_sec; my $end_sec; +my $time_diff; + +$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1), (3,1); COMMIT;"); +$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;"); +is( $ntrans, 2, 'Slow node can see mix node data change'); +$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;"); +is( $ntrans, 2, 'Fast node can see mix node data change'); + +$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; INSERT INTO t VALUES(1,1); COMMIT;"); +$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT count(*) FROM t; COMMIT;"); +is( $ntrans, 3, 'CURRENTLY FAILED:Data change to fast node on slow node, and slow node can see data change'); + +# READ COMMITED mode ignores the time skew. +$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 1"); +$ntrans = $node2->safe_psql('postgres', "SELECT ntrans FROM summary"); +note("ntrans: $ntrans\n"); +is( $ntrans, 1, 'See committed values in the READ COMMITTED mode'); + +# Access from the future +$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = ntrans + 1; COMMIT;"); +$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;"); +note("ntrans: $ntrans\n"); +is( $ntrans, 1, 'Do not see values, committed in the future at the REPEATABLE READ mode'); + +# But... +$node1->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 0"); +$node2->safe_psql('postgres', "ALTER SYSTEM SET csn_time_shift = 5"); +$node1->restart(); +$node2->restart(); + +# Check READ COMMITED mode +$node2->safe_psql('postgres', "UPDATE summary SET ntrans = 2"); +$ntrans = $node1->safe_psql('postgres', "SELECT ntrans FROM summary"); +note("ntrans: $ntrans\n"); +is( $ntrans, 2, 'See committed values in the READ COMMITTED mode, step 2'); + +# Node from the future will wait for a time before UPDATE table. +($st_sec) = localtime(); +$node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 3; COMMIT;"); +($end_sec) = localtime(); $time_diff = $end_sec - $st_sec; +$ntrans = $node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;"); +note("ntrans: $ntrans, Test time: $time_diff seconds"); +is( ($ntrans == 3), 1, 'The test execution time correlates with the time offset.'); + +# Node from the future will wait for a time before SELECT from a table. +$node1->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; UPDATE summary SET ntrans = 4; COMMIT;"); +($st_sec) = localtime(); +$ntrans = $node2->safe_psql('postgres', "START TRANSACTION ISOLATION LEVEL REPEATABLE READ; SELECT ntrans FROM summary; COMMIT;"); +($end_sec) = localtime(); $time_diff = $end_sec - $st_sec; +note("ntrans: $ntrans, Test time: $time_diff seconds ($end_sec, $st_sec)"); +is( ($ntrans == 4), 1, 'See values, committed in the past. The test execution time correlates with the time offset.'); + +$node1->safe_psql('postgres', "UPDATE summary SET ntrans = 0, value = 0"); +$q1 = File::Temp->new(); +append_to_file($q1, q{ + UPDATE summary SET value = value + 1, ntrans = ntrans + 1; +}); +$q2 = File::Temp->new(); +append_to_file($q2, q{ + START TRANSACTION ISOLATION LEVEL REPEATABLE READ; + UPDATE summary SET value = value + (SELECT SUM(ntrans) FROM summary); + UPDATE summary SET value = value - (SELECT SUM(ntrans) FROM summary); + COMMIT; +}); +$seconds = 3; +$pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q1, 'postgres' ); +$pgb_handle2 = $node2->pgbench_async(-n, -c => 1, -T => $seconds, -f => $q2, 'postgres' ); +$node1->pgbench_await($pgb_handle1); +$node2->pgbench_await($pgb_handle2); + +$count1 = $node1->safe_psql('postgres', "SELECT SUM(value) FROM summary"); +$count2 = $node1->safe_psql('postgres', "SELECT SUM(ntrans) FROM summary"); +note("$count1, $count2"); +is( ( ($count1 > 0) and ($count1 == $count2)), 1, 'Skew test'); + +$node1->stop(); +$node2->stop(); diff --git a/src/test/modules/csnsnapshot/t/004_read_committed.pl b/src/test/modules/csnsnapshot/t/004_read_committed.pl new file mode 100644 index 0000000000..ba27536a7a --- /dev/null +++ b/src/test/modules/csnsnapshot/t/004_read_committed.pl @@ -0,0 +1,97 @@ +use strict; +use warnings; + +use TestLib; +use Test::More tests => 2; +use PostgresNode; + +my $node1 = PostgresNode->new('csn1'); +$node1->init; +$node1->append_conf('postgresql.conf', qq{ + max_prepared_transactions = 20 + shared_preload_libraries = 'postgres_fdw' + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + postgres_fdw.use_csn_snapshots = true + csn_time_shift = 0 + }); +$node1->start; +my $node2 = PostgresNode->new('csn2'); +$node2->init; +$node2->append_conf('postgresql.conf', qq{ + max_prepared_transactions = 20 + shared_preload_libraries = 'postgres_fdw' + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + postgres_fdw.use_csn_snapshots = true + csn_time_shift = 0 + }); +$node2->start; + +# Create foreign servers +$node1->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node2->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; +"); +$node2->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node1->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; +"); + +# Create sharded table +$node1->safe_psql('postgres', " + CREATE TABLE dept1(name TEXT); + CREATE FOREIGN TABLE dept2 (name TEXT) SERVER remote; +"); +$node2->safe_psql('postgres', " + CREATE TABLE dept2(name TEXT); + CREATE FOREIGN TABLE dept1 (name TEXT) SERVER remote; + CREATE TABLE results(success_tx int); + INSERT INTO results (success_tx) VALUES (0); +"); + +# Fill the table +$node1->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Jonathan')"); +$node1->safe_psql('postgres', "INSERT INTO dept2 (name) VALUES ('Hoshi')"); +$node2->safe_psql('postgres', "INSERT INTO dept1 (name) VALUES ('Leonard')"); +my $count1 = $node1->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a"); +my $count2 = $node2->safe_psql('postgres', "SELECT count(*) FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a"); +note("$count1, $count2"); +is( (($count1 == 3) and ($count1 == $count2)), 1, 'Correct insert'); + +# Queries +my $q1 = File::Temp->new(); +append_to_file($q1, q{ + BEGIN; + SELECT count(*) AS cnt FROM dept1; \gset + \if :cnt > 0 + INSERT INTO dept2 (SELECT * FROM dept1); + DELETE FROM dept1; + \else + INSERT INTO dept1 (SELECT * FROM dept2); + DELETE FROM dept2; + \endif + + COMMIT; +}); +my $q2 = File::Temp->new(); +append_to_file($q2, q{ + SELECT count(*) AS cnt FROM ((SELECT * FROM dept1) UNION (SELECT * FROM dept2)) AS a; \gset + \if :cnt = 3 + UPDATE results SET success_tx = success_tx + 1; + \endif +}); +my $transactions = 1000; +my $pgb_handle1 = $node1->pgbench_async(-n, -c => 1, -t => $transactions, -f => $q1, 'postgres' ); +my $pgb_handle2 = $node2->pgbench_async(-n, -c => 20, -t => $transactions, -f => $q2, 'postgres' ); +$node1->pgbench_await($pgb_handle1); +$node2->pgbench_await($pgb_handle2); + +$count2 = $node2->safe_psql('postgres', "SELECT success_tx FROM results"); +note("$count2"); +is( $count2, 20*$transactions, 'Correct READ COMMITTED updates'); + +$node1->stop(); +$node2->stop(); diff --git a/src/test/modules/csnsnapshot/t/005_basic_visibility.pl b/src/test/modules/csnsnapshot/t/005_basic_visibility.pl new file mode 100644 index 0000000000..93fa348e7b --- /dev/null +++ b/src/test/modules/csnsnapshot/t/005_basic_visibility.pl @@ -0,0 +1,181 @@ +use strict; +use warnings; + +use PostgreSQL::Test::Cluster; +use PostgreSQL::Test::Utils; +use Test::More tests => 3; + +my $node1 = PostgreSQL::Test::Cluster->new('csn1'); +$node1->init; +$node1->append_conf('postgresql.conf', qq{ + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + max_prepared_transactions = 30 + csn_time_shift = 0 + shared_preload_libraries = 'postgres_fdw' + postgres_fdw.use_csn_snapshots = true + log_statement = none + default_transaction_isolation = 'REPEATABLE READ' + log_min_messages = LOG + }); + +my $node2 = PostgreSQL::Test::Cluster->new('csn2'); +$node2->init; +$node2->append_conf('postgresql.conf', qq{ + enable_csn_snapshot = on + csn_snapshot_defer_time = 10 + max_prepared_transactions = 30 + csn_time_shift = 0 + shared_preload_libraries = 'postgres_fdw' + postgres_fdw.use_csn_snapshots = true + log_statement = none + default_transaction_isolation = 'REPEATABLE READ' + log_min_messages = LOG + }); +$node1->start; +$node2->start; + +$node1->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node2->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; + CREATE TABLE test(key int, x bigint) PARTITION BY LIST (key); + CREATE TABLE t1 PARTITION OF test FOR VALUES IN (1); + CREATE FOREIGN TABLE t2 PARTITION OF test FOR VALUES IN (2) SERVER remote; +"); +$node2->safe_psql('postgres', " + CREATE EXTENSION postgres_fdw; + CREATE SERVER remote FOREIGN DATA WRAPPER postgres_fdw OPTIONS (port '".$node1->port."'); + CREATE USER MAPPING FOR PUBLIC SERVER remote; + CREATE TABLE test(key int, x bigint) PARTITION BY LIST (key); + CREATE FOREIGN TABLE t1 PARTITION OF test FOR VALUES IN (1) SERVER remote; + CREATE TABLE t2 PARTITION OF test FOR VALUES IN (2); +"); +$node1->safe_psql('postgres', " + INSERT INTO test (key, x) VALUES (1, -1); + INSERT INTO test (key, x) VALUES (2, 1); +"); + +$node1->safe_psql('postgres', "VACUUM FULL"); +$node2->safe_psql('postgres', "VACUUM FULL"); + +# ############################################################################## +# +# Tests +# +# ############################################################################## + +my $updates = File::Temp->new(); +append_to_file($updates, q{ + BEGIN; + UPDATE test SET x = x + 1; + UPDATE test SET x = x - 1; + END; +}); + +my $local_update1 = File::Temp->new(); +append_to_file($local_update1, q{ + BEGIN; + UPDATE t1 SET x = x + 1; + UPDATE t1 SET x = x - 1; + END; +}); +my $local_update2 = File::Temp->new(); +append_to_file($local_update2, q{ + BEGIN; + UPDATE t2 SET x = x + 1; + UPDATE t2 SET x = x - 1; + END; +}); + +my ($pgb_handle1, $pgb_handle2, $pgb_handle3, $sum1, $sum2, $errors, $selects, $started); +my $test_time = 30; +my $result; + +# ############################################################################## +# +# Concurrent local UPDATE and global SELECT +# +# ############################################################################## +$errors = 0; +$selects = 0; +$started = time(); +$pgb_handle1 = $node1->pgbench_async(-n, -c => 5, -T => $test_time, -f => $local_update1, 'postgres' ); +while (time() - $started < $test_time) +{ + $result = $node2->safe_psql('postgres', " + SELECT 'sum=' || sum(x), (SELECT x FROM t1), (SELECT x FROM t2) + FROM test;"); + + if ( index($result, "sum=0") < 0 ) + { + diag("[$selects] Isolation error. result = [ $result ]"); + $errors++; + $node1->stop(); + $node2->stop(); + exit(1); + } + $selects++; +} +$node1->pgbench_await($pgb_handle1); +note("TOTAL: selects = $selects, errors = $errors"); +is($errors == 0, 1, 'Local updates'); +#exit(1); + +# ############################################################################## +# +# Global UPDATE and global SELECT +# +# ############################################################################## +$errors = 0; +$selects = 0; +$started = time(); +$pgb_handle1 = $node1->pgbench_async(-n, -c => 5, -T => $test_time, -f => $updates, 'postgres' ); +while (time() - $started < $test_time) +{ + $result = $node2->safe_psql('postgres', " + SELECT 'sum=' || sum(x), (SELECT x FROM t1), (SELECT x FROM t2) + FROM test;"); + + if ( index($result, "sum=0") < 0 ) + { + diag("[$selects] Isolation error. result = [ $result ]"); + $errors++; + } + $selects++; +} +$node1->pgbench_await($pgb_handle1); +note("TOTAL: selects = $selects, errors = $errors"); +is($errors == 0, 1, 'Distributed updates'); + +# ############################################################################## +# +# Local UPDATEs, global UPDATE and global SELECT +# +# ############################################################################## +$errors = 0; +$selects = 0; +$started = time(); +$pgb_handle1 = $node1->pgbench_async(-n, -c => 2, -T => $test_time, -f => $updates, 'postgres' ); +$pgb_handle2 = $node2->pgbench_async(-n, -c => 2, -T => $test_time, -f => $local_update2, 'postgres' ); +$pgb_handle3 = $node1->pgbench_async(-n, -c => 2, -T => $test_time, -f => $local_update1, 'postgres' ); +while (time() - $started < $test_time) +{ + $sum1 = $node1->safe_psql('postgres', "SELECT sum(x) FROM test;"); + $sum2 = $node2->safe_psql('postgres', "SELECT sum(x) FROM test;"); + + if ( ($sum1 ne 0) or ($sum2 ne 0) ) + { + diag("[$selects] Isolation error. Sums = [ $sum1, $sum2 ]"); + $errors++; + } + $selects++; +} +$node1->pgbench_await($pgb_handle1); +$node1->pgbench_await($pgb_handle3); +$node2->pgbench_await($pgb_handle2); +note("TOTAL: selects = $selects, errors = $errors"); +is($errors == 0, 1, 'Mix of local and distributed updates'); + +$node1->stop(); +$node2->stop(); diff --git a/src/test/modules/snapshot_too_old/sto.conf b/src/test/modules/snapshot_too_old/sto.conf index 7eeaeeb0dc..3177cc0e15 100644 --- a/src/test/modules/snapshot_too_old/sto.conf +++ b/src/test/modules/snapshot_too_old/sto.conf @@ -1,2 +1,3 @@ autovacuum = off old_snapshot_threshold = 0 +enable_csn_snapshot = false diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out index 2088857615..010b3b3144 100644 --- a/src/test/regress/expected/sysviews.out +++ b/src/test/regress/expected/sysviews.out @@ -104,6 +104,8 @@ select name, setting from pg_settings where name like 'enable%'; --------------------------------+--------- enable_async_append | on enable_bitmapscan | on + enable_csn_snapshot | on + enable_csn_wal | on enable_gathermerge | on enable_hashagg | on enable_hashjoin | on @@ -122,7 +124,7 @@ select name, setting from pg_settings where name like 'enable%'; enable_seqscan | on enable_sort | on enable_tidscan | on -(20 rows) +(22 rows) -- Test that the pg_timezone_names and pg_timezone_abbrevs views are -- more-or-less working. We can't test their contents in any great detail -- 2.25.1 --------------D1AC7FD3DF5DDBE5D0DD522A--