commit_ts.c File Reference

#include "postgres.h"
#include "access/commit_ts.h"
#include "access/htup_details.h"
#include "access/slru.h"
#include "access/transam.h"
#include "access/xloginsert.h"
#include "access/xlogutils.h"
#include "catalog/pg_type.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "pg_trace.h"
#include "storage/shmem.h"
#include "utils/builtins.h"
#include "utils/snapmgr.h"
#include "utils/timestamp.h"

Include dependency graph for commit_ts.c:

Go to the source code of this file.

Data Structures
struct	CommitTimestampEntry
struct	CommitTimestampShared

Macros
#define	SizeOfCommitTimestampEntry
#define	COMMIT_TS_XACTS_PER_PAGE    (BLCKSZ / SizeOfCommitTimestampEntry)
#define	TransactionIdToCTsPage(xid)    ((xid) / (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
#define	TransactionIdToCTsEntry(xid)    ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
#define	CommitTsCtl   (&CommitTsCtlData)

Typedefs
typedef struct CommitTimestampEntry	CommitTimestampEntry
typedef struct CommitTimestampShared	CommitTimestampShared

Functions
static void	SetXidCommitTsInPage (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz ts, RepOriginId nodeid, int pageno)
static void	TransactionIdSetCommitTs (TransactionId xid, TimestampTz ts, RepOriginId nodeid, int slotno)
static void	error_commit_ts_disabled (void)
static int	ZeroCommitTsPage (int pageno, bool writeXlog)
static bool	CommitTsPagePrecedes (int page1, int page2)
static void	ActivateCommitTs (void)
static void	DeactivateCommitTs (void)
static void	WriteZeroPageXlogRec (int pageno)
static void	WriteTruncateXlogRec (int pageno, TransactionId oldestXid)
void	TransactionTreeSetCommitTsData (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz timestamp, RepOriginId nodeid)
bool	TransactionIdGetCommitTsData (TransactionId xid, TimestampTz *ts, RepOriginId *nodeid)
TransactionId	GetLatestCommitTsData (TimestampTz *ts, RepOriginId *nodeid)
Datum	pg_xact_commit_timestamp (PG_FUNCTION_ARGS)
Datum	pg_last_committed_xact (PG_FUNCTION_ARGS)
Datum	pg_xact_commit_timestamp_origin (PG_FUNCTION_ARGS)
Size	CommitTsShmemBuffers (void)
Size	CommitTsShmemSize (void)
void	CommitTsShmemInit (void)
void	BootStrapCommitTs (void)
void	StartupCommitTs (void)
void	CompleteCommitTsInitialization (void)
void	CommitTsParameterChange (bool newvalue, bool oldvalue)
void	CheckPointCommitTs (void)
void	ExtendCommitTs (TransactionId newestXact)
void	TruncateCommitTs (TransactionId oldestXact)
void	SetCommitTsLimit (TransactionId oldestXact, TransactionId newestXact)
void	AdvanceOldestCommitTsXid (TransactionId oldestXact)
void	commit_ts_redo (XLogReaderState *record)
int	committssyncfiletag (const FileTag *ftag, char *path)

Variables
static SlruCtlData	CommitTsCtlData
static CommitTimestampShared *	commitTsShared
bool	track_commit_timestamp

Macro Definition Documentation

COMMIT_TS_XACTS_PER_PAGE

#define COMMIT_TS_XACTS_PER_PAGE    (BLCKSZ / SizeOfCommitTimestampEntry)

Definition at line 65 of file commit_ts.c.

CommitTsCtl

#define CommitTsCtl   (&CommitTsCtlData)

Definition at line 78 of file commit_ts.c.

SizeOfCommitTimestampEntry

#define SizeOfCommitTimestampEntry

Value:

                                    (offsetof(CommitTimestampEntry, nodeid) + \
                                    sizeof(RepOriginId))

Definition at line 62 of file commit_ts.c.

TransactionIdToCTsEntry

#define TransactionIdToCTsEntry

(

xid

)

