Skip to content

G
Gpdb

Greenplum / Gpdb

通知 7
Star 1
Fork 0
G
Gpdb

体验新版 GitCode，发现更多精彩内容 >>

提交 a52e567c 编写于 作者: A Adam Berlin 提交者： Adam Berlin
浏览文件
操作

Thread the desired distributed transaction context through for getting snapshot data. 

No longer rely on a global variable to determine the distributed snapshot context.
上级 92e9a194
隐藏空白更改
内联 并排
Showing with 96 addition and 70 deletion
src/backend/access/transam/xact.c
浏览文件 @ a52e567c
......@@ -2123,29 +2123,29 @@ AtSubCleanup_Memory(void)
* DOH: this totally ignores subtransactions for now!
*/
void
SetSharedTransactionId_writer(void)
SetSharedTransactionId_writer(DtxContext distributedTransactionContext)
{
Assert(SharedLocalSnapshotSlot != NULL);
Assert(LWLockHeldByMe(SharedLocalSnapshotSlot->slotLock));
Assert(DistributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE ||
DistributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER ||
DistributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER ||
DistributedTransactionContext == DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT);
Assert(distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE ||
distributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER ||
distributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER ||
distributedTransactionContext == DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT);
ereportif(Debug_print_full_dtm, LOG,
(errmsg("%s setting shared xid %u -> %u",
DtxContextToString(DistributedTransactionContext),
DtxContextToString(distributedTransactionContext),
SharedLocalSnapshotSlot->xid,
TopTransactionStateData.transactionId)));
SharedLocalSnapshotSlot->xid = TopTransactionStateData.transactionId;
}
void
SetSharedTransactionId_reader(TransactionId xid, CommandId cid)
SetSharedTransactionId_reader(TransactionId xid, CommandId cid, DtxContext distributedTransactionContext)
{
Assert(DistributedTransactionContext == DTX_CONTEXT_QE_READER ||
DistributedTransactionContext == DTX_CONTEXT_QE_ENTRY_DB_SINGLETON);
Assert(distributedTransactionContext == DTX_CONTEXT_QE_READER ||
distributedTransactionContext == DTX_CONTEXT_QE_ENTRY_DB_SINGLETON);
/*
* For DTX_CONTEXT_QE_READER or DTX_CONTEXT_QE_ENTRY_DB_SINGLETON, during
......
src/backend/cdb/dispatcher/cdbdisp_dtx.c
浏览文件 @ a52e567c
......@@ -165,6 +165,7 @@ qdSerializeDtxContextInfo(int *size, bool wantSnapshot, bool inCursor,
Snapshot snapshot = NULL;
int serializedLen;
DtxContextInfo *pDtxContextInfo = NULL;
DtxContext currentDistributedTransactionContext = DistributedTransactionContext;
/*
* If 'wantSnapshot' is set, then serialize the ActiveSnapshot. The
......@@ -179,7 +180,7 @@ qdSerializeDtxContextInfo(int *size, bool wantSnapshot, bool inCursor,
snapshot = GetActiveSnapshot();
}
switch (DistributedTransactionContext)
switch (currentDistributedTransactionContext)
{
case DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE:
case DTX_CONTEXT_LOCAL_ONLY:
......@@ -191,8 +192,11 @@ qdSerializeDtxContextInfo(int *size, bool wantSnapshot, bool inCursor,
if (DistributedTransactionContext ==
DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE && snapshot != NULL)
{
updateSharedLocalSnapshot(&TempQDDtxContextInfo, snapshot,
"qdSerializeDtxContextInfo");
updateSharedLocalSnapshot(
&TempQDDtxContextInfo,
currentDistributedTransactionContext,
snapshot,
"qdSerializeDtxContextInfo");
}
pDtxContextInfo = &TempQDDtxContextInfo;
......
src/backend/storage/ipc/procarray.c
浏览文件 @ a52e567c
......@@ -1513,7 +1513,10 @@ GetLocalOldestXmin(Relation rel, bool ignoreVacuum)
}
void
updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo, Snapshot snapshot, char *debugCaller)
updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo,
DtxContext distributedTransactionContext,
Snapshot snapshot,
char *debugCaller)
{
int combocidSize;
......@@ -1523,7 +1526,7 @@ updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo, Snapshot snapshot, cha
ereport((Debug_print_full_dtm ? LOG : DEBUG5),
(errmsg("updateSharedLocalSnapshot for DistributedTransactionContext = '%s' passed local snapshot (xmin: %u xmax: %u xcnt: %u) curcid: %d",
DtxContextToString(DistributedTransactionContext),
DtxContextToString(distributedTransactionContext),
snapshot->xmin,
snapshot->xmax,
snapshot->xcnt,
......@@ -1559,7 +1562,7 @@ updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo, Snapshot snapshot, cha
(errmsg("updateSharedLocalSnapshot: combocidsize is now %d max %d segmateSync %d->%d",
combocidSize, MaxComboCids, SharedLocalSnapshotSlot->segmateSync, dtxContextInfo->segmateSync)));
SetSharedTransactionId_writer();
SetSharedTransactionId_writer(distributedTransactionContext);
SharedLocalSnapshotSlot->QDcid = dtxContextInfo->curcid;
SharedLocalSnapshotSlot->QDxid = dtxContextInfo->distributedXid;
......@@ -1570,7 +1573,7 @@ updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo, Snapshot snapshot, cha
ereport((Debug_print_full_dtm ? LOG : DEBUG5),
(errmsg("updateSharedLocalSnapshot for DistributedTransactionContext = '%s' setting shared local snapshot xid = %u (xmin: %u xmax: %u xcnt: %u) curcid: %d, QDxid = %u, QDcid = %u",
DtxContextToString(DistributedTransactionContext),
DtxContextToString(distributedTransactionContext),
SharedLocalSnapshotSlot->xid,
SharedLocalSnapshotSlot->snapshot.xmin,
SharedLocalSnapshotSlot->snapshot.xmax,
......@@ -1587,14 +1590,14 @@ updateSharedLocalSnapshot(DtxContextInfo *dtxContextInfo, Snapshot snapshot, cha
getDistributedTransactionId(),
SharedLocalSnapshotSlot->slotid,
debugCaller,
DtxContextToString(DistributedTransactionContext))));
DtxContextToString(distributedTransactionContext))));
LWLockRelease(SharedLocalSnapshotSlot->slotLock);
}
static int
GetDistributedSnapshotMaxCount(void)
GetDistributedSnapshotMaxCount(DtxContext distributedTransactionContext)
{
switch (DistributedTransactionContext)
switch (distributedTransactionContext)
{
case DTX_CONTEXT_LOCAL_ONLY:
case DTX_CONTEXT_QD_RETRY_PHASE_2:
......@@ -1616,12 +1619,12 @@ GetDistributedSnapshotMaxCount(void)
case DTX_CONTEXT_QE_PREPARED:
elog(FATAL, "Unexpected segment distribute transaction context: '%s'",
DtxContextToString(DistributedTransactionContext));
DtxContextToString(distributedTransactionContext));
break;
default:
elog(FATAL, "Unrecognized DTX transaction context: %d",
(int) DistributedTransactionContext);
(int) distributedTransactionContext);
break;
}
......@@ -1633,13 +1636,13 @@ GetDistributedSnapshotMaxCount(void)
* information that the QE sent us (if any).
*/
static void
FillInDistributedSnapshot(Snapshot snapshot)
FillInDistributedSnapshot(Snapshot snapshot, DtxContext distributedTransactionContext)
{
ereport((Debug_print_full_dtm ? LOG : DEBUG5),
(errmsg("FillInDistributedSnapshot DTX Context = '%s'",
DtxContextToString(DistributedTransactionContext))));
DtxContextToString(distributedTransactionContext))));
switch (DistributedTransactionContext)
switch (distributedTransactionContext)
{
case DTX_CONTEXT_LOCAL_ONLY:
case DTX_CONTEXT_QD_RETRY_PHASE_2:
......@@ -1657,7 +1660,7 @@ FillInDistributedSnapshot(Snapshot snapshot)
* while holding ProcArrayLock, not here.
*/
elog(ERROR, "FillInDistributedSnapshot called in context '%s'",
DtxContextToString(DistributedTransactionContext));
DtxContextToString(distributedTransactionContext));
break;
case DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER:
......@@ -1690,12 +1693,12 @@ FillInDistributedSnapshot(Snapshot snapshot)
case DTX_CONTEXT_QE_PREPARED:
elog(FATAL, "Unexpected segment distribute transaction context: '%s'",
DtxContextToString(DistributedTransactionContext));
DtxContextToString(distributedTransactionContext));
break;
default:
elog(FATAL, "Unrecognized DTX transaction context: %d",
(int) DistributedTransactionContext);
(int) distributedTransactionContext);
break;
}
......@@ -1898,7 +1901,7 @@ DistributedSnapshotMappedEntry_Compare(const void *p1, const void *p2)
* create distributed snapshot based on current visible distributed transaction
*/
static bool
CreateDistributedSnapshot(DistributedSnapshot *ds)
CreateDistributedSnapshot(DistributedSnapshot *ds, DtxContext distributedTransactionContext)
{
int i;
int count;
......@@ -2013,7 +2016,7 @@ CreateDistributedSnapshot(DistributedSnapshot *ds)
"[Distributed Snapshot #%u] *Create* (gxid = %u, '%s')",
distribSnapshotId,
MyTmGxact->gxid,
DtxContextToString(DistributedTransactionContext));
DtxContextToString(distributedTransactionContext));
return true;
}
......@@ -2076,7 +2079,7 @@ GetMaxSnapshotSubxidCount(void)
* not statically allocated (see xip allocation below).
*/
Snapshot
GetSnapshotData(Snapshot snapshot)
GetSnapshotData(Snapshot snapshot, DtxContext distributedTransactionContext)
{
ProcArrayStruct *arrayP = procArray;
TransactionId xmin;
......@@ -2144,7 +2147,7 @@ GetSnapshotData(Snapshot snapshot)
if (snapshot->distribSnapshotWithLocalMapping.ds.inProgressXidArray == NULL)
{
int maxCount = GetDistributedSnapshotMaxCount();
int maxCount = GetDistributedSnapshotMaxCount(distributedTransactionContext);
if (maxCount > 0)
{
snapshot->distribSnapshotWithLocalMapping.ds.inProgressXidArray =
......@@ -2187,8 +2190,8 @@ GetSnapshotData(Snapshot snapshot)
* its never for a very long time.
*
*/
if (DistributedTransactionContext == DTX_CONTEXT_QE_READER ||
DistributedTransactionContext == DTX_CONTEXT_QE_ENTRY_DB_SINGLETON)
if (distributedTransactionContext == DTX_CONTEXT_QE_READER ||
distributedTransactionContext == DTX_CONTEXT_QE_ENTRY_DB_SINGLETON)
{
/* the pg_usleep() call below is in units of us (microseconds), interconnect
* timeout is in seconds. Start with 1 millisecond. */
......@@ -2205,7 +2208,7 @@ GetSnapshotData(Snapshot snapshot)
* making copies from the QEDtxContextInfo structure sent by
* the QD.
*/
FillInDistributedSnapshot(snapshot);
FillInDistributedSnapshot(snapshot, distributedTransactionContext);
/*
* If we're a cursor-reader, we get out snapshot from the
......@@ -2214,7 +2217,9 @@ GetSnapshotData(Snapshot snapshot)
*/
if (QEDtxContextInfo.cursorContext)
{
readSharedLocalSnapshot_forCursor(snapshot);
readSharedLocalSnapshot_forCursor(
snapshot,
distributedTransactionContext);
return snapshot;
}
......@@ -2224,7 +2229,7 @@ GetSnapshotData(Snapshot snapshot)
QEDtxContextInfo.distributedSnapshot.distribSnapshotId,
QEDtxContextInfo.distributedXid,
QEDtxContextInfo.curcid,
DtxContextToString(DistributedTransactionContext))));
DtxContextToString(distributedTransactionContext))));
/*
* This is the second phase of the handshake we started in
......@@ -2291,7 +2296,8 @@ GetSnapshotData(Snapshot snapshot)
usedComboCids, comboCidCnt, segmateSync)));
SetSharedTransactionId_reader(SharedLocalSnapshotSlot->xid,
SharedLocalSnapshotSlot->snapshot.curcid);
SharedLocalSnapshotSlot->snapshot.curcid,
distributedTransactionContext);
ereport((Debug_print_snapshot_dtm ? LOG : DEBUG5),
(errmsg("Reader qExec setting shared local snapshot to: xmin: %d xmax: %d curcid: %d",
......@@ -2337,7 +2343,7 @@ GetSnapshotData(Snapshot snapshot)
QEDtxContextInfo.distributedXid, SharedLocalSnapshotSlot->QDxid,
QEDtxContextInfo.curcid,
SharedLocalSnapshotSlot->QDcid, SharedLocalSnapshotSlot->ready,
DtxContextToString(DistributedTransactionContext),
DtxContextToString(distributedTransactionContext),
SharedLocalSnapshotSlot->slotindex, SharedSnapshotDump())));
}
else if (warning_sleep_time_us > 1000 * 1000)
......@@ -2352,7 +2358,7 @@ GetSnapshotData(Snapshot snapshot)
SharedLocalSnapshotSlot->QDxid,
QEDtxContextInfo.curcid,
SharedLocalSnapshotSlot->QDcid,
DtxContextToString(DistributedTransactionContext))));
DtxContextToString(distributedTransactionContext))));
ereport(LOG,
......@@ -2368,7 +2374,7 @@ GetSnapshotData(Snapshot snapshot)
SharedLocalSnapshotSlot->QDcid,
SharedLocalSnapshotSlot->ready,
SharedLocalSnapshotSlot->slotindex,
DtxContextToString(DistributedTransactionContext))));
DtxContextToString(distributedTransactionContext))));
warning_sleep_time_us = 0;
}
......@@ -2379,8 +2385,8 @@ GetSnapshotData(Snapshot snapshot)
}
/* We must not be a reader. */
Assert(DistributedTransactionContext != DTX_CONTEXT_QE_READER);
Assert(DistributedTransactionContext != DTX_CONTEXT_QE_ENTRY_DB_SINGLETON);
Assert(distributedTransactionContext != DTX_CONTEXT_QE_READER);
Assert(distributedTransactionContext != DTX_CONTEXT_QE_ENTRY_DB_SINGLETON);
/*
* It is sufficient to get shared lock on ProcArrayLock, even if we are
......@@ -2436,9 +2442,11 @@ GetSnapshotData(Snapshot snapshot)
* including distributed transactions in the local snapshot via their
* local xids.
*/
if (DistributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE)
if (distributedTransactionContext == DTX_CONTEXT_QD_DISTRIBUTED_CAPABLE)
{
snapshot->haveDistribSnapshot = CreateDistributedSnapshot(&snapshot->distribSnapshotWithLocalMapping.ds);
snapshot->haveDistribSnapshot = CreateDistributedSnapshot(
&snapshot->distribSnapshotWithLocalMapping.ds,
distributedTransactionContext);
ereport((Debug_print_full_dtm ? LOG : DEBUG5),
(errmsg("Got distributed snapshot from DistributedSnapshotWithLocalXids_Create = %s",
......@@ -2625,7 +2633,7 @@ GetSnapshotData(Snapshot snapshot)
*
* (We do this after releasing ProcArrayLock, to reduce contention.)
*/
FillInDistributedSnapshot(snapshot);
FillInDistributedSnapshot(snapshot, distributedTransactionContext);
/*
* In computing RecentGlobalXmin, also take distributed snapshots into
......@@ -2693,12 +2701,16 @@ GetSnapshotData(Snapshot snapshot)
* snapshot for the "Local Implicit using Distributed Snapshot" case, too.
*/
if ((DistributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER ||
DistributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER ||
DistributedTransactionContext == DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT) &&
if ((distributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_EXPLICIT_WRITER ||
distributedTransactionContext == DTX_CONTEXT_QE_TWO_PHASE_IMPLICIT_WRITER ||
distributedTransactionContext == DTX_CONTEXT_QE_AUTO_COMMIT_IMPLICIT) &&
SharedLocalSnapshotSlot != NULL)
{
updateSharedLocalSnapshot(&QEDtxContextInfo, snapshot, "GetSnapshotData");
updateSharedLocalSnapshot(
&QEDtxContextInfo,
distributedTransactionContext,
snapshot,
"GetSnapshotData");
}
ereport((Debug_print_snapshot_dtm ? LOG : DEBUG5),
......
src/backend/storage/ipc/test/procarray_test.c
浏览文件 @ a52e567c
......@@ -73,7 +73,7 @@ test__CreateDistributedSnapshot(void **state)
* Basic case, no other in progress transaction in system
*/
memset(ds->inProgressXidArray, 0, SIZE_OF_IN_PROGRESS_ARRAY);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping, DTX_CONTEXT_LOCAL_ONLY);
/* perform all the validations */
assert_true(ds->xminAllDistributedSnapshots == 20);
......@@ -101,7 +101,7 @@ test__CreateDistributedSnapshot(void **state)
procArray->numProcs = 3;
memset(ds->inProgressXidArray, 0, SIZE_OF_IN_PROGRESS_ARRAY);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping, DTX_CONTEXT_LOCAL_ONLY);
/* perform all the validations */
assert_true(ds->xminAllDistributedSnapshots == 5);
......@@ -129,7 +129,7 @@ test__CreateDistributedSnapshot(void **state)
procArray->numProcs = 5;
memset(ds->inProgressXidArray, 0, SIZE_OF_IN_PROGRESS_ARRAY);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping);
CreateDistributedSnapshot(&distribSnapshotWithLocalMapping, DTX_CONTEXT_LOCAL_ONLY);
/* perform all the validations */
assert_true(ds->xminAllDistributedSnapshots == 5);
......
src/backend/storage/lmgr/predicate.c
浏览文件 @ a52e567c
......@@ -1704,7 +1704,7 @@ GetSerializableTransactionSnapshotInt(Snapshot snapshot,
/* Get the snapshot, or check that it's safe to use */
if (!TransactionIdIsValid(sourcexid))
snapshot = GetSnapshotData(snapshot);
snapshot = GetSnapshotData(snapshot, DistributedTransactionContext);
else if (!ProcArrayInstallImportedXmin(snapshot->xmin, sourcexid))
{
ReleasePredXact(sxact);
......
src/backend/utils/cache/relcache.c
浏览文件 @ a52e567c
......@@ -341,7 +341,9 @@ ScanPgRelation(Oid targetRelId, bool indexOK, bool force_non_historic)
* relfilenode of non mapped system relations during decoding.
*/
if (force_non_historic)
snapshot = GetNonHistoricCatalogSnapshot(RelationRelationId);
snapshot = GetNonHistoricCatalogSnapshot(
RelationRelationId,
DistributedTransactionContext);
else
snapshot = GetCatalogSnapshot(RelationRelationId);
......
src/backend/utils/time/sharedsnapshot.c
浏览文件 @ a52e567c
......@@ -747,7 +747,7 @@ dumpSharedLocalSnapshot_forCursor(void)
}
void
readSharedLocalSnapshot_forCursor(Snapshot snapshot)
readSharedLocalSnapshot_forCursor(Snapshot snapshot, DtxContext distributedTransactionContext)
{
BufFile *f;
char *fname=NULL;
......@@ -872,7 +872,10 @@ readSharedLocalSnapshot_forCursor(Snapshot snapshot)
/* we're done with file. */
BufFileClose(f);
SetSharedTransactionId_reader(localXid, snapshot->curcid);
SetSharedTransactionId_reader(
localXid,
snapshot->curcid,
distributedTransactionContext);
return;
}
......
src/backend/utils/time/snapmgr.c
浏览文件 @ a52e567c
......@@ -208,7 +208,7 @@ GetTransactionSnapshot(void)
if (IsolationIsSerializable())
CurrentSnapshot = GetSerializableTransactionSnapshot(&CurrentSnapshotData);
else
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, DistributedTransactionContext);
/* Make a saved copy */
CurrentSnapshot = CopySnapshot(CurrentSnapshot);
FirstXactSnapshot = CurrentSnapshot;
......@@ -217,7 +217,7 @@ GetTransactionSnapshot(void)
RegisteredSnapshots++;
}
else
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, DistributedTransactionContext);
FirstSnapshotSet = true;
return CurrentSnapshot;
......@@ -243,7 +243,7 @@ GetTransactionSnapshot(void)
/* Don't allow catalog snapshot to be older than xact snapshot. */
InvalidateCatalogSnapshot();
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, DistributedTransactionContext);
elog((Debug_print_snapshot_dtm ? LOG : DEBUG5),
"[Distributed Snapshot #%u] (gxid = %u, '%s')",
......@@ -272,7 +272,7 @@ GetLatestSnapshot(void)
if (!FirstSnapshotSet)
return GetTransactionSnapshot();
SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData);
SecondarySnapshot = GetSnapshotData(&SecondarySnapshotData, DistributedTransactionContext);
return SecondarySnapshot;
}
......@@ -305,7 +305,7 @@ GetCatalogSnapshot(Oid relid)
DtxContext saveDistributedTransactionContext = DistributedTransactionContext;
DistributedTransactionContext = DTX_CONTEXT_LOCAL_ONLY;
Snapshot snapshot = GetNonHistoricCatalogSnapshot(relid);
Snapshot snapshot = GetNonHistoricCatalogSnapshot(relid, DistributedTransactionContext);
DistributedTransactionContext = saveDistributedTransactionContext;
return snapshot;
......@@ -318,7 +318,7 @@ GetCatalogSnapshot(Oid relid)
* up.
*/
Snapshot
GetNonHistoricCatalogSnapshot(Oid relid)
GetNonHistoricCatalogSnapshot(Oid relid, DtxContext distributedTransactionContext)
{
/*
* If the caller is trying to scan a relation that has no syscache, no
......@@ -335,7 +335,9 @@ GetNonHistoricCatalogSnapshot(Oid relid)
if (CatalogSnapshot == NULL)
{
/* Get new snapshot. */
CatalogSnapshot = GetSnapshotData(&CatalogSnapshotData);
CatalogSnapshot = GetSnapshotData(
&CatalogSnapshotData,
distributedTransactionContext);
/*
* Make sure the catalog snapshot will be accounted for in decisions
......@@ -439,7 +441,7 @@ SetTransactionSnapshot(Snapshot sourcesnap, TransactionId sourcexid)
* two variables in exported snapshot files, but it seems better to have
* snapshot importers compute reasonably up-to-date values for them.)
*/
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData);
CurrentSnapshot = GetSnapshotData(&CurrentSnapshotData, DistributedTransactionContext);
/*
* Now copy appropriate fields from the source snapshot.
......
src/include/access/xact.h
浏览文件 @ a52e567c
......@@ -19,6 +19,7 @@
#include "storage/relfilenode.h"
#include "cdb/cdbpublic.h"
#include "cdb/cdbtm.h"
/*
* Xact isolation levels
......@@ -226,8 +227,8 @@ typedef struct xl_xact_distributed_forget
*/
/* Greenplum Database specific */
extern void SetSharedTransactionId_writer(void);
extern void SetSharedTransactionId_reader(TransactionId xid, CommandId cid);
extern void SetSharedTransactionId_writer(DtxContext distributedTransactionContext);
extern void SetSharedTransactionId_reader(TransactionId xid, CommandId cid, DtxContext distributedTransactionContext);
extern bool IsTransactionState(void);
extern bool IsAbortInProgress(void);
extern bool IsCommitInProgress(void);
......
src/include/storage/procarray.h
浏览文件 @ a52e567c
......@@ -19,6 +19,7 @@
#include "utils/snapshot.h"
#include "cdb/cdbpublic.h"
#include "cdb/cdbtm.h"
struct DtxContextInfo; /* cdb/cdbdtxcontextinfo.h */
struct SnapshotData; /* utils/tqual.h */
......@@ -47,7 +48,7 @@ extern void ExpireOldKnownAssignedTransactionIds(TransactionId xid);
extern int GetMaxSnapshotXidCount(void);
extern int GetMaxSnapshotSubxidCount(void);
extern Snapshot GetSnapshotData(Snapshot snapshot);
extern Snapshot GetSnapshotData(Snapshot snapshot, DtxContext distributedTransactionContext);
extern bool ProcArrayInstallImportedXmin(TransactionId xmin,
TransactionId sourcexid);
......@@ -89,7 +90,7 @@ extern void XidCacheRemoveRunningXids(TransactionId xid,
extern PGPROC *FindProcByGpSessionId(long gp_session_id);
extern void UpdateSerializableCommandId(CommandId curcid);
extern void updateSharedLocalSnapshot(struct DtxContextInfo *dtxContextInfo, struct SnapshotData *snapshot, char* debugCaller);
extern void updateSharedLocalSnapshot(struct DtxContextInfo *dtxContextInfo, DtxContext distributedTransactionContext, struct SnapshotData *snapshot, char* debugCaller);
extern void GetSlotTableDebugInfo(void **snapshotArray, int *maxSlots);
......
src/include/utils/sharedsnapshot.h
浏览文件 @ a52e567c
......@@ -50,7 +50,7 @@ extern void addSharedSnapshot(char *creatorDescription, int id);
extern void lookupSharedSnapshot(char *lookerDescription, char *creatorDescription, int id);
extern void dumpSharedLocalSnapshot_forCursor(void);
extern void readSharedLocalSnapshot_forCursor(Snapshot snapshot);
extern void readSharedLocalSnapshot_forCursor(Snapshot snapshot, DtxContext distributedTransactionContext);
extern void AtEOXact_SharedSnapshot(void);
......
src/include/utils/snapmgr.h
浏览文件 @ a52e567c
......@@ -16,6 +16,7 @@
#include "fmgr.h"
#include "utils/resowner.h"
#include "utils/snapshot.h"
#include "cdb/cdbtm.h"
extern bool FirstSnapshotSet;
......@@ -30,7 +31,7 @@ extern Snapshot GetLatestSnapshot(void);
extern void SnapshotSetCommandId(CommandId curcid);
extern Snapshot GetCatalogSnapshot(Oid relid);
extern Snapshot GetNonHistoricCatalogSnapshot(Oid relid);
extern Snapshot GetNonHistoricCatalogSnapshot(Oid relid, DtxContext distributedTransactionContext);
extern void InvalidateCatalogSnapshot(void);
extern void InvalidateCatalogSnapshotConditionally(void);
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或