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 "funcapi.h"
#include "miscadmin.h"
#include "storage/shmem.h"
#include "utils/fmgrprotos.h"
#include "utils/guc_hooks.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	TransactionIdToCTsEntry(xid)    ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
#define	CommitTsCtl   (&CommitTsCtlData)

Typedefs
typedef struct CommitTimestampEntry	CommitTimestampEntry
typedef struct CommitTimestampShared	CommitTimestampShared

Functions
static int64	TransactionIdToCTsPage (TransactionId xid)
static void	SetXidCommitTsInPage (TransactionId xid, int nsubxids, TransactionId *subxids, TimestampTz ts, RepOriginId nodeid, int64 pageno)
static void	TransactionIdSetCommitTs (TransactionId xid, TimestampTz ts, RepOriginId nodeid, int slotno)
static void	error_commit_ts_disabled (void)
static bool	CommitTsPagePrecedes (int64 page1, int64 page2)
static void	ActivateCommitTs (void)
static void	DeactivateCommitTs (void)
static void	WriteTruncateXlogRec (int64 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)
static int	CommitTsShmemBuffers (void)
Size	CommitTsShmemSize (void)
void	CommitTsShmemInit (void)
bool	check_commit_ts_buffers (int *newval, void **extra, GucSource source)
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 63 of file commit_ts.c.

CommitTsCtl

#define CommitTsCtl   (&CommitTsCtlData)

Definition at line 85 of file commit_ts.c.

SizeOfCommitTimestampEntry

#define SizeOfCommitTimestampEntry

Value:

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

Definition at line 60 of file commit_ts.c.

TransactionIdToCTsEntry

#define TransactionIdToCTsEntry

(

xid

)