((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)

Definition at line 70 of file commit_ts.c.

TransactionIdToCTsPage

#define TransactionIdToCTsPage

(

xid

)

((xid) / (TransactionId) COMMIT_TS_XACTS_PER_PAGE)

Definition at line 68 of file commit_ts.c.

Typedef Documentation

CommitTimestampEntry

typedef struct CommitTimestampEntry CommitTimestampEntry

CommitTimestampShared

typedef struct CommitTimestampShared CommitTimestampShared

Function Documentation

ActivateCommitTs()

static void ActivateCommitTs ( void  )

static

Definition at line 662 of file commit_ts.c.

{
    TransactionId xid;
    int         pageno;
 
    /* If we've done this already, there's nothing to do */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (commitTsShared->commitTsActive)
    {
        LWLockRelease(CommitTsLock);
        return;
    }
    LWLockRelease(CommitTsLock);
 
    xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
    pageno = TransactionIdToCTsPage(xid);
 
    /*
     * Re-Initialize our idea of the latest page number.
     */
    LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
    CommitTsCtl->shared->latest_page_number = pageno;
    LWLockRelease(CommitTsSLRULock);
 
    /*
     * If CommitTs is enabled, but it wasn't in the previous server run, we
     * need to set the oldest and newest values to the next Xid; that way, we
     * will not try to read data that might not have been set.
     *
     * XXX does this have a problem if a server is started with commitTs
     * enabled, then started with commitTs disabled, then restarted with it
     * enabled again?  It doesn't look like it does, because there should be a
     * checkpoint that sets the value to InvalidTransactionId at end of
     * recovery; and so any chance of injecting new transactions without
     * CommitTs values would occur after the oldestCommitTsXid has been set to
     * Invalid temporarily.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (ShmemVariableCache->oldestCommitTsXid == InvalidTransactionId)
    {
        ShmemVariableCache->oldestCommitTsXid =
            ShmemVariableCache->newestCommitTsXid = ReadNextTransactionId();
    }
    LWLockRelease(CommitTsLock);
 
    /* Create the current segment file, if necessary */
    if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno))
    {
        int         slotno;
 
        LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
        slotno = ZeroCommitTsPage(pageno, false);
        SimpleLruWritePage(CommitTsCtl, slotno);
        Assert(!CommitTsCtl->shared->page_dirty[slotno]);
        LWLockRelease(CommitTsSLRULock);
    }
 
    /* Change the activation status in shared memory. */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->commitTsActive = true;
    LWLockRelease(CommitTsLock);
}

References Assert(), CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::newestCommitTsXid, VariableCacheData::nextXid, VariableCacheData::oldestCommitTsXid, ReadNextTransactionId(), ShmemVariableCache, SimpleLruDoesPhysicalPageExist(), SimpleLruWritePage(), TransactionIdToCTsPage, XidFromFullTransactionId, and ZeroCommitTsPage().

Referenced by CommitTsParameterChange(), CompleteCommitTsInitialization(), and StartupCommitTs().

AdvanceOldestCommitTsXid()

void AdvanceOldestCommitTsXid

(

TransactionId

oldestXact

)

Definition at line 887 of file commit_ts.c.

{
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId &&
        TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, oldestXact))
        ShmemVariableCache->oldestCommitTsXid = oldestXact;
    LWLockRelease(CommitTsLock);
}

References InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::oldestCommitTsXid, ShmemVariableCache, and TransactionIdPrecedes().

Referenced by commit_ts_redo(), and vac_truncate_clog().

BootStrapCommitTs()

void BootStrapCommitTs

(

void

)

Definition at line 553 of file commit_ts.c.

{
    /*
     * Nothing to do here at present, unlike most other SLRU modules; segments
     * are created when the server is started with this module enabled. See
     * ActivateCommitTs.
     */
}

Referenced by BootStrapXLOG().

CheckPointCommitTs()

void CheckPointCommitTs

(

void

)

Definition at line 774 of file commit_ts.c.

{
    /*
     * Write dirty CommitTs pages to disk.  This may result in sync requests
     * queued for later handling by ProcessSyncRequests(), as part of the
     * checkpoint.
     */
    SimpleLruWriteAll(CommitTsCtl, true);
}

References CommitTsCtl, and SimpleLruWriteAll().

Referenced by CheckPointGuts().

commit_ts_redo()

void commit_ts_redo

(

XLogReaderState *

record

)

Definition at line 967 of file commit_ts.c.

{
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;
 
    /* Backup blocks are not used in commit_ts records */
    Assert(!XLogRecHasAnyBlockRefs(record));
 
    if (info == COMMIT_TS_ZEROPAGE)
    {
        int         pageno;
        int         slotno;
 
        memcpy(&pageno, XLogRecGetData(record), sizeof(int));
 
        LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
 
        slotno = ZeroCommitTsPage(pageno, false);
        SimpleLruWritePage(CommitTsCtl, slotno);
        Assert(!CommitTsCtl->shared->page_dirty[slotno]);
 
        LWLockRelease(CommitTsSLRULock);
    }
    else if (info == COMMIT_TS_TRUNCATE)
    {
        xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) XLogRecGetData(record);
 
        AdvanceOldestCommitTsXid(trunc->oldestXid);
 
        /*
         * During XLOG replay, latest_page_number isn't set up yet; insert a
         * suitable value to bypass the sanity test in SimpleLruTruncate.
         */
        CommitTsCtl->shared->latest_page_number = trunc->pageno;
 
        SimpleLruTruncate(CommitTsCtl, trunc->pageno);
    }
    else
        elog(PANIC, "commit_ts_redo: unknown op code %u", info);
}

