commit_ts.h File Reference

#include "access/xlog.h"
#include "datatype/timestamp.h"
#include "replication/origin.h"
#include "utils/guc.h"

Include dependency graph for commit_ts.h:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Data Structures
struct	xl_commit_ts_set
struct	xl_commit_ts_truncate

Macros
#define	COMMIT_TS_ZEROPAGE   0x00
#define	COMMIT_TS_TRUNCATE   0x10
#define	COMMIT_TS_SETTS   0x20
#define	SizeOfCommitTsSet
#define	SizeOfCommitTsTruncate

Typedefs
typedef struct xl_commit_ts_set	xl_commit_ts_set
typedef struct xl_commit_ts_truncate	xl_commit_ts_truncate

Functions
bool	check_track_commit_timestamp (bool *newval, void **extra, GucSource source)
void	TransactionTreeSetCommitTsData (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz timestamp, RepOriginId nodeid, bool write_xlog)
bool	TransactionIdGetCommitTsData (TransactionId xid, TimestampTz *ts, RepOriginId *nodeid)
TransactionId	GetLatestCommitTsData (TimestampTz *ts, RepOriginId *nodeid)
Size	CommitTsShmemBuffers (void)
Size	CommitTsShmemSize (void)
void	CommitTsShmemInit (void)
void	BootStrapCommitTs (void)
void	StartupCommitTs (void)
void	CommitTsParameterChange (bool newvalue, bool oldvalue)
void	CompleteCommitTsInitialization (void)
void	ShutdownCommitTs (void)
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)
void	commit_ts_desc (StringInfo buf, XLogReaderState *record)
const char *	commit_ts_identify (uint8 info)

Variables
PGDLLIMPORT bool	track_commit_timestamp

Macro Definition Documentation

COMMIT_TS_SETTS

#define COMMIT_TS_SETTS   0x20

Definition at line 51 of file commit_ts.h.

Referenced by commit_ts_desc(), commit_ts_identify(), commit_ts_redo(), and WriteSetTimestampXlogRec().

COMMIT_TS_TRUNCATE

#define COMMIT_TS_TRUNCATE   0x10

Definition at line 50 of file commit_ts.h.

Referenced by commit_ts_desc(), commit_ts_identify(), commit_ts_redo(), and WriteTruncateXlogRec().

COMMIT_TS_ZEROPAGE

#define COMMIT_TS_ZEROPAGE   0x00

Definition at line 49 of file commit_ts.h.

Referenced by commit_ts_desc(), commit_ts_identify(), commit_ts_redo(), and WriteZeroPageXlogRec().

SizeOfCommitTsSet

#define SizeOfCommitTsSet

Value:

(offsetof(xl_commit_ts_set, mainxid) + \
                             sizeof(TransactionId))

Definition at line 61 of file commit_ts.h.

Referenced by commit_ts_desc(), and commit_ts_redo().

SizeOfCommitTsTruncate

#define SizeOfCommitTsTruncate

Value:

(offsetof(xl_commit_ts_truncate, oldestXid) + \
                                 sizeof(TransactionId))

Definition at line 70 of file commit_ts.h.

Referenced by WriteTruncateXlogRec().

Typedef Documentation

xl_commit_ts_set

typedef struct xl_commit_ts_set xl_commit_ts_set

xl_commit_ts_truncate

typedef struct xl_commit_ts_truncate xl_commit_ts_truncate

Function Documentation

AdvanceOldestCommitTsXid()

void AdvanceOldestCommitTsXid

(

TransactionId

oldestXact

)

Definition at line 875 of file commit_ts.c.

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

Referenced by commit_ts_redo(), and vac_truncate_clog().

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

BootStrapCommitTs()

void BootStrapCommitTs

(

void

)

Definition at line 523 of file commit_ts.c.

Referenced by BootStrapXLOG().

{
    /*
     * 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.
     */
}

check_track_commit_timestamp()

bool check_track_commit_timestamp	(	bool *	newval,
		void **	extra,
		GucSource	source
	)

CheckPointCommitTs()

void CheckPointCommitTs

(

void

)

Definition at line 760 of file commit_ts.c.

References CommitTsCtl, fsync_fname(), and SimpleLruFlush().

Referenced by CheckPointGuts().

{
    /* Flush dirty CommitTs pages to disk */
    SimpleLruFlush(CommitTsCtl, true);

    /*
     * fsync pg_commit_ts to ensure that any files flushed previously are
     * durably on disk.
     */
    fsync_fname("pg_commit_ts", true);
}

commit_ts_desc()

void commit_ts_desc	(	StringInfo	buf,
		XLogReaderState *	record
	)

Definition at line 22 of file committsdesc.c.