((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)

Definition at line 77 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 681 of file commit_ts.c.

{
    TransactionId xid;
    int64       pageno;
 
    /*
     * During bootstrap, we should not register commit timestamps so skip the
     * activation in this case.
     */
    if (IsBootstrapProcessingMode())
        return;
 
    /* 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(TransamVariables->nextXid);
    pageno = TransactionIdToCTsPage(xid);
 
    /*
     * Re-Initialize our idea of the latest page number.
     */
    pg_atomic_write_u64(&CommitTsCtl->shared->latest_page_number, pageno);
 
    /*
     * 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 (TransamVariables->oldestCommitTsXid == InvalidTransactionId)
    {
        TransamVariables->oldestCommitTsXid =
            TransamVariables->newestCommitTsXid = ReadNextTransactionId();
    }
    LWLockRelease(CommitTsLock);
 
    /* Create the current segment file, if necessary */
    if (!SimpleLruDoesPhysicalPageExist(CommitTsCtl, pageno))
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
 
    /* Change the activation status in shared memory. */
    LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
    commitTsShared->commitTsActive = true;
    LWLockRelease(CommitTsLock);
}

References CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, InvalidTransactionId, IsBootstrapProcessingMode, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, TransamVariablesData::nextXid, TransamVariablesData::oldestCommitTsXid, pg_atomic_write_u64(), ReadNextTransactionId(), SimpleLruDoesPhysicalPageExist(), SimpleLruZeroAndWritePage(), TransactionIdToCTsPage(), TransamVariables, and XidFromFullTransactionId.

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

AdvanceOldestCommitTsXid()

void AdvanceOldestCommitTsXid

(

TransactionId

oldestXact

)

Definition at line 914 of file commit_ts.c.

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

References InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::oldestCommitTsXid, TransactionIdPrecedes(), and TransamVariables.

Referenced by commit_ts_redo(), and vac_truncate_clog().

BootStrapCommitTs()

void BootStrapCommitTs

(

void

)

Definition at line 594 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().

check_commit_ts_buffers()

bool check_commit_ts_buffers	(	int *	newval,
		void **	extra,
		GucSource	source
	)

Definition at line 582 of file commit_ts.c.

{
    return check_slru_buffers("commit_timestamp_buffers", newval);
}

References check_slru_buffers(), and newval.

CheckPointCommitTs()

void CheckPointCommitTs

(

void

)

Definition at line 794 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 983 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)
    {
        int64       pageno;
 
        memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
        SimpleLruZeroAndWritePage(CommitTsCtl, pageno);
    }
    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.
         */
        pg_atomic_write_u64(&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, xl_commit_ts_truncate::oldestXid, xl_commit_ts_truncate::pageno, PANIC, pg_atomic_write_u64(), SimpleLruTruncate(), SimpleLruZeroAndWritePage(), XLogRecGetData, XLogRecGetInfo, and XLogRecHasAnyBlockRefs.

CommitTsPagePrecedes()

static bool CommitTsPagePrecedes ( int64  page1, int64  page2  )

static

Definition at line 948 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 640 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()

static int CommitTsShmemBuffers ( void  )

static

Definition at line 504 of file commit_ts.c.

{
    /* auto-tune based on shared buffers */
    if (commit_timestamp_buffers == 0)
        return SimpleLruAutotuneBuffers(512, 1024);
 
    return Min(Max(16, commit_timestamp_buffers), SLRU_MAX_ALLOWED_BUFFERS);
}

References commit_timestamp_buffers, Max, Min, SimpleLruAutotuneBuffers(), and SLRU_MAX_ALLOWED_BUFFERS.

Referenced by CommitTsShmemInit(), and CommitTsShmemSize().

CommitTsShmemInit()

void CommitTsShmemInit

(

void

)

Definition at line 528 of file commit_ts.c.

{
    bool        found;
 
    /* If auto-tuning is requested, now is the time to do it */
    if (commit_timestamp_buffers == 0)
    {
        char        buf[32];
 
        snprintf(buf, sizeof(buf), "%d", CommitTsShmemBuffers());
        SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                        PGC_S_DYNAMIC_DEFAULT);
 
        /*
         * We prefer to report this value's source as PGC_S_DYNAMIC_DEFAULT.
         * However, if the DBA explicitly set commit_timestamp_buffers = 0 in
         * the config file, then PGC_S_DYNAMIC_DEFAULT will fail to override
         * that and we must force the matter with PGC_S_OVERRIDE.
         */
        if (commit_timestamp_buffers == 0)  /* failed to apply it? */
            SetConfigOption("commit_timestamp_buffers", buf, PGC_POSTMASTER,
                            PGC_S_OVERRIDE);
    }
    Assert(commit_timestamp_buffers != 0);
 
    CommitTsCtl->PagePrecedes = CommitTsPagePrecedes;
    SimpleLruInit(CommitTsCtl, "commit_timestamp", CommitTsShmemBuffers(), 0,
                  "pg_commit_ts", LWTRANCHE_COMMITTS_BUFFER,
                  LWTRANCHE_COMMITTS_SLRU,
                  SYNC_HANDLER_COMMIT_TS,
                  false);
    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(), buf, commit_timestamp_buffers, COMMIT_TS_XACTS_PER_PAGE, CommitTimestampShared::commitTsActive, CommitTsCtl, CommitTsPagePrecedes(), commitTsShared, CommitTsShmemBuffers(), CommitTimestampShared::dataLastCommit, InvalidRepOriginId, InvalidTransactionId, IsUnderPostmaster, CommitTimestampEntry::nodeid, PGC_POSTMASTER, PGC_S_DYNAMIC_DEFAULT, PGC_S_OVERRIDE, SetConfigOption(), ShmemInitStruct(), SimpleLruInit(), SlruPagePrecedesUnitTests, snprintf, SYNC_HANDLER_COMMIT_TS, CommitTimestampEntry::time, TIMESTAMP_NOBEGIN, and CommitTimestampShared::xidLastCommit.

Referenced by CreateOrAttachShmemStructs().

CommitTsShmemSize()

Size CommitTsShmemSize

(

void

)

Definition at line 517 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 1020 of file commit_ts.c.

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

References CommitTsCtl, and SlruSyncFileTag().

CompleteCommitTsInitialization()

void CompleteCommitTsInitialization

(

void

)

Definition at line 618 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 752 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;
 
    TransamVariables->oldestCommitTsXid = InvalidTransactionId;
    TransamVariables->newestCommitTsXid = InvalidTransactionId;
 
    /*
     * 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.)
     *
     * Note that we do this with CommitTsLock acquired in exclusive mode. This
     * is very heavy-handed, but since this routine can only be called in the
     * replica and should happen very rarely, we don't worry too much about
     * it.  Note also that no process should be consulting this SLRU if we
     * have just deactivated it.
     */
    (void) SlruScanDirectory(CommitTsCtl, SlruScanDirCbDeleteAll, NULL);
 
    LWLockRelease(CommitTsLock);
}

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

Referenced by CommitTsParameterChange(), and CompleteCommitTsInitialization().

error_commit_ts_disabled()

static void error_commit_ts_disabled ( void  )

static

Definition at line 379 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 816 of file commit_ts.c.

{
    int64       pageno;
    LWLock     *lock;
 
    /*
     * 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);
 
    lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
 
    LWLockAcquire(lock, LW_EXCLUSIVE);
 
    /* Zero the page ... */
    SimpleLruZeroPage(CommitTsCtl, pageno);
 
    /* and make a WAL entry about that, unless we're in REDO */
    if (!InRecovery)
        XLogSimpleInsertInt64(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE, pageno);
 
    LWLockRelease(lock);
}