References AdvanceOldestCommitTsXid(), Assert(), COMMIT_TS_TRUNCATE, COMMIT_TS_ZEROPAGE, CommitTsCtl, elog(), LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), xl_commit_ts_truncate::oldestXid, xl_commit_ts_truncate::pageno, PANIC, SimpleLruTruncate(), SimpleLruWritePage(), XLogRecGetData, XLogRecGetInfo, XLogRecHasAnyBlockRefs, XLR_INFO_MASK, and ZeroCommitTsPage().

CommitTsPagePrecedes()

static bool CommitTsPagePrecedes ( int  page1, int  page2  )

static

Definition at line 921 of file commit_ts.c.

{
    TransactionId xid1;
    TransactionId xid2;
 
    xid1 = ((TransactionId) page1) * COMMIT_TS_XACTS_PER_PAGE;
    xid1 += FirstNormalTransactionId + 1;
    xid2 = ((TransactionId) page2) * COMMIT_TS_XACTS_PER_PAGE;
    xid2 += FirstNormalTransactionId + 1;
 
    return (TransactionIdPrecedes(xid1, xid2) &&
            TransactionIdPrecedes(xid1, xid2 + COMMIT_TS_XACTS_PER_PAGE - 1));
}

References COMMIT_TS_XACTS_PER_PAGE, FirstNormalTransactionId, and TransactionIdPrecedes().

Referenced by CommitTsShmemInit().

CommitTsParameterChange()

void CommitTsParameterChange	(	bool	newvalue,
		bool	oldvalue
	)

Definition at line 621 of file commit_ts.c.

{
    /*
     * If the commit_ts module is disabled in this server and we get word from
     * the primary server that it is enabled there, activate it so that we can
     * replay future WAL records involving it; also mark it as active on
     * pg_control.  If the old value was already set, we already did this, so
     * don't do anything.
     *
     * If the module is disabled in the primary, disable it here too, unless
     * the module is enabled locally.
     *
     * Note this only runs in the recovery process, so an unlocked read is
     * fine.
     */
    if (newvalue)
    {
        if (!commitTsShared->commitTsActive)
            ActivateCommitTs();
    }
    else if (commitTsShared->commitTsActive)
        DeactivateCommitTs();
}

References ActivateCommitTs(), CommitTimestampShared::commitTsActive, commitTsShared, and DeactivateCommitTs().

Referenced by xlog_redo().

CommitTsShmemBuffers()

Size CommitTsShmemBuffers

(

void

)

Definition at line 498 of file commit_ts.c.

{
    return Min(256, Max(4, NBuffers / 256));
}

References Max, Min, and NBuffers.

Referenced by CommitTsShmemInit(), and CommitTsShmemSize().

CommitTsShmemInit()

void CommitTsShmemInit

(

void

)

Definition at line 518 of file commit_ts.c.

{
    bool        found;
 
    CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
    SimpleLruInit(CommitTsCtl, "CommitTs", CommitTsShmemBuffers(), 0,
                  CommitTsSLRULock, "pg_commit_ts",
                  LWTRANCHE_COMMITTS_BUFFER,
                  SYNC_HANDLER_COMMIT_TS);
    SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);
 
    commitTsShared = ShmemInitStruct("CommitTs shared",
                                     sizeof(CommitTimestampShared),
                                     &found);
 
    if (!IsUnderPostmaster)
    {
        Assert(!found);
 
        commitTsShared->xidLastCommit = InvalidTransactionId;
        TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
        commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId;
        commitTsShared->commitTsActive = false;
    }
    else
        Assert(found);
}

References Assert(), COMMIT_TS_XACTS_PER_PAGE, CommitTimestampShared::commitTsActive, CommitTsCtl, CommitTsPagePrecedes(), commitTsShared, CommitTsShmemBuffers(), CommitTimestampShared::dataLastCommit, InvalidRepOriginId, InvalidTransactionId, IsUnderPostmaster, LWTRANCHE_COMMITTS_BUFFER, CommitTimestampEntry::nodeid, ShmemInitStruct(), SimpleLruInit(), SlruPagePrecedesUnitTests, SYNC_HANDLER_COMMIT_TS, CommitTimestampEntry::time, TIMESTAMP_NOBEGIN, and CommitTimestampShared::xidLastCommit.