References appendStringInfo(), COMMIT_TS_SETTS, COMMIT_TS_TRUNCATE, COMMIT_TS_ZEROPAGE, i, xl_commit_ts_set::mainxid, xl_commit_ts_set::nodeid, xl_commit_ts_truncate::oldestXid, xl_commit_ts_truncate::pageno, palloc(), pfree(), SizeOfCommitTsSet, xl_commit_ts_set::timestamp, timestamptz_to_str(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetInfo, and XLR_INFO_MASK.

{
    char       *rec = XLogRecGetData(record);
    uint8       info = XLogRecGetInfo(record) & ~XLR_INFO_MASK;

    if (info == COMMIT_TS_ZEROPAGE)
    {
        int         pageno;

        memcpy(&pageno, rec, sizeof(int));
        appendStringInfo(buf, "%d", pageno);
    }
    else if (info == COMMIT_TS_TRUNCATE)
    {
        xl_commit_ts_truncate *trunc = (xl_commit_ts_truncate *) rec;

        appendStringInfo(buf, "pageno %d, oldestXid %u",
                         trunc->pageno, trunc->oldestXid);
    }
    else if (info == COMMIT_TS_SETTS)
    {
        xl_commit_ts_set *xlrec = (xl_commit_ts_set *) rec;
        int         nsubxids;

        appendStringInfo(buf, "set %s/%d for: %u",
                         timestamptz_to_str(xlrec->timestamp),
                         xlrec->nodeid,
                         xlrec->mainxid);
        nsubxids = ((XLogRecGetDataLen(record) - SizeOfCommitTsSet) /
                    sizeof(TransactionId));
        if (nsubxids > 0)
        {
            int         i;
            TransactionId *subxids;

            subxids = palloc(sizeof(TransactionId) * nsubxids);
            memcpy(subxids,
                   XLogRecGetData(record) + SizeOfCommitTsSet,
                   sizeof(TransactionId) * nsubxids);
            for (i = 0; i < nsubxids; i++)
                appendStringInfo(buf, ", %u", subxids[i]);
            pfree(subxids);
        }
    }
}

commit_ts_identify()

const char* commit_ts_identify

(

uint8

info

)

Definition at line 69 of file committsdesc.c.

References COMMIT_TS_SETTS, COMMIT_TS_TRUNCATE, and COMMIT_TS_ZEROPAGE.

{
    switch (info)
    {
        case COMMIT_TS_ZEROPAGE:
            return "ZEROPAGE";
        case COMMIT_TS_TRUNCATE:
            return "TRUNCATE";
        case COMMIT_TS_SETTS:
            return "SETTS";
        default:
            return NULL;
    }
}

commit_ts_redo()

void commit_ts_redo

(

XLogReaderState *

record

)

Definition at line 963 of file commit_ts.c.

References AdvanceOldestCommitTsXid(), Assert, COMMIT_TS_SETTS, COMMIT_TS_TRUNCATE, COMMIT_TS_ZEROPAGE, CommitTsCtl, elog, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), xl_commit_ts_set::mainxid, xl_commit_ts_set::nodeid, xl_commit_ts_truncate::oldestXid, xl_commit_ts_truncate::pageno, palloc(), PANIC, pfree(), SimpleLruTruncate(), SimpleLruWritePage(), SizeOfCommitTsSet, xl_commit_ts_set::timestamp, TransactionTreeSetCommitTsData(), XLogRecGetData, XLogRecGetDataLen, XLogRecGetInfo, XLogRecHasAnyBlockRefs, XLR_INFO_MASK, and ZeroCommitTsPage().

{
    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(CommitTsControlLock, LW_EXCLUSIVE);

        slotno = ZeroCommitTsPage(pageno, false);
        SimpleLruWritePage(CommitTsCtl, slotno);
        Assert(!CommitTsCtl->shared->page_dirty[slotno]);

        LWLockRelease(CommitTsControlLock);
    }
    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 if (info == COMMIT_TS_SETTS)
    {
        xl_commit_ts_set *setts = (xl_commit_ts_set *) XLogRecGetData(record);
        int         nsubxids;
        TransactionId *subxids;

        nsubxids = ((XLogRecGetDataLen(record) - SizeOfCommitTsSet) /
                    sizeof(TransactionId));
        if (nsubxids > 0)
        {
            subxids = palloc(sizeof(TransactionId) * nsubxids);
            memcpy(subxids,
                   XLogRecGetData(record) + SizeOfCommitTsSet,
                   sizeof(TransactionId) * nsubxids);
        }
        else
            subxids = NULL;

        TransactionTreeSetCommitTsData(setts->mainxid, nsubxids, subxids,
                                       setts->timestamp, setts->nodeid, true);
        if (subxids)
            pfree(subxids);
    }
    else
        elog(PANIC, "commit_ts_redo: unknown op code %u", info);
}

