Received: from malur.postgresql.org ([217.196.149.56]) by arkaria.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_CBC_SHA384:256) (Exim 4.89) (envelope-from ) id 1fDY7s-00087Z-2C for pgsql-hackers@arkaria.postgresql.org; Tue, 01 May 2018 16:27:37 +0000 Received: from localhost ([127.0.0.1] helo=malur.postgresql.org) by malur.postgresql.org with esmtp (Exim 4.89) (envelope-from ) id 1fDY7q-00075P-RB for pgsql-hackers@arkaria.postgresql.org; Tue, 01 May 2018 16:27:34 +0000 Received: from makus.postgresql.org ([2001:4800:1501:1::229]) by malur.postgresql.org with esmtps (TLS1.2:ECDHE_RSA_AES_256_CBC_SHA384:256) (Exim 4.89) (envelope-from ) id 1fDY7p-00075F-Nl for pgsql-hackers@lists.postgresql.org; Tue, 01 May 2018 16:27:34 +0000 Received: from mail.postgrespro.ru ([93.174.131.138]) by makus.postgresql.org with esmtp (Exim 4.89) (envelope-from ) id 1fDY7k-00018q-DH for pgsql-hackers@postgresql.org; Tue, 01 May 2018 16:27:32 +0000 Received: from localhost (localhost [127.0.0.1]) by mail.postgrespro.ru (Postfix) with ESMTP id 9CD4D21C1F65 for ; Tue, 1 May 2018 19:27:25 +0300 (MSK) X-Virus-Scanned: Debian amavisd-new at postgrespro.ru X-Spam-Flag: NO X-Spam-Score: 0 X-Spam-Level: X-Spam-Status: No, score=x tagged_above=-99 required=4 WHITELISTED tests=[] autolearn=unavailable Received: from [192.168.0.138] (unknown [37.204.27.226]) (using TLSv1.2 with cipher ECDHE-RSA-AES256-GCM-SHA384 (256/256 bits)) (Client did not present a certificate) by mail.postgrespro.ru (Postfix) with ESMTPSA id 4742821C1ADB for ; Tue, 1 May 2018 19:27:25 +0300 (MSK) DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/simple; d=postgrespro.ru; s=mail; t=1525192045; bh=yaXb/XIZD6250GtQvwNglkiJp3aFDL17lnTJi3NS68k=; h=From:Subject:Date:To; b=EI7KgNo5zWwgJBZbwWC9azN63bbruWJvtKYq6yknMq3i0b3SYMtFwYZs/H+jGKu6n xsIYUJ0/bFLS0Df1gz8E6EQYFtKnA1E+smm8yCyB9K0vqGOa6Kl+fN2KL00+cS3lfg CeL6evx+t9D3q+yeSBiH9IzRhRHnhr23iinOSklw= From: Stas Kelvich Content-Type: multipart/mixed; boundary="Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55" Mime-Version: 1.0 (Mac OS X Mail 11.2 \(3445.5.20\)) Subject: Global snapshots Message-Id: <21BC916B-80A1-43BF-8650-3363CCDAE09C@postgrespro.ru> Date: Tue, 1 May 2018 19:27:22 +0300 To: PostgreSQL Hackers X-Mailer: Apple Mail (2.3445.5.20) List-Id: List-Help: List-Subscribe: List-Post: List-Owner: List-Archive: Precedence: bulk --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55 Content-Transfer-Encoding: quoted-printable Content-Type: text/plain; charset=us-ascii Global snapshots =3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D=3D Here proposed a set of patches that allow achieving proper snapshot = isolation semantics in the case of cross-node transactions. Provided = infrastructure to synchronize snapshots between different Postgres instances and a way to atomically commit such transaction with respect to other concurrent = global and local transactions. Such global transactions can be coordinated outside = of Postgres by using provided SQL functions or through postgres_fdw, which = make use of this functions on remote nodes transparently. Background ---------- Several years ago was proposed extensible transaction manager (XTM) = patch [1,2,3] which allowed extensions to hook and override = transaction-related functions like a xid assignment, snapshot acquiring, transaction tree commit/abort, etc. Also, two transaction management extensions were = created based on that API: pg_dtm, pg_tsdtm [4]. The first one, pg_dtm, was inspired by Postgres-XL and represents a = direct generalization of Postgres MVCC to the cross-node scenario: there is = standalone service (DTM) that maintains a xid counter and an array of running = backends on all nodes. Every node keeps an open connection to a DTM and asks for = xids and Postgres-style snapshots by a network call. While this approach is = reasonable for low transaction rates (which is common for OLAP-like load) it = quickly gasps for OLTP case. The second one, pg_tsdtm, based on Clock-SI paper [5] implements = distributed snapshot synchronization protocol without the necessity of central = service like DTM. It makes use of CSN-based visibility and requires two steps of communication during transaction commit. Since commit between nodes each = of which can abort transaction is usually done by 2PC protocol two steps of communication are anyway already needed. The current patch is a reimplementation of pg_tsdtm moved into core = directly without using any API. The decision to drop XTM API was based on two = thoughts. At first XTM API itself needed a huge pile of work to be done to became = an actual API instead of the set of hooks over current implementation of = MVCC with extensions responsible for handling random static variables like RecentGlobalXmin and friend, which other guts make use of. For the = second in all of our test pg_tsdtm was better and faster, but that wasn't obvious from = the beginning of work on XTM/DTM. So we decided that it is better to focus = on the good implementation of Clock-SI approach. Algorithm --------- Clock-SI is described in [5] and here I provide a small overview, which supposedly should be enough to catch the idea. Assume that each node = runs Commit Sequence Number (CSN) based visibility: database tracks one counter for = each transaction start (xid) and another counter for each transaction commit = (csn). In such setting, a snapshot is just a single number -- a copy of current = CSN at the moment when the snapshot was taken. Visibility rules are boiled down = to checking whether current tuple's CSN is less than our snapshot's csn. = Also it worth of mentioning that for the last 5 years there is an active = proposal to switch Postgres to CSN-based visibility [6]. Next, let's assume that CSN is current physical time on the node and = call it GlobalCSN. If the physical time on different nodes would be perfectly synchronized then such snapshot exported on one node can be used on = other nodes. But unfortunately physical time never perfectly sync and can drift, so = that fact should be taken into mind. Also, there is no easy notion of lock or = atomic operation in the distributed environment, so commit atomicity on = different nodes with respect to concurrent snapshot acquisition should be handled = somehow. Clock-SI addresses that in following way: 1) To achieve commit atomicity of different nodes intermediate step is introduced: at first running transaction is marked as InDoubt on all = nodes, and only after that each node commit it and stamps with a given = GlobalCSN. All readers who ran into tuples of an InDoubt transaction should wait = until it ends and recheck visibility. The same technique was used in [6] to = achieve atomicity of subtransactions tree commit. 2) When coordinator is marking transactions as InDoubt on other nodes it collects ProposedGlobalCSN from each participant which is local time = at that nodes. Next, it selects the maximal value of all ProposedGlobalCSN's = and commits the transaction on all nodes with that maximal GlobaCSN. Even = if that value is greater than current time on this node due to clock drift. = So the GlobalCSN for the given transaction will be the same on all nodes. 3) When local transaction imports foreign global snapshot with some = GlobalCSN and current time on this node is smaller then incoming GlobalCSN then = the transaction should wait until this GlobalCSN time comes on the local = clock. Rules 2) and 3) provide protection against time drift. Paper [5] proves = that this is enough to guarantee a proper Snapshot Isolation. Implementation -------------- Main design decision of this patch is trying to affect the performance = of local transaction as low as possible while providing a way to make global transactions. GUC variable track_global_snapshots enables/disables this = feature. Patch 0001-GlobalCSNLog introduces new SLRU instance that maps xids to GlobalCSN. GlobalCSNLog code is pretty straightforward and more or less = copied from SUBTRANS log which is also not persistent. Also, I kept an eye on = the corresponding part of Heikki's original patch for CSN's in [6] and = commit_ts.c. Patch 0002-Global-snapshots provides infrastructure to snapshot sync = functions and global commit functions. It consists of several parts which would be = enabled when GUC track_global_snapshots is on: * Each Postgres snapshot acquisition is accompanied by taking current = GlobalCSN under the same shared ProcArray lock. * Each transaction commit also writes current GlobalCSN into = GlobalCSNLog. To avoid writes to SLRU under exclusive ProcArray lock (which would be = the major hit on commit performance) trick with intermediate InDoubt state is = used: before calling ProcArrayEndTransaction() backend writes InDoubt state = in SLRU, then inside of ProcArrayEndTransaction() under a ProcArray lock = GlobalCSN is assigned, and after the lock is released assigned GlobalCSN value is = written to GlobalCSNLog SLRU. This approach ensures XIP-based snapshots and GlobalCSN-based are going to see the same subset of tuple versions = without putting too much extra contention on ProcArray lock. * XidInMVCCSnapshot can use both XIP-based and GlobalCSN-based snapshot. = If the current transaction is local one then at first XIP-based check is = performed, then if the tuple is visible don't do any further checks; if this xid = is in-progress we need to fetch GlobalCSN from SLRU and recheck it with GlobalCSN-based visibility rules, as it may be part of global InDoubt transaction. So, for local transactions XidInMVCCSnapshot() will fetch = SLRU entry only for in-progress transactions. This can be optimized = further: it is possible to store a flag in a snapshot which indicates whether there = were any active global transactions when this snapshot was taken. Then if there = were no global transactions during snapshot creation SLRU access in XidInMVCCSnapshot() can be skipped at all. Otherwise, if current = backend have the snapshot that was imported from foreign node then we use only GlobalCSN-based visibility rules (as we don't have any information = about how XIP array looked like when that GlobalCSN was taken). * Import/export routines provided for global snapshots. Export just = returns current snapshot's global_csn. Import, on the other hand, is more = complicated. Since imported GlobalCSN usually points in past we should hold = OldestXmin and ensure that tuple versions for given GlobalCSN don't pass through = OldestXmin boundary. To achieve that, mechanism like one in SnapshotTooOld is = created: on each snapshot creation current oldestXmin is written to a sparse ring = buffer which holds oldestXmin entries for a last global_snapshot_defer_time = seconds. GetOldestXmin() is delaying its results to the oldest entry in this = ring buffer. If we asked to import snapshot that is later then current_time = - global_snapshot_defer_time then we just error out with "global = snapshot too old" message. Otherwise, we have enough info to set proper xmin in our = proc entry to defuse GetOldestXmin(). * Following routines for commit provided: * pg_global_snaphot_prepare(gid) sets InDoubt state for given global tx = and return proposed GlobalCSN. * pg_global_snaphot_assign(gid, global_csn) assign given global_csn to = given global tx. Consequent COMMIT PREPARED will use that. Import/export and commit routines are made as SQL functions. IMO it's = better to transform them to IMPORT GLOBAL SNAPSHOT / EXPORT GLOBAL SNAPSHOT / = PREPARE GLOBAL / COMMIT GLOBAL PREPARED. But at this stage, I don't want to = clutter this patch with new grammar. Patch 0003-postgres_fdw-support-for-global-snapshots uses the previous infrastructure to achieve isolation in the generated transaction. Right = now it is a minimalistic implementation of 2PC like one in [7], but which = doesn't care about writing something about remote prepares in WAL on coordinator and = doesn't care about any recovery. The main usage is to test things in global = snapshot patch, as it easier to do with TAP tests over several = postgres_fdw-connected instances. Tests are included. Usage ----- The distributed transaction can be coordinated by an external = coordinator. In this case normal scenario would be following: 1) Start transaction on the first node, do some queries if needed, call pg_global_snaphot_export(). 2) On the other node open transaction and call pg_global_snaphot_import(global_csn). global_csn is from previous = export. ... do some useful work ... 3) Issue PREPARE for all participant. 4) Call pg_global_snaphot_prepare(gid) on all participant and store = returned global_csn's; select maximal over of returned global_csn's. 5) Call pg_global_snaphot_assign(gid, max_global_csn) on all = participant. 6) Issue COMMIT PREPARED for all participant. As it was said earlier steps 4) and 5) can be melded into 3) and 5) respectively, but let's go for grammar changes after and if there will = be agreement on the overall concept. postgres_fdw in 003-GlobalSnapshotsForPostgresFdw does the same thing, = but transparently to the user. Tests ----- Each XidInMVCCSnapshot() in this patch is coated with assert that checks = that same things are visible under XIP-based check and GlobalCSN one. Patch 003-GlobalSnapshotsForPostgresFdw includes numerous variants of = banking test. It spin-offs several Postgres instances and several concurrent = pgbenches which simulate cross-node bank account transfers, while test constantly = checks that total balance is correct. Also, there are tests that imported = snapshot is going to see the same checksum of data as it was during import. I think this feature also deserves separate test module that will wobble = the clock time, but that isn't done yet. Attributions ------------ Original XTM, pg_dtm, pg_tsdtm were written by Konstantin Knizhnik, = Constantin Pan and me. This version is mostly on me, with some inputs by Konstantin Knizhnik and Arseny Sher. [1] = https://www.postgresql.org/message-id/flat/20160223164335.GA11285%40momjia= n.us (originally thread was about fdw-based sharding, but later got = "hijacked" by xtm) [2] = https://www.postgresql.org/message-id/flat/F2766B97-555D-424F-B29F-E0CA0F6= D1D74%40postgrespro.ru [3] = https://www.postgresql.org/message-id/flat/56E3CAAE.6060407%40postgrespro.= ru [4] https://wiki.postgresql.org/wiki/DTM [5] https://dl.acm.org/citation.cfm?id=3D2553434 [6] = https://www.postgresql.org/message-id/flat/CA%2BCSw_tEpJ%3Dmd1zgxPkjH6CWDn= TDft4gBi%3D%2BP9SnoC%2BWy3pKdA%40mail.gmail.com [7] = https://www.postgresql.org/message-id/flat/CAFjFpRfPoDAzf3x-fs86nDwJ4FAwn2= cZ%2BxdmbdDPSChU-kt7%2BQ%40mail.gmail.com --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55 Content-Disposition: attachment; filename=0001-GlobalCSNLog-SLRU.patch Content-Type: application/octet-stream; x-unix-mode=0644; name="0001-GlobalCSNLog-SLRU.patch" Content-Transfer-Encoding: quoted-printable =46rom=20627cb29efebc4d14cdbbf82d82185b8311856fc0=20Mon=20Sep=2017=20= 00:00:00=202001=0AFrom:=20Stas=20Kelvich=20=0ADate:=20= Wed,=2025=20Apr=202018=2016:05:46=20+0300=0ASubject:=20[PATCH=201/3]=20= GlobalCSNLog=20SLRU=0A=0A---=0A=20src/backend/access/transam/Makefile=20=20= =20=20=20=20=20=20=20|=20=20=203=20+-=0A=20= src/backend/access/transam/global_csn_log.c=20|=20439=20= ++++++++++++++++++++++++++++=0A=20src/backend/access/transam/twophase.c=20= =20=20=20=20=20=20|=20=20=201=20+=0A=20= src/backend/access/transam/varsup.c=20=20=20=20=20=20=20=20=20|=20=20=20= 2=20+=0A=20src/backend/access/transam/xlog.c=20=20=20=20=20=20=20=20=20=20= =20|=20=2012=20+=0A=20src/backend/storage/ipc/ipci.c=20=20=20=20=20=20=20= =20=20=20=20=20=20=20|=20=20=203=20+=0A=20= src/backend/storage/ipc/procarray.c=20=20=20=20=20=20=20=20=20|=20=20=20= 3=20+=0A=20src/backend/storage/lmgr/lwlocknames.txt=20=20=20=20|=20=20=20= 1=20+=0A=20src/backend/utils/misc/guc.c=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20=20|=20=20=209=20+=0A=20src/backend/utils/probes.d=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20=20=20=20|=20=20=202=20+=0A=20= src/bin/initdb/initdb.c=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20=20|=20=20=203=20+-=0A=20src/include/access/global_csn_log.h=20=20= =20=20=20=20=20=20=20|=20=2030=20++=0A=20src/include/storage/lwlock.h=20=20= =20=20=20=20=20=20=20=20=20=20=20=20=20=20|=20=20=201=20+=0A=20= src/include/utils/snapshot.h=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20|=20=20=203=20+=0A=2014=20files=20changed,=20510=20insertions(+),=202=20= deletions(-)=0A=20create=20mode=20100644=20= src/backend/access/transam/global_csn_log.c=0A=20create=20mode=20100644=20= src/include/access/global_csn_log.h=0A=0Adiff=20--git=20= a/src/backend/access/transam/Makefile=20= b/src/backend/access/transam/Makefile=0Aindex=2016fbe47269..03aa360ea3=20= 100644=0A---=20a/src/backend/access/transam/Makefile=0A+++=20= b/src/backend/access/transam/Makefile=0A@@=20-12,7=20+12,8=20@@=20subdir=20= =3D=20src/backend/access/transam=0A=20top_builddir=20=3D=20../../../..=0A= =20include=20$(top_builddir)/src/Makefile.global=0A=20=0A-OBJS=20=3D=20= clog.o=20commit_ts.o=20generic_xlog.o=20multixact.o=20parallel.o=20= rmgr.o=20slru.o=20\=0A+OBJS=20=3D=20clog.o=20commit_ts.o=20= global_csn_log.o=20generic_xlog.o=20\=0A+=09multixact.o=20parallel.o=20= rmgr.o=20slru.o=20\=0A=20=09subtrans.o=20timeline.o=20transam.o=20= twophase.o=20twophase_rmgr.o=20varsup.o=20\=0A=20=09xact.o=20xlog.o=20= xlogarchive.o=20xlogfuncs.o=20\=0A=20=09xloginsert.o=20xlogreader.o=20= xlogutils.o=0Adiff=20--git=20= a/src/backend/access/transam/global_csn_log.c=20= b/src/backend/access/transam/global_csn_log.c=0Anew=20file=20mode=20= 100644=0Aindex=200000000000..d9d66528e4=0A---=20/dev/null=0A+++=20= b/src/backend/access/transam/global_csn_log.c=0A@@=20-0,0=20+1,439=20@@=0A= +/*-----------------------------------------------------------------------= ------=0A+=20*=0A+=20*=20global_csn_log.c=0A+=20*=09=09Track=20global=20= commit=20sequence=20numbers=20of=20finished=20transactions=0A+=20*=0A+=20= *=20Implementation=20of=20cross-node=20transaction=20isolation=20relies=20= on=20commit=20sequence=0A+=20*=20number=20(CSN)=20based=20visibility=20= rules.=20=20This=20module=20provides=20SLRU=20to=20store=0A+=20*=20CSN=20= for=20each=20transaction.=20=20This=20mapping=20need=20to=20be=20kept=20= only=20for=20xid's=0A+=20*=20greater=20then=20oldestXid,=20but=20that=20= can=20require=20arbitrary=20large=20amounts=20of=0A+=20*=20memory=20in=20= case=20of=20long-lived=20transactions.=20=20Because=20of=20same=20= lifetime=20and=0A+=20*=20persistancy=20requirements=20this=20module=20is=20= quite=20similar=20to=20subtrans.c=0A+=20*=0A+=20*=20Portions=20Copyright=20= (c)=201996-2018,=20PostgreSQL=20Global=20Development=20Group=0A+=20*=20= Portions=20Copyright=20(c)=201994,=20Regents=20of=20the=20University=20= of=20California=0A+=20*=0A+=20*=20= src/backend/access/transam/global_csn_log.c=0A+=20*=0A+=20= *-------------------------------------------------------------------------= ----=0A+=20*/=0A+#include=20"postgres.h"=0A+=0A+#include=20= "access/global_csn_log.h"=0A+#include=20"access/slru.h"=0A+#include=20= "access/subtrans.h"=0A+#include=20"access/transam.h"=0A+#include=20= "miscadmin.h"=0A+#include=20"pg_trace.h"=0A+#include=20"utils/snapmgr.h"=0A= +=0A+bool=20track_global_snapshots;=0A+=0A+/*=0A+=20*=20Defines=20for=20= GlobalCSNLog=20page=20sizes.=20=20A=20page=20is=20the=20same=20BLCKSZ=20= as=20is=20used=0A+=20*=20everywhere=20else=20in=20Postgres.=0A+=20*=0A+=20= *=20Note:=20because=20TransactionIds=20are=2032=20bits=20and=20wrap=20= around=20at=200xFFFFFFFF,=0A+=20*=20GlobalCSNLog=20page=20numbering=20= also=20wraps=20around=20at=0A+=20*=20= 0xFFFFFFFF/GLOBAL_CSN_LOG_XACTS_PER_PAGE,=20and=20GlobalCSNLog=20segment=20= numbering=20at=0A+=20*=20= 0xFFFFFFFF/CLOG_XACTS_PER_PAGE/SLRU_PAGES_PER_SEGMENT.=20=20We=20need=20= take=20no=0A+=20*=20explicit=20notice=20of=20that=20fact=20in=20this=20= module,=20except=20when=20comparing=20segment=0A+=20*=20and=20page=20= numbers=20in=20TruncateGlobalCSNLog=20(see=20GlobalCSNLogPagePrecedes).=0A= +=20*/=0A+=0A+/*=20We=20store=20the=20commit=20GlobalCSN=20for=20each=20= xid=20*/=0A+#define=20GCSNLOG_XACTS_PER_PAGE=20(BLCKSZ=20/=20= sizeof(GlobalCSN))=0A+=0A+#define=20TransactionIdToPage(xid)=09((xid)=20= /=20(TransactionId)=20GCSNLOG_XACTS_PER_PAGE)=0A+#define=20= TransactionIdToPgIndex(xid)=20((xid)=20%=20(TransactionId)=20= GCSNLOG_XACTS_PER_PAGE)=0A+=0A+/*=0A+=20*=20Link=20to=20shared-memory=20= data=20structures=20for=20CLOG=20control=0A+=20*/=0A+static=20= SlruCtlData=20GlobalCSNLogCtlData;=0A+#define=20GlobalCsnlogCtl=20= (&GlobalCSNLogCtlData)=0A+=0A+static=20int=09ZeroGlobalCSNLogPage(int=20= pageno);=0A+static=20bool=20GlobalCSNLogPagePrecedes(int=20page1,=20int=20= page2);=0A+static=20void=20GlobalCSNLogSetPageStatus(TransactionId=20= xid,=20int=20nsubxids,=0A+=09=09=09=09=09=09=09=09=09=20=20TransactionId=20= *subxids,=0A+=09=09=09=09=09=09=09=09=09=20=20GlobalCSN=20csn,=20int=20= pageno);=0A+static=20void=20GlobalCSNLogSetCSNInSlot(TransactionId=20= xid,=20GlobalCSN=20csn,=0A+=09=09=09=09=09=09=09=09=09=20=20int=20= slotno);=0A+=0A+/*=0A+=20*=20GlobalCSNLogSetCSN=0A+=20*=0A+=20*=20Record=20= GlobalCSN=20of=20transaction=20and=20its=20subtransaction=20tree.=0A+=20= *=0A+=20*=20xid=20is=20a=20single=20xid=20to=20set=20status=20for.=20= This=20will=20typically=20be=20the=20top=20level=0A+=20*=20transactionid=20= for=20a=20top=20level=20commit=20or=20abort.=20It=20can=20also=20be=20a=0A= +=20*=20subtransaction=20when=20we=20record=20transaction=20aborts.=0A+=20= *=0A+=20*=20subxids=20is=20an=20array=20of=20xids=20of=20length=20= nsubxids,=20representing=20subtransactions=0A+=20*=20in=20the=20tree=20= of=20xid.=20In=20various=20cases=20nsubxids=20may=20be=20zero.=0A+=20*=0A= +=20*=20csn=20is=20the=20commit=20sequence=20number=20of=20the=20= transaction.=20It=20should=20be=0A+=20*=20AbortedGlobalCSN=20for=20abort=20= cases.=0A+=20*/=0A+void=0A+GlobalCSNLogSetCSN(TransactionId=20xid,=20int=20= nsubxids,=0A+=09=09=09=09=09=20TransactionId=20*subxids,=20GlobalCSN=20= csn)=0A+{=0A+=09int=09=09=09pageno;=0A+=09int=09=09=09i=20=3D=200;=0A+=09= int=09=09=09offset=20=3D=200;=0A+=0A+=09/*=20Callers=20of=20= GlobalCSNLogSetCSN()=20must=20check=20GUC=20params=20*/=0A+=09= Assert(track_global_snapshots);=0A+=0A+=09= Assert(TransactionIdIsValid(xid));=0A+=0A+=09pageno=20=3D=20= TransactionIdToPage(xid);=09=09/*=20get=20page=20of=20parent=20*/=0A+=09= for=20(;;)=0A+=09{=0A+=09=09int=09=09=09num_on_page=20=3D=200;=0A+=0A+=09= =09while=20(i=20<=20nsubxids=20&&=20TransactionIdToPage(subxids[i])=20=3D=3D= =20pageno)=0A+=09=09{=0A+=09=09=09num_on_page++;=0A+=09=09=09i++;=0A+=09=09= }=0A+=0A+=09=09GlobalCSNLogSetPageStatus(xid,=0A+=09=09=09=09=09=09=09= num_on_page,=20subxids=20+=20offset,=0A+=09=09=09=09=09=09=09csn,=20= pageno);=0A+=09=09if=20(i=20>=3D=20nsubxids)=0A+=09=09=09break;=0A+=0A+=09= =09offset=20=3D=20i;=0A+=09=09pageno=20=3D=20= TransactionIdToPage(subxids[offset]);=0A+=09=09xid=20=3D=20= InvalidTransactionId;=0A+=09}=0A+}=0A+=0A+/*=0A+=20*=20Record=20the=20= final=20state=20of=20transaction=20entries=20in=20the=20csn=20log=20for=0A= +=20*=20all=20entries=20on=20a=20single=20page.=20=20Atomic=20only=20on=20= this=20page.=0A+=20*=0A+=20*=20Otherwise=20API=20is=20same=20as=20= TransactionIdSetTreeStatus()=0A+=20*/=0A+static=20void=0A= +GlobalCSNLogSetPageStatus(TransactionId=20xid,=20int=20nsubxids,=0A+=09=09= =09=09=09=09=20=20=20TransactionId=20*subxids,=0A+=09=09=09=09=09=09=20=20= =20GlobalCSN=20csn,=20int=20pageno)=0A+{=0A+=09int=09=09=09slotno;=0A+=09= int=09=09=09i;=0A+=0A+=09LWLockAcquire(GlobalCSNLogControlLock,=20= LW_EXCLUSIVE);=0A+=0A+=09slotno=20=3D=20= SimpleLruReadPage(GlobalCsnlogCtl,=20pageno,=20true,=20xid);=0A+=0A+=09= /*=20Subtransactions=20first,=20if=20needed=20...=20*/=0A+=09for=20(i=20= =3D=200;=20i=20<=20nsubxids;=20i++)=0A+=09{=0A+=09=09= Assert(GlobalCsnlogCtl->shared->page_number[slotno]=20=3D=3D=20= TransactionIdToPage(subxids[i]));=0A+=09=09= GlobalCSNLogSetCSNInSlot(subxids[i],=09csn,=20slotno);=0A+=09}=0A+=0A+=09= /*=20...=20then=20the=20main=20transaction=20*/=0A+=09if=20= (TransactionIdIsValid(xid))=0A+=09=09GlobalCSNLogSetCSNInSlot(xid,=20= csn,=20slotno);=0A+=0A+=09GlobalCsnlogCtl->shared->page_dirty[slotno]=20= =3D=20true;=0A+=0A+=09LWLockRelease(GlobalCSNLogControlLock);=0A+}=0A+=0A= +/*=0A+=20*=20Sets=20the=20commit=20status=20of=20a=20single=20= transaction.=0A+=20*/=0A+static=20void=0A= +GlobalCSNLogSetCSNInSlot(TransactionId=20xid,=20GlobalCSN=20csn,=20int=20= slotno)=0A+{=0A+=09int=09=09=09entryno=20=3D=20= TransactionIdToPgIndex(xid);=0A+=09GlobalCSN=20*ptr;=0A+=0A+=09= Assert(LWLockHeldByMe(GlobalCSNLogControlLock));=0A+=0A+=09ptr=20=3D=20= (GlobalCSN=20*)=20(GlobalCsnlogCtl->shared->page_buffer[slotno]=20+=20= entryno=20*=20sizeof(XLogRecPtr));=0A+=0A+=09*ptr=20=3D=20csn;=0A+}=0A+=0A= +/*=0A+=20*=20Interrogate=20the=20state=20of=20a=20transaction=20in=20= the=20log.=0A+=20*=0A+=20*=20NB:=20this=20is=20a=20low-level=20routine=20= and=20is=20NOT=20the=20preferred=20entry=20point=0A+=20*=20for=20most=20= uses;=20TransactionIdGetGlobalCSN()=20in=20global_snapshot.c=20is=20the=0A= +=20*=20intended=20caller.=0A+=20*/=0A+GlobalCSN=0A= +GlobalCSNLogGetCSN(TransactionId=20xid)=0A+{=0A+=09int=09=09=09pageno=20= =3D=20TransactionIdToPage(xid);=0A+=09int=09=09=09entryno=20=3D=20= TransactionIdToPgIndex(xid);=0A+=09int=09=09=09slotno;=0A+=09GlobalCSN=20= *ptr;=0A+=09GlobalCSN=09global_csn;=0A+=0A+=09/*=20Callers=20of=20= GlobalCSNLogGetCSN()=20must=20check=20GUC=20params=20*/=0A+=09= Assert(track_global_snapshots);=0A+=0A+=09/*=20Can't=20ask=20about=20= stuff=20that=20might=20not=20be=20around=20anymore=20*/=0A+=09= Assert(TransactionIdFollowsOrEquals(xid,=20TransactionXmin));=0A+=0A+=09= /*=20lock=20is=20acquired=20by=20SimpleLruReadPage_ReadOnly=20*/=0A+=0A+=09= slotno=20=3D=20SimpleLruReadPage_ReadOnly(GlobalCsnlogCtl,=20pageno,=20= xid);=0A+=09ptr=20=3D=20(GlobalCSN=20*)=20= (GlobalCsnlogCtl->shared->page_buffer[slotno]=20+=20entryno=20*=20= sizeof(XLogRecPtr));=0A+=09global_csn=20=3D=20*ptr;=0A+=0A+=09= LWLockRelease(GlobalCSNLogControlLock);=0A+=0A+=09return=20global_csn;=0A= +}=0A+=0A+/*=0A+=20*=20Number=20of=20shared=20GlobalCSNLog=20buffers.=0A= +=20*/=0A+static=20Size=0A+GlobalCSNLogShmemBuffers(void)=0A+{=0A+=09= return=20Min(32,=20Max(4,=20NBuffers=20/=20512));=0A+}=0A+=0A+/*=0A+=20*=20= Reserve=20shared=20memory=20for=20GlobalCsnlogCtl.=0A+=20*/=0A+Size=0A= +GlobalCSNLogShmemSize(void)=0A+{=0A+=09if=20(!track_global_snapshots)=0A= +=09=09return=200;=0A+=0A+=09return=20= SimpleLruShmemSize(GlobalCSNLogShmemBuffers(),=200);=0A+}=0A+=0A+/*=0A+=20= *=20Initialization=20of=20shared=20memory=20for=20GlobalCSNLog.=0A+=20*/=0A= +void=0A+GlobalCSNLogShmemInit(void)=0A+{=0A+=09if=20= (!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09= GlobalCsnlogCtl->PagePrecedes=20=3D=20GlobalCSNLogPagePrecedes;=0A+=09= SimpleLruInit(GlobalCsnlogCtl,=20"GlobalCSNLog=20Ctl",=20= GlobalCSNLogShmemBuffers(),=200,=0A+=09=09=09=09=20=20= GlobalCSNLogControlLock,=20"pg_global_csn",=20= LWTRANCHE_GLOBAL_CSN_LOG_BUFFERS);=0A+}=0A+=0A+/*=0A+=20*=20This=20func=20= must=20be=20called=20ONCE=20on=20system=20install.=20=20It=20creates=20= the=20initial=0A+=20*=20GlobalCSNLog=20segment.=20=20The=20pg_global_csn=20= directory=20is=20assumed=20to=20have=20been=0A+=20*=20created=20by=20= initdb,=20and=20GlobalCSNLogShmemInit=20must=20have=20been=20called=20= already.=0A+=20*/=0A+void=0A+BootStrapGlobalCSNLog(void)=0A+{=0A+=09int=09= =09=09slotno;=0A+=0A+=09if=20(!track_global_snapshots)=0A+=09=09return;=0A= +=0A+=09LWLockAcquire(GlobalCSNLogControlLock,=20LW_EXCLUSIVE);=0A+=0A+=09= /*=20Create=20and=20zero=20the=20first=20page=20of=20the=20commit=20log=20= */=0A+=09slotno=20=3D=20ZeroGlobalCSNLogPage(0);=0A+=0A+=09/*=20Make=20= sure=20it's=20written=20out=20*/=0A+=09= SimpleLruWritePage(GlobalCsnlogCtl,=20slotno);=0A+=09= Assert(!GlobalCsnlogCtl->shared->page_dirty[slotno]);=0A+=0A+=09= LWLockRelease(GlobalCSNLogControlLock);=0A+}=0A+=0A+/*=0A+=20*=20= Initialize=20(or=20reinitialize)=20a=20page=20of=20GlobalCSNLog=20to=20= zeroes.=0A+=20*=0A+=20*=20The=20page=20is=20not=20actually=20written,=20= just=20set=20up=20in=20shared=20memory.=0A+=20*=20The=20slot=20number=20= of=20the=20new=20page=20is=20returned.=0A+=20*=0A+=20*=20Control=20lock=20= must=20be=20held=20at=20entry,=20and=20will=20be=20held=20at=20exit.=0A+=20= */=0A+static=20int=0A+ZeroGlobalCSNLogPage(int=20pageno)=0A+{=0A+=09= Assert(LWLockHeldByMe(GlobalCSNLogControlLock));=0A+=09return=20= SimpleLruZeroPage(GlobalCsnlogCtl,=20pageno);=0A+}=0A+=0A+/*=0A+=20*=20= This=20must=20be=20called=20ONCE=20during=20postmaster=20or=20= standalone-backend=20startup,=0A+=20*=20after=20StartupXLOG=20has=20= initialized=20ShmemVariableCache->nextXid.=0A+=20*=0A+=20*=20= oldestActiveXID=20is=20the=20oldest=20XID=20of=20any=20prepared=20= transaction,=20or=20nextXid=0A+=20*=20if=20there=20are=20none.=0A+=20*/=0A= +void=0A+StartupGlobalCSNLog(TransactionId=20oldestActiveXID)=0A+{=0A+=09= int=09=09=09startPage;=0A+=09int=09=09=09endPage;=0A+=0A+=09if=20= (!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09/*=0A+=09=20*=20= Since=20we=20don't=20expect=20pg_global_csn=20to=20be=20valid=20across=20= crashes,=20we=0A+=09=20*=20initialize=20the=20currently-active=20page(s)=20= to=20zeroes=20during=20startup.=0A+=09=20*=20Whenever=20we=20advance=20= into=20a=20new=20page,=20ExtendGlobalCSNLog=20will=20likewise=0A+=09=20*=20= zero=20the=20new=20page=20without=20regard=20to=20whatever=20was=20= previously=20on=20disk.=0A+=09=20*/=0A+=09= LWLockAcquire(GlobalCSNLogControlLock,=20LW_EXCLUSIVE);=0A+=0A+=09= startPage=20=3D=20TransactionIdToPage(oldestActiveXID);=0A+=09endPage=20= =3D=20TransactionIdToPage(ShmemVariableCache->nextXid);=0A+=0A+=09while=20= (startPage=20!=3D=20endPage)=0A+=09{=0A+=09=09(void)=20= ZeroGlobalCSNLogPage(startPage);=0A+=09=09startPage++;=0A+=09=09/*=20= must=20account=20for=20wraparound=20*/=0A+=09=09if=20(startPage=20>=20= TransactionIdToPage(MaxTransactionId))=0A+=09=09=09startPage=20=3D=200;=0A= +=09}=0A+=09(void)=20ZeroGlobalCSNLogPage(startPage);=0A+=0A+=09= LWLockRelease(GlobalCSNLogControlLock);=0A+}=0A+=0A+/*=0A+=20*=20This=20= must=20be=20called=20ONCE=20during=20postmaster=20or=20= standalone-backend=20shutdown=0A+=20*/=0A+void=0A= +ShutdownGlobalCSNLog(void)=0A+{=0A+=09if=20(!track_global_snapshots)=0A= +=09=09return;=0A+=0A+=09/*=0A+=09=20*=20Flush=20dirty=20GlobalCSNLog=20= pages=20to=20disk.=0A+=09=20*=0A+=09=20*=20This=20is=20not=20actually=20= necessary=20from=20a=20correctness=20point=20of=20view.=20We=20do=0A+=09=20= *=20it=20merely=20as=20a=20debugging=20aid.=0A+=09=20*/=0A+=09= TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_START(false);=0A+=09= SimpleLruFlush(GlobalCsnlogCtl,=20false);=0A+=09= TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_DONE(false);=0A+}=0A+=0A+/*=0A+=20= *=20Perform=20a=20checkpoint=20---=20either=20during=20shutdown,=20or=20= on-the-fly=0A+=20*/=0A+void=0A+CheckPointGlobalCSNLog(void)=0A+{=0A+=09= if=20(!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09/*=0A+=09=20*=20= Flush=20dirty=20GlobalCSNLog=20pages=20to=20disk.=0A+=09=20*=0A+=09=20*=20= This=20is=20not=20actually=20necessary=20from=20a=20correctness=20point=20= of=20view.=20We=20do=0A+=09=20*=20it=20merely=20to=20improve=20the=20= odds=20that=20writing=20of=20dirty=20pages=20is=20done=20by=0A+=09=20*=20= the=20checkpoint=20process=20and=20not=20by=20backends.=0A+=09=20*/=0A+=09= TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_START(true);=0A+=09= SimpleLruFlush(GlobalCsnlogCtl,=20true);=0A+=09= TRACE_POSTGRESQL_GLOBALCSNLOG_CHECKPOINT_DONE(true);=0A+}=0A+=0A+/*=0A+=20= *=20Make=20sure=20that=20GlobalCSNLog=20has=20room=20for=20a=20= newly-allocated=20XID.=0A+=20*=0A+=20*=20NB:=20this=20is=20called=20= while=20holding=20XidGenLock.=20=20We=20want=20it=20to=20be=20very=20= fast=0A+=20*=20most=20of=20the=20time;=20even=20when=20it's=20not=20so=20= fast,=20no=20actual=20I/O=20need=20happen=0A+=20*=20unless=20we're=20= forced=20to=20write=20out=20a=20dirty=20clog=20or=20xlog=20page=20to=20= make=20room=0A+=20*=20in=20shared=20memory.=0A+=20*/=0A+void=0A= +ExtendGlobalCSNLog(TransactionId=20newestXact)=0A+{=0A+=09int=09=09=09= pageno;=0A+=0A+=09if=20(!track_global_snapshots)=0A+=09=09return;=0A+=0A= +=09/*=0A+=09=20*=20No=20work=20except=20at=20first=20XID=20of=20a=20= page.=20=20But=20beware:=20just=20after=0A+=09=20*=20wraparound,=20the=20= first=20XID=20of=20page=20zero=20is=20FirstNormalTransactionId.=0A+=09=20= */=0A+=09if=20(TransactionIdToPgIndex(newestXact)=20!=3D=200=20&&=0A+=09=09= !TransactionIdEquals(newestXact,=20FirstNormalTransactionId))=0A+=09=09= return;=0A+=0A+=09pageno=20=3D=20TransactionIdToPage(newestXact);=0A+=0A= +=09LWLockAcquire(GlobalCSNLogControlLock,=20LW_EXCLUSIVE);=0A+=0A+=09/*=20= Zero=20the=20page=20and=20make=20an=20XLOG=20entry=20about=20it=20*/=0A+=09= ZeroGlobalCSNLogPage(pageno);=0A+=0A+=09= LWLockRelease(GlobalCSNLogControlLock);=0A+}=0A+=0A+/*=0A+=20*=20Remove=20= all=20GlobalCSNLog=20segments=20before=20the=20one=20holding=20the=20= passed=0A+=20*=20transaction=20ID.=0A+=20*=0A+=20*=20This=20is=20= normally=20called=20during=20checkpoint,=20with=20oldestXact=20being=20= the=0A+=20*=20oldest=20TransactionXmin=20of=20any=20running=20= transaction.=0A+=20*/=0A+void=0A+TruncateGlobalCSNLog(TransactionId=20= oldestXact)=0A+{=0A+=09int=09=09=09cutoffPage;=0A+=0A+=09if=20= (!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09/*=0A+=09=20*=20The=20= cutoff=20point=20is=20the=20start=20of=20the=20segment=20containing=20= oldestXact.=20We=0A+=09=20*=20pass=20the=20*page*=20containing=20= oldestXact=20to=20SimpleLruTruncate.=20We=20step=0A+=09=20*=20back=20one=20= transaction=20to=20avoid=20passing=20a=20cutoff=20page=20that=20hasn't=20= been=0A+=09=20*=20created=20yet=20in=20the=20rare=20case=20that=20= oldestXact=20would=20be=20the=20first=20item=20on=0A+=09=20*=20a=20page=20= and=20oldestXact=20=3D=3D=20next=20XID.=20=20In=20that=20case,=20if=20we=20= didn't=20subtract=0A+=09=20*=20one,=20we'd=20trigger=20= SimpleLruTruncate's=20wraparound=20detection.=0A+=09=20*/=0A+=09= TransactionIdRetreat(oldestXact);=0A+=09cutoffPage=20=3D=20= TransactionIdToPage(oldestXact);=0A+=0A+=09= SimpleLruTruncate(GlobalCsnlogCtl,=20cutoffPage);=0A+}=0A+=0A+/*=0A+=20*=20= Decide=20which=20of=20two=20GlobalCSNLog=20page=20numbers=20is=20"older"=20= for=20truncation=0A+=20*=20purposes.=0A+=20*=0A+=20*=20We=20need=20to=20= use=20comparison=20of=20TransactionIds=20here=20in=20order=20to=20do=20= the=20right=0A+=20*=20thing=20with=20wraparound=20XID=20arithmetic.=20=20= However,=20if=20we=20are=20asked=20about=0A+=20*=20page=20number=20zero,=20= we=20don't=20want=20to=20hand=20InvalidTransactionId=20to=0A+=20*=20= TransactionIdPrecedes:=20it'll=20get=20weird=20about=20permanent=20xact=20= IDs.=20=20So,=0A+=20*=20offset=20both=20xids=20by=20= FirstNormalTransactionId=20to=20avoid=20that.=0A+=20*/=0A+static=20bool=0A= +GlobalCSNLogPagePrecedes(int=20page1,=20int=20page2)=0A+{=0A+=09= TransactionId=20xid1;=0A+=09TransactionId=20xid2;=0A+=0A+=09xid1=20=3D=20= ((TransactionId)=20page1)=20*=20GCSNLOG_XACTS_PER_PAGE;=0A+=09xid1=20+=3D=20= FirstNormalTransactionId;=0A+=09xid2=20=3D=20((TransactionId)=20page2)=20= *=20GCSNLOG_XACTS_PER_PAGE;=0A+=09xid2=20+=3D=20= FirstNormalTransactionId;=0A+=0A+=09return=20TransactionIdPrecedes(xid1,=20= xid2);=0A+}=0Adiff=20--git=20a/src/backend/access/transam/twophase.c=20= b/src/backend/access/transam/twophase.c=0Aindex=20cdd8156ce4..a144e2d6a9=20= 100644=0A---=20a/src/backend/access/transam/twophase.c=0A+++=20= b/src/backend/access/transam/twophase.c=0A@@=20-77,6=20+77,7=20@@=0A=20= #include=20=0A=20=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_csn_log.h"=0A=20#include=20= "access/htup_details.h"=0A=20#include=20"access/subtrans.h"=0A=20= #include=20"access/transam.h"=0Adiff=20--git=20= a/src/backend/access/transam/varsup.c=20= b/src/backend/access/transam/varsup.c=0Aindex=20394843f7e9..4035b90d5e=20= 100644=0A---=20a/src/backend/access/transam/varsup.c=0A+++=20= b/src/backend/access/transam/varsup.c=0A@@=20-15,6=20+15,7=20@@=0A=20=0A=20= #include=20"access/clog.h"=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_csn_log.h"=0A=20#include=20= "access/subtrans.h"=0A=20#include=20"access/transam.h"=0A=20#include=20= "access/xact.h"=0A@@=20-169,6=20+170,7=20@@=20GetNewTransactionId(bool=20= isSubXact)=0A=20=09=20*=20Extend=20pg_subtrans=20and=20pg_commit_ts=20= too.=0A=20=09=20*/=0A=20=09ExtendCLOG(xid);=0A+=09= ExtendGlobalCSNLog(xid);=0A=20=09ExtendCommitTs(xid);=0A=20=09= ExtendSUBTRANS(xid);=0A=20=0Adiff=20--git=20= a/src/backend/access/transam/xlog.c=20= b/src/backend/access/transam/xlog.c=0Aindex=20c0923d97f2..749b0c1ce2=20= 100644=0A---=20a/src/backend/access/transam/xlog.c=0A+++=20= b/src/backend/access/transam/xlog.c=0A@@=20-24,6=20+24,7=20@@=0A=20=0A=20= #include=20"access/clog.h"=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_csn_log.h"=0A=20#include=20= "access/multixact.h"=0A=20#include=20"access/rewriteheap.h"=0A=20= #include=20"access/subtrans.h"=0A@@=20-5189,6=20+5190,7=20@@=20= BootStrapXLOG(void)=0A=20=0A=20=09/*=20Bootstrap=20the=20commit=20log,=20= too=20*/=0A=20=09BootStrapCLOG();=0A+=09BootStrapGlobalCSNLog();=0A=20=09= BootStrapCommitTs();=0A=20=09BootStrapSUBTRANS();=0A=20=09= BootStrapMultiXact();=0A@@=20-6963,6=20+6965,7=20@@=20StartupXLOG(void)=0A= =20=09=09=09=20*=20maintained=20during=20recovery=20and=20need=20not=20= be=20started=20yet.=0A=20=09=09=09=20*/=0A=20=09=09=09StartupCLOG();=0A+=09= =09=09StartupGlobalCSNLog(oldestActiveXID);=0A=20=09=09=09= StartupSUBTRANS(oldestActiveXID);=0A=20=0A=20=09=09=09/*=0A@@=20-7748,6=20= +7751,7=20@@=20StartupXLOG(void)=0A=20=09if=20(standbyState=20=3D=3D=20= STANDBY_DISABLED)=0A=20=09{=0A=20=09=09StartupCLOG();=0A+=09=09= StartupGlobalCSNLog(oldestActiveXID);=0A=20=09=09= StartupSUBTRANS(oldestActiveXID);=0A=20=09}=0A=20=0A@@=20-8400,6=20= +8404,7=20@@=20ShutdownXLOG(int=20code,=20Datum=20arg)=0A=20=09=09= CreateCheckPoint(CHECKPOINT_IS_SHUTDOWN=20|=20CHECKPOINT_IMMEDIATE);=0A=20= =09}=0A=20=09ShutdownCLOG();=0A+=09ShutdownGlobalCSNLog();=0A=20=09= ShutdownCommitTs();=0A=20=09ShutdownSUBTRANS();=0A=20=09= ShutdownMultiXact();=0A@@=20-8977,7=20+8982,10=20@@=20= CreateCheckPoint(int=20flags)=0A=20=09=20*=20StartupSUBTRANS=20hasn't=20= been=20called=20yet.=0A=20=09=20*/=0A=20=09if=20(!RecoveryInProgress())=0A= +=09{=0A=20=09=09TruncateSUBTRANS(GetOldestXmin(NULL,=20= PROCARRAY_FLAGS_DEFAULT));=0A+=09=09= TruncateGlobalCSNLog(GetOldestXmin(NULL,=20PROCARRAY_FLAGS_DEFAULT));=0A= +=09}=0A=20=0A=20=09/*=20Real=20work=20is=20done,=20but=20log=20and=20= update=20stats=20before=20releasing=20lock.=20*/=0A=20=09= LogCheckpointEnd(false);=0A@@=20-9053,6=20+9061,7=20@@=20static=20void=0A= =20CheckPointGuts(XLogRecPtr=20checkPointRedo,=20int=20flags)=0A=20{=0A=20= =09CheckPointCLOG();=0A+=09CheckPointGlobalCSNLog();=0A=20=09= CheckPointCommitTs();=0A=20=09CheckPointSUBTRANS();=0A=20=09= CheckPointMultiXact();=0A@@=20-9339,7=20+9348,10=20@@=20= CreateRestartPoint(int=20flags)=0A=20=09=20*=20this=20because=20= StartupSUBTRANS=20hasn't=20been=20called=20yet.=0A=20=09=20*/=0A=20=09if=20= (EnableHotStandby)=0A+=09{=0A=20=09=09= TruncateSUBTRANS(GetOldestXmin(NULL,=20PROCARRAY_FLAGS_DEFAULT));=0A+=09=09= TruncateGlobalCSNLog(GetOldestXmin(NULL,=20PROCARRAY_FLAGS_DEFAULT));=0A= +=09}=0A=20=0A=20=09/*=20Real=20work=20is=20done,=20but=20log=20and=20= update=20before=20releasing=20lock.=20*/=0A=20=09LogCheckpointEnd(true);=0A= diff=20--git=20a/src/backend/storage/ipc/ipci.c=20= b/src/backend/storage/ipc/ipci.c=0Aindex=200c86a581c0..2af468fc6a=20= 100644=0A---=20a/src/backend/storage/ipc/ipci.c=0A+++=20= b/src/backend/storage/ipc/ipci.c=0A@@=20-16,6=20+16,7=20@@=0A=20=0A=20= #include=20"access/clog.h"=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_csn_log.h"=0A=20#include=20"access/heapam.h"=0A= =20#include=20"access/multixact.h"=0A=20#include=20"access/nbtree.h"=0A= @@=20-127,6=20+128,7=20@@=20CreateSharedMemoryAndSemaphores(bool=20= makePrivate,=20int=20port)=0A=20=09=09size=20=3D=20add_size(size,=20= ProcGlobalShmemSize());=0A=20=09=09size=20=3D=20add_size(size,=20= XLOGShmemSize());=0A=20=09=09size=20=3D=20add_size(size,=20= CLOGShmemSize());=0A+=09=09size=20=3D=20add_size(size,=20= GlobalCSNLogShmemSize());=0A=20=09=09size=20=3D=20add_size(size,=20= CommitTsShmemSize());=0A=20=09=09size=20=3D=20add_size(size,=20= SUBTRANSShmemSize());=0A=20=09=09size=20=3D=20add_size(size,=20= TwoPhaseShmemSize());=0A@@=20-219,6=20+221,7=20@@=20= CreateSharedMemoryAndSemaphores(bool=20makePrivate,=20int=20port)=0A=20=09= =20*/=0A=20=09XLOGShmemInit();=0A=20=09CLOGShmemInit();=0A+=09= GlobalCSNLogShmemInit();=0A=20=09CommitTsShmemInit();=0A=20=09= SUBTRANSShmemInit();=0A=20=09MultiXactShmemInit();=0Adiff=20--git=20= a/src/backend/storage/ipc/procarray.c=20= b/src/backend/storage/ipc/procarray.c=0Aindex=20afe1c03aa3..bc59fca21c=20= 100644=0A---=20a/src/backend/storage/ipc/procarray.c=0A+++=20= b/src/backend/storage/ipc/procarray.c=0A@@=20-46,6=20+46,7=20@@=0A=20= #include=20=0A=20=0A=20#include=20"access/clog.h"=0A+#include=20= "access/global_csn_log.h"=0A=20#include=20"access/subtrans.h"=0A=20= #include=20"access/transam.h"=0A=20#include=20"access/twophase.h"=0A@@=20= -833,6=20+834,7=20@@=20ProcArrayApplyRecoveryInfo(RunningTransactions=20= running)=0A=20=09while=20(TransactionIdPrecedes(latestObservedXid,=20= running->nextXid))=0A=20=09{=0A=20=09=09= ExtendSUBTRANS(latestObservedXid);=0A+=09=09= ExtendGlobalCSNLog(latestObservedXid);=0A=20=09=09= TransactionIdAdvance(latestObservedXid);=0A=20=09}=0A=20=09= TransactionIdRetreat(latestObservedXid);=09/*=20=3D=20running->nextXid=20= -=201=20*/=0A@@=20-3201,6=20+3203,7=20@@=20= RecordKnownAssignedTransactionIds(TransactionId=20xid)=0A=20=09=09while=20= (TransactionIdPrecedes(next_expected_xid,=20xid))=0A=20=09=09{=0A=20=09=09= =09TransactionIdAdvance(next_expected_xid);=0A+=09=09=09= ExtendGlobalCSNLog(next_expected_xid);=0A=20=09=09=09= ExtendSUBTRANS(next_expected_xid);=0A=20=09=09}=0A=20=09=09= Assert(next_expected_xid=20=3D=3D=20xid);=0Adiff=20--git=20= a/src/backend/storage/lmgr/lwlocknames.txt=20= b/src/backend/storage/lmgr/lwlocknames.txt=0Aindex=20= e6025ecedb..9615058f29=20100644=0A---=20= a/src/backend/storage/lmgr/lwlocknames.txt=0A+++=20= b/src/backend/storage/lmgr/lwlocknames.txt=0A@@=20-50,3=20+50,4=20@@=20= OldSnapshotTimeMapLock=09=09=09=0942=0A=20BackendRandomLock=09=09=09=09=09= 43=0A=20LogicalRepWorkerLock=09=09=09=0944=0A=20CLogTruncationLock=09=09=09= =09=0945=0A+GlobalCSNLogControlLock=09=09=09=0946=0Adiff=20--git=20= a/src/backend/utils/misc/guc.c=20b/src/backend/utils/misc/guc.c=0Aindex=20= 6eae3d62cc..2699a7e6b4=20100644=0A---=20a/src/backend/utils/misc/guc.c=0A= +++=20b/src/backend/utils/misc/guc.c=0A@@=20-1001,6=20+1001,15=20@@=20= static=20struct=20config_bool=20ConfigureNamesBool[]=20=3D=0A=20=09=09= false,=0A=20=09=09NULL,=20NULL,=20NULL=0A=20=09},=0A+=09{=0A+=09=09= {"track_global_snapshots",=20PGC_POSTMASTER,=20RESOURCES_MEM,=0A+=09=09=09= gettext_noop("Enable=20global=20snapshot=20tracking."),=0A+=09=09=09= gettext_noop("Used=20to=20achieve=20REPEATEBLE=20READ=20isolation=20= level=20for=20postgres_fdw=20transactions.")=0A+=09=09},=0A+=09=09= &track_global_snapshots,=0A+=09=09true,=20/*=20XXX:=20set=20true=20to=20= simplify=20tesing.=20XXX2:=20Seems=20that=20RESOURCES_MEM=20isn't=20the=20= best=20catagory=20*/=0A+=09=09NULL,=20NULL,=20NULL=0A+=09},=0A=20=09{=0A=20= =09=09{"ssl",=20PGC_SIGHUP,=20CONN_AUTH_SSL,=0A=20=09=09=09= gettext_noop("Enables=20SSL=20connections."),=0Adiff=20--git=20= a/src/backend/utils/probes.d=20b/src/backend/utils/probes.d=0Aindex=20= ad06e8e2ea..5ebe2ad888=20100644=0A---=20a/src/backend/utils/probes.d=0A= +++=20b/src/backend/utils/probes.d=0A@@=20-77,6=20+77,8=20@@=20provider=20= postgresql=20{=0A=20=09probe=20clog__checkpoint__done(bool);=0A=20=09= probe=20subtrans__checkpoint__start(bool);=0A=20=09probe=20= subtrans__checkpoint__done(bool);=0A+=09probe=20= globalcsnlog__checkpoint__start(bool);=0A+=09probe=20= globalcsnlog__checkpoint__done(bool);=0A=20=09probe=20= multixact__checkpoint__start(bool);=0A=20=09probe=20= multixact__checkpoint__done(bool);=0A=20=09probe=20= twophase__checkpoint__start();=0Adiff=20--git=20= a/src/bin/initdb/initdb.c=20b/src/bin/initdb/initdb.c=0Aindex=20= b39115c346..eafdca4276=20100644=0A---=20a/src/bin/initdb/initdb.c=0A+++=20= b/src/bin/initdb/initdb.c=0A@@=20-220,7=20+220,8=20@@=20static=20const=20= char=20*const=20subdirs[]=20=3D=20{=0A=20=09"pg_xact",=0A=20=09= "pg_logical",=0A=20=09"pg_logical/snapshots",=0A-=09= "pg_logical/mappings"=0A+=09"pg_logical/mappings",=0A+=09"pg_global_csn"=0A= =20};=0A=20=0A=20=0Adiff=20--git=20a/src/include/access/global_csn_log.h=20= b/src/include/access/global_csn_log.h=0Anew=20file=20mode=20100644=0A= index=200000000000..417c26c8a3=0A---=20/dev/null=0A+++=20= b/src/include/access/global_csn_log.h=0A@@=20-0,0=20+1,30=20@@=0A+/*=0A+=20= *=20global_csn_log.h=0A+=20*=0A+=20*=20Commit-Sequence-Number=20log.=0A+=20= *=0A+=20*=20Portions=20Copyright=20(c)=201996-2014,=20PostgreSQL=20= Global=20Development=20Group=0A+=20*=20Portions=20Copyright=20(c)=20= 1994,=20Regents=20of=20the=20University=20of=20California=0A+=20*=0A+=20= *=20src/include/access/global_csn_log.h=0A+=20*/=0A+#ifndef=20CSNLOG_H=0A= +#define=20CSNLOG_H=0A+=0A+#include=20"access/xlog.h"=0A+#include=20= "utils/snapshot.h"=0A+=0A+extern=20void=20= GlobalCSNLogSetCSN(TransactionId=20xid,=20int=20nsubxids,=0A+=09=09=09=09= =09=09=09=20=20=20TransactionId=20*subxids,=20GlobalCSN=20csn);=0A= +extern=20GlobalCSN=20GlobalCSNLogGetCSN(TransactionId=20xid);=0A+=0A= +extern=20Size=20GlobalCSNLogShmemSize(void);=0A+extern=20void=20= GlobalCSNLogShmemInit(void);=0A+extern=20void=20= BootStrapGlobalCSNLog(void);=0A+extern=20void=20= StartupGlobalCSNLog(TransactionId=20oldestActiveXID);=0A+extern=20void=20= ShutdownGlobalCSNLog(void);=0A+extern=20void=20= CheckPointGlobalCSNLog(void);=0A+extern=20void=20= ExtendGlobalCSNLog(TransactionId=20newestXact);=0A+extern=20void=20= TruncateGlobalCSNLog(TransactionId=20oldestXact);=0A+=0A+#endif=20=20=20= /*=20CSNLOG_H=20*/=0A\=20No=20newline=20at=20end=20of=20file=0Adiff=20= --git=20a/src/include/storage/lwlock.h=20b/src/include/storage/lwlock.h=0A= index=20c21bfe2f66..ab330b71c2=20100644=0A---=20= a/src/include/storage/lwlock.h=0A+++=20b/src/include/storage/lwlock.h=0A= @@=20-198,6=20+198,7=20@@=20typedef=20enum=20BuiltinTrancheIds=0A=20=09= LWTRANCHE_CLOG_BUFFERS=20=3D=20NUM_INDIVIDUAL_LWLOCKS,=0A=20=09= LWTRANCHE_COMMITTS_BUFFERS,=0A=20=09LWTRANCHE_SUBTRANS_BUFFERS,=0A+=09= LWTRANCHE_GLOBAL_CSN_LOG_BUFFERS,=0A=20=09LWTRANCHE_MXACTOFFSET_BUFFERS,=0A= =20=09LWTRANCHE_MXACTMEMBER_BUFFERS,=0A=20=09LWTRANCHE_ASYNC_BUFFERS,=0A= diff=20--git=20a/src/include/utils/snapshot.h=20= b/src/include/utils/snapshot.h=0Aindex=20a8a5a8f4c0..318d41e6f7=20100644=0A= ---=20a/src/include/utils/snapshot.h=0A+++=20= b/src/include/utils/snapshot.h=0A@@=20-24,6=20+24,9=20@@=20typedef=20= struct=20SnapshotData=20*Snapshot;=0A=20=0A=20#define=20InvalidSnapshot=09= =09((Snapshot)=20NULL)=0A=20=0A+typedef=20uint64=20GlobalCSN;=0A+extern=20= bool=20track_global_snapshots;=0A+=0A=20/*=0A=20=20*=20We=20use=20= SnapshotData=20structures=20to=20represent=20both=20"regular"=20(MVCC)=0A= =20=20*=20snapshots=20and=20"special"=20snapshots=20that=20have=20= non-MVCC=20semantics.=0A--=20=0A2.16.2=0A=0A= --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55 Content-Disposition: attachment; filename=0002-Global-snapshots.patch Content-Type: application/octet-stream; x-unix-mode=0644; name="0002-Global-snapshots.patch" Content-Transfer-Encoding: quoted-printable =46rom=20f023c9acaea4d40b5d2d71a543d1d8563e0d5079=20Mon=20Sep=2017=20= 00:00:00=202001=0AFrom:=20Stas=20Kelvich=20=0ADate:=20= Wed,=2025=20Apr=202018=2016:22:30=20+0300=0ASubject:=20[PATCH=202/3]=20= Global=20snapshots=0A=0A---=0A=20src/backend/access/transam/Makefile=20=20= =20=20=20=20=20=20=20=20=20|=20=20=202=20+-=0A=20= src/backend/access/transam/global_snapshot.c=20=20|=20747=20= ++++++++++++++++++++++++++=0A=20src/backend/access/transam/twophase.c=20=20= =20=20=20=20=20=20=20|=20156=20++++++=0A=20= src/backend/access/transam/xact.c=20=20=20=20=20=20=20=20=20=20=20=20=20= |=20=2029=20+=0A=20src/backend/access/transam/xlog.c=20=20=20=20=20=20=20= =20=20=20=20=20=20|=20=20=202=20+=0A=20src/backend/storage/ipc/ipci.c=20=20= =20=20=20=20=20=20=20=20=20=20=20=20=20=20|=20=20=203=20+=0A=20= src/backend/storage/ipc/procarray.c=20=20=20=20=20=20=20=20=20=20=20|=20= 103=20+++-=0A=20src/backend/storage/lmgr/lwlocknames.txt=20=20=20=20=20=20= |=20=20=201=20+=0A=20src/backend/storage/lmgr/proc.c=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20|=20=20=205=20+=0A=20= src/backend/utils/misc/guc.c=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20|=20=2011=20+=0A=20= src/backend/utils/misc/postgresql.conf.sample=20|=20=20=202=20+=0A=20= src/backend/utils/time/snapmgr.c=20=20=20=20=20=20=20=20=20=20=20=20=20=20= |=20=2098=20++++=0A=20src/backend/utils/time/tqual.c=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20=20|=20=2065=20++-=0A=20= src/include/access/global_snapshot.h=20=20=20=20=20=20=20=20=20=20|=20=20= 72=20+++=0A=20src/include/access/twophase.h=20=20=20=20=20=20=20=20=20=20= =20=20=20=20=20=20=20|=20=20=201=20+=0A=20= src/include/catalog/pg_proc.dat=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20|=20=2014=20+=0A=20src/include/datatype/timestamp.h=20=20=20=20=20=20=20= =20=20=20=20=20=20=20|=20=20=203=20+=0A=20src/include/fmgr.h=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20|=20= =20=201=20+=0A=20src/include/portability/instr_time.h=20=20=20=20=20=20=20= =20=20=20|=20=2010=20+=0A=20src/include/storage/proc.h=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20=20=20=20|=20=2015=20+=0A=20= src/include/storage/procarray.h=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20|=20=20=208=20+=0A=20src/include/utils/snapmgr.h=20=20=20=20=20=20=20=20= =20=20=20=20=20=20=20=20=20=20=20|=20=20=203=20+=0A=20= src/include/utils/snapshot.h=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20|=20=20=208=20+=0A=2023=20files=20changed,=201352=20= insertions(+),=207=20deletions(-)=0A=20create=20mode=20100644=20= src/backend/access/transam/global_snapshot.c=0A=20create=20mode=20100644=20= src/include/access/global_snapshot.h=0A=0Adiff=20--git=20= a/src/backend/access/transam/Makefile=20= b/src/backend/access/transam/Makefile=0Aindex=2003aa360ea3..8ef677cada=20= 100644=0A---=20a/src/backend/access/transam/Makefile=0A+++=20= b/src/backend/access/transam/Makefile=0A@@=20-12,7=20+12,7=20@@=20subdir=20= =3D=20src/backend/access/transam=0A=20top_builddir=20=3D=20../../../..=0A= =20include=20$(top_builddir)/src/Makefile.global=0A=20=0A-OBJS=20=3D=20= clog.o=20commit_ts.o=20global_csn_log.o=20generic_xlog.o=20\=0A+OBJS=20=3D= =20clog.o=20commit_ts.o=20global_csn_log.o=20global_snapshot.o=20= generic_xlog.o=20\=0A=20=09multixact.o=20parallel.o=20rmgr.o=20slru.o=20= \=0A=20=09subtrans.o=20timeline.o=20transam.o=20twophase.o=20= twophase_rmgr.o=20varsup.o=20\=0A=20=09xact.o=20xlog.o=20xlogarchive.o=20= xlogfuncs.o=20\=0Adiff=20--git=20= a/src/backend/access/transam/global_snapshot.c=20= b/src/backend/access/transam/global_snapshot.c=0Anew=20file=20mode=20= 100644=0Aindex=200000000000..b0801b5b3c=0A---=20/dev/null=0A+++=20= b/src/backend/access/transam/global_snapshot.c=0A@@=20-0,0=20+1,747=20@@=0A= +/*-----------------------------------------------------------------------= --=0A+=20*=0A+=20*=20global_snapshot.c=0A+=20*=09=09Support=20for=20= cross-node=20snapshot=20isolation.=0A+=20*=0A+=20*=20Portions=20= Copyright=20(c)=201996-2018,=20PostgreSQL=20Global=20Development=20Group=0A= +=20*=20Portions=20Copyright=20(c)=201994,=20Regents=20of=20the=20= University=20of=20California=0A+=20*=0A+=20*=20= src/backend/access/transam/global_snapshot.c=0A+=20*=0A+=20= *-------------------------------------------------------------------------= =0A+=20*/=0A+#include=20"postgres.h"=0A+=0A+#include=20= "access/global_csn_log.h"=0A+#include=20"access/global_snapshot.h"=0A= +#include=20"access/transam.h"=0A+#include=20"access/twophase.h"=0A= +#include=20"access/xact.h"=0A+#include=20"portability/instr_time.h"=0A= +#include=20"storage/lmgr.h"=0A+#include=20"storage/proc.h"=0A+#include=20= "storage/procarray.h"=0A+#include=20"storage/shmem.h"=0A+#include=20= "storage/spin.h"=0A+#include=20"utils/builtins.h"=0A+#include=20= "utils/guc.h"=0A+#include=20"utils/snapmgr.h"=0A+#include=20= "miscadmin.h"=0A+=0A+/*=20Raise=20a=20warning=20if=20imported=20= global_csn=20exceeds=20ours=20by=20this=20value.=20*/=0A+#define=20= SNAP_DESYNC_COMPLAIN=20(1*NSECS_PER_SEC)=20/*=201=20second=20*/=0A+=0A= +/*=0A+=20*=20GlobalSnapshotState=0A+=20*=0A+=20*=20Do=20not=20trust=20= local=20clocks=20to=20be=20strictly=20monotonical=20and=20save=20last=20= acquired=0A+=20*=20value=20so=20later=20we=20can=20compare=20next=20= timestamp=20with=20it.=20Accessed=20through=0A+=20*=20= GlobalSnapshotGenerate()=20and=20GlobalSnapshotSync().=0A+=20*/=0A= +typedef=20struct=0A+{=0A+=09GlobalCSN=09=09=20last_global_csn;=0A+=09= volatile=20slock_t=20lock;=0A+}=20GlobalSnapshotState;=0A+=0A+static=20= GlobalSnapshotState=20*gsState;=0A+=0A+=0A+/*=0A+=20*=20GUC=20to=20delay=20= advance=20of=20oldestXid=20for=20this=20amount=20of=20time.=20Also=20= determines=0A+=20*=20the=20size=20GlobalSnapshotXidMap=20circular=20= buffer.=0A+=20*/=0A+int=20global_snapshot_defer_time;=0A+=0A+/*=0A+=20*=20= Enables=20this=20module.=0A+=20*/=0A+extern=20bool=20= track_global_snapshots;=0A+=0A+/*=0A+=20*=20GlobalSnapshotXidMap=0A+=20*=0A= +=20*=20To=20be=20able=20to=20install=20global=20snapshot=20that=20= points=20to=20past=20we=20need=20to=20keep=0A+=20*=20old=20versions=20of=20= tuples=20and=20therefore=20delay=20advance=20of=20oldestXid.=20=20Here=20= we=0A+=20*=20keep=20track=20of=20correspondence=20between=20snapshot's=20= global_csn=20and=20oldestXid=0A+=20*=20that=20was=20set=20at=20the=20= time=20when=20the=20snapshot=20was=20taken.=20=20Much=20like=20the=0A+=20= *=20snapshot=20too=20old's=20OldSnapshotControlData=20does,=20but=20with=20= finer=20granularity=0A+=20*=20to=20seconds.=0A+=20*=0A+=20*=20Different=20= strategies=20can=20be=20employed=20to=20hold=20oldestXid=20(e.g.=20we=20= can=20track=0A+=20*=20oldest=20global_csn-based=20snapshot=20among=20= cluster=20nodes=20and=20map=20it=20oldestXid=0A+=20*=20on=20each=20node)=20= but=20here=20implemented=20one=20that=20tries=20to=20avoid=20cross-node=0A= +=20*=20communications=20which=20are=20tricky=20in=20case=20of=20= postgres_fdw.=0A+=20*=0A+=20*=20On=20each=20snapshot=20acquisition=20= GlobalSnapshotMapXmin()=20is=20called=20and=20stores=0A+=20*=20= correspondence=20between=20current=20global_csn=20and=20oldestXmin=20in=20= a=20sparse=20way:=0A+=20*=20global_csn=20is=20rounded=20to=20seconds=20= (and=20here=20we=20use=20the=20fact=20that=20global_csn=0A+=20*=20is=20= just=20a=20timestamp)=20and=20oldestXmin=20is=20stored=20in=20the=20= circular=20buffer=20where=0A+=20*=20rounded=20global_csn=20acts=20as=20= an=20offset=20from=20current=20circular=20buffer=20head.=0A+=20*=20Size=20= of=20the=20circular=20buffer=20is=20controlled=20by=20= global_snapshot_defer_time=20GUC.=0A+=20*=0A+=20*=20When=20global=20= snapshot=20arrives=20from=20different=20node=20we=20check=20that=20its=0A= +=20*=20global_csn=20is=20still=20in=20our=20map,=20otherwise=20we'll=20= error=20out=20with=20"snapshot=20too=0A+=20*=20old"=20message.=20=20If=20= global_csn=20is=20successfully=20mapped=20to=20oldestXid=20we=20move=0A+=20= *=20backend's=20pgxact->xmin=20to=20proc->originalXmin=20and=20fill=20= pgxact->xmin=20to=0A+=20*=20mapped=20oldestXid.=20=20That=20way=20= GetOldestXmin()=20can=20take=20into=20account=20backends=0A+=20*=20with=20= imported=20global=20snapshot=20and=20old=20tuple=20versions=20will=20be=20= preserved.=0A+=20*=0A+=20*=20Also=20while=20calculating=20oldestXmin=20= for=20our=20map=20in=20presence=20of=20imported=0A+=20*=20global=20= snapshots=20we=20should=20use=20proc->originalXmin=20instead=20of=20= pgxact->xmin=0A+=20*=20that=20was=20set=20during=20import.=20=20= Otherwise,=20we=20can=20create=20a=20feedback=20loop:=0A+=20*=20xmin's=20= of=20imported=20global=20snapshots=20were=20calculated=20using=20our=20= map=20and=20new=0A+=20*=20entries=20in=20map=20going=20to=20be=20= calculated=20based=20on=20that=20xmin's,=20and=20there=20is=0A+=20*=20a=20= risk=20to=20stuck=20forever=20with=20one=20non-increasing=20oldestXmin.=20= =20All=20other=0A+=20*=20callers=20of=20GetOldestXmin()=20are=20using=20= pgxact->xmin=20so=20the=20old=20tuple=20versions=0A+=20*=20are=20= preserved.=0A+=20*/=0A+typedef=20struct=20GlobalSnapshotXidMap=0A+{=0A+=09= int=09=09=09=09=20head;=09=09=09=09/*=20offset=20of=20current=20freshest=20= value=20*/=0A+=09int=09=09=09=09=20size;=09=09=09=09/*=20total=20size=20= of=20circular=20buffer=20*/=0A+=09GlobalCSN_atomic=20last_csn_seconds;=09= /*=20last=20rounded=20global_csn=20that=20changed=0A+=09=09=09=09=09=09=09= =09=09=09=20*=20xmin_by_second[]=20*/=0A+=09TransactionId=20=20=20= *xmin_by_second;=09/*=20circular=20buffer=20of=20oldestXmin's=20*/=0A+}=0A= +GlobalSnapshotXidMap;=0A+=0A+static=20GlobalSnapshotXidMap=20*gsXidMap;=0A= +=0A+=0A+/*=20Estimate=20shared=20memory=20space=20needed=20*/=0A+Size=0A= +GlobalSnapshotShmemSize(void)=0A+{=0A+=09Size=09size=20=3D=200;=0A+=0A+=09= if=20(track_global_snapshots=20||=20global_snapshot_defer_time=20>=200)=0A= +=09{=0A+=09=09size=20+=3D=20MAXALIGN(sizeof(GlobalSnapshotState));=0A+=09= }=0A+=0A+=09if=20(global_snapshot_defer_time=20>=200)=0A+=09{=0A+=09=09= size=20+=3D=20sizeof(GlobalSnapshotXidMap);=0A+=09=09size=20+=3D=20= global_snapshot_defer_time*sizeof(TransactionId);=0A+=09=09size=20=3D=20= MAXALIGN(size);=0A+=09}=0A+=0A+=09return=20size;=0A+}=0A+=0A+/*=20Init=20= shared=20memory=20structures=20*/=0A+void=0A+GlobalSnapshotShmemInit()=0A= +{=0A+=09bool=20found;=0A+=0A+=09if=20(track_global_snapshots=20||=20= global_snapshot_defer_time=20>=200)=0A+=09{=0A+=09=09gsState=20=3D=20= ShmemInitStruct("gsState",=0A+=09=09=09=09=09=09=09=09= sizeof(GlobalSnapshotState),=0A+=09=09=09=09=09=09=09=09&found);=0A+=09=09= if=20(!found)=0A+=09=09{=0A+=09=09=09gsState->last_global_csn=20=3D=200;=0A= +=09=09=09SpinLockInit(&gsState->lock);=0A+=09=09}=0A+=09}=0A+=0A+=09if=20= (global_snapshot_defer_time=20>=200)=0A+=09{=0A+=09=09gsXidMap=20=3D=20= ShmemInitStruct("gsXidMap",=0A+=09=09=09=09=09=09=09=09=20=20=20= sizeof(GlobalSnapshotXidMap),=0A+=09=09=09=09=09=09=09=09=20=20=20= &found);=0A+=09=09if=20(!found)=0A+=09=09{=0A+=09=09=09int=20i;=0A+=0A+=09= =09=09pg_atomic_init_u64(&gsXidMap->last_csn_seconds,=200);=0A+=09=09=09= gsXidMap->head=20=3D=200;=0A+=09=09=09gsXidMap->size=20=3D=20= global_snapshot_defer_time;=0A+=09=09=09gsXidMap->xmin_by_second=20=3D=0A= +=09=09=09=09=09=09=09ShmemAlloc(sizeof(TransactionId)*gsXidMap->size);=0A= +=0A+=09=09=09for=20(i=20=3D=200;=20i=20<=20gsXidMap->size;=20i++)=0A+=09= =09=09=09gsXidMap->xmin_by_second[i]=20=3D=20InvalidTransactionId;=0A+=09= =09}=0A+=09}=0A+}=0A+=0A+/*=0A+=20*=20GlobalSnapshotStartup=0A+=20*=0A+=20= *=20Set=20gsXidMap=20etnries=20to=20oldestActiveXID=20during=20startup.=0A= +=20*/=0A+void=0A+GlobalSnapshotStartup(TransactionId=20oldestActiveXID)=0A= +{=0A+=09/*=0A+=09=20*=20Run=20only=20if=20we=20have=20initialized=20= shared=20memory=20and=20gsXidMap=0A+=09=20*=20is=20enabled.=0A+=09=20*/=0A= +=09if=20(IsNormalProcessingMode()=20&&=20global_snapshot_defer_time=20>=20= 0)=0A+=09{=0A+=09=09int=20i;=0A+=0A+=09=09= Assert(TransactionIdIsValid(oldestActiveXID));=0A+=09=09for=20(i=20=3D=20= 0;=20i=20<=20gsXidMap->size;=20i++)=0A+=09=09=09= gsXidMap->xmin_by_second[i]=20=3D=20oldestActiveXID;=0A+=09=09= ProcArraySetGlobalSnapshotXmin(oldestActiveXID);=0A+=09}=0A+}=0A+=0A+/*=0A= +=20*=20GlobalSnapshotMapXmin=0A+=20*=0A+=20*=20Maintain=20circular=20= buffer=20of=20oldestXmins=20for=20several=20seconds=20in=20past.=20This=0A= +=20*=20buffer=20allows=20to=20shift=20oldestXmin=20in=20the=20past=20= when=20backend=20is=20importing=0A+=20*=20global=20transaction.=20= Otherwise=20old=20versions=20of=20tuples=20that=20were=20needed=20for=0A= +=20*=20this=20transaction=20can=20be=20recycled=20by=20other=20= processes=20(vacuum,=20HOT,=20etc).=0A+=20*=0A+=20*=20Called=20upon=20= each=20snapshot=20creation=20after=20ProcArrayLock=20is=20released.=20= Such=0A+=20*=20usage=20creates=20a=20race=20condition.=20It=20is=20= possible=20that=20backend=20who=20got=0A+=20*=20glabal_csn=20called=20= GlobalSnapshotMapXmin()=20only=20after=20other=20backends=20managed=0A+=20= *=20to=20get=20snapshot=20and=20complete=20GlobalSnapshotMapXmin()=20= call.=20To=20address=20that=0A+=20*=20race=20we=20do=20two=20thigs:=0A+=20= *=0A+=20*=09=09*=20snapshot_global_csn=20is=20always=20rounded=20up=20to=20= next=20second.=20So=20that=20is=0A+=20*=09=09=20=20okay=20if=20call=20to=20= GlobalSnapshotMapXmin()=20with=20later=20global_csn=20will=0A+=20*=09=09=20= =20succeed=20first=20--=20it=20anyway=20will=20be=20taken=20into=20= account=20for=20a=20next=0A+=20*=09=09=20=20second.=0A+=20*=0A+=20*=09=09= *=20if=20GlobalSnapshotMapXmin()=20founds=20a=20gap=20in=20several=20= seconds=20between=0A+=20*=09=09=20=20current=20call=20and=20latest=20= completed=20call=20then=20it=20should=20fill=20that=20gap=0A+=20*=09=09=20= =20with=20latest=20known=20values=20instead=20of=20new=20one.=20= Otherwise=20it=20is=20possible=0A+=20*=09=09=20=20(however=20highly=20= unlikely)=20that=20this=20gap=20also=20happend=20between=20taking=0A+=20= *=09=09=20=20snapshot=20and=20call=20to=20GlobalSnapshotMapXmin()=20for=20= some=20backend.=20And=20we=0A+=20*=09=09=20=20are=20at=20risk=20to=20= fill=20circullar=20buffer=20with=20oldestXmin's=20that=20are=0A+=20*=09=09= =20=20bigger=20then=20they=20actually=20were.=0A+=20*/=0A+void=0A= +GlobalSnapshotMapXmin(GlobalCSN=20snapshot_global_csn)=0A+{=0A+=09int=20= offset,=20gap,=20i;=0A+=09GlobalCSN=20csn_seconds;=0A+=09GlobalCSN=20= last_csn_seconds;=0A+=09volatile=20TransactionId=20oldest_deferred_xmin;=0A= +=09TransactionId=20current_oldest_xmin,=20previous_oldest_xmin;=0A+=0A+=09= /*=20Callers=20should=20check=20config=20values=20*/=0A+=09= Assert(global_snapshot_defer_time=20>=200);=0A+=09Assert(gsXidMap=20!=3D=20= NULL);=0A+=0A+=09/*=0A+=09=20*=20We=20don't=20have=20guarantee=20that=20= process=20who=20called=20us=20first=20for=20this=0A+=09=20*=20= csn_seconds=20is=20actually=20one=20who=20took=20snapshot=20firt=20in=20= this=20second.=0A+=09=20*=20So=20just=20round=20up=20global_csn=20to=20= the=20next=20second=20--=20snapshots=20for=20next=0A+=09=20*=20second=20= would=20have=20oldestXmin=20greater=20or=20equal=20then=20ours=20anyway.=0A= +=09=20*/=0A+=09csn_seconds=20=3D=20(snapshot_global_csn=20/=20= NSECS_PER_SEC=20+=201);=0A+=0A+=09/*=0A+=09=20*=20Fast-path=20check.=20= Avoid=20taking=20exclusive=20GlobalSnapshotXidMapLock=20lock=0A+=09=20*=20= if=20oldestXid=20was=20already=20written=20to=20xmin_by_second[]=20for=20= this=20rounded=0A+=09=20*=20global_csn.=0A+=09=20*/=0A+=09if=20= (pg_atomic_read_u64(&gsXidMap->last_csn_seconds)=20>=3D=20csn_seconds)=0A= +=09=09return;=0A+=0A+=09/*=20Ok,=20we=20have=20new=20entry=20(or=20= entries)=20*/=0A+=09LWLockAcquire(GlobalSnapshotXidMapLock,=20= LW_EXCLUSIVE);=0A+=0A+=09/*=20Re-check=20last_csn_seconds=20under=20lock=20= */=0A+=09last_csn_seconds=20=3D=20= pg_atomic_read_u64(&gsXidMap->last_csn_seconds);=0A+=09if=20= (last_csn_seconds=20>=3D=20csn_seconds)=0A+=09{=0A+=09=09= LWLockRelease(GlobalSnapshotXidMapLock);=0A+=09=09return;=0A+=09}=0A+=09= pg_atomic_write_u64(&gsXidMap->last_csn_seconds,=20csn_seconds);=0A+=0A+=09= /*=0A+=09=20*=20Count=20oldest_xmin.=0A+=09=20*=0A+=09=20*=20It=20was=20= possible=20to=20calculate=20oldest_xmin=20during=20corresponding=20= snapshot=0A+=09=20*=20creation,=20but=20GetSnapshotData()=20= intentionally=20reads=20only=20PgXact,=20but=20not=0A+=09=20*=20PgProc.=20= And=20we=20need=20info=20about=20originalXmin=20(see=20comment=20to=20= gsXidMap)=0A+=09=20*=20which=20is=20stored=20in=20PgProc=20because=20of=20= threats=20in=20comments=20around=20PgXact=0A+=09=20*=20about=20extending=20= it=20with=20new=20fields.=20So=20just=20calculate=20oldest_xmin=20again,=0A= +=09=20*=20that=20anyway=20happens=20quite=20rarely.=0A+=09=20*/=0A+=09= current_oldest_xmin=20=3D=20GetOldestXmin(NULL,=20= PROCARRAY_NON_IMPORTED_XMIN);=0A+=0A+=09previous_oldest_xmin=20=3D=20= gsXidMap->xmin_by_second[gsXidMap->head];=0A+=0A+=09= Assert(TransactionIdIsNormal(current_oldest_xmin));=0A+=09= Assert(TransactionIdIsNormal(previous_oldest_xmin));=0A+=09= Assert(TransactionIdFollowsOrEquals(current_oldest_xmin,=20= previous_oldest_xmin));=0A+=0A+=09gap=20=3D=20csn_seconds=20-=20= last_csn_seconds;=0A+=09offset=20=3D=20csn_seconds=20%=20gsXidMap->size;=0A= +=0A+=09/*=20Sanity=20check=20before=20we=20update=20head=20and=20gap=20= */=0A+=09Assert(=20gap=20>=3D=201=20);=0A+=09Assert(=20(gsXidMap->head=20= +=20gap)=20%=20gsXidMap->size=20=3D=3D=20offset=20);=0A+=0A+=09gap=20=3D=20= gap=20>=20gsXidMap->size=20?=20gsXidMap->size=20:=20gap;=0A+=09= gsXidMap->head=20=3D=20offset;=0A+=0A+=09/*=20Fill=20new=20entry=20with=20= current_oldest_xmin=20*/=0A+=09gsXidMap->xmin_by_second[offset]=20=3D=20= current_oldest_xmin;=0A+=09offset=20=3D=20(offset=20+=20gsXidMap->size=20= -=201)=20%=20gsXidMap->size;=0A+=0A+=09/*=0A+=09=20*=20If=20we=20have=20= gap=20then=20fill=20it=20with=20previous_oldest_xmin=20for=20reasons=0A+=09= =20*=20outlined=20in=20comment=20above=20this=20function.=0A+=09=20*/=0A= +=09for=20(i=20=3D=201;=20i=20<=20gap;=20i++)=0A+=09{=0A+=09=09= gsXidMap->xmin_by_second[offset]=20=3D=20previous_oldest_xmin;=0A+=09=09= offset=20=3D=20(offset=20+=20gsXidMap->size=20-=201)=20%=20= gsXidMap->size;=0A+=09}=0A+=0A+=09oldest_deferred_xmin=20=3D=0A+=09=09= gsXidMap->xmin_by_second[=20(gsXidMap->head=20+=201)=20%=20= gsXidMap->size=20];=0A+=0A+=09LWLockRelease(GlobalSnapshotXidMapLock);=0A= +=0A+=09/*=0A+=09=20*=20Advance=20procArray->global_snapshot_xmin=20= after=20we=20released=0A+=09=20*=20GlobalSnapshotXidMapLock.=0A+=09=20*/=0A= +=09ProcArraySetGlobalSnapshotXmin(oldest_deferred_xmin);=0A+}=0A+=0A+=0A= +/*=0A+=20*=20GlobalSnapshotToXmin=0A+=20*=0A+=20*=20Get=20oldestXmin=20= that=20took=20place=20when=20snapshot_global_csn=20was=20taken.=0A+=20*/=0A= +TransactionId=0A+GlobalSnapshotToXmin(GlobalCSN=20snapshot_global_csn)=0A= +{=0A+=09TransactionId=20xmin;=0A+=09GlobalCSN=20csn_seconds;=0A+=09= volatile=20GlobalCSN=20last_csn_seconds;=0A+=0A+=09/*=20Callers=20should=20= check=20config=20values=20*/=0A+=09Assert(global_snapshot_defer_time=20>=20= 0);=0A+=09Assert(gsXidMap=20!=3D=20NULL);=0A+=0A+=09/*=20Round=20down=20= to=20get=20conservative=20estimates=20*/=0A+=09csn_seconds=20=3D=20= (snapshot_global_csn=20/=20NSECS_PER_SEC);=0A+=0A+=09= LWLockAcquire(GlobalSnapshotXidMapLock,=20LW_SHARED);=0A+=09= last_csn_seconds=20=3D=20= pg_atomic_read_u64(&gsXidMap->last_csn_seconds);=0A+=09if=20(csn_seconds=20= >=20last_csn_seconds)=0A+=09{=0A+=09=09/*=20we=20don't=20have=20entry=20= for=20this=20global_csn=20yet,=20return=20latest=20known=20*/=0A+=09=09= xmin=20=3D=20gsXidMap->xmin_by_second[gsXidMap->head];=0A+=09}=0A+=09= else=20if=20(last_csn_seconds=20-=20csn_seconds=20<=20gsXidMap->size)=0A= +=09{=0A+=09=09/*=20we=20are=20good,=20retrieve=20value=20from=20our=20= map=20*/=0A+=09=09Assert(last_csn_seconds=20%=20gsXidMap->size=20=3D=3D=20= gsXidMap->head);=0A+=09=09xmin=20=3D=20= gsXidMap->xmin_by_second[csn_seconds=20%=20gsXidMap->size];=0A+=09}=0A+=09= else=0A+=09{=0A+=09=09/*=20requested=20global_csn=20is=20too=20old,=20= let=20caller=20know=20*/=0A+=09=09xmin=20=3D=20InvalidTransactionId;=0A+=09= }=0A+=09LWLockRelease(GlobalSnapshotXidMapLock);=0A+=0A+=09return=20= xmin;=0A+}=0A+=0A+/*=0A+=20*=20GlobalSnapshotGenerate=0A+=20*=0A+=20*=20= Generate=20GlobalCSN=20which=20is=20actually=20a=20local=20time.=20Also=20= we=20are=20forcing=0A+=20*=20this=20time=20to=20be=20always=20= increasing.=20Since=20now=20it=20is=20not=20uncommon=20to=20have=0A+=20*=20= millions=20of=20read=20transactions=20per=20second=20we=20are=20trying=20= to=20use=20nanoseconds=0A+=20*=20if=20such=20time=20resolution=20is=20= available.=0A+=20*/=0A+GlobalCSN=0A+GlobalSnapshotGenerate(bool=20= locked)=0A+{=0A+=09instr_time=09current_time;=0A+=09GlobalCSN=09= global_csn;=0A+=0A+=09Assert(track_global_snapshots=20||=20= global_snapshot_defer_time=20>=200);=0A+=0A+=09/*=0A+=09=20*=20TODO:=20= create=20some=20macro=20that=20add=20small=20random=20shift=20to=20= current=20time.=0A+=09=20*/=0A+=09INSTR_TIME_SET_CURRENT(current_time);=0A= +=09global_csn=20=3D=20(GlobalCSN)=20= INSTR_TIME_GET_NANOSEC(current_time);=0A+=0A+=09/*=20TODO:=20change=20to=20= atomics?=20*/=0A+=09if=20(!locked)=0A+=09=09= SpinLockAcquire(&gsState->lock);=0A+=0A+=09if=20(global_csn=20<=3D=20= gsState->last_global_csn)=0A+=09=09global_csn=20=3D=20= ++gsState->last_global_csn;=0A+=09else=0A+=09=09gsState->last_global_csn=20= =3D=20global_csn;=0A+=0A+=09if=20(!locked)=0A+=09=09= SpinLockRelease(&gsState->lock);=0A+=0A+=09return=20global_csn;=0A+}=0A+=0A= +/*=0A+=20*=20GlobalSnapshotSync=0A+=20*=0A+=20*=20Due=20to=20time=20= desynchronization=20on=20different=20nodes=20we=20can=20receive=20= global_csn=0A+=20*=20which=20is=20greater=20than=20global_csn=20on=20= this=20node.=20To=20preserve=20proper=20isolation=0A+=20*=20this=20node=20= needs=20to=20wait=20when=20such=20global_csn=20comes=20on=20local=20= clock.=0A+=20*=0A+=20*=20This=20should=20happend=20relatively=20rare=20= if=20nodes=20have=20running=20NTP/PTP/etc.=0A+=20*=20Complain=20if=20= wait=20time=20is=20more=20than=20SNAP_SYNC_COMPLAIN.=0A+=20*/=0A+void=0A= +GlobalSnapshotSync(GlobalCSN=20remote_gcsn)=0A+{=0A+=09GlobalCSN=09= local_gcsn;=0A+=09GlobalCSN=09delta;=0A+=0A+=09= Assert(track_global_snapshots);=0A+=0A+=09for(;;)=0A+=09{=0A+=09=09= SpinLockAcquire(&gsState->lock);=0A+=09=09if=20(gsState->last_global_csn=20= >=20remote_gcsn)=0A+=09=09{=0A+=09=09=09/*=20Everything=20is=20fine=20*/=0A= +=09=09=09SpinLockRelease(&gsState->lock);=0A+=09=09=09return;=0A+=09=09= }=0A+=09=09else=20if=20((local_gcsn=20=3D=20= GlobalSnapshotGenerate(true))=20>=3D=20remote_gcsn)=0A+=09=09{=0A+=09=09=09= /*=0A+=09=09=09=20*=20Everything=20is=20fine=20too,=20but=20= last_global_csn=20wasn't=20updated=20for=0A+=09=09=09=20*=20some=20time.=0A= +=09=09=09=20*/=0A+=09=09=09SpinLockRelease(&gsState->lock);=0A+=09=09=09= return;=0A+=09=09}=0A+=09=09SpinLockRelease(&gsState->lock);=0A+=0A+=09=09= /*=20Okay=20we=20need=20to=20sleep=20now=20*/=0A+=09=09delta=20=3D=20= remote_gcsn=20-=20local_gcsn;=0A+=09=09if=20(delta=20>=20= SNAP_DESYNC_COMPLAIN)=0A+=09=09=09ereport(WARNING,=0A+=09=09=09=09= (errmsg("remote=20global=20snapshot=20exceeds=20ours=20by=20more=20than=20= a=20second"),=0A+=09=09=09=09=20errhint("Consider=20running=20NTPd=20on=20= servers=20participating=20in=20global=20transaction")));=0A+=0A+=09=09/*=20= TODO:=20report=20this=20sleeptime=20somewhere?=20*/=0A+=09=09= pg_usleep((long)=20(delta/NSECS_PER_USEC));=0A+=0A+=09=09/*=0A+=09=09=20= *=20Loop=20that=20checks=20to=20ensure=20that=20we=20actually=20slept=20= for=20specified=0A+=09=09=20*=20amount=20of=20time.=0A+=09=09=20*/=0A+=09= }=0A+=0A+=09Assert(false);=20/*=20Should=20not=20happend=20*/=0A+=09= return;=0A+}=0A+=0A+/*=0A+=20*=20TransactionIdGetGlobalCSN=0A+=20*=0A+=20= *=20Get=20GlobalCSN=20for=20specified=20TransactionId=20taking=20care=20= about=20special=20xids,=0A+=20*=20xids=20beyond=20TransactionXmin=20and=20= InDoubt=20states.=0A+=20*/=0A+GlobalCSN=0A= +TransactionIdGetGlobalCSN(TransactionId=20xid)=0A+{=0A+=09GlobalCSN=20= global_csn;=0A+=0A+=09Assert(track_global_snapshots);=0A+=0A+=09/*=20= Handle=20permanent=20TransactionId's=20for=20which=20we=20don't=20have=20= mapping=20*/=0A+=09if=20(!TransactionIdIsNormal(xid))=0A+=09{=0A+=09=09= if=20(xid=20=3D=3D=20InvalidTransactionId)=0A+=09=09=09return=20= AbortedGlobalCSN;=0A+=09=09if=20(xid=20=3D=3D=20FrozenTransactionId=20||=20= xid=20=3D=3D=20BootstrapTransactionId)=0A+=09=09=09return=20= FrozenGlobalCSN;=0A+=09=09Assert(false);=20/*=20Should=20not=20happend=20= */=0A+=09}=0A+=0A+=09/*=0A+=09=20*=20For=20xids=20which=20less=20then=20= TransactionXmin=20GlobalCSNLog=20can=20be=20already=0A+=09=20*=20trimmed=20= but=20we=20know=20that=20such=20transaction=20is=20definetly=20not=20= concurrently=0A+=09=20*=20running=20according=20to=20any=20snapshot=20= including=20timetravel=20ones.=20Callers=0A+=09=20*=20should=20check=20= TransactionDidCommit=20after.=0A+=09=20*/=0A+=09if=20= (TransactionIdPrecedes(xid,=20TransactionXmin))=0A+=09=09return=20= FrozenGlobalCSN;=0A+=0A+=09/*=20Read=20GlobalCSN=20from=20SLRU=20*/=0A+=09= global_csn=20=3D=20GlobalCSNLogGetCSN(xid);=0A+=0A+=09/*=0A+=09=20*=20If=20= we=20faced=20InDoubt=20state=20then=20transaction=20is=20beeing=20= committed=20and=20we=0A+=09=20*=20should=20wait=20until=20GlobalCSN=20= will=20be=20assigned=20so=20that=20visibility=20check=0A+=09=20*=20could=20= decide=20whether=20tuple=20is=20in=20snapshot.=20See=20also=20comments=20= in=0A+=09=20*=20GlobalSnapshotPrecommit().=0A+=09=20*/=0A+=09if=20= (GlobalCSNIsInDoubt(global_csn))=0A+=09{=0A+=09=09XactLockTableWait(xid,=20= NULL,=20NULL,=20XLTW_None);=0A+=09=09global_csn=20=3D=20= GlobalCSNLogGetCSN(xid);=0A+=09=09Assert(GlobalCSNIsNormal(global_csn)=20= ||=0A+=09=09=09=09GlobalCSNIsAborted(global_csn));=0A+=09}=0A+=0A+=09= Assert(GlobalCSNIsNormal(global_csn)=20||=0A+=09=09=09= GlobalCSNIsInProgress(global_csn)=20||=0A+=09=09=09= GlobalCSNIsAborted(global_csn));=0A+=0A+=09return=20global_csn;=0A+}=0A+=0A= +/*=0A+=20*=20XidInvisibleInGlobalSnapshot=0A+=20*=0A+=20*=20Version=20= of=20XidInMVCCSnapshot=20for=20global=20transactions.=20For=20= non-imported=0A+=20*=20global=20snapshots=20this=20should=20give=20same=20= results=20as=20XidInLocalMVCCSnapshot=0A+=20*=20(except=20that=20aborts=20= will=20be=20shown=20as=20invisible=20without=20going=20to=20clog)=20and=20= to=0A+=20*=20ensure=20such=20behaviour=20XidInMVCCSnapshot=20is=20coated=20= with=20asserts=20that=20checks=0A+=20*=20identicalness=20of=20= XidInvisibleInGlobalSnapshot/XidInLocalMVCCSnapshot=20in=0A+=20*=20case=20= of=20ordinary=20snapshot.=0A+=20*/=0A+bool=0A= +XidInvisibleInGlobalSnapshot(TransactionId=20xid,=20Snapshot=20= snapshot)=0A+{=0A+=09GlobalCSN=20csn;=0A+=0A+=09= Assert(track_global_snapshots);=0A+=0A+=09csn=20=3D=20= TransactionIdGetGlobalCSN(xid);=0A+=0A+=09if=20(GlobalCSNIsNormal(csn))=0A= +=09{=0A+=09=09if=20(csn=20<=20snapshot->global_csn)=0A+=09=09=09return=20= false;=0A+=09=09else=0A+=09=09=09return=20true;=0A+=09}=0A+=09else=20if=20= (GlobalCSNIsFrozen(csn))=0A+=09{=0A+=09=09/*=20It=20is=20bootstrap=20or=20= frozen=20transaction=20*/=0A+=09=09return=20false;=0A+=09}=0A+=09else=0A= +=09{=0A+=09=09/*=20It=20is=20aborted=20or=20in-progress=20*/=0A+=09=09= Assert(GlobalCSNIsAborted(csn)=20||=20GlobalCSNIsInProgress(csn));=0A+=09= =09if=20(GlobalCSNIsAborted(csn))=0A+=09=09=09= Assert(TransactionIdDidAbort(xid));=0A+=09=09return=20true;=0A+=09}=0A+}=0A= +=0A+=0A= +/************************************************************************= *****=0A+=20*=20Functions=20to=20handle=20distributed=20commit=20on=20= transaction=20coordinator:=0A+=20*=20GlobalSnapshotPrepareCurrent()=20/=20= GlobalSnapshotAssignCsnCurrent().=0A+=20*=20Correspoding=20functions=20= for=20remote=20nodes=20are=20defined=20in=20twophase.c:=0A+=20*=20= pg_global_snaphot_prepare/pg_global_snaphot_assign.=0A+=20= **************************************************************************= ***/=0A+=0A+=0A+/*=0A+=20*=20GlobalSnapshotPrepareCurrent=0A+=20*=0A+=20= *=20Set=20InDoubt=20state=20for=20currently=20active=20transaction=20and=20= return=20commit's=0A+=20*=20global=20snapshot.=0A+=20*/=0A+GlobalCSN=0A= +GlobalSnapshotPrepareCurrent()=0A+{=0A+=09TransactionId=20xid=20=3D=20= GetCurrentTransactionIdIfAny();=0A+=0A+=09if=20(!track_global_snapshots)=0A= +=09=09ereport(ERROR,=0A+=09=09=09= (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09=09= errmsg("could=20not=20prepare=20transaction=20for=20global=20commit"),=0A= +=09=09=09=09errhint("Make=20sure=20the=20configuration=20parameter=20= \"%s\"=20is=20enabled.",=0A+=09=09=09=09=09=09= "track_global_snapshots")));=0A+=0A+=09if=20(TransactionIdIsValid(xid))=0A= +=09{=0A+=09=09TransactionId=20*subxids;=0A+=09=09int=20nsubxids=20=3D=20= xactGetCommittedChildren(&subxids);=0A+=09=09GlobalCSNLogSetCSN(xid,=20= nsubxids,=0A+=09=09=09=09=09=09=09=09=09subxids,=20InDoubtGlobalCSN);=0A= +=09}=0A+=0A+=09/*=20Nothing=20to=20write=20if=20we=20don't=20heve=20xid=20= */=0A+=0A+=09return=20GlobalSnapshotGenerate(false);=0A+}=0A+=0A+/*=0A+=20= *=20GlobalSnapshotAssignCsnCurrent=0A+=20*=0A+=20*=20Asign=20GlobalCSN=20= for=20currently=20active=20transaction.=20GlobalCSN=20is=20supposedly=0A= +=20*=20maximal=20among=20of=20values=20returned=20by=20= GlobalSnapshotPrepareCurrent=20and=0A+=20*=20pg_global_snaphot_prepare.=0A= +=20*/=0A+void=0A+GlobalSnapshotAssignCsnCurrent(GlobalCSN=20global_csn)=0A= +{=0A+=09if=20(!track_global_snapshots)=0A+=09=09ereport(ERROR,=0A+=09=09= =09(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09=09= errmsg("could=20not=20prepare=20transaction=20for=20global=20commit"),=0A= +=09=09=09=09errhint("Make=20sure=20the=20configuration=20parameter=20= \"%s\"=20is=20enabled.",=0A+=09=09=09=09=09=09= "track_global_snapshots")));=0A+=0A+=09if=20= (!GlobalCSNIsNormal(global_csn))=0A+=09=09ereport(ERROR,=0A+=09=09=09=09= (errcode(ERRCODE_INVALID_PARAMETER_VALUE),=0A+=09=09=09=09=20= errmsg("pg_global_snaphot_assign=20expects=20normal=20global_csn")));=0A= +=0A+=09/*=20Skip=20emtpty=20transactions=20*/=0A+=09if=20= (!TransactionIdIsValid(GetCurrentTransactionIdIfAny()))=0A+=09=09return;=0A= +=0A+=09/*=20Set=20global_csn=20and=20defuse=20ProcArrayEndTransaction=20= from=20assigning=20one=20*/=0A+=09= pg_atomic_write_u64(&MyProc->assignedGlobalCsn,=20global_csn);=0A+}=0A+=0A= +=0A= +/************************************************************************= *****=0A+=20*=20Functions=20to=20handle=20global=20and=20local=20= transactions=20commit.=0A+=20*=0A+=20*=20For=20local=20transactions=20= GlobalSnapshotPrecommit=20sets=20InDoubt=20state=20before=0A+=20*=20= ProcArrayEndTransaction=20is=20called=20and=20transaction=20data=20= potetntially=20becomes=0A+=20*=20visible=20to=20other=20backends.=20= ProcArrayEndTransaction=20(or=20ProcArrayRemove=20in=0A+=20*=20twophase=20= case)=20then=20acquires=20global_csn=20under=20ProcArray=20lock=20and=20= stores=20it=0A+=20*=20in=20proc->assignedGlobalCsn.=20It's=20important=20= that=20global_csn=20for=20commit=20is=0A+=20*=20generated=20under=20= ProcArray=20lock,=20otherwise=20global=20and=20local=20snapshots=20won't=0A= +=20*=20be=20equivalent.=20Consequent=20call=20to=20GlobalSnapshotCommit=20= will=20write=0A+=20*=20proc->assignedGlobalCsn=20to=20GlobalCSNLog.=0A+=20= *=0A+=20*=20Same=20rules=20applies=20to=20global=20transaction,=20except=20= that=20global_csn=20is=20already=0A+=20*=20assigned=20by=20= GlobalSnapshotAssignCsnCurrent/pg_global_snaphot_assign=20and=0A+=20*=20= GlobalSnapshotPrecommit=20is=20basically=20no-op.=0A+=20*=0A+=20*=20= GlobalSnapshotAbort=20is=20slightly=20different=20comparing=20to=20= commit=20because=20abort=0A+=20*=20can=20skip=20InDoubt=20phase=20and=20= can=20be=20called=20for=20transaction=20subtree.=0A+=20= **************************************************************************= ***/=0A+=0A+=0A+/*=0A+=20*=20GlobalSnapshotAbort=0A+=20*=0A+=20*=20Abort=20= transaction=20in=20GlobalCsnLog.=20We=20can=20skip=20InDoubt=20state=20= for=20aborts=0A+=20*=20since=20no=20concurrent=20transactions=20allowed=20= to=20see=20aborted=20data=20anyway.=0A+=20*/=0A+void=0A= +GlobalSnapshotAbort(PGPROC=20*proc,=20TransactionId=20xid,=0A+=09=09=09=09= =09int=20nsubxids,=20TransactionId=20*subxids)=0A+{=0A+=09if=20= (!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09= GlobalCSNLogSetCSN(xid,=20nsubxids,=20subxids,=20AbortedGlobalCSN);=0A+=0A= +=09/*=0A+=09=20*=20Clean=20assignedGlobalCsn=20anyway,=20as=20it=20was=20= possibly=20set=20in=0A+=09=20*=20GlobalSnapshotAssignCsnCurrent.=0A+=09=20= */=0A+=09pg_atomic_write_u64(&proc->assignedGlobalCsn,=20= InProgressGlobalCSN);=0A+}=0A+=0A+/*=0A+=20*=20GlobalSnapshotPrecommit=0A= +=20*=0A+=20*=20Set=20InDoubt=20status=20for=20local=20transaction=20= that=20we=20are=20going=20to=20commit.=0A+=20*=20This=20step=20is=20= needed=20to=20achieve=20consistency=20between=20local=20snapshots=20and=0A= +=20*=20global=20csn-based=20snapshots.=20We=20don't=20hold=20ProcArray=20= lock=20while=20writing=0A+=20*=20csn=20for=20transaction=20in=20SLRU=20= but=20instead=20we=20set=20InDoubt=20status=20before=0A+=20*=20= transaction=20is=20deleted=20from=20ProcArray=20so=20the=20readers=20who=20= will=20read=20csn=0A+=20*=20in=20the=20gap=20between=20ProcArray=20= removal=20and=20GlobalCSN=20assignment=20can=20wait=0A+=20*=20until=20= GlobalCSN=20is=20finally=20assigned.=20See=20also=20= TransactionIdGetGlobalCSN().=0A+=20*=0A+=20*=20For=20global=20= transaction=20this=20does=20nothing=20as=20InDoubt=20state=20was=20= written=0A+=20*=20earlier.=0A+=20*=0A+=20*=20This=20should=20be=20called=20= only=20from=20parallel=20group=20leader=20before=20backend=20is=0A+=20*=20= deleted=20from=20ProcArray.=0A+=20*/=0A+void=0A= +GlobalSnapshotPrecommit(PGPROC=20*proc,=20TransactionId=20xid,=0A+=09=09= =09=09=09int=20nsubxids,=20TransactionId=20*subxids)=0A+{=0A+=09= GlobalCSN=20oldAssignedGlobalCsn=20=3D=20InProgressGlobalCSN;=0A+=09bool=20= in_progress;=0A+=0A+=09if=20(!track_global_snapshots)=0A+=09=09return;=0A= +=0A+=09/*=20Set=20InDoubt=20status=20if=20it=20is=20local=20transaction=20= */=0A+=09in_progress=20=3D=20= pg_atomic_compare_exchange_u64(&proc->assignedGlobalCsn,=0A+=09=09=09=09=09= =09=09=09=09=09=09=09=20&oldAssignedGlobalCsn,=0A+=09=09=09=09=09=09=09=09= =09=09=09=09=20InDoubtGlobalCSN);=0A+=09if=20(in_progress)=0A+=09{=0A+=09= =09Assert(GlobalCSNIsInProgress(oldAssignedGlobalCsn));=0A+=09=09= GlobalCSNLogSetCSN(xid,=20nsubxids,=0A+=09=09=09=09=09=09=20=20=20= subxids,=20InDoubtGlobalCSN);=0A+=09}=0A+=09else=0A+=09{=0A+=09=09/*=20= Otherwise=20we=20should=20have=20valid=20GlobalCSN=20by=20this=20time=20= */=0A+=09=09Assert(GlobalCSNIsNormal(oldAssignedGlobalCsn));=0A+=09=09/*=20= Also=20global=20transaction=20should=20already=20be=20in=20InDoubt=20= state=20*/=0A+=09=09Assert(GlobalCSNIsInDoubt(GlobalCSNLogGetCSN(xid)));=0A= +=09}=0A+}=0A+=0A+/*=0A+=20*=20GlobalSnapshotCommit=0A+=20*=0A+=20*=20= Write=20GlobalCSN=20that=20were=20acquired=20earlier=20to=20= GlobalCsnLog.=20Should=20be=0A+=20*=20preceded=20by=20= GlobalSnapshotPrecommit()=20so=20readers=20can=20wait=20until=20we=20= finally=0A+=20*=20finished=20writing=20to=20SLRU.=0A+=20*=0A+=20*=20= Should=20be=20called=20after=20ProcArrayEndTransaction,=20but=20before=20= releasing=0A+=20*=20transaction=20locks,=20so=20that=20= TransactionIdGetGlobalCSN=20can=20wait=20on=20this=0A+=20*=20lock=20for=20= GlobalCSN.=0A+=20*/=0A+void=0A+GlobalSnapshotCommit(PGPROC=20*proc,=20= TransactionId=20xid,=0A+=09=09=09=09=09int=20nsubxids,=20TransactionId=20= *subxids)=0A+{=0A+=09volatile=20GlobalCSN=20assigned_global_csn;=0A+=0A+=09= if=20(!track_global_snapshots)=0A+=09=09return;=0A+=0A+=09if=20= (!TransactionIdIsValid(xid))=0A+=09{=0A+=09=09assigned_global_csn=20=3D=20= pg_atomic_read_u64(&proc->assignedGlobalCsn);=0A+=09=09= Assert(GlobalCSNIsInProgress(assigned_global_csn));=0A+=09=09return;=0A+=09= }=0A+=0A+=09/*=20Finally=20write=20resulting=20GlobalCSN=20in=20SLRU=20= */=0A+=09assigned_global_csn=20=3D=20= pg_atomic_read_u64(&proc->assignedGlobalCsn);=0A+=09= Assert(GlobalCSNIsNormal(assigned_global_csn));=0A+=09= GlobalCSNLogSetCSN(xid,=20nsubxids,=0A+=09=09=09=09=09=09=20=20=20= subxids,=20assigned_global_csn);=0A+=0A+=09/*=20Reset=20for=20next=20= transaction=20*/=0A+=09pg_atomic_write_u64(&proc->assignedGlobalCsn,=20= InProgressGlobalCSN);=0A+}=0Adiff=20--git=20= a/src/backend/access/transam/twophase.c=20= b/src/backend/access/transam/twophase.c=0Aindex=20a144e2d6a9..f3fa3283f8=20= 100644=0A---=20a/src/backend/access/transam/twophase.c=0A+++=20= b/src/backend/access/transam/twophase.c=0A@@=20-77,6=20+77,7=20@@=0A=20= #include=20=0A=20=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_snapshot.h"=0A=20#include=20= "access/global_csn_log.h"=0A=20#include=20"access/htup_details.h"=0A=20= #include=20"access/subtrans.h"=0A@@=20-1508,8=20+1509,34=20@@=20= FinishPreparedTransaction(const=20char=20*gid,=20bool=20isCommit)=0A=20=09= =09=09=09=09=09=09=09=09=20=20=20hdr->nabortrels,=20abortrels,=0A=20=09=09= =09=09=09=09=09=09=09=20=20=20gid);=0A=20=0A+=09/*=0A+=09=20*=20= GlobalSnapshot=20callbacks=20that=20should=20be=20called=20right=20= before=20we=20are=0A+=09=20*=20going=20to=20become=20visible.=20Details=20= in=20comments=20to=20this=20functions.=0A+=09=20*/=0A+=09if=20(isCommit)=0A= +=09=09GlobalSnapshotPrecommit(proc,=20xid,=20hdr->nsubxacts,=20= children);=0A+=09else=0A+=09=09GlobalSnapshotAbort(proc,=20xid,=20= hdr->nsubxacts,=20children);=0A+=0A+=0A=20=09ProcArrayRemove(proc,=20= latestXid);=0A=20=0A+=09/*=0A+=09=20*=20Stamp=20our=20transaction=20with=20= GlobalCSN=20in=20GlobalCsnLog.=0A+=09=20*=20Should=20be=20called=20after=20= ProcArrayEndTransaction,=20but=20before=20releasing=0A+=09=20*=20= transaction=20locks,=20since=20TransactionIdGetGlobalCSN=20relies=20on=0A= +=09=20*=20XactLockTableWait=20to=20await=20global_csn.=0A+=09=20*/=0A+=09= if=20(isCommit)=0A+=09{=0A+=09=09GlobalSnapshotCommit(proc,=20xid,=20= hdr->nsubxacts,=20children);=0A+=09}=0A+=09else=0A+=09{=0A+=09=09= GlobalCSN=20global_csn=20=3D=20= pg_atomic_read_u64(&proc->assignedGlobalCsn);=0A+=09=09= Assert(GlobalCSNIsInProgress(global_csn));=0A+=09}=0A+=0A=20=09/*=0A=20=09= =20*=20In=20case=20we=20fail=20while=20running=20the=20callbacks,=20mark=20= the=20gxact=20invalid=20so=0A=20=09=20*=20no=20one=20else=20will=20try=20= to=20commit/rollback,=20and=20so=20it=20will=20be=20recycled=20if=0A@@=20= -2486,3=20+2513,132=20@@=20PrepareRedoRemove(TransactionId=20xid,=20bool=20= giveWarning)=0A=20=0A=20=09return;=0A=20}=0A+=0A+/*=0A+=20*=20= GlobalSnapshotPrepareTwophase=0A+=20*=0A+=20*=20Set=20InDoubt=20state=20= for=20currently=20active=20transaction=20and=20return=20commit's=0A+=20*=20= global=20snapshot.=0A+=20*=0A+=20*=20This=20function=20is=20a=20= counterpart=20of=20GlobalSnapshotPrepareCurrent()=20for=0A+=20*=20= twophase=20transactions.=0A+=20*/=0A+static=20GlobalCSN=0A= +GlobalSnapshotPrepareTwophase(const=20char=20*gid)=0A+{=0A+=09= GlobalTransaction=20gxact;=0A+=09PGXACT=09=20=20=20*pgxact;=0A+=09char=09= =20=20=20*buf;=0A+=09TransactionId=20xid;=0A+=09xl_xact_parsed_prepare=20= parsed;=0A+=0A+=09if=20(!track_global_snapshots)=0A+=09=09ereport(ERROR,=0A= +=09=09=09(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09= =09errmsg("could=20not=20prepare=20transaction=20for=20global=20= commit"),=0A+=09=09=09=09errhint("Make=20sure=20the=20configuration=20= parameter=20\"%s\"=20is=20enabled.",=0A+=09=09=09=09=09=09= "track_global_snapshots")));=0A+=0A+=09/*=0A+=09=20*=20Validate=20the=20= GID,=20and=20lock=20the=20GXACT=20to=20ensure=20that=20two=20backends=20= do=20not=0A+=09=20*=20try=20to=20access=20the=20same=20GID=20at=20once.=0A= +=09=20*/=0A+=09gxact=20=3D=20LockGXact(gid,=20GetUserId());=0A+=09= pgxact=20=3D=20&ProcGlobal->allPgXact[gxact->pgprocno];=0A+=09xid=20=3D=20= pgxact->xid;=0A+=0A+=09if=20(gxact->ondisk)=0A+=09=09buf=20=3D=20= ReadTwoPhaseFile(xid,=20true);=0A+=09else=0A+=09=09= XlogReadTwoPhaseData(gxact->prepare_start_lsn,=20&buf,=20NULL);=0A+=0A+=09= ParsePrepareRecord(0,=20buf,=20&parsed);=0A+=0A+=09= GlobalCSNLogSetCSN(xid,=20parsed.nsubxacts,=0A+=09=09=09=09=09= parsed.subxacts,=20InDoubtGlobalCSN);=0A+=0A+=09/*=20Unlock=20our=20= GXACT=20*/=0A+=09LWLockAcquire(TwoPhaseStateLock,=20LW_EXCLUSIVE);=0A+=09= gxact->locking_backend=20=3D=20InvalidBackendId;=0A+=09= LWLockRelease(TwoPhaseStateLock);=0A+=0A+=09pfree(buf);=0A+=0A+=09return=20= GlobalSnapshotGenerate(false);=0A+}=0A+=0A+/*=0A+=20*=20SQL=20interface=20= to=20GlobalSnapshotPrepareTwophase()=0A+=20*=0A+=20*=20TODO:=20Rewrite=20= this=20as=20PREPARE=20TRANSACTION=20'gid'=20RETURNING=20SNAPSHOT=0A+=20= */=0A+Datum=0A+pg_global_snaphot_prepare(PG_FUNCTION_ARGS)=0A+{=0A+=09= const=20char=20*gid=20=3D=20text_to_cstring(PG_GETARG_TEXT_PP(0));=0A+=09= GlobalCSN=09global_csn;=0A+=0A+=09global_csn=20=3D=20= GlobalSnapshotPrepareTwophase(gid);=0A+=0A+=09= PG_RETURN_INT64(global_csn);=0A+}=0A+=0A+=0A+/*=0A+=20*=20= TwoPhaseAssignGlobalCsn=0A+=20*=0A+=20*=20Asign=20GlobalCSN=20for=20= currently=20active=20transaction.=20GlobalCSN=20is=20supposedly=0A+=20*=20= maximal=20among=20of=20values=20returned=20by=20= GlobalSnapshotPrepareCurrent=20and=0A+=20*=20pg_global_snaphot_prepare.=0A= +=20*=0A+=20*=20This=20function=20is=20a=20counterpart=20of=20= GlobalSnapshotAssignCsnCurrent()=20for=0A+=20*=20twophase=20= transactions.=0A+=20*/=0A+static=20void=0A= +GlobalSnapshotAssignCsnTwoPhase(const=20char=20*gid,=20GlobalCSN=20= global_csn)=0A+{=0A+=09GlobalTransaction=20gxact;=0A+=09PGPROC=09=20=20=20= *proc;=0A+=0A+=09if=20(!track_global_snapshots)=0A+=09=09ereport(ERROR,=0A= +=09=09=09(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09= =09errmsg("could=20not=20prepare=20transaction=20for=20global=20= commit"),=0A+=09=09=09=09errhint("Make=20sure=20the=20configuration=20= parameter=20\"%s\"=20is=20enabled.",=0A+=09=09=09=09=09=09= "track_global_snapshots")));=0A+=0A+=09if=20= (!GlobalCSNIsNormal(global_csn))=0A+=09=09ereport(ERROR,=0A+=09=09=09=09= (errcode(ERRCODE_INVALID_PARAMETER_VALUE),=0A+=09=09=09=09=20= errmsg("pg_global_snaphot_assign=20expects=20normal=20global_csn")));=0A= +=0A+=09/*=0A+=09=20*=20Validate=20the=20GID,=20and=20lock=20the=20GXACT=20= to=20ensure=20that=20two=20backends=20do=20not=0A+=09=20*=20try=20to=20= access=20the=20same=20GID=20at=20once.=0A+=09=20*/=0A+=09gxact=20=3D=20= LockGXact(gid,=20GetUserId());=0A+=09proc=20=3D=20= &ProcGlobal->allProcs[gxact->pgprocno];=0A+=0A+=09/*=20Set=20global_csn=20= and=20defuse=20ProcArrayRemove=20from=20assigning=20one.=20*/=0A+=09= pg_atomic_write_u64(&proc->assignedGlobalCsn,=20global_csn);=0A+=0A+=09= /*=20Unlock=20our=20GXACT=20*/=0A+=09LWLockAcquire(TwoPhaseStateLock,=20= LW_EXCLUSIVE);=0A+=09gxact->locking_backend=20=3D=20InvalidBackendId;=0A= +=09LWLockRelease(TwoPhaseStateLock);=0A+}=0A+=0A+/*=0A+=20*=20SQL=20= interface=20to=20GlobalSnapshotAssignCsnTwoPhase()=0A+=20*=0A+=20*=20= TODO:=20Rewrite=20this=20as=20COMMIT=20PREPARED=20'gid'=20SNAPSHOT=20= 'global_csn'=0A+=20*/=0A+Datum=0A= +pg_global_snaphot_assign(PG_FUNCTION_ARGS)=0A+{=0A+=09const=20char=20= *gid=20=3D=20text_to_cstring(PG_GETARG_TEXT_PP(0));=0A+=09GlobalCSN=09= global_csn=20=3D=20PG_GETARG_INT64(1);=0A+=0A+=09= GlobalSnapshotAssignCsnTwoPhase(gid,=20global_csn);=0A+=09= PG_RETURN_VOID();=0A+}=0Adiff=20--git=20= a/src/backend/access/transam/xact.c=20= b/src/backend/access/transam/xact.c=0Aindex=20f4e5ea84b9..44d57e2031=20= 100644=0A---=20a/src/backend/access/transam/xact.c=0A+++=20= b/src/backend/access/transam/xact.c=0A@@=20-21,6=20+21,7=20@@=0A=20= #include=20=0A=20=0A=20#include=20"access/commit_ts.h"=0A= +#include=20"access/global_snapshot.h"=0A=20#include=20= "access/multixact.h"=0A=20#include=20"access/parallel.h"=0A=20#include=20= "access/subtrans.h"=0A@@=20-1341,6=20+1342,14=20@@=20= RecordTransactionCommit(void)=0A=20=0A=20=09/*=20Reset=20XactLastRecEnd=20= until=20the=20next=20transaction=20writes=20something=20*/=0A=20=09= XactLastRecEnd=20=3D=200;=0A+=0A+=09/*=0A+=09=20*=20Mark=20our=20= transaction=20as=20InDoubt=20in=20GlobalCsnLog=20and=20get=20ready=20for=0A= +=09=20*=20commit.=0A+=09=20*/=0A+=09if=20(markXidCommitted)=0A+=09=09= GlobalSnapshotPrecommit(MyProc,=20xid,=20nchildren,=20children);=0A+=0A=20= cleanup:=0A=20=09/*=20Clean=20up=20local=20data=20*/=0A=20=09if=20(rels)=0A= @@=20-1602,6=20+1611,11=20@@=20RecordTransactionAbort(bool=20isSubXact)=0A= =20=09=20*/=0A=20=09TransactionIdAbortTree(xid,=20nchildren,=20= children);=0A=20=0A+=09/*=0A+=09=20*=20Mark=20our=20transaction=20as=20= Aborted=20in=20GlobalCsnLog.=0A+=09=20*/=0A+=09= GlobalSnapshotAbort(MyProc,=20xid,=20nchildren,=20children);=0A+=0A=20=09= END_CRIT_SECTION();=0A=20=0A=20=09/*=20Compute=20latestXid=20while=20we=20= have=20the=20child=20XIDs=20handy=20*/=0A@@=20-2059,6=20+2073,21=20@@=20= CommitTransaction(void)=0A=20=09=20*/=0A=20=09= ProcArrayEndTransaction(MyProc,=20latestXid);=0A=20=0A+=09/*=0A+=09=20*=20= Stamp=20our=20transaction=20with=20GlobalCSN=20in=20GlobalCsnLog.=0A+=09=20= *=20Should=20be=20called=20after=20ProcArrayEndTransaction,=20but=20= before=20releasing=0A+=09=20*=20transaction=20locks.=0A+=09=20*/=0A+=09= if=20(!is_parallel_worker)=0A+=09{=0A+=09=09TransactionId=20=20xid=20=3D=20= GetTopTransactionIdIfAny();=0A+=09=09TransactionId=20*subxids;=0A+=09=09= int=09=09=09=20=20=20nsubxids;=0A+=0A+=09=09nsubxids=20=3D=20= xactGetCommittedChildren(&subxids);=0A+=09=09= GlobalSnapshotCommit(MyProc,=20xid,=20nsubxids,=20subxids);=0A+=09}=0A+=0A= =20=09/*=0A=20=09=20*=20This=20is=20all=20post-commit=20cleanup.=20=20= Note=20that=20if=20an=20error=20is=20raised=20here,=0A=20=09=20*=20it's=20= too=20late=20to=20abort=20the=20transaction.=20=20This=20should=20be=20= just=0Adiff=20--git=20a/src/backend/access/transam/xlog.c=20= b/src/backend/access/transam/xlog.c=0Aindex=20749b0c1ce2..71151fc3df=20= 100644=0A---=20a/src/backend/access/transam/xlog.c=0A+++=20= b/src/backend/access/transam/xlog.c=0A@@=20-6967,6=20+6967,7=20@@=20= StartupXLOG(void)=0A=20=09=09=09StartupCLOG();=0A=20=09=09=09= StartupGlobalCSNLog(oldestActiveXID);=0A=20=09=09=09= StartupSUBTRANS(oldestActiveXID);=0A+=09=09=09= GlobalSnapshotStartup(oldestActiveXID);=0A=20=0A=20=09=09=09/*=0A=20=09=09= =09=20*=20If=20we're=20beginning=20at=20a=20shutdown=20checkpoint,=20we=20= know=20that=0A@@=20-7753,6=20+7754,7=20@@=20StartupXLOG(void)=0A=20=09=09= StartupCLOG();=0A=20=09=09StartupGlobalCSNLog(oldestActiveXID);=0A=20=09=09= StartupSUBTRANS(oldestActiveXID);=0A+=09=09= GlobalSnapshotStartup(oldestActiveXID);=0A=20=09}=0A=20=0A=20=09/*=0A= diff=20--git=20a/src/backend/storage/ipc/ipci.c=20= b/src/backend/storage/ipc/ipci.c=0Aindex=202af468fc6a..3a04f6a824=20= 100644=0A---=20a/src/backend/storage/ipc/ipci.c=0A+++=20= b/src/backend/storage/ipc/ipci.c=0A@@=20-22,6=20+22,7=20@@=0A=20#include=20= "access/nbtree.h"=0A=20#include=20"access/subtrans.h"=0A=20#include=20= "access/twophase.h"=0A+#include=20"access/global_snapshot.h"=0A=20= #include=20"commands/async.h"=0A=20#include=20"miscadmin.h"=0A=20= #include=20"pgstat.h"=0A@@=20-147,6=20+148,7=20@@=20= CreateSharedMemoryAndSemaphores(bool=20makePrivate,=20int=20port)=0A=20=09= =09size=20=3D=20add_size(size,=20WalSndShmemSize());=0A=20=09=09size=20=3D= =20add_size(size,=20WalRcvShmemSize());=0A=20=09=09size=20=3D=20= add_size(size,=20ApplyLauncherShmemSize());=0A+=09=09size=20=3D=20= add_size(size,=20GlobalSnapshotShmemSize());=0A=20=09=09size=20=3D=20= add_size(size,=20SnapMgrShmemSize());=0A=20=09=09size=20=3D=20= add_size(size,=20BTreeShmemSize());=0A=20=09=09size=20=3D=20= add_size(size,=20SyncScanShmemSize());=0A@@=20-273,6=20+275,7=20@@=20= CreateSharedMemoryAndSemaphores(bool=20makePrivate,=20int=20port)=0A=20=09= SyncScanShmemInit();=0A=20=09AsyncShmemInit();=0A=20=09= BackendRandomShmemInit();=0A+=09GlobalSnapshotShmemInit();=0A=20=0A=20= #ifdef=20EXEC_BACKEND=0A=20=0Adiff=20--git=20= a/src/backend/storage/ipc/procarray.c=20= b/src/backend/storage/ipc/procarray.c=0Aindex=20bc59fca21c..c6cb63fe97=20= 100644=0A---=20a/src/backend/storage/ipc/procarray.c=0A+++=20= b/src/backend/storage/ipc/procarray.c=0A@@=20-47,6=20+47,7=20@@=0A=20=0A=20= #include=20"access/clog.h"=0A=20#include=20"access/global_csn_log.h"=0A= +#include=20"access/global_snapshot.h"=0A=20#include=20= "access/subtrans.h"=0A=20#include=20"access/transam.h"=0A=20#include=20= "access/twophase.h"=0A@@=20-91,6=20+92,8=20@@=20typedef=20struct=20= ProcArrayStruct=0A=20=09TransactionId=20replication_slot_xmin;=0A=20=09= /*=20oldest=20catalog=20xmin=20of=20any=20replication=20slot=20*/=0A=20=09= TransactionId=20replication_slot_catalog_xmin;=0A+=09/*=20xmin=20of=20= oldest=20active=20global=20snapshot=20*/=0A+=09TransactionId=20= global_snapshot_xmin;=0A=20=0A=20=09/*=20indexes=20into=20allPgXact[],=20= has=20PROCARRAY_MAXPROCS=20entries=20*/=0A=20=09int=09=09=09= pgprocnos[FLEXIBLE_ARRAY_MEMBER];=0A@@=20-246,6=20+249,7=20@@=20= CreateSharedProcArray(void)=0A=20=09=09procArray->lastOverflowedXid=20=3D=20= InvalidTransactionId;=0A=20=09=09procArray->replication_slot_xmin=20=3D=20= InvalidTransactionId;=0A=20=09=09= procArray->replication_slot_catalog_xmin=20=3D=20InvalidTransactionId;=0A= +=09=09procArray->global_snapshot_xmin=20=3D=20InvalidTransactionId;=0A=20= =09}=0A=20=0A=20=09allProcs=20=3D=20ProcGlobal->allProcs;=0A@@=20-352,6=20= +356,17=20@@=20ProcArrayRemove(PGPROC=20*proc,=20TransactionId=20= latestXid)=0A=20=09=09if=20= (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,=0A=20=09=09= =09=09=09=09=09=09=20=20latestXid))=0A=20=09=09=09= ShmemVariableCache->latestCompletedXid=20=3D=20latestXid;=0A+=0A+=09=09= /*=0A+=09=09=20*=20Assign=20global=20csn=20while=20holding=20= ProcArrayLock=20for=20non-global=0A+=09=09=20*=20COMMIT=20PREPARED.=20= After=20lock=20is=20released=20consequent=0A+=09=09=20*=20= GlobalSnapshotCommit()=20will=20write=20this=20value=20to=20= GlobalCsnLog.=0A+=09=09=20*=0A+=09=09=20*=20In=20case=20of=20global=20= commit=20proc->assignedGlobalCsn=20is=20already=20set=0A+=09=09=20*=20by=20= prior=20AssignGlobalCsn().=0A+=09=09=20*/=0A+=09=09if=20= (GlobalCSNIsInDoubt(pg_atomic_read_u64(&proc->assignedGlobalCsn)))=0A+=09= =09=09pg_atomic_write_u64(&proc->assignedGlobalCsn,=20= GlobalSnapshotGenerate(false));=0A=20=09}=0A=20=09else=0A=20=09{=0A@@=20= -432,6=20+447,8=20@@=20ProcArrayEndTransaction(PGPROC=20*proc,=20= TransactionId=20latestXid)=0A=20=0A=20=09=09proc->lxid=20=3D=20= InvalidLocalTransactionId;=0A=20=09=09pgxact->xmin=20=3D=20= InvalidTransactionId;=0A+=09=09proc->originalXmin=20=3D=20= InvalidTransactionId;=0A+=0A=20=09=09/*=20must=20be=20cleared=20with=20= xid/xmin:=20*/=0A=20=09=09pgxact->vacuumFlags=20&=3D=20= ~PROC_VACUUM_STATE_MASK;=0A=20=09=09pgxact->delayChkpt=20=3D=20false;=20= /*=20be=20sure=20this=20is=20cleared=20in=20abort=20*/=0A@@=20-454,6=20= +471,8=20@@=20ProcArrayEndTransactionInternal(PGPROC=20*proc,=20PGXACT=20= *pgxact,=0A=20=09pgxact->xid=20=3D=20InvalidTransactionId;=0A=20=09= proc->lxid=20=3D=20InvalidLocalTransactionId;=0A=20=09pgxact->xmin=20=3D=20= InvalidTransactionId;=0A+=09proc->originalXmin=20=3D=20= InvalidTransactionId;=0A+=0A=20=09/*=20must=20be=20cleared=20with=20= xid/xmin:=20*/=0A=20=09pgxact->vacuumFlags=20&=3D=20= ~PROC_VACUUM_STATE_MASK;=0A=20=09pgxact->delayChkpt=20=3D=20false;=20/*=20= be=20sure=20this=20is=20cleared=20in=20abort=20*/=0A@@=20-467,6=20= +486,20=20@@=20ProcArrayEndTransactionInternal(PGPROC=20*proc,=20PGXACT=20= *pgxact,=0A=20=09if=20= (TransactionIdPrecedes(ShmemVariableCache->latestCompletedXid,=0A=20=09=09= =09=09=09=09=09=20=20latestXid))=0A=20=09=09= ShmemVariableCache->latestCompletedXid=20=3D=20latestXid;=0A+=0A+=09/*=0A= +=09=20*=20Assign=20global=20csn=20while=20holding=20ProcArrayLock=20for=20= non-global=0A+=09=20*=20COMMIT.=20After=20lock=20is=20released=20= consequent=20GlobalSnapshotFinish()=20will=0A+=09=20*=20write=20this=20= value=20to=20GlobalCsnLog.=0A+=09=20*=0A+=09=20*=20In=20case=20of=20= global=20commit=20MyProc->assignedGlobalCsn=20is=20already=20set=0A+=09=20= *=20by=20prior=20AssignGlobalCsn().=0A+=09=20*=0A+=09=20*=20TODO:=20in=20= case=20of=20group=20commit=20we=20can=20generate=20one=20GlobalSnapshot=20= for=0A+=09=20*=20whole=20group=20to=20save=20time=20on=20timestamp=20= aquisition.=0A+=09=20*/=0A+=09if=20= (GlobalCSNIsInDoubt(pg_atomic_read_u64(&proc->assignedGlobalCsn)))=0A+=09= =09pg_atomic_write_u64(&proc->assignedGlobalCsn,=20= GlobalSnapshotGenerate(false));=0A=20}=0A=20=0A=20/*=0A@@=20-616,6=20= +649,7=20@@=20ProcArrayClearTransaction(PGPROC=20*proc)=0A=20=09= pgxact->xid=20=3D=20InvalidTransactionId;=0A=20=09proc->lxid=20=3D=20= InvalidLocalTransactionId;=0A=20=09pgxact->xmin=20=3D=20= InvalidTransactionId;=0A+=09proc->originalXmin=20=3D=20= InvalidTransactionId;=0A=20=09proc->recoveryConflictPending=20=3D=20= false;=0A=20=0A=20=09/*=20redundant,=20but=20just=20in=20case=20*/=0A@@=20= -1323,6=20+1357,7=20@@=20GetOldestXmin(Relation=20rel,=20int=20flags)=0A=20= =0A=20=09volatile=20TransactionId=20replication_slot_xmin=20=3D=20= InvalidTransactionId;=0A=20=09volatile=20TransactionId=20= replication_slot_catalog_xmin=20=3D=20InvalidTransactionId;=0A+=09= volatile=20TransactionId=20global_snapshot_xmin=20=3D=20= InvalidTransactionId;=0A=20=0A=20=09/*=0A=20=09=20*=20If=20we're=20not=20= computing=20a=20relation=20specific=20limit,=20or=20if=20a=20shared=0A@@=20= -1359,8=20+1394,9=20@@=20GetOldestXmin(Relation=20rel,=20int=20flags)=0A=20= =09=09=09proc->databaseId=20=3D=3D=20MyDatabaseId=20||=0A=20=09=09=09= proc->databaseId=20=3D=3D=200)=09/*=20always=20include=20WalSender=20*/=0A= =20=09=09{=0A-=09=09=09/*=20Fetch=20xid=20just=20once=20-=20see=20= GetNewTransactionId=20*/=0A+=09=09=09/*=20Fetch=20both=20xids=20just=20= once=20-=20see=20GetNewTransactionId=20*/=0A=20=09=09=09TransactionId=20= xid=20=3D=20pgxact->xid;=0A+=09=09=09TransactionId=20original_xmin=20=3D=20= proc->originalXmin;=0A=20=0A=20=09=09=09/*=20First=20consider=20the=20= transaction's=20own=20Xid,=20if=20any=20*/=0A=20=09=09=09if=20= (TransactionIdIsNormal(xid)=20&&=0A@@=20-1373,8=20+1409,17=20@@=20= GetOldestXmin(Relation=20rel,=20int=20flags)=0A=20=09=09=09=20*=20We=20= must=20check=20both=20Xid=20and=20Xmin=20because=20a=20transaction=20= might=0A=20=09=09=09=20*=20have=20an=20Xmin=20but=20not=20(yet)=20an=20= Xid;=20conversely,=20if=20it=20has=20an=0A=20=09=09=09=20*=20Xid,=20that=20= could=20determine=20some=20not-yet-set=20Xmin.=0A+=09=09=09=20*=0A+=09=09= =09=20*=20In=20case=20of=20oldestXmin=20calculation=20for=20= GlobalSnapshotMapXmin()=0A+=09=09=09=20*=20pgxact->xmin=20should=20be=20= changed=20to=20proc->originalXmin.=20Details=0A+=09=09=09=20*=20in=20= commets=20to=20GlobalSnapshotMapXmin.=0A=20=09=09=09=20*/=0A-=09=09=09= xid=20=3D=20pgxact->xmin;=20/*=20Fetch=20just=20once=20*/=0A+=09=09=09if=20= ((flags=20&=20PROCARRAY_NON_IMPORTED_XMIN)=20&&=0A+=09=09=09=09=09= TransactionIdIsValid(original_xmin))=0A+=09=09=09=09xid=20=3D=20= original_xmin;=0A+=09=09=09else=0A+=09=09=09=09xid=20=3D=20pgxact->xmin;=20= /*=20Fetch=20just=20once=20*/=0A+=0A=20=09=09=09if=20= (TransactionIdIsNormal(xid)=20&&=0A=20=09=09=09=09= TransactionIdPrecedes(xid,=20result))=0A=20=09=09=09=09result=20=3D=20= xid;=0A@@=20-1384,6=20+1429,7=20@@=20GetOldestXmin(Relation=20rel,=20int=20= flags)=0A=20=09/*=20fetch=20into=20volatile=20var=20while=20= ProcArrayLock=20is=20held=20*/=0A=20=09replication_slot_xmin=20=3D=20= procArray->replication_slot_xmin;=0A=20=09replication_slot_catalog_xmin=20= =3D=20procArray->replication_slot_catalog_xmin;=0A+=09= global_snapshot_xmin=20=3D=20procArray->global_snapshot_xmin;=0A=20=0A=20= =09if=20(RecoveryInProgress())=0A=20=09{=0A@@=20-1425,6=20+1471,11=20@@=20= GetOldestXmin(Relation=20rel,=20int=20flags)=0A=20=09=09=09result=20=3D=20= FirstNormalTransactionId;=0A=20=09}=0A=20=0A+=09if=20(!(flags=20&=20= PROCARRAY_NON_IMPORTED_XMIN)=20&&=0A+=09=09= TransactionIdIsValid(global_snapshot_xmin)=20&&=0A+=09=09= NormalTransactionIdPrecedes(global_snapshot_xmin,=20result))=0A+=09=09= result=20=3D=20global_snapshot_xmin;=0A+=0A=20=09/*=0A=20=09=20*=20Check=20= whether=20there=20are=20replication=20slots=20requiring=20an=20older=20= xmin.=0A=20=09=20*/=0A@@=20-1520,6=20+1571,8=20@@=20= GetSnapshotData(Snapshot=20snapshot)=0A=20=09bool=09=09suboverflowed=20=3D= =20false;=0A=20=09volatile=20TransactionId=20replication_slot_xmin=20=3D=20= InvalidTransactionId;=0A=20=09volatile=20TransactionId=20= replication_slot_catalog_xmin=20=3D=20InvalidTransactionId;=0A+=09= volatile=20TransactionId=20global_snapshot_xmin=20=3D=20= InvalidTransactionId;=0A+=09volatile=20GlobalCSN=09=20=20=20global_csn=20= =3D=20FrozenGlobalCSN;=0A=20=0A=20=09Assert(snapshot=20!=3D=20NULL);=0A=20= =0A@@=20-1708,10=20+1761,18=20@@=20GetSnapshotData(Snapshot=20snapshot)=0A= =20=09/*=20fetch=20into=20volatile=20var=20while=20ProcArrayLock=20is=20= held=20*/=0A=20=09replication_slot_xmin=20=3D=20= procArray->replication_slot_xmin;=0A=20=09replication_slot_catalog_xmin=20= =3D=20procArray->replication_slot_catalog_xmin;=0A+=09= global_snapshot_xmin=20=3D=20procArray->global_snapshot_xmin;=0A=20=0A=20= =09if=20(!TransactionIdIsValid(MyPgXact->xmin))=0A=20=09=09= MyPgXact->xmin=20=3D=20TransactionXmin=20=3D=20xmin;=0A=20=0A+=09/*=0A+=09= =20*=20Take=20GlobalCSN=20under=20ProcArrayLock=20so=20the=20= local/global=20snapshot=20stays=0A+=09=20*=20synchronized.=0A+=09=20*/=0A= +=09if=20(track_global_snapshots=20||=20global_snapshot_defer_time=20>=20= 0)=0A+=09=09global_csn=20=3D=20GlobalSnapshotGenerate(false);=0A+=0A=20=09= LWLockRelease(ProcArrayLock);=0A=20=0A=20=09/*=0A@@=20-1727,6=20+1788,10=20= @@=20GetSnapshotData(Snapshot=20snapshot)=0A=20=09if=20= (!TransactionIdIsNormal(RecentGlobalXmin))=0A=20=09=09RecentGlobalXmin=20= =3D=20FirstNormalTransactionId;=0A=20=0A+=09if=20= (TransactionIdIsValid(global_snapshot_xmin)=20&&=0A+=09=09= TransactionIdPrecedes(global_snapshot_xmin,=20RecentGlobalXmin))=0A+=09=09= RecentGlobalXmin=20=3D=20global_snapshot_xmin;=0A+=0A=20=09/*=20Check=20= whether=20there's=20a=20replication=20slot=20requiring=20an=20older=20= xmin.=20*/=0A=20=09if=20(TransactionIdIsValid(replication_slot_xmin)=20= &&=0A=20=09=09NormalTransactionIdPrecedes(replication_slot_xmin,=20= RecentGlobalXmin))=0A@@=20-1782,6=20+1847,11=20@@=20= GetSnapshotData(Snapshot=20snapshot)=0A=20=09=09= MaintainOldSnapshotTimeMapping(snapshot->whenTaken,=20xmin);=0A=20=09}=0A= =20=0A+=09snapshot->imported_global_csn=20=3D=20false;=0A+=09= snapshot->global_csn=20=3D=20global_csn;=0A+=09if=20= (global_snapshot_defer_time=20>=200)=0A+=09=09= GlobalSnapshotMapXmin(snapshot->global_csn);=0A+=0A=20=09return=20= snapshot;=0A=20}=0A=20=0A@@=20-2999,6=20+3069,35=20@@=20= ProcArrayGetReplicationSlotXmin(TransactionId=20*xmin,=0A=20=09= LWLockRelease(ProcArrayLock);=0A=20}=0A=20=0A+/*=0A+=20*=20= ProcArraySetGlobalSnapshotXmin=0A+=20*/=0A+void=0A= +ProcArraySetGlobalSnapshotXmin(TransactionId=20xmin)=0A+{=0A+=09= LWLockAcquire(ProcArrayLock,=20LW_SHARED);=0A+=0A+=09if=20= (TransactionIdFollows(xmin,=20procArray->global_snapshot_xmin))=0A+=09=09= procArray->global_snapshot_xmin=20=3D=20xmin;=0A+=0A+=09= LWLockRelease(ProcArrayLock);=0A+}=0A+=0A+/*=0A+=20*=20= ProcArrayGetGlobalSnapshotXmin=0A+=20*/=0A+TransactionId=0A= +ProcArrayGetGlobalSnapshotXmin(bool=20locked)=0A+{=0A+=09volatile=20= TransactionId=20xmin;=0A+=0A+=09if=20(!locked)=0A+=09=09= LWLockAcquire(ProcArrayLock,=20LW_SHARED);=0A+=09xmin=20=3D=20= procArray->global_snapshot_xmin;=0A+=09if=20(!locked)=0A+=09=09= LWLockRelease(ProcArrayLock);=0A+=09return=20xmin;=0A+}=0A=20=0A=20= #define=20XidCacheRemove(i)=20\=0A=20=09do=20{=20\=0Adiff=20--git=20= a/src/backend/storage/lmgr/lwlocknames.txt=20= b/src/backend/storage/lmgr/lwlocknames.txt=0Aindex=20= 9615058f29..8ef731d560=20100644=0A---=20= a/src/backend/storage/lmgr/lwlocknames.txt=0A+++=20= b/src/backend/storage/lmgr/lwlocknames.txt=0A@@=20-51,3=20+51,4=20@@=20= BackendRandomLock=09=09=09=09=0943=0A=20LogicalRepWorkerLock=09=09=09=09= 44=0A=20CLogTruncationLock=09=09=09=09=0945=0A=20GlobalCSNLogControlLock=09= =09=09=0946=0A+GlobalSnapshotXidMapLock=09=09=0947=0Adiff=20--git=20= a/src/backend/storage/lmgr/proc.c=20b/src/backend/storage/lmgr/proc.c=0A= index=206f30e082b2..ed497185be=20100644=0A---=20= a/src/backend/storage/lmgr/proc.c=0A+++=20= b/src/backend/storage/lmgr/proc.c=0A@@=20-37,6=20+37,7=20@@=0A=20=0A=20= #include=20"access/transam.h"=0A=20#include=20"access/twophase.h"=0A= +#include=20"access/global_snapshot.h"=0A=20#include=20"access/xact.h"=0A= =20#include=20"miscadmin.h"=0A=20#include=20"pgstat.h"=0A@@=20-417,6=20= +418,9=20@@=20InitProcess(void)=0A=20=09MyProc->clogGroupMemberLsn=20=3D=20= InvalidXLogRecPtr;=0A=20=09pg_atomic_init_u32(&MyProc->clogGroupNext,=20= INVALID_PGPROCNO);=0A=20=0A+=09MyProc->originalXmin=20=3D=20= InvalidTransactionId;=0A+=09= pg_atomic_init_u64(&MyProc->assignedGlobalCsn,=20InProgressGlobalCSN);=0A= +=0A=20=09/*=0A=20=09=20*=20Acquire=20ownership=20of=20the=20PGPROC's=20= latch,=20so=20that=20we=20can=20use=20WaitLatch=0A=20=09=20*=20on=20it.=20= =20That=20allows=20us=20to=20repoint=20the=20process=20latch,=20which=20= so=20far=0A@@=20-559,6=20+563,7=20@@=20InitAuxiliaryProcess(void)=0A=20=09= MyProc->lwWaitMode=20=3D=200;=0A=20=09MyProc->waitLock=20=3D=20NULL;=0A=20= =09MyProc->waitProcLock=20=3D=20NULL;=0A+=09MyProc->originalXmin=20=3D=20= InvalidTransactionId;=0A=20#ifdef=20USE_ASSERT_CHECKING=0A=20=09{=0A=20=09= =09int=09=09=09i;=0Adiff=20--git=20a/src/backend/utils/misc/guc.c=20= b/src/backend/utils/misc/guc.c=0Aindex=202699a7e6b4..2abc4be86c=20100644=0A= ---=20a/src/backend/utils/misc/guc.c=0A+++=20= b/src/backend/utils/misc/guc.c=0A@@=20-28,6=20+28,7=20@@=0A=20=0A=20= #include=20"access/commit_ts.h"=0A=20#include=20"access/gin.h"=0A= +#include=20"access/global_snapshot.h"=0A=20#include=20"access/rmgr.h"=0A= =20#include=20"access/transam.h"=0A=20#include=20"access/twophase.h"=0A= @@=20-2332,6=20+2333,16=20@@=20static=20struct=20config_int=20= ConfigureNamesInt[]=20=3D=0A=20=09=09NULL,=20NULL,=20NULL=0A=20=09},=0A=20= =0A+=09{=0A+=09=09{"global_snapshot_defer_time",=20PGC_POSTMASTER,=20= REPLICATION_MASTER,=0A+=09=09=09gettext_noop("Minimal=20age=20of=20= records=20which=20allowed=20to=20be=20vacuumed,=20in=20seconds."),=0A+=09= =09=09NULL=0A+=09=09},=0A+=09=09&global_snapshot_defer_time,=0A+=09=090,=20= 0,=20INT_MAX,=0A+=09=09NULL,=20NULL,=20NULL=0A+=09},=0A+=0A=20=09/*=0A=20= =09=20*=20See=20also=20CheckRequiredParameterValues()=20if=20this=20= parameter=20changes=0A=20=09=20*/=0Adiff=20--git=20= a/src/backend/utils/misc/postgresql.conf.sample=20= b/src/backend/utils/misc/postgresql.conf.sample=0Aindex=20= 3d88e80a20..206c1806e0=20100644=0A---=20= a/src/backend/utils/misc/postgresql.conf.sample=0A+++=20= b/src/backend/utils/misc/postgresql.conf.sample=0A@@=20-255,6=20+255,8=20= @@=0A=20=09=09=09=09#=20and=20comma-separated=20list=20of=20= application_name=0A=20=09=09=09=09#=20from=20standby(s);=20'*'=20=3D=20= all=0A=20#vacuum_defer_cleanup_age=20=3D=200=09#=20number=20of=20xacts=20= by=20which=20cleanup=20is=20delayed=0A+#global_snapshot_defer_time=20=3D=20= 0=09#=20minimal=20age=20of=20records=20which=20allowed=20to=20be=0A+=09=09= =09=09#=20vacuumed,=20in=20seconds=0A=20=0A=20#=20-=20Standby=20Servers=20= -=0A=20=0Adiff=20--git=20a/src/backend/utils/time/snapmgr.c=20= b/src/backend/utils/time/snapmgr.c=0Aindex=204b45d3cccd..135a5a6881=20= 100644=0A---=20a/src/backend/utils/time/snapmgr.c=0A+++=20= b/src/backend/utils/time/snapmgr.c=0A@@=20-48,6=20+48,7=20@@=0A=20= #include=20=0A=20#include=20=0A=20=0A+#include=20= "access/global_snapshot.h"=0A=20#include=20"access/transam.h"=0A=20= #include=20"access/xact.h"=0A=20#include=20"access/xlog.h"=0A@@=20-245,6=20= +246,8=20@@=20typedef=20struct=20SerializedSnapshotData=0A=20=09= CommandId=09curcid;=0A=20=09TimestampTz=20whenTaken;=0A=20=09XLogRecPtr=09= lsn;=0A+=09GlobalCSN=09global_csn;=0A+=09bool=09=09imported_global_csn;=0A= =20}=20SerializedSnapshotData;=0A=20=0A=20Size=0A@@=20-992,7=20+995,9=20= @@=20SnapshotResetXmin(void)=0A=20=09=09=09=09=09=09=09=09=09=09= pairingheap_first(&RegisteredSnapshots));=0A=20=0A=20=09if=20= (TransactionIdPrecedes(MyPgXact->xmin,=20minSnapshot->xmin))=0A+=09{=0A=20= =09=09MyPgXact->xmin=20=3D=20minSnapshot->xmin;=0A+=09}=0A=20}=0A=20=0A=20= /*=0A@@=20-2081,6=20+2086,8=20@@=20SerializeSnapshot(Snapshot=20= snapshot,=20char=20*start_address)=0A=20=09serialized_snapshot.curcid=20= =3D=20snapshot->curcid;=0A=20=09serialized_snapshot.whenTaken=20=3D=20= snapshot->whenTaken;=0A=20=09serialized_snapshot.lsn=20=3D=20= snapshot->lsn;=0A+=09serialized_snapshot.global_csn=20=3D=20= snapshot->global_csn;=0A+=09serialized_snapshot.imported_global_csn=20=3D=20= snapshot->imported_global_csn;=0A=20=0A=20=09/*=0A=20=09=20*=20Ignore=20= the=20SubXID=20array=20if=20it=20has=20overflowed,=20unless=20the=20= snapshot=20was=0A@@=20-2155,6=20+2162,8=20@@=20RestoreSnapshot(char=20= *start_address)=0A=20=09snapshot->curcid=20=3D=20= serialized_snapshot.curcid;=0A=20=09snapshot->whenTaken=20=3D=20= serialized_snapshot.whenTaken;=0A=20=09snapshot->lsn=20=3D=20= serialized_snapshot.lsn;=0A+=09snapshot->global_csn=20=3D=20= serialized_snapshot.global_csn;=0A+=09snapshot->imported_global_csn=20=3D=20= serialized_snapshot.imported_global_csn;=0A=20=0A=20=09/*=20Copy=20XIDs,=20= if=20present.=20*/=0A=20=09if=20(serialized_snapshot.xcnt=20>=200)=0A@@=20= -2192,3=20+2201,92=20@@=20RestoreTransactionSnapshot(Snapshot=20= snapshot,=20void=20*master_pgproc)=0A=20{=0A=20=09= SetTransactionSnapshot(snapshot,=20NULL,=20InvalidPid,=20master_pgproc);=0A= =20}=0A+=0A+/*=0A+=20*=20ExportGlobalSnapshot=0A+=20*=0A+=20*=20Export=20= global_csn=20so=20that=20caller=20can=20expand=20this=20transaction=20to=20= other=0A+=20*=20nodes.=0A+=20*=0A+=20*=20TODO:=20it's=20better=20to=20do=20= this=20through=20EXPORT/IMPORT=20SNAPSHOT=20syntax=20and=0A+=20*=20add=20= some=20additional=20checks=20that=20transaction=20did=20not=20yet=20= acquired=20xid,=20but=0A+=20*=20for=20current=20iteration=20of=20this=20= patch=20I=20don't=20want=20to=20hack=20on=20parser.=0A+=20*/=0A= +GlobalCSN=0A+ExportGlobalSnapshot()=0A+{=0A+=09if=20= (!track_global_snapshots)=0A+=09=09ereport(ERROR,=0A+=09=09=09= (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09=20= errmsg("could=20not=20export=20global=20snapshot"),=0A+=09=09=09=20= errhint("Make=20sure=20the=20configuration=20parameter=20\"%s\"=20is=20= enabled.",=0A+=09=09=09=09=09=20"track_global_snapshots")));=0A+=0A+=09= return=20CurrentSnapshot->global_csn;=0A+}=0A+=0A+/*=20SQL=20accessor=20= to=20ExportGlobalSnapshot()=20*/=0A+Datum=0A= +pg_global_snaphot_export(PG_FUNCTION_ARGS)=0A+{=0A+=09GlobalCSN=09= global_csn=20=3D=20ExportGlobalSnapshot();=0A+=09= PG_RETURN_UINT64(global_csn);=0A+}=0A+=0A+/*=0A+=20*=20= ImportGlobalSnapshot=0A+=20*=0A+=20*=20Import=20global_csn=20and=20= retract=20this=20backends=20xmin=20to=20the=20value=20that=20was=0A+=20*=20= actual=20when=20we=20had=20such=20global_csn.=0A+=20*=0A+=20*=20TODO:=20= it's=20better=20to=20do=20this=20through=20EXPORT/IMPORT=20SNAPSHOT=20= syntax=20and=0A+=20*=20add=20some=20additional=20checks=20that=20= transaction=20did=20not=20yet=20acquired=20xid,=20but=0A+=20*=20for=20= current=20iteration=20of=20this=20patch=20I=20don't=20want=20to=20hack=20= on=20parser.=0A+=20*/=0A+void=0A+ImportGlobalSnapshot(GlobalCSN=20= snap_global_csn)=0A+{=0A+=09volatile=20TransactionId=20xmin;=0A+=0A+=09= if=20(!track_global_snapshots)=0A+=09=09ereport(ERROR,=0A+=09=09=09= (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),=0A+=09=09=09=20= errmsg("could=20not=20import=20global=20snapshot"),=0A+=09=09=09=20= errhint("Make=20sure=20the=20configuration=20parameter=20\"%s\"=20is=20= enabled.",=0A+=09=09=09=09=09=20"track_global_snapshots")));=0A+=0A+=09= /*=0A+=09=20*=20Call=20GlobalSnapshotToXmin=20under=20ProcArrayLock=20to=20= avoid=20situation=20that=0A+=09=20*=20resulting=20xmin=20will=20be=20= evicted=20from=20map=20before=20we=20will=20set=20it=20into=20our=0A+=09=20= *=20backend's=20xmin.=0A+=09=20*/=0A+=09LWLockAcquire(ProcArrayLock,=20= LW_SHARED);=0A+=09xmin=20=3D=20GlobalSnapshotToXmin(snap_global_csn);=0A= +=09if=20(!TransactionIdIsValid(xmin))=0A+=09{=0A+=09=09= LWLockRelease(ProcArrayLock);=0A+=09=09elog(ERROR,=20= "GlobalSnapshotToXmin:=20global=20snapshot=20too=20old");=0A+=09}=0A+=09= MyProc->originalXmin=20=3D=20MyPgXact->xmin;=0A+=09MyPgXact->xmin=20=3D=20= TransactionXmin=20=3D=20xmin;=0A+=09LWLockRelease(ProcArrayLock);=0A+=0A= +=09CurrentSnapshot->xmin=20=3D=20xmin;=20/*=20defuse=20= SnapshotResetXmin()=20*/=0A+=09CurrentSnapshot->global_csn=20=3D=20= snap_global_csn;=0A+=09CurrentSnapshot->imported_global_csn=20=3D=20= true;=0A+=09GlobalSnapshotSync(snap_global_csn);=0A+=0A+=09= Assert(TransactionIdPrecedesOrEquals(RecentGlobalXmin,=20xmin));=0A+=09= Assert(TransactionIdPrecedesOrEquals(RecentGlobalDataXmin,=20xmin));=0A= +}=0A+=0A+/*=20SQL=20accessor=20to=20ImportGlobalSnapshot()=20*/=0A= +Datum=0A+pg_global_snaphot_import(PG_FUNCTION_ARGS)=0A+{=0A+=09= GlobalCSN=09global_csn=20=3D=20PG_GETARG_UINT64(0);=0A+=09= ImportGlobalSnapshot(global_csn);=0A+=09PG_RETURN_VOID();=0A+}=0A+=0A+=0A= diff=20--git=20a/src/backend/utils/time/tqual.c=20= b/src/backend/utils/time/tqual.c=0Aindex=20f7c4c9188c..f2fbc77fa8=20= 100644=0A---=20a/src/backend/utils/time/tqual.c=0A+++=20= b/src/backend/utils/time/tqual.c=0A@@=20-63,6=20+63,7=20@@=0A=20=0A=20= #include=20"postgres.h"=0A=20=0A+#include=20"access/global_snapshot.h"=0A= =20#include=20"access/htup_details.h"=0A=20#include=20= "access/multixact.h"=0A=20#include=20"access/subtrans.h"=0A@@=20-1462,8=20= +1463,8=20@@=20HeapTupleIsSurelyDead(HeapTuple=20htup,=20TransactionId=20= OldestXmin)=0A=20}=0A=20=0A=20/*=0A-=20*=20XidInMVCCSnapshot=0A-=20*=09=09= Is=20the=20given=20XID=20still-in-progress=20according=20to=20the=20= snapshot?=0A+=20*=20XidInLocalMVCCSnapshot=0A+=20*=09=09Is=20the=20given=20= XID=20still-in-progress=20according=20to=20the=20local=20snapshot?=0A=20=20= *=0A=20=20*=20Note:=20GetSnapshotData=20never=20stores=20either=20top=20= xid=20or=20subxids=20of=20our=20own=0A=20=20*=20backend=20into=20a=20= snapshot,=20so=20these=20xids=20will=20not=20be=20reported=20as=20= "running"=0A@@=20-1471,8=20+1472,8=20@@=20= HeapTupleIsSurelyDead(HeapTuple=20htup,=20TransactionId=20OldestXmin)=0A=20= =20*=20TransactionIdIsCurrentTransactionId=20first,=20except=20when=20= it's=20known=20the=0A=20=20*=20XID=20could=20not=20be=20ours=20anyway.=0A= =20=20*/=0A-bool=0A-XidInMVCCSnapshot(TransactionId=20xid,=20Snapshot=20= snapshot)=0A+static=20bool=0A+XidInLocalMVCCSnapshot(TransactionId=20= xid,=20Snapshot=20snapshot)=0A=20{=0A=20=09uint32=09=09i;=0A=20=0A@@=20= -1583,6=20+1584,62=20@@=20XidInMVCCSnapshot(TransactionId=20xid,=20= Snapshot=20snapshot)=0A=20=09return=20false;=0A=20}=0A=20=0A+/*=0A+=20*=20= XidInMVCCSnapshot=0A+=20*=0A+=20*=20Check=20whether=20this=20xid=20is=20= in=20snapshot,=20taking=20into=20account=20fact=20that=0A+=20*=20= snapshot=20can=20be=20global.=20When=20track_global_snapshots=20is=20= switched=20off=0A+=20*=20just=20call=20XidInLocalMVCCSnapshot().=0A+=20= */=0A+bool=0A+XidInMVCCSnapshot(TransactionId=20xid,=20Snapshot=20= snapshot)=0A+{=0A+=09bool=20in_snapshot;=0A+=0A+=09if=20= (snapshot->imported_global_csn)=0A+=09{=0A+=09=09= Assert(track_global_snapshots);=0A+=09=09/*=20No=20point=20to=20using=20= snapshot=20info=20except=20CSN=20*/=0A+=09=09return=20= XidInvisibleInGlobalSnapshot(xid,=20snapshot);=0A+=09}=0A+=0A+=09= in_snapshot=20=3D=20XidInLocalMVCCSnapshot(xid,=20snapshot);=0A+=0A+=09= if=20(!track_global_snapshots)=0A+=09{=0A+=09=09= Assert(GlobalCSNIsFrozen(snapshot->global_csn));=0A+=09=09return=20= in_snapshot;=0A+=09}=0A+=0A+=09if=20(in_snapshot)=0A+=09{=0A+=09=09/*=0A= +=09=09=20*=20This=20xid=20may=20be=20already=20in=20unknown=20state=20= and=20in=20that=20case=0A+=09=09=20*=20we=20must=20wait=20and=20recheck.=0A= +=09=09=20*=0A+=09=09=20*=20TODO:=20this=20check=20can=20be=20skipped=20= if=20we=20know=20for=20sure=20that=20there=20were=0A+=09=09=20*=20no=20= global=20transactions=20when=20this=20snapshot=20was=20taken.=20That=20= requires=0A+=09=09=20*=20some=20changes=20to=20mechanisms=20of=20global=20= snapshots=20exprot/import=20(if=0A+=09=09=20*=20backend=20set=20xmin=20= then=20we=20should=20have=20a-priori=20knowledge=20that=20this=0A+=09=09=20= *=20transaction=20going=20to=20be=20global=20or=20local=20--=20right=20= now=20this=20is=20not=0A+=09=09=20*=20enforced).=20Leave=20that=20for=20= future=20and=20don't=20complicate=20this=20patch.=0A+=09=09=20*/=0A+=09=09= return=20XidInvisibleInGlobalSnapshot(xid,=20snapshot);=0A+=09}=0A+=09= else=0A+=09{=0A+#ifdef=20USE_ASSERT_CHECKING=0A+=09=09/*=20Check=20that=20= global=20snapshot=20gives=20the=20same=20results=20as=20local=20one=20*/=0A= +=09=09if=20(XidInvisibleInGlobalSnapshot(xid,=20snapshot))=0A+=09=09{=0A= +=09=09=09GlobalCSN=20gcsn=20=3D=20TransactionIdGetGlobalCSN(xid);=0A+=09= =09=09Assert(GlobalCSNIsAborted(gcsn));=0A+=09=09}=0A+#endif=0A+=09=09= return=20false;=0A+=09}=0A+}=0A+=0A=20/*=0A=20=20*=20Is=20the=20tuple=20= really=20only=20locked?=20=20That=20is,=20is=20it=20not=20updated?=0A=20=20= *=0Adiff=20--git=20a/src/include/access/global_snapshot.h=20= b/src/include/access/global_snapshot.h=0Anew=20file=20mode=20100644=0A= index=200000000000..246b180cfd=0A---=20/dev/null=0A+++=20= b/src/include/access/global_snapshot.h=0A@@=20-0,0=20+1,72=20@@=0A= +/*-----------------------------------------------------------------------= --=0A+=20*=0A+=20*=20global_snapshot.h=0A+=20*=09=20=20Support=20for=20= cross-node=20snapshot=20isolation.=0A+=20*=0A+=20*=20Portions=20= Copyright=20(c)=201996-2018,=20PostgreSQL=20Global=20Development=20Group=0A= +=20*=20Portions=20Copyright=20(c)=201994,=20Regents=20of=20the=20= University=20of=20California=0A+=20*=0A+=20*=20= src/include/access/global_snapshot.h=0A+=20*=0A+=20= *-------------------------------------------------------------------------= =0A+=20*/=0A+#ifndef=20GLOBAL_SNAPSHOT_H=0A+#define=20GLOBAL_SNAPSHOT_H=0A= +=0A+#include=20"port/atomics.h"=0A+#include=20"storage/lock.h"=0A= +#include=20"utils/snapshot.h"=0A+#include=20"utils/guc.h"=0A+=0A+/*=0A+=20= *=20snapshot.h=20is=20used=20in=20frontend=20code=20so=20atomic=20= variant=20of=20GlobalCSN=20type=0A+=20*=20is=20defined=20here.=0A+=20*/=0A= +typedef=20pg_atomic_uint64=20GlobalCSN_atomic;=0A+=0A+#define=20= InProgressGlobalCSN=09=20UINT64CONST(0x0)=0A+#define=20AbortedGlobalCSN=09= =20UINT64CONST(0x1)=0A+#define=20FrozenGlobalCSN=09=09=20= UINT64CONST(0x2)=0A+#define=20InDoubtGlobalCSN=09=20UINT64CONST(0x3)=0A= +#define=20FirstNormalGlobalCSN=20UINT64CONST(0x4)=0A+=0A+#define=20= GlobalCSNIsInProgress(csn)=09((csn)=20=3D=3D=20InProgressGlobalCSN)=0A= +#define=20GlobalCSNIsAborted(csn)=09=09((csn)=20=3D=3D=20= AbortedGlobalCSN)=0A+#define=20GlobalCSNIsFrozen(csn)=09=09((csn)=20=3D=3D= =20FrozenGlobalCSN)=0A+#define=20GlobalCSNIsInDoubt(csn)=09=09((csn)=20= =3D=3D=20InDoubtGlobalCSN)=0A+#define=20GlobalCSNIsNormal(csn)=09=09= ((csn)=20>=3D=20FirstNormalGlobalCSN)=0A+=0A+=0A+extern=20int=20= global_snapshot_defer_time;=0A+=0A+=0A+extern=20Size=20= GlobalSnapshotShmemSize(void);=0A+extern=20void=20= GlobalSnapshotShmemInit(void);=0A+extern=20void=20= GlobalSnapshotStartup(TransactionId=20oldestActiveXID);=0A+=0A+extern=20= void=20GlobalSnapshotMapXmin(GlobalCSN=20snapshot_global_csn);=0A+extern=20= TransactionId=20GlobalSnapshotToXmin(GlobalCSN=20snapshot_global_csn);=0A= +=0A+extern=20GlobalCSN=20GlobalSnapshotGenerate(bool=20locked);=0A+=0A= +extern=20bool=20XidInvisibleInGlobalSnapshot(TransactionId=20xid,=20= Snapshot=20snapshot);=0A+=0A+extern=20void=20= GlobalSnapshotSync(GlobalCSN=20remote_gcsn);=0A+=0A+extern=20GlobalCSN=20= TransactionIdGetGlobalCSN(TransactionId=20xid);=0A+=0A+extern=20= GlobalCSN=20GlobalSnapshotPrepareGlobal(const=20char=20*gid);=0A+extern=20= void=20GlobalSnapshotAssignCsnGlobal(const=20char=20*gid,=0A+=09=09=09=09= =09=09=09=09=09=09=20=20GlobalCSN=20global_csn);=0A+=0A+extern=20= GlobalCSN=20GlobalSnapshotPrepareCurrent(void);=0A+extern=20void=20= GlobalSnapshotAssignCsnCurrent(GlobalCSN=20global_csn);=0A+=0A+extern=20= void=20GlobalSnapshotAbort(PGPROC=20*proc,=20TransactionId=20xid,=20int=20= nsubxids,=0A+=09=09=09=09=09=09=09=09TransactionId=20*subxids);=0A= +extern=20void=20GlobalSnapshotPrecommit(PGPROC=20*proc,=20TransactionId=20= xid,=20int=20nsubxids,=0A+=09=09=09=09=09=09=09=09=09TransactionId=20= *subxids);=0A+extern=20void=20GlobalSnapshotCommit(PGPROC=20*proc,=20= TransactionId=20xid,=20int=20nsubxids,=0A+=09=09=09=09=09=09=09=09=09= TransactionId=20*subxids);=0A+=0A+#endif=09=09=09=09=09=09=09/*=20= GLOBAL_SNAPSHOT_H=20*/=0Adiff=20--git=20a/src/include/access/twophase.h=20= b/src/include/access/twophase.h=0Aindex=200e932daa48..f8b774f393=20= 100644=0A---=20a/src/include/access/twophase.h=0A+++=20= b/src/include/access/twophase.h=0A@@=20-18,6=20+18,7=20@@=0A=20#include=20= "access/xact.h"=0A=20#include=20"datatype/timestamp.h"=0A=20#include=20= "storage/lock.h"=0A+#include=20"utils/snapshot.h"=0A=20=0A=20/*=0A=20=20= *=20GlobalTransactionData=20is=20defined=20in=20twophase.c;=20other=20= places=20have=20no=0Adiff=20--git=20a/src/include/catalog/pg_proc.dat=20= b/src/include/catalog/pg_proc.dat=0Aindex=2066c6c224a8..6d9774901f=20= 100644=0A---=20a/src/include/catalog/pg_proc.dat=0A+++=20= b/src/include/catalog/pg_proc.dat=0A@@=20-10202,4=20+10202,18=20@@=0A=20=20= =20proisstrict=20=3D>=20'f',=20prorettype=20=3D>=20'bool',=20proargtypes=20= =3D>=20'oid=20int4=20int4=20any',=0A=20=20=20proargmodes=20=3D>=20= '{i,i,i,v}',=20prosrc=20=3D>=20'satisfies_hash_partition'=20},=0A=20=0A= +#=20global=20transaction=20handling=0A+{=20oid=20=3D>=20'3430',=20descr=20= =3D>=20'export=20global=20transaction=20snapshot',=0A+=20=20proname=20=3D>= =20'pg_global_snaphot_export',=20provolatile=20=3D>=20'v',=20proparallel=20= =3D>=20'u',=0A+=20=20prorettype=20=3D>=20'int8',=20proargtypes=20=3D>=20= '',=20prosrc=20=3D>=20'pg_global_snaphot_export'=20},=0A+{=20oid=20=3D>=20= '3431',=20descr=20=3D>=20'import=20global=20transaction=20snapshot',=0A+=20= =20proname=20=3D>=20'pg_global_snaphot_import',=20provolatile=20=3D>=20= 'v',=20proparallel=20=3D>=20'u',=0A+=20=20prorettype=20=3D>=20'void',=20= proargtypes=20=3D>=20'int8',=20prosrc=20=3D>=20= 'pg_global_snaphot_import'=20},=0A+{=20oid=20=3D>=20'3432',=20descr=20=3D>= =20'prepare=20distributed=20transaction=20for=20commit,=20get=20= global_csn',=0A+=20=20proname=20=3D>=20'pg_global_snaphot_prepare',=20= provolatile=20=3D>=20'v',=20proparallel=20=3D>=20'u',=0A+=20=20= prorettype=20=3D>=20'int8',=20proargtypes=20=3D>=20'text',=20prosrc=20=3D>= =20'pg_global_snaphot_prepare'=20},=0A+{=20oid=20=3D>=20'3433',=20descr=20= =3D>=20'assign=20global_csn=20to=20distributed=20transaction',=0A+=20=20= proname=20=3D>=20'pg_global_snaphot_assign',=20provolatile=20=3D>=20'v',=20= proparallel=20=3D>=20'u',=0A+=20=20prorettype=20=3D>=20'void',=20= proargtypes=20=3D>=20'text=20int8',=20prosrc=20=3D>=20= 'pg_global_snaphot_assign'=20},=0A+=0A=20]=0Adiff=20--git=20= a/src/include/datatype/timestamp.h=20b/src/include/datatype/timestamp.h=0A= index=20f5b6026ef5..75ec93b46b=20100644=0A---=20= a/src/include/datatype/timestamp.h=0A+++=20= b/src/include/datatype/timestamp.h=0A@@=20-93,6=20+93,9=20@@=20typedef=20= struct=0A=20#define=20USECS_PER_MINUTE=20INT64CONST(60000000)=0A=20= #define=20USECS_PER_SEC=09INT64CONST(1000000)=0A=20=0A+#define=20= NSECS_PER_SEC=09INT64CONST(1000000000)=0A+#define=20NSECS_PER_USEC=09= INT64CONST(1000)=0A+=0A=20/*=0A=20=20*=20We=20allow=20numeric=20timezone=20= offsets=20up=20to=2015:59:59=20either=20way=20from=20Greenwich.=0A=20=20= *=20Currently,=20the=20record=20holders=20for=20wackiest=20offsets=20in=20= actual=20use=20are=20zones=0Adiff=20--git=20a/src/include/fmgr.h=20= b/src/include/fmgr.h=0Aindex=20101f513ba6..3026e71f83=20100644=0A---=20= a/src/include/fmgr.h=0A+++=20b/src/include/fmgr.h=0A@@=20-250,6=20+250,7=20= @@=20extern=20struct=20varlena=20*pg_detoast_datum_packed(struct=20= varlena=20*datum);=0A=20#define=20PG_GETARG_FLOAT4(n)=20=20= DatumGetFloat4(PG_GETARG_DATUM(n))=0A=20#define=20PG_GETARG_FLOAT8(n)=20=20= DatumGetFloat8(PG_GETARG_DATUM(n))=0A=20#define=20PG_GETARG_INT64(n)=09=20= DatumGetInt64(PG_GETARG_DATUM(n))=0A+#define=20PG_GETARG_UINT64(n)=09=20= DatumGetUInt64(PG_GETARG_DATUM(n))=0A=20/*=20use=20this=20if=20you=20= want=20the=20raw,=20possibly-toasted=20input=20datum:=20*/=0A=20#define=20= PG_GETARG_RAW_VARLENA_P(n)=09((struct=20varlena=20*)=20= PG_GETARG_POINTER(n))=0A=20/*=20use=20this=20if=20you=20want=20the=20= input=20datum=20de-toasted:=20*/=0Adiff=20--git=20= a/src/include/portability/instr_time.h=20= b/src/include/portability/instr_time.h=0Aindex=20f968444671..ebc3836c9c=20= 100644=0A---=20a/src/include/portability/instr_time.h=0A+++=20= b/src/include/portability/instr_time.h=0A@@=20-138,6=20+138,9=20@@=20= typedef=20struct=20timespec=20instr_time;=0A=20#define=20= INSTR_TIME_GET_MICROSEC(t)=20\=0A=20=09(((uint64)=20(t).tv_sec=20*=20= (uint64)=201000000)=20+=20(uint64)=20((t).tv_nsec=20/=201000))=0A=20=0A= +#define=20INSTR_TIME_GET_NANOSEC(t)=20\=0A+=09(((uint64)=20(t).tv_sec=20= *=20(uint64)=201000000000)=20+=20(uint64)=20((t).tv_nsec))=0A+=0A=20= #else=09=09=09=09=09=09=09/*=20!HAVE_CLOCK_GETTIME=20*/=0A=20=0A=20/*=20= Use=20gettimeofday()=20*/=0A@@=20-202,6=20+205,10=20@@=20typedef=20= struct=20timeval=20instr_time;=0A=20#define=20INSTR_TIME_GET_MICROSEC(t)=20= \=0A=20=09(((uint64)=20(t).tv_sec=20*=20(uint64)=201000000)=20+=20= (uint64)=20(t).tv_usec)=0A=20=0A+#define=20INSTR_TIME_GET_NANOSEC(t)=20\=0A= +=09(((uint64)=20(t).tv_sec=20*=20(uint64)=201000000000)=20+=20\=0A+=09=09= (uint64)=20(t).tv_usec=20*=20(uint64)=201000)=0A+=0A=20#endif=09=09=09=09= =09=09=09/*=20HAVE_CLOCK_GETTIME=20*/=0A=20=0A=20#else=09=09=09=09=09=09=09= /*=20WIN32=20*/=0A@@=20-234,6=20+241,9=20@@=20typedef=20LARGE_INTEGER=20= instr_time;=0A=20#define=20INSTR_TIME_GET_MICROSEC(t)=20\=0A=20=09= ((uint64)=20(((double)=20(t).QuadPart=20*=201000000.0)=20/=20= GetTimerFrequency()))=0A=20=0A+#define=20INSTR_TIME_GET_NANOSEC(t)=20\=0A= +=09((uint64)=20(((double)=20(t).QuadPart=20*=201000000000.0)=20/=20= GetTimerFrequency()))=0A+=0A=20static=20inline=20double=0A=20= GetTimerFrequency(void)=0A=20{=0Adiff=20--git=20= a/src/include/storage/proc.h=20b/src/include/storage/proc.h=0Aindex=20= 5c19a61dcf..7e24d539cc=20100644=0A---=20a/src/include/storage/proc.h=0A= +++=20b/src/include/storage/proc.h=0A@@=20-15,8=20+15,10=20@@=0A=20= #define=20_PROC_H_=0A=20=0A=20#include=20"access/clog.h"=0A+#include=20= "access/global_snapshot.h"=0A=20#include=20"access/xlogdefs.h"=0A=20= #include=20"lib/ilist.h"=0A+#include=20"utils/snapshot.h"=0A=20#include=20= "storage/latch.h"=0A=20#include=20"storage/lock.h"=0A=20#include=20= "storage/pg_sema.h"=0A@@=20-57,6=20+59,7=20@@=20struct=20XidCache=0A=20= #define=09=09PROC_IN_LOGICAL_DECODING=090x10=09/*=20currently=20doing=20= logical=0A=20=09=09=09=09=09=09=09=09=09=09=09=09=20*=20decoding=20= outside=20xact=20*/=0A=20#define=09=09PROC_RESERVED=09=09=09=090x20=09/*=20= reserved=20for=20procarray=20*/=0A+#define=09=09PROC_RESERVED2=09=09=09=09= 0x40=09/*=20reserved=20for=20procarray=20*/=0A=20=0A=20/*=20flags=20= reset=20at=20EOXact=20*/=0A=20#define=09=09PROC_VACUUM_STATE_MASK=20\=0A= @@=20-200,6=20+203,18=20@@=20struct=20PGPROC=0A=20=09PGPROC=09=20=20=20= *lockGroupLeader;=09/*=20lock=20group=20leader,=20if=20I'm=20a=20member=20= */=0A=20=09dlist_head=09lockGroupMembers;=09/*=20list=20of=20members,=20= if=20I'm=20a=20leader=20*/=0A=20=09dlist_node=09lockGroupLink;=09/*=20my=20= member=20link,=20if=20I'm=20a=20member=20*/=0A+=0A+=09/*=0A+=09=20*=20= assignedGlobalCsn=20holds=20GlobalCSN=20for=20this=20transaction.=20=20= It=20is=20generated=0A+=09=20*=20under=20a=20ProcArray=20lock=20and=20= later=20is=20writter=20to=20a=20GlobalCSNLog.=20=20This=0A+=09=20*=20= variable=20defined=20as=20atomic=20only=20for=20case=20of=20group=20= commit,=20in=20all=20other=0A+=09=20*=20scenarios=20only=20backend=20= responsible=20for=20this=20proc=20entry=20is=20working=20with=0A+=09=20*=20= this=20variable.=0A+=09=20*/=0A+=09GlobalCSN_atomic=20assignedGlobalCsn;=0A= +=0A+=09/*=20Original=20xmin=20of=20this=20backend=20before=20global=20= snapshot=20was=20imported=20*/=0A+=09TransactionId=20originalXmin;=0A=20= };=0A=20=0A=20/*=20NOTE:=20"typedef=20struct=20PGPROC=20PGPROC"=20= appears=20in=20storage/lock.h.=20*/=0Adiff=20--git=20= a/src/include/storage/procarray.h=20b/src/include/storage/procarray.h=0A= index=2075bab2985f..4c749d6123=20100644=0A---=20= a/src/include/storage/procarray.h=0A+++=20= b/src/include/storage/procarray.h=0A@@=20-36,6=20+36,10=20@@=0A=20=0A=20= #define=09=09PROCARRAY_SLOTS_XMIN=09=09=090x20=09/*=20replication=20slot=20= xmin,=0A=20=09=09=09=09=09=09=09=09=09=09=09=09=09=20*=20catalog_xmin=20= */=0A+=0A+#define=09=09PROCARRAY_NON_IMPORTED_XMIN=09=090x40=09/*=20use=20= originalXmin=20instead=0A+=09=09=09=09=09=09=09=09=09=09=09=09=09=20*=20= of=20xmin=20to=20properly=0A+=09=09=09=09=09=09=09=09=09=09=09=09=09=20*=20= maintain=20gsXidMap=20*/=0A=20/*=0A=20=20*=20Only=20flags=20in=20= PROCARRAY_PROC_FLAGS_MASK=20are=20considered=20when=20matching=0A=20=20*=20= PGXACT->vacuumFlags.=20Other=20flags=20are=20used=20for=20different=20= purposes=20and=0A@@=20-124,4=20+128,8=20@@=20extern=20void=20= ProcArraySetReplicationSlotXmin(TransactionId=20xmin,=0A=20extern=20void=20= ProcArrayGetReplicationSlotXmin(TransactionId=20*xmin,=0A=20=09=09=09=09=09= =09=09=09TransactionId=20*catalog_xmin);=0A=20=0A+extern=20void=20= ProcArraySetGlobalSnapshotXmin(TransactionId=20xmin);=0A+=0A+extern=20= TransactionId=20ProcArrayGetGlobalSnapshotXmin(bool=20locked);=0A+=0A=20= #endif=09=09=09=09=09=09=09/*=20PROCARRAY_H=20*/=0Adiff=20--git=20= a/src/include/utils/snapmgr.h=20b/src/include/utils/snapmgr.h=0Aindex=20= 83806f3040..1a066fd8d8=20100644=0A---=20a/src/include/utils/snapmgr.h=0A= +++=20b/src/include/utils/snapmgr.h=0A@@=20-87,6=20+87,9=20@@=20extern=20= void=20AtSubCommit_Snapshot(int=20level);=0A=20extern=20void=20= AtSubAbort_Snapshot(int=20level);=0A=20extern=20void=20= AtEOXact_Snapshot(bool=20isCommit,=20bool=20resetXmin);=0A=20=0A+extern=20= GlobalCSN=20ExportGlobalSnapshot(void);=0A+extern=20void=20= ImportGlobalSnapshot(GlobalCSN=20snap_global_csn);=0A+=0A=20extern=20= void=20ImportSnapshot(const=20char=20*idstr);=0A=20extern=20bool=20= XactHasExportedSnapshots(void);=0A=20extern=20void=20= DeleteAllExportedSnapshotFiles(void);=0Adiff=20--git=20= a/src/include/utils/snapshot.h=20b/src/include/utils/snapshot.h=0Aindex=20= 318d41e6f7..1563407824=20100644=0A---=20a/src/include/utils/snapshot.h=0A= +++=20b/src/include/utils/snapshot.h=0A@@=20-115,6=20+115,14=20@@=20= typedef=20struct=20SnapshotData=0A=20=0A=20=09TimestampTz=20whenTaken;=09= =09/*=20timestamp=20when=20snapshot=20was=20taken=20*/=0A=20=09= XLogRecPtr=09lsn;=09=09=09/*=20position=20in=20the=20WAL=20stream=20when=20= taken=20*/=0A+=0A+=09/*=0A+=09=20*=20GlobalCSN=20for=20cross-node=20= snapshot=20isolation=20support.=0A+=09=20*=20Will=20be=20used=20only=20= if=20track_global_snapshots=20is=20enabled.=0A+=09=20*/=0A+=09GlobalCSN=09= global_csn;=0A+=09/*=20Did=20we=20have=20our=20own=20global_csn=20or=20= imported=20one=20from=20different=20node=20*/=0A+=09bool=09=09= imported_global_csn;=0A=20}=20SnapshotData;=0A=20=0A=20/*=0A--=20=0A= 2.16.2=0A=0A= --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55 Content-Disposition: attachment; filename=0003-postgres_fdw-support-for-global-snapshots.patch Content-Type: application/octet-stream; x-unix-mode=0644; name="0003-postgres_fdw-support-for-global-snapshots.patch" Content-Transfer-Encoding: quoted-printable =46rom=20f123454bdb9943697cfb60cb2c154b553675c06b=20Mon=20Sep=2017=20= 00:00:00=202001=0AFrom:=20Stas=20Kelvich=20=0ADate:=20= Wed,=2025=20Apr=202018=2016:39:09=20+0300=0ASubject:=20[PATCH=203/3]=20= postgres_fdw=20support=20for=20global=20snapshots=0A=0A---=0A=20= contrib/postgres_fdw/Makefile=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20= =20=20=20|=20=20=206=20+=0A=20contrib/postgres_fdw/connection.c=20=20=20=20= =20=20=20=20=20=20=20=20=20=20|=20286=20++++++++++++++++++++++---=0A=20= contrib/postgres_fdw/postgres_fdw.c=20=20=20=20=20=20=20=20=20=20=20=20|=20= =2012=20++=0A=20contrib/postgres_fdw/postgres_fdw.h=20=20=20=20=20=20=20=20= =20=20=20=20|=20=20=202=20+=0A=20= contrib/postgres_fdw/t/001_bank_coordinator.pl=20|=20262=20= ++++++++++++++++++++++=0A=20= contrib/postgres_fdw/t/002_bank_participant.pl=20|=20238=20= ++++++++++++++++++++=0A=20contrib/postgres_fdw/t/003_bank_shared.pl=20=20= =20=20=20=20|=20155=20++++++++++++++=0A=20src/test/perl/PostgresNode.pm=20= =20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20=20|=20=2031=20+++=0A=20= 8=20files=20changed,=20965=20insertions(+),=2027=20deletions(-)=0A=20= create=20mode=20100644=20contrib/postgres_fdw/t/001_bank_coordinator.pl=0A= =20create=20mode=20100644=20= contrib/postgres_fdw/t/002_bank_participant.pl=0A=20create=20mode=20= 100644=20contrib/postgres_fdw/t/003_bank_shared.pl=0A=0Adiff=20--git=20= a/contrib/postgres_fdw/Makefile=20b/contrib/postgres_fdw/Makefile=0A= index=2085394b4f1f..3721bdd618=20100644=0A---=20= a/contrib/postgres_fdw/Makefile=0A+++=20b/contrib/postgres_fdw/Makefile=0A= @@=20-23,3=20+23,9=20@@=20top_builddir=20=3D=20../..=0A=20include=20= $(top_builddir)/src/Makefile.global=0A=20include=20= $(top_srcdir)/contrib/contrib-global.mk=0A=20endif=0A+=0A+check:=20= tapcheck=0A+=0A+tapcheck:=0A+=09env=20= DESTDIR=3D'$(abs_top_builddir)'/tmp_install=20$(MAKE)=20install=0A+=09= $(prove_check)=0Adiff=20--git=20a/contrib/postgres_fdw/connection.c=20= b/contrib/postgres_fdw/connection.c=0Aindex=20fe4893a8e0..d6d74c3318=20= 100644=0A---=20a/contrib/postgres_fdw/connection.c=0A+++=20= b/contrib/postgres_fdw/connection.c=0A@@=20-14,9=20+14,11=20@@=0A=20=0A=20= #include=20"postgres_fdw.h"=0A=20=0A+#include=20= "access/global_snapshot.h"=0A=20#include=20"access/htup_details.h"=0A=20= #include=20"catalog/pg_user_mapping.h"=0A=20#include=20"access/xact.h"=0A= +#include=20"access/xlog.h"=20/*=20GetSystemIdentifier()=20*/=0A=20= #include=20"mb/pg_wchar.h"=0A=20#include=20"miscadmin.h"=0A=20#include=20= "pgstat.h"=0A@@=20-24,6=20+26,8=20@@=0A=20#include=20"utils/hsearch.h"=0A= =20#include=20"utils/inval.h"=0A=20#include=20"utils/memutils.h"=0A= +#include=20"utils/snapmgr.h"=0A+#include=20"utils/snapshot.h"=0A=20= #include=20"utils/syscache.h"=0A=20=0A=20=0A@@=20-65,6=20+69,21=20@@=20= typedef=20struct=20ConnCacheEntry=0A=20=20*/=0A=20static=20HTAB=20= *ConnectionHash=20=3D=20NULL;=0A=20=0A+/*=0A+=20*=20FdwTransactionState=0A= +=20*=0A+=20*=20Holds=20number=20of=20open=20remote=20transactions=20and=20= shared=20state=0A+=20*=20needed=20for=20all=20connection=20entries.=0A+=20= */=0A+typedef=20struct=20FdwTransactionState=0A+{=0A+=09char=09=09*gid;=0A= +=09int=09=09=09nparticipants;=0A+=09GlobalCSN=09global_csn;=0A+=09bool=09= =09two_phase_commit;=0A+}=20FdwTransactionState;=0A+static=20= FdwTransactionState=20*fdwTransState;=0A+=0A=20/*=20for=20assigning=20= cursor=20numbers=20and=20prepared=20statement=20numbers=20*/=0A=20static=20= unsigned=20int=20cursor_number=20=3D=200;=0A=20static=20unsigned=20int=20= prep_stmt_number=20=3D=200;=0A@@=20-72,6=20+91,9=20@@=20static=20= unsigned=20int=20prep_stmt_number=20=3D=200;=0A=20/*=20tracks=20whether=20= any=20work=20is=20needed=20in=20callback=20functions=20*/=0A=20static=20= bool=20xact_got_connection=20=3D=20false;=0A=20=0A+/*=20counter=20of=20= prepared=20tx=20made=20by=20this=20backend=20*/=0A+static=20int=20= two_phase_xact_count=20=3D=200;=0A+=0A=20/*=20prototypes=20of=20private=20= functions=20*/=0A=20static=20PGconn=20*connect_pg_server(ForeignServer=20= *server,=20UserMapping=20*user);=0A=20static=20void=20= disconnect_pg_server(ConnCacheEntry=20*entry);=0A@@=20-80,6=20+102,7=20= @@=20static=20void=20configure_remote_session(PGconn=20*conn);=0A=20= static=20void=20do_sql_command(PGconn=20*conn,=20const=20char=20*sql);=0A= =20static=20void=20begin_remote_xact(ConnCacheEntry=20*entry);=0A=20= static=20void=20pgfdw_xact_callback(XactEvent=20event,=20void=20*arg);=0A= +static=20void=20deallocate_prepared_stmts(ConnCacheEntry=20*entry);=0A=20= static=20void=20pgfdw_subxact_callback(SubXactEvent=20event,=0A=20=09=09=09= =09=09=20=20=20SubTransactionId=20mySubid,=0A=20=09=09=09=09=09=20=20=20= SubTransactionId=20parentSubid,=0A@@=20-136,6=20+159,15=20@@=20= GetConnection(UserMapping=20*user,=20bool=20will_prep_stmt)=0A=20=09=09=09= =09=09=09=09=09=09=20=20pgfdw_inval_callback,=20(Datum)=200);=0A=20=09}=0A= =20=0A+=09/*=20allocate=20FdwTransactionState=20*/=0A+=09if=20= (fdwTransState=20=3D=3D=20NULL)=0A+=09{=0A+=09=09MemoryContext=20oldcxt;=0A= +=09=09oldcxt=20=3D=20MemoryContextSwitchTo(CacheMemoryContext);=0A+=09=09= fdwTransState=20=3D=20palloc0(sizeof(FdwTransactionState));=0A+=09=09= MemoryContextSwitchTo(oldcxt);=0A+=09}=0A+=0A=20=09/*=20Set=20flag=20= that=20we=20did=20GetConnection=20during=20the=20current=20transaction=20= */=0A=20=09xact_got_connection=20=3D=20true;=0A=20=0A@@=20-388,7=20= +420,8=20@@=20configure_remote_session(PGconn=20*conn)=0A=20}=0A=20=0A=20= /*=0A-=20*=20Convenience=20subroutine=20to=20issue=20a=20= non-data-returning=20SQL=20command=20to=20remote=0A+=20*=20Convenience=20= subroutine=20to=20issue=20a=20non-data-returning=20SQL=20command=20or=0A= +=20*=20statement=20to=20remote=20node.=0A=20=20*/=0A=20static=20void=0A=20= do_sql_command(PGconn=20*conn,=20const=20char=20*sql)=0A@@=20-398,7=20= +431,8=20@@=20do_sql_command(PGconn=20*conn,=20const=20char=20*sql)=0A=20= =09if=20(!PQsendQuery(conn,=20sql))=0A=20=09=09pgfdw_report_error(ERROR,=20= NULL,=20conn,=20false,=20sql);=0A=20=09res=20=3D=20= pgfdw_get_result(conn,=20sql);=0A-=09if=20(PQresultStatus(res)=20!=3D=20= PGRES_COMMAND_OK)=0A+=09if=20(PQresultStatus(res)=20!=3D=20= PGRES_COMMAND_OK=20&&=0A+=09=09=09PQresultStatus(res)=20!=3D=20= PGRES_TUPLES_OK)=0A=20=09=09pgfdw_report_error(ERROR,=20res,=20conn,=20= true,=20sql);=0A=20=09PQclear(res);=0A=20}=0A@@=20-434,6=20+468,23=20@@=20= begin_remote_xact(ConnCacheEntry=20*entry)=0A=20=09=09= do_sql_command(entry->conn,=20sql);=0A=20=09=09entry->xact_depth=20=3D=20= 1;=0A=20=09=09entry->changing_xact_state=20=3D=20false;=0A+=0A+=09=09if=20= (UseGlobalSnapshots)=0A+=09=09{=0A+=09=09=09char=20import_sql[128];=0A+=0A= +=09=09=09/*=20Export=20our=20snapshot=20*/=0A+=09=09=09if=20= (fdwTransState->global_csn=20=3D=3D=200)=0A+=09=09=09=09= fdwTransState->global_csn=20=3D=20ExportGlobalSnapshot();=0A+=0A+=09=09=09= snprintf(import_sql,=20sizeof(import_sql),=0A+=09=09=09=09"SELECT=20= pg_global_snaphot_import("UINT64_FORMAT")",=0A+=09=09=09=09= fdwTransState->global_csn);=0A+=0A+=09=09=09do_sql_command(entry->conn,=20= import_sql);=0A+=09=09}=0A+=0A+=09=09fdwTransState->nparticipants=20+=3D=20= 1;=0A=20=09}=0A=20=0A=20=09/*=0A@@=20-643,6=20+694,94=20@@=20= pgfdw_report_error(int=20elevel,=20PGresult=20*res,=20PGconn=20*conn,=0A=20= =09=09PQclear(res);=0A=20}=0A=20=0A+/*=20Callback=20typedef=20for=20= BroadcastStmt=20*/=0A+typedef=20bool=20(*BroadcastCmdResHandler)=20= (PGresult=20*result,=20void=20*arg);=0A+=0A+/*=20Broadcast=20sql=20in=20= parallel=20to=20all=20ConnectionHash=20entries=20*/=0A+static=20bool=0A= +BroadcastStmt(char=20const=20*=20sql,=20unsigned=20expectedStatus,=0A+=09= =09=09=09BroadcastCmdResHandler=20handler,=20void=20*arg)=0A+{=0A+=09= HASH_SEQ_STATUS=20scan;=0A+=09ConnCacheEntry=20*entry;=0A+=09bool=09=09= allOk=20=3D=20true;=0A+=0A+=09/*=20Broadcast=20sql=20*/=0A+=09= hash_seq_init(&scan,=20ConnectionHash);=0A+=09while=20((entry=20=3D=20= (ConnCacheEntry=20*)=20hash_seq_search(&scan)))=0A+=09{=0A+=09=09= pgfdw_reject_incomplete_xact_state_change(entry);=0A+=0A+=09=09if=20= (entry->xact_depth=20>=200=20&&=20entry->conn=20!=3D=20NULL)=0A+=09=09{=0A= +=09=09=09if=20(!PQsendQuery(entry->conn,=20sql))=0A+=09=09=09{=0A+=09=09= =09=09PGresult=20=20=20*res=20=3D=20PQgetResult(entry->conn);=0A+=0A+=09=09= =09=09elog(WARNING,=20"Failed=20to=20send=20command=20%s",=20sql);=0A+=09= =09=09=09pgfdw_report_error(ERROR,=20res,=20entry->conn,=20true,=20sql);=0A= +=09=09=09=09PQclear(res);=0A+=09=09=09}=0A+=09=09}=0A+=09}=0A+=0A+=09/*=20= Collect=20responses=20*/=0A+=09hash_seq_init(&scan,=20ConnectionHash);=0A= +=09while=20((entry=20=3D=20(ConnCacheEntry=20*)=20= hash_seq_search(&scan)))=0A+=09{=0A+=09=09if=20(entry->xact_depth=20>=20= 0=20&&=20entry->conn=20!=3D=20NULL)=0A+=09=09{=0A+=09=09=09PGresult=20=20= =20*result=20=3D=20PQgetResult(entry->conn);=0A+=0A+=09=09=09if=20= (PQresultStatus(result)=20!=3D=20expectedStatus=20||=0A+=09=09=09=09= (handler=20&&=20!handler(result,=20arg)))=0A+=09=09=09{=0A+=09=09=09=09= elog(WARNING,=20"Failed=20command=20%s:=20status=3D%d,=20expected=20= status=3D%d",=20sql,=20PQresultStatus(result),=20expectedStatus);=0A+=09=09= =09=09pgfdw_report_error(ERROR,=20result,=20entry->conn,=20true,=20sql);=0A= +=09=09=09=09allOk=20=3D=20false;=0A+=09=09=09}=0A+=09=09=09= PQclear(result);=0A+=09=09=09PQgetResult(entry->conn);=09/*=20consume=20= NULL=20result=20*/=0A+=09=09}=0A+=09}=0A+=0A+=09return=20allOk;=0A+}=0A+=0A= +/*=20Wrapper=20for=20broadcasting=20commands=20*/=0A+static=20bool=0A= +BroadcastCmd(char=20const=20*sql)=0A+{=0A+=09return=20= BroadcastStmt(sql,=20PGRES_COMMAND_OK,=20NULL,=20NULL);=0A+}=0A+=0A+/*=20= Wrapper=20for=20broadcasting=20statements=20*/=0A+static=20bool=0A= +BroadcastFunc(char=20const=20*sql)=0A+{=0A+=09return=20= BroadcastStmt(sql,=20PGRES_TUPLES_OK,=20NULL,=20NULL);=0A+}=0A+=0A+/*=20= Callback=20for=20selecting=20maximal=20csn=20*/=0A+static=20bool=0A= +MaxCsnCB(PGresult=20*result,=20void=20*arg)=0A+{=0A+=09char=09=09=20=20=20= *resp;=0A+=09GlobalCSN=09=20=20=20*max_csn=20=3D=20(GlobalCSN=20*)=20= arg;=0A+=09GlobalCSN=09=09csn=20=3D=200;=0A+=0A+=09resp=20=3D=20= PQgetvalue(result,=200,=200);=0A+=0A+=09if=20(resp=20=3D=3D=20NULL=20||=20= (*resp)=20=3D=3D=20'\0'=20||=0A+=09=09=09sscanf(resp,=20UINT64_FORMAT,=20= &csn)=20!=3D=201)=0A+=09=09return=20false;=0A+=0A+=09if=20(*max_csn=20<=20= csn)=0A+=09=09*max_csn=20=3D=20csn;=0A+=0A+=09return=20true;=0A+}=0A+=0A=20= /*=0A=20=20*=20pgfdw_xact_callback=20---=20cleanup=20at=20= main-transaction=20end.=0A=20=20*/=0A@@=20-656,6=20+795,84=20@@=20= pgfdw_xact_callback(XactEvent=20event,=20void=20*arg)=0A=20=09if=20= (!xact_got_connection)=0A=20=09=09return;=0A=20=0A+=09/*=20Handle=20= possible=20two-phase=20commit=20*/=0A+=09if=20(event=20=3D=3D=20= XACT_EVENT_PARALLEL_PRE_COMMIT=20||=20event=20=3D=3D=20= XACT_EVENT_PRE_COMMIT)=0A+=09{=0A+=09=09bool=20include_local_tx=20=3D=20= false;=0A+=0A+=09=09/*=20Should=20we=20take=20into=20account=20this=20= node?=20*/=0A+=09=09if=20= (TransactionIdIsValid(GetCurrentTransactionIdIfAny()))=0A+=09=09{=0A+=09=09= =09include_local_tx=20=3D=20true;=0A+=09=09=09= fdwTransState->nparticipants=20+=3D=201;=0A+=09=09}=0A+=0A+=09=09/*=20= Switch=20to=202PC=20mode=20there=20were=20more=20than=20one=20= participant=20*/=0A+=09=09if=20(UseGlobalSnapshots=20&&=20= fdwTransState->nparticipants=20>=201)=0A+=09=09=09= fdwTransState->two_phase_commit=20=3D=20true;=0A+=0A+=09=09if=20= (fdwTransState->two_phase_commit)=0A+=09=09{=0A+=09=09=09GlobalCSN=09= max_csn=20=3D=20InProgressGlobalCSN,=0A+=09=09=09=09=09=09my_csn=20=3D=20= InProgressGlobalCSN;=0A+=09=09=09bool=09res;=0A+=09=09=09char=20=20=20= *sql;=0A+=0A+=09=09=09fdwTransState->gid=20=3D=20= psprintf("fdw:%llx:%lld:%d:%d",=0A+=09=09=09=09=09=09=09(long=20long)=20= GetSystemIdentifier(),=0A+=09=09=09=09=09=09=09(long=20long)=20= GetCurrentTimestamp(),=0A+=09=09=09=09=09=09=09MyProcPid,=0A+=09=09=09=09= =09=09=09++two_phase_xact_count);=0A+=0A+=09=09=09/*=20Broadcast=20= PREPARE=20*/=0A+=09=09=09sql=20=3D=20psprintf("PREPARE=20TRANSACTION=20= '%s'",=20fdwTransState->gid);=0A+=09=09=09res=20=3D=20BroadcastCmd(sql);=0A= +=09=09=09if=20(!res)=0A+=09=09=09=09goto=20error;=0A+=0A+=09=09=09/*=20= Broadcast=20pg_global_snaphot_prepare()=20*/=0A+=09=09=09if=20= (include_local_tx)=0A+=09=09=09=09my_csn=20=3D=20= GlobalSnapshotPrepareCurrent();=0A+=0A+=09=09=09sql=20=3D=20= psprintf("SELECT=20pg_global_snaphot_prepare('%s')",=0A+=09=09=09=09=09=09= =09=09=09=09=09=09=09=09fdwTransState->gid);=0A+=09=09=09res=20=3D=20= BroadcastStmt(sql,=20PGRES_TUPLES_OK,=20MaxCsnCB,=20&max_csn);=0A+=09=09=09= if=20(!res)=0A+=09=09=09=09goto=20error;=0A+=0A+=09=09=09/*=20select=20= maximal=20global=20csn=20*/=0A+=09=09=09if=20(my_csn=20>=20max_csn)=0A+=09= =09=09=09max_csn=20=3D=20my_csn;=0A+=0A+=09=09=09/*=20Broadcast=20= pg_global_snaphot_assign()=20*/=0A+=09=09=09if=20(include_local_tx)=0A+=09= =09=09=09GlobalSnapshotAssignCsnCurrent(max_csn);=0A+=09=09=09sql=20=3D=20= psprintf("SELECT=20pg_global_snaphot_assign('%s',"UINT64_FORMAT")",=0A+=09= =09=09=09=09=09=09fdwTransState->gid,=20max_csn);=0A+=09=09=09res=20=3D=20= BroadcastFunc(sql);=0A+=0A+error:=0A+=09=09=09if=20(!res)=0A+=09=09=09{=0A= +=09=09=09=09sql=20=3D=20psprintf("ABORT=20PREPARED=20'%s'",=20= fdwTransState->gid);=0A+=09=09=09=09BroadcastCmd(sql);=0A+=09=09=09=09= elog(ERROR,=20"Failed=20to=20PREPARE=20transaction=20on=20remote=20= node");=0A+=09=09=09}=0A+=0A+=09=09=09/*=0A+=09=09=09=20*=20Do=20not=20= fall=20down.=20Consequent=20COMMIT=20event=20will=20clean=20thing=20up.=0A= +=09=09=09=20*/=0A+=09=09=09return;=0A+=09=09}=0A+=09}=0A+=0A+=09/*=20= COMMIT=20open=20transaction=20of=20we=20were=20doing=202PC=20*/=0A+=09if=20= (fdwTransState->two_phase_commit=20&&=0A+=09=09(event=20=3D=3D=20= XACT_EVENT_PARALLEL_COMMIT=20||=20event=20=3D=3D=20XACT_EVENT_COMMIT))=0A= +=09{=0A+=09=09BroadcastCmd(psprintf("COMMIT=20PREPARED=20'%s'",=20= fdwTransState->gid));=0A+=09}=0A+=0A=20=09/*=0A=20=09=20*=20Scan=20all=20= connection=20cache=20entries=20to=20find=20open=20remote=20transactions,=20= and=0A=20=09=20*=20close=20them.=0A@@=20-663,8=20+880,6=20@@=20= pgfdw_xact_callback(XactEvent=20event,=20void=20*arg)=0A=20=09= hash_seq_init(&scan,=20ConnectionHash);=0A=20=09while=20((entry=20=3D=20= (ConnCacheEntry=20*)=20hash_seq_search(&scan)))=0A=20=09{=0A-=09=09= PGresult=20=20=20*res;=0A-=0A=20=09=09/*=20Ignore=20cache=20entry=20if=20= no=20open=20connection=20right=20now=20*/=0A=20=09=09if=20(entry->conn=20= =3D=3D=20NULL)=0A=20=09=09=09continue;=0A@@=20-681,6=20+896,7=20@@=20= pgfdw_xact_callback(XactEvent=20event,=20void=20*arg)=0A=20=09=09=09{=0A=20= =09=09=09=09case=20XACT_EVENT_PARALLEL_PRE_COMMIT:=0A=20=09=09=09=09case=20= XACT_EVENT_PRE_COMMIT:=0A+=09=09=09=09=09= Assert(!fdwTransState->two_phase_commit);=0A=20=0A=20=09=09=09=09=09/*=0A= =20=09=09=09=09=09=20*=20If=20abort=20cleanup=20previously=20failed=20= for=20this=20connection,=0A@@=20-693,28=20+909,7=20@@=20= pgfdw_xact_callback(XactEvent=20event,=20void=20*arg)=0A=20=09=09=09=09=09= do_sql_command(entry->conn,=20"COMMIT=20TRANSACTION");=0A=20=09=09=09=09=09= entry->changing_xact_state=20=3D=20false;=0A=20=0A-=09=09=09=09=09/*=0A-=09= =09=09=09=09=20*=20If=20there=20were=20any=20errors=20in=20= subtransactions,=20and=20we=0A-=09=09=09=09=09=20*=20made=20prepared=20= statements,=20do=20a=20DEALLOCATE=20ALL=20to=20make=0A-=09=09=09=09=09=20= *=20sure=20we=20get=20rid=20of=20all=20prepared=20statements.=20This=20= is=0A-=09=09=09=09=09=20*=20annoying=20and=20not=20terribly=20= bulletproof,=20but=20it's=0A-=09=09=09=09=09=20*=20probably=20not=20= worth=20trying=20harder.=0A-=09=09=09=09=09=20*=0A-=09=09=09=09=09=20*=20= DEALLOCATE=20ALL=20only=20exists=20in=208.3=20and=20later,=20so=20this=0A= -=09=09=09=09=09=20*=20constrains=20how=20old=20a=20server=20= postgres_fdw=20can=0A-=09=09=09=09=09=20*=20communicate=20with.=20=20We=20= intentionally=20ignore=20errors=20in=0A-=09=09=09=09=09=20*=20the=20= DEALLOCATE,=20so=20that=20we=20can=20hobble=20along=20to=20some=0A-=09=09= =09=09=09=20*=20extent=20with=20older=20servers=20(leaking=20prepared=20= statements=0A-=09=09=09=09=09=20*=20as=20we=20go;=20but=20we=20don't=20= really=20support=20update=20operations=0A-=09=09=09=09=09=20*=20pre-8.3=20= anyway).=0A-=09=09=09=09=09=20*/=0A-=09=09=09=09=09if=20= (entry->have_prep_stmt=20&&=20entry->have_error)=0A-=09=09=09=09=09{=0A-=09= =09=09=09=09=09res=20=3D=20PQexec(entry->conn,=20"DEALLOCATE=20ALL");=0A= -=09=09=09=09=09=09PQclear(res);=0A-=09=09=09=09=09}=0A-=09=09=09=09=09= entry->have_prep_stmt=20=3D=20false;=0A-=09=09=09=09=09entry->have_error=20= =3D=20false;=0A+=09=09=09=09=09deallocate_prepared_stmts(entry);=0A=20=09= =09=09=09=09break;=0A=20=09=09=09=09case=20XACT_EVENT_PRE_PREPARE:=0A=20=0A= @@=20-729,10=20+924,15=20@@=20pgfdw_xact_callback(XactEvent=20event,=20= void=20*arg)=0A=20=09=09=09=09=09=20*/=0A=20=09=09=09=09=09= ereport(ERROR,=0A=20=09=09=09=09=09=09=09= (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),=0A-=09=09=09=09=09=09=09=20= errmsg("cannot=20prepare=20a=20transaction=20that=20modified=20remote=20= tables")));=0A+=09=09=09=09=09=09=09=09errmsg("cannot=20prepare=20a=20= transaction=20that=20modified=20remote=20tables")));=0A=20=09=09=09=09=09= break;=0A=20=09=09=09=09case=20XACT_EVENT_PARALLEL_COMMIT:=0A=20=09=09=09= =09case=20XACT_EVENT_COMMIT:=0A+=09=09=09=09=09if=20= (fdwTransState->two_phase_commit)=0A+=09=09=09=09=09=09= deallocate_prepared_stmts(entry);=0A+=09=09=09=09=09else=20/*=20= Pre-commit=20should=20have=20closed=20the=20open=20transaction=20*/=0A+=09= =09=09=09=09=09elog(ERROR,=20"missed=20cleaning=20up=20connection=20= during=20pre-commit");=0A+=09=09=09=09=09break;=0A=20=09=09=09=09case=20= XACT_EVENT_PREPARE:=0A=20=09=09=09=09=09/*=20Pre-commit=20should=20have=20= closed=20the=20open=20transaction=20*/=0A=20=09=09=09=09=09elog(ERROR,=20= "missed=20cleaning=20up=20connection=20during=20pre-commit");=0A@@=20= -828,6=20+1028,38=20@@=20pgfdw_xact_callback(XactEvent=20event,=20void=20= *arg)=0A=20=0A=20=09/*=20Also=20reset=20cursor=20numbering=20for=20next=20= transaction=20*/=0A=20=09cursor_number=20=3D=200;=0A+=0A+=09/*=20Reset=20= fdwTransState=20*/=0A+=09memset(fdwTransState,=20'\0',=20= sizeof(FdwTransactionState));=0A+}=0A+=0A+/*=0A+=20*=20If=20there=20were=20= any=20errors=20in=20subtransactions,=20and=20we=0A+=20*=20made=20= prepared=20statements,=20do=20a=20DEALLOCATE=20ALL=20to=20make=0A+=20*=20= sure=20we=20get=20rid=20of=20all=20prepared=20statements.=20This=20is=0A= +=20*=20annoying=20and=20not=20terribly=20bulletproof,=20but=20it's=0A+=20= *=20probably=20not=20worth=20trying=20harder.=0A+=20*=0A+=20*=20= DEALLOCATE=20ALL=20only=20exists=20in=208.3=20and=20later,=20so=20this=0A= +=20*=20constrains=20how=20old=20a=20server=20postgres_fdw=20can=0A+=20*=20= communicate=20with.=20=20We=20intentionally=20ignore=20errors=20in=0A+=20= *=20the=20DEALLOCATE,=20so=20that=20we=20can=20hobble=20along=20to=20= some=0A+=20*=20extent=20with=20older=20servers=20(leaking=20prepared=20= statements=0A+=20*=20as=20we=20go;=20but=20we=20don't=20really=20support=20= update=20operations=0A+=20*=20pre-8.3=20anyway).=0A+=20*/=0A+static=20= void=0A+deallocate_prepared_stmts(ConnCacheEntry=20*entry)=0A+{=0A+=09= PGresult=20=20=20*res;=0A+=0A+=09if=20(entry->have_prep_stmt=20&&=20= entry->have_error)=0A+=09{=0A+=09=09res=20=3D=20PQexec(entry->conn,=20= "DEALLOCATE=20ALL");=0A+=09=09PQclear(res);=0A+=09}=0A+=09= entry->have_prep_stmt=20=3D=20false;=0A+=09entry->have_error=20=3D=20= false;=0A=20}=0A=20=0A=20/*=0Adiff=20--git=20= a/contrib/postgres_fdw/postgres_fdw.c=20= b/contrib/postgres_fdw/postgres_fdw.c=0Aindex=20312581b741..d1fcacc4ad=20= 100644=0A---=20a/contrib/postgres_fdw/postgres_fdw.c=0A+++=20= b/contrib/postgres_fdw/postgres_fdw.c=0A@@=20-268,6=20+268,9=20@@=20= typedef=20struct=0A=20=09List=09=20=20=20*already_used;=09/*=20= expressions=20already=20dealt=20with=20*/=0A=20}=20= ec_member_foreign_arg;=0A=20=0A+bool=09=09UseGlobalSnapshots;=0A+void=09=09= _PG_init(void);=0A+=0A=20/*=0A=20=20*=20SQL=20functions=0A=20=20*/=0A@@=20= -5772,3=20+5775,12=20@@=20find_em_expr_for_rel(EquivalenceClass=20*ec,=20= RelOptInfo=20*rel)=0A=20=09/*=20We=20didn't=20find=20any=20suitable=20= equivalence=20class=20expression=20*/=0A=20=09return=20NULL;=0A=20}=0A+=0A= +void=0A+_PG_init(void)=0A+{=0A+=09= DefineCustomBoolVariable("postgres_fdw.use_tsdtm",=0A+=09=09=09=09=09=09=09= =20"Use=20global=20snapshots=20for=20FDW=20transactions",=20NULL,=0A+=09=09= =09=09=09=09=09=20&UseGlobalSnapshots,=20false,=20PGC_USERSET,=200,=20= NULL,=0A+=09=09=09=09=09=09=09=20NULL,=20NULL);=0A+}=0Adiff=20--git=20= a/contrib/postgres_fdw/postgres_fdw.h=20= b/contrib/postgres_fdw/postgres_fdw.h=0Aindex=20d37cc88b6e..f39d360157=20= 100644=0A---=20a/contrib/postgres_fdw/postgres_fdw.h=0A+++=20= b/contrib/postgres_fdw/postgres_fdw.h=0A@@=20-184,4=20+184,6=20@@=20= extern=20const=20char=20*get_jointype_name(JoinType=20jointype);=0A=20= extern=20bool=20is_builtin(Oid=20objectId);=0A=20extern=20bool=20= is_shippable(Oid=20objectId,=20Oid=20classId,=20PgFdwRelationInfo=20= *fpinfo);=0A=20=0A+extern=20bool=20UseGlobalSnapshots;=0A+=0A=20#endif=09= =09=09=09=09=09=09/*=20POSTGRES_FDW_H=20*/=0Adiff=20--git=20= a/contrib/postgres_fdw/t/001_bank_coordinator.pl=20= b/contrib/postgres_fdw/t/001_bank_coordinator.pl=0Anew=20file=20mode=20= 100644=0Aindex=200000000000..4e2e8677c7=0A---=20/dev/null=0A+++=20= b/contrib/postgres_fdw/t/001_bank_coordinator.pl=0A@@=20-0,0=20+1,262=20= @@=0A+use=20strict;=0A+use=20warnings;=0A+=0A+use=20PostgresNode;=0A+use=20= TestLib;=0A+use=20Test::More=20tests=20=3D>=203;=0A+=0A+my=20$master=20=3D= =20get_new_node("master");=0A+$master->init;=0A= +$master->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09log_checkpoints=20=3D=20true=0A= +=09postgres_fdw.use_tsdtm=20=3D=20on=0A+));=0A+$master->start;=0A+=0A= +my=20$shard1=20=3D=20get_new_node("shard1");=0A+$shard1->init;=0A= +$shard1->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09global_snapshot_defer_time=20= =3D=2015=0A+=09track_global_snapshots=20=3D=20on=0A+));=0A= +$shard1->start;=0A+=0A+my=20$shard2=20=3D=20get_new_node("shard2");=0A= +$shard2->init;=0A+$shard2->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09global_snapshot_defer_time=20= =3D=2015=0A+=09track_global_snapshots=20=3D=20on=0A+));=0A= +$shard2->start;=0A+=0A= +#########################################################################= ######=0A+#=20Prepare=20nodes=0A= +#########################################################################= ######=0A+=0A+$master->safe_psql('postgres',=20qq[=0A+=09CREATE=20= EXTENSION=20postgres_fdw;=0A+=09CREATE=20TABLE=20accounts(id=20integer=20= primary=20key,=20amount=20integer);=0A+=09CREATE=20TABLE=20= global_transactions(tx_time=20timestamp);=0A+]);=0A+=0A+foreach=20my=20= $node=20($shard1,=20$shard2)=0A+{=0A+=09my=20$port=20=3D=20$node->port;=0A= +=09my=20$host=20=3D=20$node->host;=0A+=0A+=09= $node->safe_psql('postgres',=0A+=09=09=09"CREATE=20TABLE=20accounts(id=20= integer=20primary=20key,=20amount=20integer)");=0A+=0A+=09= $master->safe_psql('postgres',=20qq[=0A+=09=09CREATE=20SERVER=20= shard_$port=20FOREIGN=20DATA=20WRAPPER=20postgres_fdw=20options(dbname=20= 'postgres',=20host=20'$host',=20port=20'$port');=0A+=09=09CREATE=20= FOREIGN=20TABLE=20accounts_fdw_$port()=20inherits=20(accounts)=20server=20= shard_$port=20options(table_name=20'accounts');=0A+=09=09CREATE=20USER=20= MAPPING=20for=20CURRENT_USER=20SERVER=20shard_$port;=0A+=09])=0A+}=0A+=0A= +$shard1->safe_psql('postgres',=20qq[=0A+=09insert=20into=20accounts=20= select=202*id-1,=200=20from=20generate_series(1,=2010010)=20as=20id;=0A+=09= CREATE=20TABLE=20local_transactions(tx_time=20timestamp);=0A+]);=0A+=0A= +$shard2->safe_psql('postgres',=20qq[=0A+=09insert=20into=20accounts=20= select=202*id,=200=20from=20generate_series(1,=2010010)=20as=20id;=0A+=09= CREATE=20TABLE=20local_transactions(tx_time=20timestamp);=0A+]);=0A+=0A= +diag("master:=20@{[$master->connstr('postgres')]}");=0A+diag("shard1:=20= @{[$shard1->connstr('postgres')]}");=0A+diag("shard2:=20= @{[$shard2->connstr('postgres')]}");=0A+=0A= +#########################################################################= ######=0A+#=20pgbench=20scripts=0A= +#########################################################################= ######=0A+=0A+my=20$bank=20=3D=20File::Temp->new();=0A= +append_to_file($bank,=20q{=0A+=09\set=20id=20random(1,=2020000)=0A+=09= BEGIN;=0A+=09WITH=20upd=20AS=20(UPDATE=20accounts=20SET=20amount=20=3D=20= amount=20-=201=20WHERE=20id=20=3D=20:id=20RETURNING=20*)=0A+=09=09INSERT=20= into=20global_transactions=20SELECT=20now()=20FROM=20upd;=0A+=09UPDATE=20= accounts=20SET=20amount=20=3D=20amount=20+=201=20WHERE=20id=20=3D=20(:id=20= +=201);=0A+=09COMMIT;=0A+});=0A+=0A+my=20$bank1=20=3D=20= File::Temp->new();=0A+append_to_file($bank1,=20q{=0A+=09\set=20id=20= random(1,=2010000)=0A+=09BEGIN;=0A+=09WITH=20upd=20AS=20(UPDATE=20= accounts=20SET=20amount=20=3D=20amount=20-=201=20WHERE=20id=20=3D=20= (2*:id=20+=201)=20RETURNING=20*)=0A+=09=09INSERT=20into=20= local_transactions=20SELECT=20now()=20FROM=20upd;=0A+=09UPDATE=20= accounts=20SET=20amount=20=3D=20amount=20+=201=20WHERE=20id=20=3D=20= (2*:id=20+=203);=0A+=09COMMIT;=0A+});=0A+=0A+my=20$bank2=20=3D=20= File::Temp->new();=0A+append_to_file($bank2,=20q{=0A+=09\set=20id=20= random(1,=2010000)=0A+=0A+=09BEGIN;=0A+=09WITH=20upd=20AS=20(UPDATE=20= accounts=20SET=20amount=20=3D=20amount=20-=201=20WHERE=20id=20=3D=20= 2*:id=20RETURNING=20*)=0A+=09=09INSERT=20into=20local_transactions=20= SELECT=20now()=20FROM=20upd;=0A+=09UPDATE=20accounts=20SET=20amount=20=3D=20= amount=20+=201=20WHERE=20id=20=3D=20(2*:id=20+=202);=0A+=09COMMIT;=0A= +});=0A+=0A= +#########################################################################= ######=0A+#=20Helpers=0A= +#########################################################################= ######=0A+=0A+sub=20count_and_delete_rows=0A+{=0A+=09my=20($node,=20= $table)=20=3D=20@_;=0A+=09my=20$count;=0A+=0A+=09$count=20=3D=20= $node->safe_psql('postgres',"select=20count(*)=20from=20$table");=0A+=09= $node->safe_psql('postgres',"delete=20from=20$table");=0A+=09= diag($node->name,=20":=20completed=20$count=20transactions");=0A+=09= return=20$count;=0A+}=0A+=0A= +#########################################################################= ######=0A+#=20Concurrent=20global=20transactions=0A= +#########################################################################= ######=0A+=0A+my=20($err,=20$rc);=0A+my=20$started;=0A+my=20$seconds=20=3D= =2030;=0A+my=20$selects;=0A+my=20$total=20=3D=20'0';=0A+my=20$oldtotal=20= =3D=20'0';=0A+my=20$isolation_errors=20=3D=200;=0A+=0A+=0A+my=20= $pgb_handle;=0A+=0A+$pgb_handle=20=3D=20$master->pgbench_async(-n,=20-c=20= =3D>=205,=20-T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20);=0A= +=0A+$started=20=3D=20time();=0A+$selects=20=3D=200;=0A+while=20(time()=20= -=20$started=20<=20$seconds)=0A+{=0A+=09$total=20=3D=20= $master->safe_psql('postgres',=20"select=20sum(amount)=20from=20= accounts");=0A+=09if=20(=20($total=20ne=20$oldtotal)=20and=20($total=20= ne=20'')=20)=0A+=09{=0A+=09=09$isolation_errors++;=0A+=09=09$oldtotal=20= =3D=20$total;=0A+=09=09diag("Isolation=20error.=20Total=20=3D=20= $total");=0A+=09}=0A+=09if=20($total=20ne=20'')=20{=20$selects++;=20}=0A= +}=0A+=0A+$master->pgbench_await($pgb_handle);=0A+=0A+#=20sanity=20check=0A= +diag("completed=20$selects=20selects");=0A+die=20"no=20actual=20= transactions=20happend"=20unless=20(=20$selects=20>=200=20&&=0A+=09= count_and_delete_rows($master,=20'global_transactions')=20>=200);=0A+=0A= +is($isolation_errors,=200,=20'isolation=20between=20concurrent=20global=20= transaction');=0A+=0A= +#########################################################################= ######=0A+#=20Concurrent=20global=20and=20local=20transactions=0A= +#########################################################################= ######=0A+=0A+my=20($pgb_handle1,=20$pgb_handle2,=20$pgb_handle3);=0A+=0A= +#=20global=20txses=0A+$pgb_handle1=20=3D=20$master->pgbench_async(-n,=20= -c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20= );=0A+=0A+#=20concurrent=20local=0A+$pgb_handle2=20=3D=20= $shard1->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank1,=20'postgres'=20);=0A+$pgb_handle3=20=3D=20= $shard2->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank2,=20'postgres'=20);=0A+=0A+$started=20=3D=20time();=0A= +$selects=20=3D=200;=0A+$oldtotal=20=3D=200;=0A+while=20(time()=20-=20= $started=20<=20$seconds)=0A+{=0A+=09$total=20=3D=20= $master->safe_psql('postgres',=20"select=20sum(amount)=20from=20= accounts");=0A+=09if=20(=20($total=20ne=20$oldtotal)=20and=20($total=20= ne=20'')=20)=0A+=09{=0A+=09=09$isolation_errors++;=0A+=09=09$oldtotal=20= =3D=20$total;=0A+=09=09diag("Isolation=20error.=20Total=20=3D=20= $total");=0A+=09}=0A+=09if=20($total=20ne=20'')=20{=20$selects++;=20}=0A= +}=0A+=0A+diag("selects=20=3D=20$selects");=0A= +$master->pgbench_await($pgb_handle1);=0A= +$shard1->pgbench_await($pgb_handle2);=0A= +$shard2->pgbench_await($pgb_handle3);=0A+=0A+diag("completed=20$selects=20= selects");=0A+die=20""=20unless=20(=20$selects=20>=200=20&&=0A+=09= count_and_delete_rows($master,=20'global_transactions')=20>=200=20&&=0A+=09= count_and_delete_rows($shard1,=20'local_transactions')=20>=200=20&&=0A+=09= count_and_delete_rows($shard2,=20'local_transactions')=20>=200);=0A+=0A= +is($isolation_errors,=200,=20'isolation=20between=20concurrent=20global=20= and=20local=20transactions');=0A+=0A+=0A= +#########################################################################= ######=0A+#=20Snapshot=20stability=0A= +#########################################################################= ######=0A+=0A+my=20($hashes,=20$hash1,=20$hash2);=0A+my=20= $stability_errors=20=3D=200;=0A+=0A+#=20global=20txses=0A+$pgb_handle1=20= =3D=20$master->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20= $seconds,=20-f=20=3D>=20$bank,=20'postgres'=20);=0A+#=20concurrent=20= local=0A+$pgb_handle2=20=3D=20$shard1->pgbench_async(-n,=20-c=20=3D>=20= 5,=20-T=20=3D>=20$seconds,=20-f=20=3D>=20$bank1,=20'postgres'=20);=0A= +$pgb_handle3=20=3D=20$shard2->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20= =3D>=20$seconds,=20-f=20=3D>=20$bank2,=20'postgres'=20);=0A+=0A+$selects=20= =3D=200;=0A+$started=20=3D=20time();=0A+while=20(time()=20-=20$started=20= <=20$seconds)=0A+{=0A+=09foreach=20my=20$node=20($master,=20$shard1,=20= $shard2)=0A+=09{=0A+=09=09($hash1,=20$_,=20$hash2)=20=3D=20split=20"\n",=20= $node->safe_psql('postgres',=20qq[=0A+=09=09=09begin=20isolation=20level=20= repeatable=20read;=0A+=09=09=09select=20= md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20accounts=20= order=20by=20id)=20as=20t;=0A+=09=09=09select=20pg_sleep(3);=0A+=09=09=09= select=20md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20= accounts=20order=20by=20id)=20as=20t;=0A+=09=09=09commit;=0A+=09=09]);=0A= +=0A+=09=09if=20($hash1=20ne=20$hash2)=0A+=09=09{=0A+=09=09=09= diag("oops");=0A+=09=09=09$stability_errors++;=0A+=09=09}=0A+=09=09elsif=20= ($hash1=20eq=20''=20or=20$hash2=20eq=20'')=0A+=09=09{=0A+=09=09=09die;=0A= +=09=09}=0A+=09=09else=0A+=09=09{=0A+=09=09=09$selects++;=0A+=09=09}=0A+=09= }=0A+}=0A+=0A+$master->pgbench_await($pgb_handle1);=0A= +$shard1->pgbench_await($pgb_handle2);=0A= +$shard2->pgbench_await($pgb_handle3);=0A+=0A+die=20""=20unless=20(=20= $selects=20>=200=20&&=0A+=09count_and_delete_rows($master,=20= 'global_transactions')=20>=200=20&&=0A+=09count_and_delete_rows($shard1,=20= 'local_transactions')=20>=200=20&&=0A+=09count_and_delete_rows($shard2,=20= 'local_transactions')=20>=200);=0A+=0A+is($stability_errors,=200,=20= 'snapshot=20is=20stable=20during=20concurrent=20global=20and=20local=20= transactions');=0A+=0A+$master->stop;=0A+$shard1->stop;=0A= +$shard2->stop;=0Adiff=20--git=20= a/contrib/postgres_fdw/t/002_bank_participant.pl=20= b/contrib/postgres_fdw/t/002_bank_participant.pl=0Anew=20file=20mode=20= 100644=0Aindex=200000000000..bc5832b54d=0A---=20/dev/null=0A+++=20= b/contrib/postgres_fdw/t/002_bank_participant.pl=0A@@=20-0,0=20+1,238=20= @@=0A+use=20strict;=0A+use=20warnings;=0A+=0A+use=20PostgresNode;=0A+use=20= TestLib;=0A+use=20Test::More=20tests=20=3D>=203;=0A+=0A+my=20$shard1=20=3D= =20get_new_node("shard1");=0A+$shard1->init;=0A= +$shard1->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09postgres_fdw.use_tsdtm=20=3D=20= on=0A+=09global_snapshot_defer_time=20=3D=2015=0A+=09= track_global_snapshots=20=3D=20on=0A+));=0A+$shard1->start;=0A+=0A+my=20= $shard2=20=3D=20get_new_node("shard2");=0A+$shard2->init;=0A= +$shard2->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09postgres_fdw.use_tsdtm=20=3D=20= on=0A+=09global_snapshot_defer_time=20=3D=2015=0A+=09= track_global_snapshots=20=3D=20on=0A+));=0A+$shard2->start;=0A+=0A= +#########################################################################= ######=0A+#=20Prepare=20nodes=0A= +#########################################################################= ######=0A+=0A+my=20@shards=20=3D=20($shard1,=20$shard2);=0A+=0A+foreach=20= my=20$node=20(@shards)=0A+{=0A+=09$node->safe_psql('postgres',=20qq[=0A+=09= =09CREATE=20EXTENSION=20postgres_fdw;=0A+=09=09CREATE=20TABLE=20= accounts(id=20integer=20primary=20key,=20amount=20integer);=0A+=09=09= CREATE=20TABLE=20accounts_local()=20inherits(accounts);=0A+=09=09CREATE=20= TABLE=20global_transactions(tx_time=20timestamp);=0A+=09=09CREATE=20= TABLE=20local_transactions(tx_time=20timestamp);=0A+=09]);=0A+=0A+=09= foreach=20my=20$neighbor=20(@shards)=0A+=09{=0A+=09=09next=20if=20= ($neighbor=20eq=20$node);=0A+=0A+=09=09my=20$port=20=3D=20= $neighbor->port;=0A+=09=09my=20$host=20=3D=20$neighbor->host;=0A+=0A+=09=09= $node->safe_psql('postgres',=20qq[=0A+=09=09=09CREATE=20SERVER=20= shard_$port=20FOREIGN=20DATA=20WRAPPER=20postgres_fdw=0A+=09=09=09=09=09= options(dbname=20'postgres',=20host=20'$host',=20port=20'$port');=0A+=09=09= =09CREATE=20FOREIGN=20TABLE=20accounts_fdw_$port()=20inherits=20= (accounts)=0A+=09=09=09=09=09server=20shard_$port=20options(table_name=20= 'accounts_local');=0A+=09=09=09CREATE=20USER=20MAPPING=20for=20= CURRENT_USER=20SERVER=20shard_$port;=0A+=09=09]);=0A+=09}=0A+}=0A+=0A= +$shard1->psql('postgres',=20"insert=20into=20accounts_local=20select=20= 2*id-1,=200=20from=20generate_series(1,=2010010)=20as=20id;");=0A= +$shard2->psql('postgres',=20"insert=20into=20accounts_local=20select=20= 2*id,=20=20=200=20from=20generate_series(1,=2010010)=20as=20id;");=0A+=0A= +#########################################################################= ######=0A+#=20pgbench=20scripts=0A= +#########################################################################= ######=0A+=0A+my=20$bank=20=3D=20File::Temp->new();=0A= +append_to_file($bank,=20q{=0A+=09\set=20id=20random(1,=2020000)=0A+=09= BEGIN;=0A+=09WITH=20upd=20AS=20(UPDATE=20accounts=20SET=20amount=20=3D=20= amount=20-=201=20WHERE=20id=20=3D=20:id=20RETURNING=20*)=0A+=09=09INSERT=20= into=20global_transactions=20SELECT=20now()=20FROM=20upd;=0A+=09UPDATE=20= accounts=20SET=20amount=20=3D=20amount=20+=201=20WHERE=20id=20=3D=20(:id=20= +=201);=0A+=09COMMIT;=0A+});=0A+=0A= +#########################################################################= ######=0A+#=20Helpers=0A= +#########################################################################= ######=0A+=0A+sub=20count_and_delete_rows=0A+{=0A+=09my=20($node,=20= $table)=20=3D=20@_;=0A+=09my=20$count;=0A+=0A+=09$count=20=3D=20= $node->safe_psql('postgres',"select=20count(*)=20from=20$table");=0A+=09= $node->safe_psql('postgres',"delete=20from=20$table");=0A+=09= diag($node->name,=20":=20completed=20$count=20transactions");=0A+=09= return=20$count;=0A+}=0A+=0A= +#########################################################################= ######=0A+#=20Concurrent=20global=20transactions=0A= +#########################################################################= ######=0A+=0A+my=20($err,=20$rc);=0A+my=20$started;=0A+my=20$seconds=20=3D= =2030;=0A+my=20$selects;=0A+my=20$total=20=3D=20'0';=0A+my=20$oldtotal=20= =3D=20'0';=0A+my=20$isolation_errors=20=3D=200;=0A+my=20$i;=0A+=0A+=0A= +my=20($pgb_handle1,=20$pgb_handle2);=0A+=0A+$pgb_handle1=20=3D=20= $shard1->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank,=20'postgres'=20);=0A+$pgb_handle2=20=3D=20= $shard2->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank,=20'postgres'=20);=0A+=0A+$started=20=3D=20time();=0A= +$selects=20=3D=200;=0A+$i=20=3D=200;=0A+while=20(time()=20-=20$started=20= <=20$seconds)=0A+{=0A+=09my=20$shard=20=3D=20$shard1;=0A+=09foreach=20my=20= $shard=20(@shards)=0A+=09{=0A+=09=09$total=20=3D=20= $shard->safe_psql('postgres',=20"select=20sum(amount)=20from=20= accounts");=0A+=09=09if=20(=20($total=20ne=20$oldtotal)=20and=20($total=20= ne=20'')=20)=0A+=09=09{=0A+=09=09=09$isolation_errors++;=0A+=09=09=09= $oldtotal=20=3D=20$total;=0A+=09=09=09diag("$i:=20Isolation=20error.=20= Total=20=3D=20$total");=0A+=09=09}=0A+=09=09if=20($total=20ne=20'')=20{=20= $selects++;=20}=0A+=09}=0A+=09$i++;=0A+}=0A+=0A= +$shard1->pgbench_await($pgb_handle1);=0A= +$shard2->pgbench_await($pgb_handle2);=0A+=0A+#=20sanity=20check=0A= +diag("completed=20$selects=20selects");=0A+die=20"no=20actual=20= transactions=20happend"=20unless=20(=20$selects=20>=200=20&&=0A+=09= count_and_delete_rows($shard1,=20'global_transactions')=20>=200=20&&=0A+=09= count_and_delete_rows($shard2,=20'global_transactions')=20>=200);=0A+=0A= +is($isolation_errors,=200,=20'isolation=20between=20concurrent=20global=20= transaction');=0A+=0A= +#########################################################################= ######=0A+#=20And=20do=20the=20same=20after=20soft=20restart=0A= +#########################################################################= ######=0A+=0A+$shard1->restart;=0A+$shard2->restart;=0A= +$shard1->poll_query_until('postgres',=20"select=20't'")=0A+=09or=20die=20= "Timed=20out=20waiting=20for=20shard1=20to=20became=20online";=0A= +$shard2->poll_query_until('postgres',=20"select=20't'")=0A+=09or=20die=20= "Timed=20out=20waiting=20for=20shard2=20to=20became=20online";=0A+=0A= +$seconds=20=3D=2015;=0A+$pgb_handle1=20=3D=20$shard1->pgbench_async(-n,=20= -c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20= );=0A+$pgb_handle2=20=3D=20$shard2->pgbench_async(-n,=20-c=20=3D>=205,=20= -T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20);=0A+=0A= +$started=20=3D=20time();=0A+$selects=20=3D=200;=0A+$i=20=3D=200;=0A+=0A= +while=20(time()=20-=20$started=20<=20$seconds)=0A+{=0A+=09my=20$shard=20= =3D=20$shard1;=0A+=09foreach=20my=20$shard=20(@shards)=0A+=09{=0A+=09=09= $total=20=3D=20$shard->safe_psql('postgres',=20"select=20sum(amount)=20= from=20accounts");=0A+=09=09if=20(=20($total=20ne=20$oldtotal)=20and=20= ($total=20ne=20'')=20)=0A+=09=09{=0A+=09=09=09$isolation_errors++;=0A+=09= =09=09$oldtotal=20=3D=20$total;=0A+=09=09=09diag("$i:=20Isolation=20= error.=20Total=20=3D=20$total");=0A+=09=09}=0A+=09=09if=20($total=20ne=20= '')=20{=20$selects++;=20}=0A+=09}=0A+=09$i++;=0A+}=0A+=0A= +$shard1->pgbench_await($pgb_handle1);=0A= +$shard2->pgbench_await($pgb_handle2);=0A+=0A+#=20sanity=20check=0A= +diag("completed=20$selects=20selects");=0A+die=20"no=20actual=20= transactions=20happend"=20unless=20(=20$selects=20>=200=20&&=0A+=09= count_and_delete_rows($shard1,=20'global_transactions')=20>=200=20&&=0A+=09= count_and_delete_rows($shard2,=20'global_transactions')=20>=200);=0A+=0A= +is($isolation_errors,=200,=20'isolation=20between=20concurrent=20global=20= transaction=20after=20restart');=0A+=0A= +#########################################################################= ######=0A+#=20And=20do=20the=20same=20after=20hard=20restart=0A= +#########################################################################= ######=0A+=0A+$shard1->teardown_node;=0A+$shard2->teardown_node;=0A= +$shard1->start;=0A+$shard2->start;=0A= +$shard1->poll_query_until('postgres',=20"select=20't'")=0A+=09or=20die=20= "Timed=20out=20waiting=20for=20shard1=20to=20became=20online";=0A= +$shard2->poll_query_until('postgres',=20"select=20't'")=0A+=09or=20die=20= "Timed=20out=20waiting=20for=20shard2=20to=20became=20online";=0A+=0A+=0A= +$seconds=20=3D=2015;=0A+$pgb_handle1=20=3D=20$shard1->pgbench_async(-n,=20= -c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20= );=0A+$pgb_handle2=20=3D=20$shard2->pgbench_async(-n,=20-c=20=3D>=205,=20= -T=20=3D>=20$seconds,=20-f=20=3D>=20$bank,=20'postgres'=20);=0A+=0A= +$started=20=3D=20time();=0A+$selects=20=3D=200;=0A+$i=20=3D=200;=0A+=0A= +while=20(time()=20-=20$started=20<=20$seconds)=0A+{=0A+=09my=20$shard=20= =3D=20$shard1;=0A+=09foreach=20my=20$shard=20(@shards)=0A+=09{=0A+=09=09= $total=20=3D=20$shard->safe_psql('postgres',=20"select=20sum(amount)=20= from=20accounts");=0A+=09=09if=20(=20($total=20ne=20$oldtotal)=20and=20= ($total=20ne=20'')=20)=0A+=09=09{=0A+=09=09=09$isolation_errors++;=0A+=09= =09=09$oldtotal=20=3D=20$total;=0A+=09=09=09diag("$i:=20Isolation=20= error.=20Total=20=3D=20$total");=0A+=09=09}=0A+=09=09if=20($total=20ne=20= '')=20{=20$selects++;=20}=0A+=09}=0A+=09$i++;=0A+}=0A+=0A= +$shard1->pgbench_await($pgb_handle1);=0A= +$shard2->pgbench_await($pgb_handle2);=0A+=0A+#=20sanity=20check=0A= +diag("completed=20$selects=20selects");=0A+die=20"no=20actual=20= transactions=20happend"=20unless=20(=20$selects=20>=200=20&&=0A+=09= count_and_delete_rows($shard1,=20'global_transactions')=20>=200=20&&=0A+=09= count_and_delete_rows($shard2,=20'global_transactions')=20>=200);=0A+=0A= +is($isolation_errors,=200,=20'isolation=20between=20concurrent=20global=20= transaction=20after=20hard=20restart');=0Adiff=20--git=20= a/contrib/postgres_fdw/t/003_bank_shared.pl=20= b/contrib/postgres_fdw/t/003_bank_shared.pl=0Anew=20file=20mode=20100644=0A= index=200000000000..db80637562=0A---=20/dev/null=0A+++=20= b/contrib/postgres_fdw/t/003_bank_shared.pl=0A@@=20-0,0=20+1,155=20@@=0A= +use=20strict;=0A+use=20warnings;=0A+=0A+use=20PostgresNode;=0A+use=20= TestLib;=0A+use=20Test::More=20tests=20=3D>=202;=0A+=0A+my=20$shard1=20=3D= =20get_new_node("shard1");=0A+$shard1->init;=0A= +$shard1->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09postgres_fdw.use_tsdtm=20=3D=20= on=0A+=09global_snapshot_defer_time=20=3D=2015=0A+=09= track_global_snapshots=20=3D=20on=0A+));=0A+$shard1->start;=0A+=0A+my=20= $shard2=20=3D=20get_new_node("shard2");=0A+$shard2->init;=0A= +$shard2->append_conf('postgresql.conf',=20qq(=0A+=09= max_prepared_transactions=20=3D=2030=0A+=09postgres_fdw.use_tsdtm=20=3D=20= on=0A+=09global_snapshot_defer_time=20=3D=2015=0A+=09= track_global_snapshots=20=3D=20on=0A+));=0A+$shard2->start;=0A+=0A= +#########################################################################= ######=0A+#=20Prepare=20nodes=0A= +#########################################################################= ######=0A+=0A+my=20@shards=20=3D=20($shard1,=20$shard2);=0A+=0A+foreach=20= my=20$node=20(@shards)=0A+{=0A+=09$node->safe_psql('postgres',=20qq[=0A+=09= =09CREATE=20EXTENSION=20postgres_fdw;=0A+=09=09CREATE=20TABLE=20= accounts(id=20integer=20primary=20key,=20amount=20integer);=0A+=09]);=0A= +=0A+=09foreach=20my=20$neighbor=20(@shards)=0A+=09{=0A+=09=09next=20if=20= ($neighbor=20eq=20$node);=0A+=0A+=09=09my=20$port=20=3D=20= $neighbor->port;=0A+=09=09my=20$host=20=3D=20$neighbor->host;=0A+=0A+=09=09= $node->safe_psql('postgres',=20qq[=0A+=09=09=09CREATE=20SERVER=20= shard_$port=20FOREIGN=20DATA=20WRAPPER=20postgres_fdw=0A+=09=09=09=09=09= options(dbname=20'postgres',=20host=20'$host',=20port=20'$port');=0A+=09=09= =09CREATE=20FOREIGN=20TABLE=0A+=09=09=09=09=09accounts_fdw(id=20integer,=20= amount=20integer)=0A+=09=09=09=09=09server=20shard_$port=20= options(table_name=20'accounts');=0A+=09=09=09CREATE=20USER=20MAPPING=20= for=20CURRENT_USER=20SERVER=20shard_$port;=0A+=09=09]);=0A+=09}=0A+=0A+}=0A= +=0A+$shard1->psql('postgres',=20"insert=20into=20accounts=20select=20= id,=200=20from=20generate_series(1,=2020020)=20as=20id;");=0A= +$shard2->psql('postgres',=20"insert=20into=20accounts=20select=20id,=20= 0=20from=20generate_series(1,=2020020)=20as=20id;");=0A+=0A+=0A+#=20= diag("shard1:=20@{[$shard1->connstr('postgres')]}");=0A+#=20= diag("shard1:=20@{[$shard2->connstr('postgres')]}");=0A+#=20sleep(3600);=0A= +=0A+=0A= +#########################################################################= ######=0A+#=20pgbench=20scripts=0A= +#########################################################################= ######=0A+=0A+my=20$bank=20=3D=20File::Temp->new();=0A= +append_to_file($bank,=20q{=0A+=09\set=20id=20random(1,=2020000)=0A+=09= BEGIN;=0A+=09UPDATE=20accounts=20SET=20amount=20=3D=20amount=20-=201=20= WHERE=20id=20=3D=20:id;=0A+=09UPDATE=20accounts=20SET=20amount=20=3D=20= amount=20+=201=20WHERE=20id=20=3D=20(:id=20+=201);=0A+=09UPDATE=20= accounts_fdw=20SET=20amount=20=3D=20amount=20-=201=20WHERE=20id=20=3D=20= :id;=0A+=09UPDATE=20accounts_fdw=20SET=20amount=20=3D=20amount=20+=201=20= WHERE=20id=20=3D=20(:id=20+=201);=0A+=09COMMIT;=0A+});=0A+=0A= +#########################################################################= ######=0A+#=20Concurrent=20global=20transactions=0A= +#########################################################################= ######=0A+=0A+my=20($hashes,=20$hash1,=20$hash2);=0A+my=20= $stability_errors=20=3D=200;=0A+my=20$selects=20=3D=200;=0A+my=20= $seconds=20=3D=2030;=0A+=0A+my=20$pgb_handle1=20=3D=20= $shard1->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank,=20'postgres'=20);=0A+#=20my=20$pgb_handle2=20=3D=20= $shard2->pgbench_async(-n,=20-c=20=3D>=205,=20-T=20=3D>=20$seconds,=20-f=20= =3D>=20$bank,=20'postgres'=20);=0A+my=20$started=20=3D=20time();=0A= +while=20(time()=20-=20$started=20<=20$seconds)=0A+{=0A+=09foreach=20my=20= $node=20($shard1,=20$shard2)=0A+=09{=0A+=09=09($hash1,=20$_,=20$hash2)=20= =3D=20split=20"\n",=20$node->safe_psql('postgres',=20qq[=0A+=09=09=09= begin=20isolation=20level=20repeatable=20read;=0A+=09=09=09select=20= md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20accounts=20= order=20by=20id)=20as=20t;=0A+=09=09=09select=20pg_sleep(1);=0A+=09=09=09= select=20md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20= accounts_fdw=20order=20by=20id)=20as=20t;=0A+=09=09=09commit;=0A+=09=09= ]);=0A+=0A+=09=09if=20($hash1=20ne=20$hash2)=0A+=09=09{=0A+=09=09=09= diag("oops");=0A+=09=09=09$stability_errors++;=0A+=09=09}=0A+=09=09elsif=20= ($hash1=20eq=20''=20or=20$hash2=20eq=20'')=0A+=09=09{=0A+=09=09=09die;=0A= +=09=09}=0A+=09=09else=0A+=09=09{=0A+=09=09=09diag("got=20hash=20= $hash1");=0A+=09=09=09$selects++;=0A+=09=09}=0A+=09}=0A+}=0A+=0A= +$shard1->pgbench_await($pgb_handle1);=0A+#=20= $shard2->pgbench_await($pgb_handle2);=0A+=0A+die=20"no=20real=20queries=20= happend"=20unless=20(=20$selects=20>=200=20);=0A+=0A= +is($stability_errors,=200,=20'snapshot=20is=20stable=20during=20= concurrent=20global=20and=20local=20transactions');=0A+=0A= +#########################################################################= ######=0A+#=20Snapshot=20should=20be=20stable=20across=20consequent=20= imports=0A= +#########################################################################= ######=0A+=0A+my=20$snap=20=3D=20$shard1->safe_psql('postgres',=20= 'select=20pg_global_snaphot_export()');=0A+my=20@resarr=20=3D=20split=20= "\n",=20$shard1->safe_psql('postgres',=20qq[=0A+=09begin=20isolation=20= level=20repeatable=20read;=0A+=09select=20= pg_global_snaphot_import($snap);=0A+=09select=20= md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20accounts=20= order=20by=20id)=20as=20t;=0A+=09commit;=0A+=0A+=09update=20accounts=20= set=20amount=20=3D=20amount=20+=201;=0A+=0A+=09begin=20isolation=20level=20= repeatable=20read;=0A+=09select=20pg_global_snaphot_import($snap);=0A+=09= select=20md5(array_agg((t.*)::text)::text)=20from=20(select=20*=20from=20= accounts=20order=20by=20id)=20as=20t;=0A+=09commit;=0A+]);=0A+=0A= +diag($resarr[1],=20"=20-=20",=20$resarr[3]);=0A+is($resarr[1],=20= $resarr[3],=20'snapshot=20is=20stable=20across=20consequent=20imports');=0A= +=0A+$shard1->stop;=0A+$shard2->stop;=0Adiff=20--git=20= a/src/test/perl/PostgresNode.pm=20b/src/test/perl/PostgresNode.pm=0A= index=203b06e78174..56c427b071=20100644=0A---=20= a/src/test/perl/PostgresNode.pm=0A+++=20b/src/test/perl/PostgresNode.pm=0A= @@=20-1757,6=20+1757,37=20@@=20sub=20pg_recvlogical_upto=0A=20=09}=0A=20= }=0A=20=0A+sub=20pgbench()=0A+{=0A+=09my=20($self,=20$node,=20@args)=20=3D= =20@_;=0A+=09my=20$pgbench_handle=20=3D=20$self->pgbench_async($node,=20= @args);=0A+=09$self->pgbench_await($pgbench_handle);=0A+}=0A+=0A+sub=20= pgbench_async()=0A+{=0A+=09my=20($self,=20@args)=20=3D=20@_;=0A+=0A+=09= my=20($in,=20$out,=20$err,=20$rc);=0A+=09$in=20=3D=20'';=0A+=09$out=20=3D=20= '';=0A+=0A+=09my=20@pgbench_command=20=3D=20(=0A+=09=09'pgbench',=0A+=09=09= -h=20=3D>=20$self->host,=0A+=09=09-p=20=3D>=20$self->port,=0A+=09=09= @args=0A+=09);=0A+=09my=20$handle=20=3D=20= IPC::Run::start(\@pgbench_command,=20$in,=20$out);=0A+=09return=20= $handle;=0A+}=0A+=0A+sub=20pgbench_await()=0A+{=0A+=09my=20($self,=20= $pgbench_handle)=20=3D=20@_;=0A+=09IPC::Run::finish($pgbench_handle)=20= ||=20BAIL_OUT("pgbench=20exited=20with=20$?");=0A+}=0A+=0A=20=3Dpod=0A=20= =0A=20=3Dback=0A--=20=0A2.16.2=0A=0A= --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55 Content-Transfer-Encoding: 7bit Content-Type: text/plain; charset=us-ascii --Apple-Mail=_103B2265-CBEC-42BC-8DE1-951872D45D55--