Referenced by CreateSharedMemoryAndSemaphores().

CommitTsShmemSize()

Size CommitTsShmemSize

(

void

)

Definition at line 507 of file commit_ts.c.

{
    return SimpleLruShmemSize(CommitTsShmemBuffers(), 0) +
        sizeof(CommitTimestampShared);
}

References CommitTsShmemBuffers(), and SimpleLruShmemSize().

Referenced by CalculateShmemSize().

committssyncfiletag()

int committssyncfiletag	(	const FileTag *	ftag,
		char *	path
	)

Definition at line 1011 of file commit_ts.c.

{
    return SlruSyncFileTag(CommitTsCtl, ftag, path);
}

References CommitTsCtl, and SlruSyncFileTag().

CompleteCommitTsInitialization()

void CompleteCommitTsInitialization

(

void

)

Definition at line 599 of file commit_ts.c.

{
    /*
     * If the feature is not enabled, turn it off for good.  This also removes
     * any leftover data.
     *
     * Conversely, we activate the module if the feature is enabled.  This is
     * necessary for primary and standby as the activation depends on the
     * control file contents at the beginning of recovery or when a
     * XLOG_PARAMETER_CHANGE is replayed.
     */
    if (!track_commit_timestamp)
        DeactivateCommitTs();
    else
        ActivateCommitTs();
}

References ActivateCommitTs(), DeactivateCommitTs(), and track_commit_timestamp.

Referenced by StartupXLOG().

DeactivateCommitTs()

static void DeactivateCommitTs ( void  )

static

Definition at line 736 of file commit_ts.c.

{
    /*
     * Cleanup the status in the shared memory.
     *
     * We reset everything in the commitTsShared record to prevent user from
     * getting confusing data about last committed transaction on the standby
     * when the module was activated repeatedly on the primary.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
 
    commitTsShared->commitTsActive = false;
    commitTsShared->xidLastCommit = InvalidTransactionId;
    TIMESTAMP_NOBEGIN(commitTsShared->dataLastCommit.time);
    commitTsShared->dataLastCommit.nodeid = InvalidRepOriginId;
 
    ShmemVariableCache->oldestCommitTsXid = InvalidTransactionId;
    ShmemVariableCache->newestCommitTsXid = InvalidTransactionId;
 
    LWLockRelease(CommitTsLock);
 
    /*
     * Remove *all* files.  This is necessary so that there are no leftover
     * files; in the case where this feature is later enabled after running
     * with it disabled for some time there may be a gap in the file sequence.
     * (We can probably tolerate out-of-sequence files, as they are going to
     * be overwritten anyway when we wrap around, but it seems better to be
     * tidy.)
     */
    LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
    (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL);
    LWLockRelease(CommitTsSLRULock);
}

References CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, CommitTimestampShared::dataLastCommit, InvalidRepOriginId, InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::newestCommitTsXid, CommitTimestampEntry::nodeid, VariableCacheData::oldestCommitTsXid, ShmemVariableCache, SlruScanDirCbDeleteAll(), SlruScanDirectory(), CommitTimestampEntry::time, TIMESTAMP_NOBEGIN, and CommitTimestampShared::xidLastCommit.

Referenced by CommitTsParameterChange(), and CompleteCommitTsInitialization().

error_commit_ts_disabled()

static void error_commit_ts_disabled ( void  )

static

Definition at line 373 of file commit_ts.c.

{
    ereport(ERROR,
            (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
             errmsg("could not get commit timestamp data"),
             RecoveryInProgress() ?
             errhint("Make sure the configuration parameter \"%s\" is set on the primary server.",
                     "track_commit_timestamp") :
             errhint("Make sure the configuration parameter \"%s\" is set.",
                     "track_commit_timestamp")));
}

References ereport, errcode(), errhint(), errmsg(), ERROR, and RecoveryInProgress().

Referenced by GetLatestCommitTsData(), and TransactionIdGetCommitTsData().

ExtendCommitTs()

void ExtendCommitTs

(

TransactionId

newestXact

)

Definition at line 796 of file commit_ts.c.