CommitTsParameterChange()

void CommitTsParameterChange	(	bool	newvalue,
		bool	oldvalue
	)

Definition at line 591 of file commit_ts.c.

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

Referenced by xlog_redo().

{
    /*
     * If the commit_ts module is disabled in this server and we get word from
     * the master 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 master, 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();
}

CommitTsShmemBuffers()

Size CommitTsShmemBuffers

(

void

)

Definition at line 470 of file commit_ts.c.

References Max, Min, and NBuffers.

Referenced by CommitTsShmemInit(), and CommitTsShmemSize().

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

CommitTsShmemInit()

void CommitTsShmemInit

(

void

)

Definition at line 490 of file commit_ts.c.

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

Referenced by CreateSharedMemoryAndSemaphores().

{
    bool        found;

    CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
    SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0,
                  CommitTsControlLock, "pg_commit_ts",
                  LWTRANCHE_COMMITTS_BUFFERS);

    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);
}

CommitTsShmemSize()

Size CommitTsShmemSize

(

void

)

Definition at line 479 of file commit_ts.c.

References CommitTsShmemBuffers(), and SimpleLruShmemSize().

Referenced by CreateSharedMemoryAndSemaphores().

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

CompleteCommitTsInitialization()

void CompleteCommitTsInitialization

(

void

)

Definition at line 569 of file commit_ts.c.

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

Referenced by StartupXLOG().

{
    /*
     * 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();
}

ExtendCommitTs()

void ExtendCommitTs

(

TransactionId

newestXact

)

Definition at line 784 of file commit_ts.c.

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

Referenced by GetNewTransactionId().

{
    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(CommitTsControlLock, LW_EXCLUSIVE);

    /* Zero the page and make an XLOG entry about it */
    ZeroCommitTsPage(pageno, !InRecovery);

    LWLockRelease(CommitTsControlLock);
}

GetLatestCommitTsData()

TransactionId GetLatestCommitTsData	(	TimestampTz *	ts,
		RepOriginId *	nodeid
	)

Definition at line 368 of file commit_ts.c.

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

Referenced by pg_last_committed_xact().

{
    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;
}

SetCommitTsLimit()

void SetCommitTsLimit	(	TransactionId	oldestXact,
		TransactionId	newestXact
	)

Definition at line 848 of file commit_ts.c.

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

Referenced by BootStrapXLOG(), and StartupXLOG().

{
    /*
     * 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);
}

ShutdownCommitTs()

void ShutdownCommitTs

(

void

)

Definition at line 744 of file commit_ts.c.

References CommitTsCtl, fsync_fname(), and SimpleLruFlush().

Referenced by ShutdownXLOG().

{
    /* Flush dirty CommitTs pages to disk */
    SimpleLruFlush(CommitTsCtl, false);

    /*
     * fsync pg_commit_ts to ensure that any files flushed previously are
     * durably on disk.
     */
    fsync_fname("pg_commit_ts", true);
}

StartupCommitTs()

void StartupCommitTs

(

void

)

Definition at line 559 of file commit_ts.c.

References ActivateCommitTs().

Referenced by StartupXLOG().

{
    ActivateCommitTs();
}

TransactionIdGetCommitTsData()

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

Definition at line 282 of file commit_ts.c.

References Assert, CommitTimestampShared::commitTsActive, CommitTsCtl, 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().

{
    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(CommitTsControlLock);
    return *ts != 0;
}

TransactionTreeSetCommitTsData()

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

Definition at line 145 of file commit_ts.c.

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

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

{
    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;

    /*
     * Comply with the WAL-before-data rule: if caller specified it wants this
     * value to be recorded in WAL, do so before touching the data.
     */
    if (write_xlog)
        WriteSetTimestampXlogRec(xid, nsubxids, subxids, timestamp, nodeid);

    /*
     * 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.
     */
    for (i = 0, headxid = xid;;)
    {
        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 + 1 >= nsubxids)
            break;

        /*
         * Set the new head and skip over it, as well as over the subxids we
         * just wrote.
         */
        headxid = subxids[j];
        i += j - i + 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);
}

TruncateCommitTs()

void TruncateCommitTs

(

TransactionId

oldestXact

)

Definition at line 822 of file commit_ts.c.

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

Referenced by vac_truncate_clog().

{
    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);
}

Variable Documentation

track_commit_timestamp

PGDLLIMPORT bool track_commit_timestamp

Definition at line 103 of file commit_ts.c.

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