References Assert(), COMMIT_TS_ZEROPAGE, CommitTimestampShared::commitTsActive, CommitTsCtl, commitTsShared, FirstNormalTransactionId, InRecovery, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), SimpleLruGetBankLock(), SimpleLruZeroPage(), TransactionIdEquals, TransactionIdToCTsEntry, TransactionIdToCTsPage(), and XLogSimpleInsertInt64().

Referenced by GetNewTransactionId().

GetLatestCommitTsData()

TransactionId GetLatestCommitTsData	(	TimestampTz *	ts,
		RepOriginId *	nodeid
	)

Definition at line 358 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 418 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 395 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 462 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 887 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 (TransamVariables->oldestCommitTsXid != InvalidTransactionId)
    {
        if (TransactionIdPrecedes(TransamVariables->oldestCommitTsXid, oldestXact))
            TransamVariables->oldestCommitTsXid = oldestXact;
        if (TransactionIdPrecedes(newestXact, TransamVariables->newestCommitTsXid))
            TransamVariables->newestCommitTsXid = newestXact;
    }
    else
    {
        Assert(TransamVariables->newestCommitTsXid == InvalidTransactionId);
        TransamVariables->oldestCommitTsXid = oldestXact;
        TransamVariables->newestCommitTsXid = newestXact;
    }
    LWLockRelease(CommitTsLock);
}

References Assert(), InvalidTransactionId, LW_EXCLUSIVE, LWLockAcquire(), LWLockRelease(), TransamVariablesData::newestCommitTsXid, TransamVariablesData::oldestCommitTsXid, TransactionIdPrecedes(), and TransamVariables.

Referenced by BootStrapXLOG(), and StartupXLOG().

SetXidCommitTsInPage()

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

static

Definition at line 220 of file commit_ts.c.

{
    LWLock     *lock = SimpleLruGetBankLock(CommitTsCtl, pageno);
    int         slotno;
    int         i;
 
    LWLockAcquire(lock, 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(lock);
}

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

Referenced by TransactionTreeSetCommitTsData().

StartupCommitTs()

void StartupCommitTs

(

void

)

Definition at line 608 of file commit_ts.c.

{
    ActivateCommitTs();
}

References ActivateCommitTs().

Referenced by StartupXLOG().

TransactionIdGetCommitTsData()

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

Definition at line 272 of file commit_ts.c.

{
    int64       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 = TransamVariables->oldestCommitTsXid;
    newestCommitTsXid = TransamVariables->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(SimpleLruGetBankLock(CommitTsCtl, pageno));
    return *ts != 0;
}

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

Referenced by GetTupleTransactionInfo(), pg_xact_commit_timestamp(), pg_xact_commit_timestamp_origin(), and update_most_recent_deletion_info().

TransactionIdSetCommitTs()

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

static

Definition at line 247 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().

TransactionIdToCTsPage()

static int64 TransactionIdToCTsPage ( TransactionId  xid )

inlinestatic

Definition at line 72 of file commit_ts.c.

{
    return xid / (int64) COMMIT_TS_XACTS_PER_PAGE;
}

References COMMIT_TS_XACTS_PER_PAGE.

Referenced by ActivateCommitTs(), ExtendCommitTs(), TransactionIdGetCommitTsData(), TransactionTreeSetCommitTsData(), and TruncateCommitTs().

TransactionTreeSetCommitTsData()

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

Definition at line 139 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 (;;)
    {
        int64       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(TransamVariables->newestCommitTsXid, newestXact))
        TransamVariables->newestCommitTsXid = newestXact;
    LWLockRelease(CommitTsLock);
}

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

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

TruncateCommitTs()

void TruncateCommitTs

(

TransactionId

oldestXact

)

Definition at line 861 of file commit_ts.c.

{
    int64       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 ( int64  pageno, TransactionId  oldestXid  )

static

Definition at line 967 of file commit_ts.c.

{
    xl_commit_ts_truncate xlrec;
 
    xlrec.pageno = pageno;
    xlrec.oldestXid = oldestXid;
 
    XLogBeginInsert();
    XLogRegisterData(&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().

Variable Documentation

CommitTsCtlData

SlruCtlData CommitTsCtlData

static

Definition at line 83 of file commit_ts.c.

commitTsShared

CommitTimestampShared* commitTsShared

static

Definition at line 105 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 109 of file commit_ts.c.

Referenced by CheckSubDeadTupleRetention(), CompleteCommitTsInitialization(), FindDeletedTupleInLocalRel(), GetTupleTransactionInfo(), InitControlFile(), and XLogReportParameters().