{
    int         pageno;
 
    /*
     * Nothing to do if module not enabled.  Note we do an unlocked read of
     * the flag here, which is okay because this routine is only called from
     * GetNewTransactionId, which is never called in a standby.
     */
    Assert(!InRecovery);
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * No work except at first XID of a page.  But beware: just after
     * wraparound, the first XID of page zero is FirstNormalTransactionId.
     */
    if (TransactionIdToCTsEntry(newestXact) != 0 &&
        !TransactionIdEquals(newestXact, FirstNormalTransactionId))
        return;
 
    pageno = TransactionIdToCTsPage(newestXact);
 
    LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
 
    /* Zero the page and make an XLOG entry about it */
    ZeroCommitTsPage(pageno, !InRecovery);
 
    LWLockRelease(CommitTsSLRULock);
}

References Assert(), CommitTimestampShared::commitTsActive, commitTsShared, FirstNormalTransactionId, InRecovery, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransactionIdEquals, TransactionIdToCTsEntry, TransactionIdToCTsPage, and ZeroCommitTsPage().

Referenced by GetNewTransactionId().

GetLatestCommitTsData()

TransactionId GetLatestCommitTsData	(	TimestampTz *	ts,
		RepOriginId *	nodeid
	)

Definition at line 352 of file commit_ts.c.

{
    TransactionId xid;
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    xid = commitTsShared->xidLastCommit;
    if (ts)
        *ts = commitTsShared->dataLastCommit.time;
    if (nodeid)
        *nodeid = commitTsShared->dataLastCommit.nodeid;
    LWLockRelease(CommitTsLock);
 
    return xid;
}

References CommitTimestampShared::commitTsActive, commitTsShared, CommitTimestampShared::dataLastCommit, error_commit_ts_disabled(), LW_SHARED, LWLockAcquire(), LWLockRelease(), CommitTimestampEntry::nodeid, CommitTimestampEntry::time, and CommitTimestampShared::xidLastCommit.

Referenced by pg_last_committed_xact().

pg_last_committed_xact()

Datum pg_last_committed_xact

(

PG_FUNCTION_ARGS

)

Definition at line 412 of file commit_ts.c.

{
    TransactionId xid;
    RepOriginId nodeid;
    TimestampTz ts;
    Datum       values[3];
    bool        nulls[3];
    TupleDesc   tupdesc;
    HeapTuple   htup;
 
    /* and construct a tuple with our data */
    xid = GetLatestCommitTsData(&ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!TransactionIdIsNormal(xid))
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TransactionIdGetDatum(xid);
        nulls[0] = false;
 
        values[1] = TimestampTzGetDatum(ts);
        nulls[1] = false;
 
        values[2] = ObjectIdGetDatum((Oid) nodeid);
        nulls[2] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}

References elog(), ERROR, get_call_result_type(), GetLatestCommitTsData(), heap_form_tuple(), HeapTupleGetDatum(), ObjectIdGetDatum(), PG_RETURN_DATUM, TimestampTzGetDatum(), TransactionIdGetDatum(), TransactionIdIsNormal, TYPEFUNC_COMPOSITE, and values.

pg_xact_commit_timestamp()

Datum pg_xact_commit_timestamp

(

PG_FUNCTION_ARGS

)

Definition at line 389 of file commit_ts.c.

{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    TimestampTz ts;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, NULL);
 
    if (!found)
        PG_RETURN_NULL();
 
    PG_RETURN_TIMESTAMPTZ(ts);
}

References PG_GETARG_TRANSACTIONID, PG_RETURN_NULL, PG_RETURN_TIMESTAMPTZ, and TransactionIdGetCommitTsData().

pg_xact_commit_timestamp_origin()

Datum pg_xact_commit_timestamp_origin

(

PG_FUNCTION_ARGS

)

Definition at line 456 of file commit_ts.c.

{
    TransactionId xid = PG_GETARG_TRANSACTIONID(0);
    RepOriginId nodeid;
    TimestampTz ts;
    Datum       values[2];
    bool        nulls[2];
    TupleDesc   tupdesc;
    HeapTuple   htup;
    bool        found;
 
    found = TransactionIdGetCommitTsData(xid, &ts, &nodeid);
 
    if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
        elog(ERROR, "return type must be a row type");
 
    if (!found)
    {
        memset(nulls, true, sizeof(nulls));
    }
    else
    {
        values[0] = TimestampTzGetDatum(ts);
        nulls[0] = false;
 
        values[1] = ObjectIdGetDatum((Oid) nodeid);
        nulls[1] = false;
    }
 
    htup = heap_form_tuple(tupdesc, values, nulls);
 
    PG_RETURN_DATUM(HeapTupleGetDatum(htup));
}

References elog(), ERROR, get_call_result_type(), heap_form_tuple(), HeapTupleGetDatum(), ObjectIdGetDatum(), PG_GETARG_TRANSACTIONID, PG_RETURN_DATUM, TimestampTzGetDatum(), TransactionIdGetCommitTsData(), TYPEFUNC_COMPOSITE, and values.

SetCommitTsLimit()

void SetCommitTsLimit	(	TransactionId	oldestXact,
		TransactionId	newestXact
	)

Definition at line 860 of file commit_ts.c.

{
    /*
     * Be careful not to overwrite values that are either further into the
     * "future" or signal a disabled committs.
     */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    if (ShmemVariableCache->oldestCommitTsXid != InvalidTransactionId)
    {
        if (TransactionIdPrecedes(ShmemVariableCache->oldestCommitTsXid, oldestXact))
            ShmemVariableCache->oldestCommitTsXid = oldestXact;
        if (TransactionIdPrecedes(newestXact, ShmemVariableCache->newestCommitTsXid))
            ShmemVariableCache->newestCommitTsXid = newestXact;
    }
    else
    {
        Assert(ShmemVariableCache->newestCommitTsXid == InvalidTransactionId);
        ShmemVariableCache->oldestCommitTsXid = oldestXact;
        ShmemVariableCache->newestCommitTsXid = newestXact;
    }
    LWLockRelease(CommitTsLock);
}

References Assert(), InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::newestCommitTsXid, VariableCacheData::oldestCommitTsXid, ShmemVariableCache, and TransactionIdPrecedes().

Referenced by BootStrapXLOG(), and StartupXLOG().

SetXidCommitTsInPage()

static void SetXidCommitTsInPage ( TransactionId  xid, int  nsubxids, TransactionId *  subxids, TimestampTz  ts, RepOriginId  nodeid, int  pageno  )

static

Definition at line 215 of file commit_ts.c.

{
    int         slotno;
    int         i;
 
    LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
 
    slotno = SimpleLruReadPage(CommitTsCtl, pageno, true, xid);
 
    TransactionIdSetCommitTs(xid, ts, nodeid, slotno);
    for (i = 0; i < nsubxids; i++)
        TransactionIdSetCommitTs(subxids[i], ts, nodeid, slotno);
 
    CommitTsCtl->shared->page_dirty[slotno] = true;
 
    LWLockRelease(CommitTsSLRULock);
}

References CommitTsCtl, i, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), SimpleLruReadPage(), and TransactionIdSetCommitTs().

Referenced by TransactionTreeSetCommitTsData().

StartupCommitTs()

void StartupCommitTs

(

void

)

Definition at line 589 of file commit_ts.c.

{
    ActivateCommitTs();
}

References ActivateCommitTs().

Referenced by StartupXLOG().

TransactionIdGetCommitTsData()

bool TransactionIdGetCommitTsData	(	TransactionId	xid,
		TimestampTz *	ts,
		RepOriginId *	nodeid
	)

Definition at line 266 of file commit_ts.c.

{
    int         pageno = TransactionIdToCTsPage(xid);
    int         entryno = TransactionIdToCTsEntry(xid);
    int         slotno;
    CommitTimestampEntry entry;
    TransactionId oldestCommitTsXid;
    TransactionId newestCommitTsXid;
 
    if (!TransactionIdIsValid(xid))
        ereport(ERROR,
                (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
                 errmsg("cannot retrieve commit timestamp for transaction %u", xid)));
    else if (!TransactionIdIsNormal(xid))
    {
        /* frozen and bootstrap xids are always committed far in the past */
        *ts = 0;
        if (nodeid)
            *nodeid = 0;
        return false;
    }
 
    LWLockAcquire(CommitTsLock, LW_SHARED);
 
    /* Error if module not enabled */
    if (!commitTsShared->commitTsActive)
        error_commit_ts_disabled();
 
    /*
     * If we're asked for the cached value, return that.  Otherwise, fall
     * through to read from SLRU.
     */
    if (commitTsShared->xidLastCommit == xid)
    {
        *ts = commitTsShared->dataLastCommit.time;
        if (nodeid)
            *nodeid = commitTsShared->dataLastCommit.nodeid;
 
        LWLockRelease(CommitTsLock);
        return *ts != 0;
    }
 
    oldestCommitTsXid = ShmemVariableCache->oldestCommitTsXid;
    newestCommitTsXid = ShmemVariableCache->newestCommitTsXid;
    /* neither is invalid, or both are */
    Assert(TransactionIdIsValid(oldestCommitTsXid) == TransactionIdIsValid(newestCommitTsXid));
    LWLockRelease(CommitTsLock);
 
    /*
     * Return empty if the requested value is outside our valid range.
     */
    if (!TransactionIdIsValid(oldestCommitTsXid) ||
        TransactionIdPrecedes(xid, oldestCommitTsXid) ||
        TransactionIdPrecedes(newestCommitTsXid, xid))
    {
        *ts = 0;
        if (nodeid)
            *nodeid = InvalidRepOriginId;
        return false;
    }
 
    /* lock is acquired by SimpleLruReadPage_ReadOnly */
    slotno = SimpleLruReadPage_ReadOnly(CommitTsCtl, pageno, xid);
    memcpy(&entry,
           CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           SizeOfCommitTimestampEntry);
 
    *ts = entry.time;
    if (nodeid)
        *nodeid = entry.nodeid;
 
    LWLockRelease(CommitTsSLRULock);
    return *ts != 0;
}

References Assert(), CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, CommitTimestampShared::dataLastCommit, ereport, errcode(), errmsg(), ERROR, error_commit_ts_disabled(), InvalidRepOriginId, LW_SHARED, LWLockAcquire(), LWLockRelease(), VariableCacheData::newestCommitTsXid, CommitTimestampEntry::nodeid, VariableCacheData::oldestCommitTsXid, ShmemVariableCache, SimpleLruReadPage_ReadOnly(), SizeOfCommitTimestampEntry, CommitTimestampEntry::time, TransactionIdIsNormal, TransactionIdIsValid, TransactionIdPrecedes(), TransactionIdToCTsEntry, TransactionIdToCTsPage, and CommitTimestampShared::xidLastCommit.

Referenced by pg_xact_commit_timestamp(), and pg_xact_commit_timestamp_origin().

TransactionIdSetCommitTs()

static void TransactionIdSetCommitTs ( TransactionId  xid, TimestampTz  ts, RepOriginId  nodeid, int  slotno  )

static

Definition at line 241 of file commit_ts.c.

{
    int         entryno = TransactionIdToCTsEntry(xid);
    CommitTimestampEntry entry;
 
    Assert(TransactionIdIsNormal(xid));
 
    entry.time = ts;
    entry.nodeid = nodeid;
 
    memcpy(CommitTsCtl->shared->page_buffer[slotno] +
           SizeOfCommitTimestampEntry * entryno,
           &entry, SizeOfCommitTimestampEntry);
}

References Assert(), CommitTsCtl, CommitTimestampEntry::nodeid, SizeOfCommitTimestampEntry, CommitTimestampEntry::time, TransactionIdIsNormal, and TransactionIdToCTsEntry.

Referenced by SetXidCommitTsInPage().

TransactionTreeSetCommitTsData()

void TransactionTreeSetCommitTsData	(	TransactionId	xid,
		int	nsubxids,
		TransactionId *	subxids,
		TimestampTz	timestamp,
		RepOriginId	nodeid
	)

Definition at line 134 of file commit_ts.c.

{
    int         i;
    TransactionId headxid;
    TransactionId newestXact;
 
    /*
     * No-op if the module is not active.
     *
     * An unlocked read here is fine, because in a standby (the only place
     * where the flag can change in flight) this routine is only called by the
     * recovery process, which is also the only process which can change the
     * flag.
     */
    if (!commitTsShared->commitTsActive)
        return;
 
    /*
     * Figure out the latest Xid in this batch: either the last subxid if
     * there's any, otherwise the parent xid.
     */
    if (nsubxids > 0)
        newestXact = subxids[nsubxids - 1];
    else
        newestXact = xid;
 
    /*
     * We split the xids to set the timestamp to in groups belonging to the
     * same SLRU page; the first element in each such set is its head.  The
     * first group has the main XID as the head; subsequent sets use the first
     * subxid not on the previous page as head.  This way, we only have to
     * lock/modify each SLRU page once.
     */
    headxid = xid;
    i = 0;
    for (;;)
    {
        int         pageno = TransactionIdToCTsPage(headxid);
        int         j;
 
        for (j = i; j < nsubxids; j++)
        {
            if (TransactionIdToCTsPage(subxids[j]) != pageno)
                break;
        }
        /* subxids[i..j] are on the same page as the head */
 
        SetXidCommitTsInPage(headxid, j - i, subxids + i, timestamp, nodeid,
                             pageno);
 
        /* if we wrote out all subxids, we're done. */
        if (j >= nsubxids)
            break;
 
        /*
         * Set the new head and skip over it, as well as over the subxids we
         * just wrote.
         */
        headxid = subxids[j];
        i = j + 1;
    }
 
    /* update the cached value in shared memory */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->xidLastCommit = xid;
    commitTsShared->dataLastCommit.time = timestamp;
    commitTsShared->dataLastCommit.nodeid = nodeid;
 
    /* and move forwards our endpoint, if needed */
    if (TransactionIdPrecedes(ShmemVariableCache->newestCommitTsXid, newestXact))
        ShmemVariableCache->newestCommitTsXid = newestXact;
    LWLockRelease(CommitTsLock);
}

References CommitTimestampShared::commitTsActive, commitTsShared, CommitTimestampShared::dataLastCommit, i, j, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), VariableCacheData::newestCommitTsXid, CommitTimestampEntry::nodeid, SetXidCommitTsInPage(), ShmemVariableCache, CommitTimestampEntry::time, TransactionIdPrecedes(), TransactionIdToCTsPage, and CommitTimestampShared::xidLastCommit.

Referenced by RecordTransactionCommit(), RecordTransactionCommitPrepared(), and xact_redo_commit().

TruncateCommitTs()

void TruncateCommitTs

(

TransactionId

oldestXact

)

Definition at line 834 of file commit_ts.c.

{
    int         cutoffPage;
 
    /*
     * The cutoff point is the start of the segment containing oldestXact. We
     * pass the *page* containing oldestXact to SimpleLruTruncate.
     */
    cutoffPage = TransactionIdToCTsPage(oldestXact);
 
    /* Check to see if there's any files that could be removed */
    if (!SlruScanDirectory(CommitTsCtl, SlruScanDirCbReportPresence,
                           &cutoffPage))
        return;                 /* nothing to remove */
 
    /* Write XLOG record */
    WriteTruncateXlogRec(cutoffPage, oldestXact);
 
    /* Now we can remove the old CommitTs segment(s) */
    SimpleLruTruncate(CommitTsCtl, cutoffPage);
}

References CommitTsCtl, SimpleLruTruncate(), SlruScanDirCbReportPresence(), SlruScanDirectory(), TransactionIdToCTsPage, and WriteTruncateXlogRec().

Referenced by vac_truncate_clog().

WriteTruncateXlogRec()

static void WriteTruncateXlogRec ( int  pageno, TransactionId  oldestXid  )

static

Definition at line 951 of file commit_ts.c.

{
    xl_commit_ts_truncate xlrec;
 
    xlrec.pageno = pageno;
    xlrec.oldestXid = oldestXid;
 
    XLogBeginInsert();
    XLogRegisterData((char *) (&xlrec), SizeOfCommitTsTruncate);
    (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_TRUNCATE);
}

References COMMIT_TS_TRUNCATE, xl_commit_ts_truncate::oldestXid, xl_commit_ts_truncate::pageno, SizeOfCommitTsTruncate, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

Referenced by TruncateCommitTs().

WriteZeroPageXlogRec()

static void WriteZeroPageXlogRec ( int  pageno )

static

Definition at line 940 of file commit_ts.c.

{
    XLogBeginInsert();
    XLogRegisterData((char *) (&pageno), sizeof(int));
    (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
}

References COMMIT_TS_ZEROPAGE, XLogBeginInsert(), XLogInsert(), and XLogRegisterData().

Referenced by ZeroCommitTsPage().

ZeroCommitTsPage()

static int ZeroCommitTsPage ( int  pageno, bool  writeXlog  )

static

Definition at line 572 of file commit_ts.c.

{
    int         slotno;
 
    slotno = SimpleLruZeroPage(CommitTsCtl, pageno);
 
    if (writeXlog)
        WriteZeroPageXlogRec(pageno);
 
    return slotno;
}

References CommitTsCtl, SimpleLruZeroPage(), and WriteZeroPageXlogRec().

Referenced by ActivateCommitTs(), commit_ts_redo(), and ExtendCommitTs().

Variable Documentation

CommitTsCtlData

SlruCtlData CommitTsCtlData

static

Definition at line 76 of file commit_ts.c.

commitTsShared

CommitTimestampShared* commitTsShared

static

Definition at line 98 of file commit_ts.c.

Referenced by ActivateCommitTs(), CommitTsParameterChange(), CommitTsShmemInit(), DeactivateCommitTs(), ExtendCommitTs(), GetLatestCommitTsData(), TransactionIdGetCommitTsData(), and TransactionTreeSetCommitTsData().

track_commit_timestamp

bool track_commit_timestamp

Definition at line 102 of file commit_ts.c.

Referenced by CompleteCommitTsInitialization(), InitControlFile(), and XLogReportParameters().