Index SLRUs by 64-bit integers rather than by 32-bit integers

We've had repeated bugs in the area of handling SLRU wraparound in the past,
some of which have caused data loss. Switching to an indexing system for SLRUs
that does not wrap around should allow us to get rid of a whole bunch
of problems and improve the overall reliability of the system.

This particular patch however only changes the indexing and doesn't address
the wraparound per se. This is going to be done in the following patches.

Author: Maxim Orlov, Aleksander Alekseev, Alexander Korotkov, Teodor Sigaev
Author: Nikita Glukhov, Pavel Borisov, Yura Sokolov
Reviewed-by: Jacob Champion, Heikki Linnakangas, Alexander Korotkov
Reviewed-by: Japin Li, Pavel Borisov, Tom Lane, Peter Eisentraut, Andres Freund
Reviewed-by: Andrey Borodin, Dilip Kumar, Aleksander Alekseev
Discussion: https://postgr.es/m/CACG%3DezZe1NQSCnfHOr78AtAZxJZeCvxrts0ygrxYwe%3DpyyjVWA%40mail.gmail.com
Discussion: https://postgr.es/m/CAJ7c6TPDOYBYrnCAeyndkBktO0WG2xSdYduTF0nxq%2BvfkmTF5Q%40mail.gmail.com

diff --git a/src/backend/access/rmgrdesc/clogdesc.c b/src/backend/access/rmgrdesc/clogdesc.c
index e60b76f9da6c301e86849b718bb46106a9f9ad98..6b367622ca4c6be7b78757f0db98efed5fb3a499 100644 (file)
--- a/src/backend/access/rmgrdesc/clogdesc.c
+++ b/src/backend/access/rmgrdesc/clogdesc.c
@@ -25,18 +25,18 @@ clog_desc(StringInfo buf, XLogReaderState *record)
 
        if (info == CLOG_ZEROPAGE)
        {
-               int                     pageno;
+               int64           pageno;
 
-               memcpy(&pageno, rec, sizeof(int));
-               appendStringInfo(buf, "page %d", pageno);
+               memcpy(&pageno, rec, sizeof(pageno));
+               appendStringInfo(buf, "page %lld", (long long) pageno);
        }
        else if (info == CLOG_TRUNCATE)
        {
                xl_clog_truncate xlrec;
 
                memcpy(&xlrec, rec, sizeof(xl_clog_truncate));
-               appendStringInfo(buf, "page %d; oldestXact %u",
-                                                xlrec.pageno, xlrec.oldestXact);
+               appendStringInfo(buf, "page %lld; oldestXact %u",
+                                                (long long) xlrec.pageno, xlrec.oldestXact);
        }
 }
 

diff --git a/src/backend/access/rmgrdesc/committsdesc.c b/src/backend/access/rmgrdesc/committsdesc.c
index e7155cd5072727cfc7d925ba97df05107a26ee38..6a1a6413f1def7e8f386c4ce47d27a0da97f608c 100644 (file)
--- a/src/backend/access/rmgrdesc/committsdesc.c
+++ b/src/backend/access/rmgrdesc/committsdesc.c
@@ -26,17 +26,17 @@ commit_ts_desc(StringInfo buf, XLogReaderState *record)
 
        if (info == COMMIT_TS_ZEROPAGE)
        {
-               int                     pageno;
+               int64           pageno;
 
-               memcpy(&pageno, rec, sizeof(int));
-               appendStringInfo(buf, "%d", pageno);
+               memcpy(&pageno, rec, sizeof(pageno));
+               appendStringInfo(buf, "%lld", (long long) 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);
+               appendStringInfo(buf, "pageno %lld, oldestXid %u",
+                                                (long long) trunc->pageno, trunc->oldestXid);
        }
 }
 

diff --git a/src/backend/access/rmgrdesc/mxactdesc.c b/src/backend/access/rmgrdesc/mxactdesc.c
index a2fa1eca181882324cd52e109b778ca2b68ac55d..e423a3da5e023c105f7fe2b47c415066c551e67f 100644 (file)
--- a/src/backend/access/rmgrdesc/mxactdesc.c
+++ b/src/backend/access/rmgrdesc/mxactdesc.c
@@ -55,10 +55,10 @@ multixact_desc(StringInfo buf, XLogReaderState *record)
        if (info == XLOG_MULTIXACT_ZERO_OFF_PAGE ||
                info == XLOG_MULTIXACT_ZERO_MEM_PAGE)
        {
-               int                     pageno;
+               int64           pageno;
 
-               memcpy(&pageno, rec, sizeof(int));
-               appendStringInfo(buf, "%d", pageno);
+               memcpy(&pageno, rec, sizeof(pageno));
+               appendStringInfo(buf, "%lld", (long long) pageno);
        }
        else if (info == XLOG_MULTIXACT_CREATE_ID)
        {

diff --git a/src/backend/access/transam/clog.c b/src/backend/access/transam/clog.c
index 4a431d5876760d3e665429ceb5dad958cb655fc1..cc60eab1e219739305bdb3f3ccb117ac0c092798 100644 (file)
--- a/src/backend/access/transam/clog.c
+++ b/src/backend/access/transam/clog.c
@@ -62,7 +62,17 @@
 #define CLOG_XACTS_PER_PAGE (BLCKSZ * CLOG_XACTS_PER_BYTE)
 #define CLOG_XACT_BITMASK      ((1 << CLOG_BITS_PER_XACT) - 1)
 
-#define TransactionIdToPage(xid)       ((xid) / (TransactionId) CLOG_XACTS_PER_PAGE)
+
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/CLOG_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToPage(TransactionId xid)
+{
+       return xid / (int64) CLOG_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToPgIndex(xid) ((xid) % (TransactionId) CLOG_XACTS_PER_PAGE)
 #define TransactionIdToByte(xid)       (TransactionIdToPgIndex(xid) / CLOG_XACTS_PER_BYTE)
 #define TransactionIdToBIndex(xid)     ((xid) % (TransactionId) CLOG_XACTS_PER_BYTE)
@@ -89,24 +99,24 @@ static SlruCtlData XactCtlData;
 #define XactCtl (&XactCtlData)
 
 
-static int     ZeroCLOGPage(int pageno, bool writeXlog);
-static bool CLOGPagePrecedes(int page1, int page2);
-static void WriteZeroPageXlogRec(int pageno);
-static void WriteTruncateXlogRec(int pageno, TransactionId oldestXact,
+static int     ZeroCLOGPage(int64 pageno, bool writeXlog);
+static bool CLOGPagePrecedes(int64 page1, int64 page2);
+static void WriteZeroPageXlogRec(int64 pageno);
+static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXact,
                                                                 Oid oldestXactDb);
 static void TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
                                                                           TransactionId *subxids, XidStatus status,
-                                                                          XLogRecPtr lsn, int pageno,
+                                                                          XLogRecPtr lsn, int64 pageno,
                                                                           bool all_xact_same_page);
 static void TransactionIdSetStatusBit(TransactionId xid, XidStatus status,
                                                                          XLogRecPtr lsn, int slotno);
 static void set_status_by_pages(int nsubxids, TransactionId *subxids,
                                                                XidStatus status, XLogRecPtr lsn);
 static bool TransactionGroupUpdateXidStatus(TransactionId xid,
-                                                                                       XidStatus status, XLogRecPtr lsn, int pageno);
+                                                                                       XidStatus status, XLogRecPtr lsn, int64 pageno);
 static void TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
                                                                                           TransactionId *subxids, XidStatus status,
-                                                                                          XLogRecPtr lsn, int pageno);
+                                                                                          XLogRecPtr lsn, int64 pageno);
 
 
 /*
@@ -162,7 +172,7 @@ void
 TransactionIdSetTreeStatus(TransactionId xid, int nsubxids,
                                                   TransactionId *subxids, XidStatus status, XLogRecPtr lsn)
 {
-       int                     pageno = TransactionIdToPage(xid);      /* get page of parent */
+       int64           pageno = TransactionIdToPage(xid);      /* get page of parent */
        int                     i;
 
        Assert(status == TRANSACTION_STATUS_COMMITTED ||
@@ -236,7 +246,7 @@ static void
 set_status_by_pages(int nsubxids, TransactionId *subxids,
                                        XidStatus status, XLogRecPtr lsn)
 {
-       int                     pageno = TransactionIdToPage(subxids[0]);
+       int64           pageno = TransactionIdToPage(subxids[0]);
        int                     offset = 0;
        int                     i = 0;
 
@@ -245,7 +255,7 @@ set_status_by_pages(int nsubxids, TransactionId *subxids,
        while (i < nsubxids)
        {
                int                     num_on_page = 0;
-               int                     nextpageno;
+               int64           nextpageno;
 
                do
                {
@@ -271,7 +281,7 @@ set_status_by_pages(int nsubxids, TransactionId *subxids,
 static void
 TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
                                                   TransactionId *subxids, XidStatus status,
-                                                  XLogRecPtr lsn, int pageno,
+                                                  XLogRecPtr lsn, int64 pageno,
                                                   bool all_xact_same_page)
 {
        /* Can't use group update when PGPROC overflows. */
@@ -337,7 +347,7 @@ TransactionIdSetPageStatus(TransactionId xid, int nsubxids,
 static void
 TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
                                                                   TransactionId *subxids, XidStatus status,
-                                                                  XLogRecPtr lsn, int pageno)
+                                                                  XLogRecPtr lsn, int64 pageno)
 {
        int                     slotno;
        int                     i;
@@ -411,7 +421,7 @@ TransactionIdSetPageStatusInternal(TransactionId xid, int nsubxids,
  */
 static bool
 TransactionGroupUpdateXidStatus(TransactionId xid, XidStatus status,
-                                                               XLogRecPtr lsn, int pageno)
+                                                               XLogRecPtr lsn, int64 pageno)
 {
        volatile PROC_HDR *procglobal = ProcGlobal;
        PGPROC     *proc = MyProc;
@@ -637,7 +647,7 @@ TransactionIdSetStatusBit(TransactionId xid, XidStatus status, XLogRecPtr lsn, i
 XidStatus
 TransactionIdGetStatus(TransactionId xid, XLogRecPtr *lsn)
 {
-       int                     pageno = TransactionIdToPage(xid);
+       int64           pageno = TransactionIdToPage(xid);
        int                     byteno = TransactionIdToByte(xid);
        int                     bshift = TransactionIdToBIndex(xid) * CLOG_BITS_PER_XACT;
        int                     slotno;
@@ -697,7 +707,7 @@ CLOGShmemInit(void)
        XactCtl->PagePrecedes = CLOGPagePrecedes;
        SimpleLruInit(XactCtl, "Xact", CLOGShmemBuffers(), CLOG_LSNS_PER_PAGE,
                                  XactSLRULock, "pg_xact", LWTRANCHE_XACT_BUFFER,
-                                 SYNC_HANDLER_CLOG);
+                                 SYNC_HANDLER_CLOG, false);
        SlruPagePrecedesUnitTests(XactCtl, CLOG_XACTS_PER_PAGE);
 }
 
@@ -734,7 +744,7 @@ BootStrapCLOG(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroCLOGPage(int pageno, bool writeXlog)
+ZeroCLOGPage(int64 pageno, bool writeXlog)
 {
        int                     slotno;
 
@@ -754,7 +764,7 @@ void
 StartupCLOG(void)
 {
        TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-       int                     pageno = TransactionIdToPage(xid);
+       int64           pageno = TransactionIdToPage(xid);
 
        LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 
@@ -773,7 +783,7 @@ void
 TrimCLOG(void)
 {
        TransactionId xid = XidFromFullTransactionId(ShmemVariableCache->nextXid);
-       int                     pageno = TransactionIdToPage(xid);
+       int64           pageno = TransactionIdToPage(xid);
 
        LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 
@@ -838,7 +848,7 @@ CheckPointCLOG(void)
 void
 ExtendCLOG(TransactionId newestXact)
 {
-       int                     pageno;
+       int64           pageno;
 
        /*
         * No work except at first XID of a page.  But beware: just after
@@ -877,7 +887,7 @@ ExtendCLOG(TransactionId newestXact)
 void
 TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid)
 {
-       int                     cutoffPage;
+       int64           cutoffPage;
 
        /*
         * The cutoff point is the start of the segment containing oldestXact. We
@@ -930,7 +940,7 @@ TruncateCLOG(TransactionId oldestXact, Oid oldestxid_datoid)
  * don't optimize that edge case.
  */
 static bool
-CLOGPagePrecedes(int page1, int page2)
+CLOGPagePrecedes(int64 page1, int64 page2)
 {
        TransactionId xid1;
        TransactionId xid2;
@@ -949,10 +959,10 @@ CLOGPagePrecedes(int page1, int page2)
  * Write a ZEROPAGE xlog record
  */
 static void
-WriteZeroPageXlogRec(int pageno)
+WriteZeroPageXlogRec(int64 pageno)
 {
        XLogBeginInsert();
-       XLogRegisterData((char *) (&pageno), sizeof(int));
+       XLogRegisterData((char *) (&pageno), sizeof(pageno));
        (void) XLogInsert(RM_CLOG_ID, CLOG_ZEROPAGE);
 }
 
@@ -963,7 +973,7 @@ WriteZeroPageXlogRec(int pageno)
  * in TruncateCLOG().
  */
 static void
-WriteTruncateXlogRec(int pageno, TransactionId oldestXact, Oid oldestXactDb)
+WriteTruncateXlogRec(int64 pageno, TransactionId oldestXact, Oid oldestXactDb)
 {
        XLogRecPtr      recptr;
        xl_clog_truncate xlrec;
@@ -991,10 +1001,10 @@ clog_redo(XLogReaderState *record)
 
        if (info == CLOG_ZEROPAGE)
        {
-               int                     pageno;
+               int64           pageno;
                int                     slotno;
 
-               memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+               memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
                LWLockAcquire(XactSLRULock, LW_EXCLUSIVE);
 

diff --git a/src/backend/access/transam/commit_ts.c b/src/backend/access/transam/commit_ts.c
index b897fabc702b6ed42512d73166e2fe26a969cf00..502a3d6068fa041c37c661f2b6a8d28ff5ef4aed 100644 (file)
--- a/src/backend/access/transam/commit_ts.c
+++ b/src/backend/access/transam/commit_ts.c
@@ -65,8 +65,17 @@ typedef struct CommitTimestampEntry
 #define COMMIT_TS_XACTS_PER_PAGE \
        (BLCKSZ / SizeOfCommitTimestampEntry)
 
-#define TransactionIdToCTsPage(xid) \
-       ((xid) / (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
+
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/COMMIT_TS_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToCTsPage(TransactionId xid)
+{
+       return xid / (int64) COMMIT_TS_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToCTsEntry(xid)   \
        ((xid) % (TransactionId) COMMIT_TS_XACTS_PER_PAGE)
 
@@ -103,16 +112,16 @@ bool              track_commit_timestamp;
 
 static void SetXidCommitTsInPage(TransactionId xid, int nsubxids,
                                                                 TransactionId *subxids, TimestampTz ts,
-                                                                RepOriginId nodeid, int pageno);
+                                                                RepOriginId nodeid, int64 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 int     ZeroCommitTsPage(int64 pageno, bool writeXlog);
+static bool CommitTsPagePrecedes(int64 page1, int64 page2);
 static void ActivateCommitTs(void);
 static void DeactivateCommitTs(void);
-static void WriteZeroPageXlogRec(int pageno);
-static void WriteTruncateXlogRec(int pageno, TransactionId oldestXid);
+static void WriteZeroPageXlogRec(int64 pageno);
+static void WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid);
 
 /*
  * TransactionTreeSetCommitTsData
@@ -170,7 +179,7 @@ TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids,
        i = 0;
        for (;;)
        {
-               int                     pageno = TransactionIdToCTsPage(headxid);
+               int64           pageno = TransactionIdToCTsPage(headxid);
                int                     j;
 
                for (j = i; j < nsubxids; j++)
@@ -214,7 +223,7 @@ TransactionTreeSetCommitTsData(TransactionId xid, int nsubxids,
 static void
 SetXidCommitTsInPage(TransactionId xid, int nsubxids,
                                         TransactionId *subxids, TimestampTz ts,
-                                        RepOriginId nodeid, int pageno)
+                                        RepOriginId nodeid, int64 pageno)
 {
        int                     slotno;
        int                     i;
@@ -266,7 +275,7 @@ bool
 TransactionIdGetCommitTsData(TransactionId xid, TimestampTz *ts,
                                                         RepOriginId *nodeid)
 {
-       int                     pageno = TransactionIdToCTsPage(xid);
+       int64           pageno = TransactionIdToCTsPage(xid);
        int                     entryno = TransactionIdToCTsEntry(xid);
        int                     slotno;
        CommitTimestampEntry entry;
@@ -523,7 +532,8 @@ CommitTsShmemInit(void)
        SimpleLruInit(CommitTsCtl, "CommitTs", CommitTsShmemBuffers(), 0,
                                  CommitTsSLRULock, "pg_commit_ts",
                                  LWTRANCHE_COMMITTS_BUFFER,
-                                 SYNC_HANDLER_COMMIT_TS);
+                                 SYNC_HANDLER_COMMIT_TS,
+                                 false);
        SlruPagePrecedesUnitTests(CommitTsCtl, COMMIT_TS_XACTS_PER_PAGE);
 
        commitTsShared = ShmemInitStruct("CommitTs shared",
@@ -569,7 +579,7 @@ BootStrapCommitTs(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroCommitTsPage(int pageno, bool writeXlog)
+ZeroCommitTsPage(int64 pageno, bool writeXlog)
 {
        int                     slotno;
 
@@ -662,7 +672,7 @@ static void
 ActivateCommitTs(void)
 {
        TransactionId xid;
-       int                     pageno;
+       int64           pageno;
 
        /* If we've done this already, there's nothing to do */
        LWLockAcquire(CommitTsLock, LW_EXCLUSIVE);
@@ -795,7 +805,7 @@ CheckPointCommitTs(void)
 void
 ExtendCommitTs(TransactionId newestXact)
 {
-       int                     pageno;
+       int64           pageno;
 
        /*
         * Nothing to do if module not enabled.  Note we do an unlocked read of
@@ -833,7 +843,7 @@ ExtendCommitTs(TransactionId newestXact)
 void
 TruncateCommitTs(TransactionId oldestXact)
 {
-       int                     cutoffPage;
+       int64           cutoffPage;
 
        /*
         * The cutoff point is the start of the segment containing oldestXact. We
@@ -918,7 +928,7 @@ AdvanceOldestCommitTsXid(TransactionId oldestXact)
  * oldestXact=N+2.1, it would be precious at oldestXact=N+2.9.
  */
 static bool
-CommitTsPagePrecedes(int page1, int page2)
+CommitTsPagePrecedes(int64 page1, int64 page2)
 {
        TransactionId xid1;
        TransactionId xid2;
@@ -937,10 +947,10 @@ CommitTsPagePrecedes(int page1, int page2)
  * Write a ZEROPAGE xlog record
  */
 static void
-WriteZeroPageXlogRec(int pageno)
+WriteZeroPageXlogRec(int64 pageno)
 {
        XLogBeginInsert();
-       XLogRegisterData((char *) (&pageno), sizeof(int));
+       XLogRegisterData((char *) (&pageno), sizeof(pageno));
        (void) XLogInsert(RM_COMMIT_TS_ID, COMMIT_TS_ZEROPAGE);
 }
 
@@ -948,7 +958,7 @@ WriteZeroPageXlogRec(int pageno)
  * Write a TRUNCATE xlog record
  */
 static void
-WriteTruncateXlogRec(int pageno, TransactionId oldestXid)
+WriteTruncateXlogRec(int64 pageno, TransactionId oldestXid)
 {
        xl_commit_ts_truncate xlrec;
 
@@ -973,10 +983,10 @@ commit_ts_redo(XLogReaderState *record)
 
        if (info == COMMIT_TS_ZEROPAGE)
        {
-               int                     pageno;
+               int64           pageno;
                int                     slotno;
 
-               memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+               memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
                LWLockAcquire(CommitTsSLRULock, LW_EXCLUSIVE);
 

diff --git a/src/backend/access/transam/multixact.c b/src/backend/access/transam/multixact.c
index 57ed34c0a88facc16f48e21dc19b83c97812b3c3..db3423f12e2f1ae60a3aae3b7a63d6261dad30e8 100644 (file)
--- a/src/backend/access/transam/multixact.c
+++ b/src/backend/access/transam/multixact.c
@@ -354,10 +354,10 @@ static void mXactCachePut(MultiXactId multi, int nmembers,
 static char *mxstatus_to_string(MultiXactStatus status);
 
 /* management of SLRU infrastructure */
-static int     ZeroMultiXactOffsetPage(int pageno, bool writeXlog);
-static int     ZeroMultiXactMemberPage(int pageno, bool writeXlog);
-static bool MultiXactOffsetPagePrecedes(int page1, int page2);
-static bool MultiXactMemberPagePrecedes(int page1, int page2);
+static int     ZeroMultiXactOffsetPage(int64 pageno, bool writeXlog);
+static int     ZeroMultiXactMemberPage(int64 pageno, bool writeXlog);
+static bool MultiXactOffsetPagePrecedes(int64 page1, int64 page2);
+static bool MultiXactMemberPagePrecedes(int64 page1, int64 page2);
 static bool MultiXactOffsetPrecedes(MultiXactOffset offset1,
                                                                        MultiXactOffset offset2);
 static void ExtendMultiXactOffset(MultiXactId multi);
@@ -366,7 +366,7 @@ static bool MultiXactOffsetWouldWrap(MultiXactOffset boundary,
                                                                         MultiXactOffset start, uint32 distance);
 static bool SetOffsetVacuumLimit(bool is_startup);
 static bool find_multixact_start(MultiXactId multi, MultiXactOffset *result);
-static void WriteMZeroPageXlogRec(int pageno, uint8 info);
+static void WriteMZeroPageXlogRec(int64 pageno, uint8 info);
 static void WriteMTruncateXlogRec(Oid oldestMultiDB,
                                                                  MultiXactId startTruncOff,
                                                                  MultiXactId endTruncOff,
@@ -864,8 +864,8 @@ static void
 RecordNewMultiXact(MultiXactId multi, MultiXactOffset offset,
                                   int nmembers, MultiXactMember *members)
 {
-       int                     pageno;
-       int                     prev_pageno;
+       int64           pageno;
+       int64           prev_pageno;
        int                     entryno;
        int                     slotno;
        MultiXactOffset *offptr;
@@ -1225,8 +1225,8 @@ int
 GetMultiXactIdMembers(MultiXactId multi, MultiXactMember **members,
                                          bool from_pgupgrade, bool isLockOnly)
 {
-       int                     pageno;
-       int                     prev_pageno;
+       int64           pageno;
+       int64           prev_pageno;
        int                     entryno;
        int                     slotno;
        MultiXactOffset *offptr;
@@ -1854,13 +1854,15 @@ MultiXactShmemInit(void)
                                  "MultiXactOffset", NUM_MULTIXACTOFFSET_BUFFERS, 0,
                                  MultiXactOffsetSLRULock, "pg_multixact/offsets",
                                  LWTRANCHE_MULTIXACTOFFSET_BUFFER,
-                                 SYNC_HANDLER_MULTIXACT_OFFSET);
+                                 SYNC_HANDLER_MULTIXACT_OFFSET,
+                                 false);
        SlruPagePrecedesUnitTests(MultiXactOffsetCtl, MULTIXACT_OFFSETS_PER_PAGE);
        SimpleLruInit(MultiXactMemberCtl,
                                  "MultiXactMember", NUM_MULTIXACTMEMBER_BUFFERS, 0,
                                  MultiXactMemberSLRULock, "pg_multixact/members",
                                  LWTRANCHE_MULTIXACTMEMBER_BUFFER,
-                                 SYNC_HANDLER_MULTIXACT_MEMBER);
+                                 SYNC_HANDLER_MULTIXACT_MEMBER,
+                                 false);
        /* doesn't call SimpleLruTruncate() or meet criteria for unit tests */
 
        /* Initialize our shared state struct */
@@ -1928,7 +1930,7 @@ BootStrapMultiXact(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroMultiXactOffsetPage(int pageno, bool writeXlog)
+ZeroMultiXactOffsetPage(int64 pageno, bool writeXlog)
 {
        int                     slotno;
 
@@ -1944,7 +1946,7 @@ ZeroMultiXactOffsetPage(int pageno, bool writeXlog)
  * Ditto, for MultiXactMember
  */
 static int
-ZeroMultiXactMemberPage(int pageno, bool writeXlog)
+ZeroMultiXactMemberPage(int64 pageno, bool writeXlog)
 {
        int                     slotno;
 
@@ -1974,7 +1976,7 @@ ZeroMultiXactMemberPage(int pageno, bool writeXlog)
 static void
 MaybeExtendOffsetSlru(void)
 {
-       int                     pageno;
+       int64           pageno;
 
        pageno = MultiXactIdToOffsetPage(MultiXactState->nextMXact);
 
@@ -2009,7 +2011,7 @@ StartupMultiXact(void)
 {
        MultiXactId multi = MultiXactState->nextMXact;
        MultiXactOffset offset = MultiXactState->nextOffset;
-       int                     pageno;
+       int64           pageno;
 
        /*
         * Initialize offset's idea of the latest page number.
@@ -2034,7 +2036,7 @@ TrimMultiXact(void)
        MultiXactOffset offset;
        MultiXactId oldestMXact;
        Oid                     oldestMXactDB;
-       int                     pageno;
+       int64           pageno;
        int                     entryno;
        int                     flagsoff;
 
@@ -2403,7 +2405,7 @@ MultiXactAdvanceOldest(MultiXactId oldestMulti, Oid oldestMultiDB)
 static void
 ExtendMultiXactOffset(MultiXactId multi)
 {
-       int                     pageno;
+       int64           pageno;
 
        /*
         * No work except at first MultiXactId of a page.  But beware: just after
@@ -2452,7 +2454,7 @@ ExtendMultiXactMember(MultiXactOffset offset, int nmembers)
                flagsbit = MXOffsetToFlagsBitShift(offset);
                if (flagsoff == 0 && flagsbit == 0)
                {
-                       int                     pageno;
+                       int64           pageno;
 
                        pageno = MXOffsetToMemberPage(offset);
 
@@ -2735,7 +2737,7 @@ static bool
 find_multixact_start(MultiXactId multi, MultiXactOffset *result)
 {
        MultiXactOffset offset;
-       int                     pageno;
+       int64           pageno;
        int                     entryno;
        int                     slotno;
        MultiXactOffset *offptr;
@@ -2854,7 +2856,7 @@ MultiXactMemberFreezeThreshold(void)
 
 typedef struct mxtruncinfo
 {
-       int                     earliestExistingPage;
+       int64           earliestExistingPage;
 } mxtruncinfo;
 
 /*
@@ -2862,7 +2864,7 @@ typedef struct mxtruncinfo
  *             This callback determines the earliest existing page number.
  */
 static bool
-SlruScanDirCbFindEarliest(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbFindEarliest(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
        mxtruncinfo *trunc = (mxtruncinfo *) data;
 
@@ -3113,7 +3115,7 @@ TruncateMultiXact(MultiXactId newOldestMulti, Oid newOldestMultiDB)
  * translational symmetry.
  */
 static bool
-MultiXactOffsetPagePrecedes(int page1, int page2)
+MultiXactOffsetPagePrecedes(int64 page1, int64 page2)
 {
        MultiXactId multi1;
        MultiXactId multi2;
@@ -3133,7 +3135,7 @@ MultiXactOffsetPagePrecedes(int page1, int page2)
  * purposes.  There is no "invalid offset number" so use the numbers verbatim.
  */
 static bool
-MultiXactMemberPagePrecedes(int page1, int page2)
+MultiXactMemberPagePrecedes(int64 page1, int64 page2)
 {
        MultiXactOffset offset1;
        MultiXactOffset offset2;
@@ -3191,10 +3193,10 @@ MultiXactOffsetPrecedes(MultiXactOffset offset1, MultiXactOffset offset2)
  * OFFSETs page (info shows which)
  */
 static void
-WriteMZeroPageXlogRec(int pageno, uint8 info)
+WriteMZeroPageXlogRec(int64 pageno, uint8 info)
 {
        XLogBeginInsert();
-       XLogRegisterData((char *) (&pageno), sizeof(int));
+       XLogRegisterData((char *) (&pageno), sizeof(pageno));
        (void) XLogInsert(RM_MULTIXACT_ID, info);
 }
 
@@ -3239,10 +3241,10 @@ multixact_redo(XLogReaderState *record)
 
        if (info == XLOG_MULTIXACT_ZERO_OFF_PAGE)
        {
-               int                     pageno;
+               int64           pageno;
                int                     slotno;
 
-               memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+               memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
                LWLockAcquire(MultiXactOffsetSLRULock, LW_EXCLUSIVE);
 
@@ -3254,10 +3256,10 @@ multixact_redo(XLogReaderState *record)
        }
        else if (info == XLOG_MULTIXACT_ZERO_MEM_PAGE)
        {
-               int                     pageno;
+               int64           pageno;
                int                     slotno;
 
-               memcpy(&pageno, XLogRecGetData(record), sizeof(int));
+               memcpy(&pageno, XLogRecGetData(record), sizeof(pageno));
 
                LWLockAcquire(MultiXactMemberSLRULock, LW_EXCLUSIVE);
 
@@ -3299,7 +3301,7 @@ multixact_redo(XLogReaderState *record)
        else if (info == XLOG_MULTIXACT_TRUNCATE_ID)
        {
                xl_multixact_truncate xlrec;
-               int                     pageno;
+               int64           pageno;
 
                memcpy(&xlrec, XLogRecGetData(record),
                           SizeOfMultiXactTruncate);

diff --git a/src/backend/access/transam/slru.c b/src/backend/access/transam/slru.c
index 9ed24e1185cd1dbd3c1a58ccc3c51759b01efb38..ac49c99c8bf2a2fa83406768bb6d898460b111f0 100644 (file)
--- a/src/backend/access/transam/slru.c
+++ b/src/backend/access/transam/slru.c
@@ -60,8 +60,33 @@
 #include "storage/fd.h"
 #include "storage/shmem.h"
 
-#define SlruFileName(ctl, path, seg) \
-       snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir, seg)
+static int     inline
+SlruFileName(SlruCtl ctl, char *path, int64 segno)
+{
+       if (ctl->long_segment_names)
+       {
+               /*
+                * We could use 16 characters here but the disadvantage would be that
+                * the SLRU segments will be hard to distinguish from WAL segments.
+                *
+                * For this reason we use 15 characters. It is enough but also means
+                * that in the future we can't decrease SLRU_PAGES_PER_SEGMENT easily.
+                */
+               Assert(segno >= 0 && segno <= INT64CONST(0xFFFFFFFFFFFFFFF));
+               return snprintf(path, MAXPGPATH, "%s/%015llX", ctl->Dir,
+                                               (long long) segno);
+       }
+       else
+       {
+               /*
+                * Despite the fact that %04X format string is used up to 24 bit
+                * integers are allowed. See SlruCorrectSegmentFilenameLength()
+                */
+               Assert(segno >= 0 && segno <= INT64CONST(0xFFFFFF));
+               return snprintf(path, MAXPGPATH, "%s/%04X", (ctl)->Dir,
+                                               (unsigned int) segno);
+       }
+}
 
 /*
  * During SimpleLruWriteAll(), we will usually not need to write more than one
@@ -75,7 +100,7 @@ typedef struct SlruWriteAllData
 {
        int                     num_files;              /* # files actually open */
        int                     fd[MAX_WRITEALL_BUFFERS];       /* their FD's */
-       int                     segno[MAX_WRITEALL_BUFFERS];    /* their log seg#s */
+       int64           segno[MAX_WRITEALL_BUFFERS];    /* their log seg#s */
 } SlruWriteAllData;
 
 typedef struct SlruWriteAllData *SlruWriteAll;
@@ -138,15 +163,16 @@ static int        slru_errno;
 static void SimpleLruZeroLSNs(SlruCtl ctl, int slotno);
 static void SimpleLruWaitIO(SlruCtl ctl, int slotno);
 static void SlruInternalWritePage(SlruCtl ctl, int slotno, SlruWriteAll fdata);
-static bool SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno);
-static bool SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno,
+static bool SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno);
+static bool SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno,
                                                                  SlruWriteAll fdata);
-static void SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid);
-static int     SlruSelectLRUPage(SlruCtl ctl, int pageno);
+static void SlruReportIOError(SlruCtl ctl, int64 pageno, TransactionId xid);
+static int     SlruSelectLRUPage(SlruCtl ctl, int64 pageno);
 
 static bool SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename,
-                                                                         int segpage, void *data);
-static void SlruInternalDeleteSegment(SlruCtl ctl, int segno);
+                                                                         int64 segpage, void *data);
+static void SlruInternalDeleteSegment(SlruCtl ctl, int64 segno);
+
 
 /*
  * Initialization of shared memory
@@ -162,7 +188,7 @@ SimpleLruShmemSize(int nslots, int nlsns)
        sz += MAXALIGN(nslots * sizeof(char *));        /* page_buffer[] */
        sz += MAXALIGN(nslots * sizeof(SlruPageStatus));        /* page_status[] */
        sz += MAXALIGN(nslots * sizeof(bool));  /* page_dirty[] */
-       sz += MAXALIGN(nslots * sizeof(int));   /* page_number[] */
+       sz += MAXALIGN(nslots * sizeof(int64)); /* page_number[] */
        sz += MAXALIGN(nslots * sizeof(int));   /* page_lru_count[] */
        sz += MAXALIGN(nslots * sizeof(LWLockPadded));  /* buffer_locks[] */
 
@@ -187,7 +213,7 @@ SimpleLruShmemSize(int nslots, int nlsns)
 void
 SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
                          LWLock *ctllock, const char *subdir, int tranche_id,
-                         SyncRequestHandler sync_handler)
+                         SyncRequestHandler sync_handler, bool long_segment_names)
 {
        SlruShared      shared;
        bool            found;
@@ -226,8 +252,8 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
                offset += MAXALIGN(nslots * sizeof(SlruPageStatus));
                shared->page_dirty = (bool *) (ptr + offset);
                offset += MAXALIGN(nslots * sizeof(bool));
-               shared->page_number = (int *) (ptr + offset);
-               offset += MAXALIGN(nslots * sizeof(int));
+               shared->page_number = (int64 *) (ptr + offset);
+               offset += MAXALIGN(nslots * sizeof(int64));
                shared->page_lru_count = (int *) (ptr + offset);
                offset += MAXALIGN(nslots * sizeof(int));
 
@@ -266,6 +292,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
         */
        ctl->shared = shared;
        ctl->sync_handler = sync_handler;
+       ctl->long_segment_names = long_segment_names;
        strlcpy(ctl->Dir, subdir, sizeof(ctl->Dir));
 }
 
@@ -278,7 +305,7 @@ SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
  * Control lock must be held at entry, and will be held at exit.
  */
 int
-SimpleLruZeroPage(SlruCtl ctl, int pageno)
+SimpleLruZeroPage(SlruCtl ctl, int64 pageno)
 {
        SlruShared      shared = ctl->shared;
        int                     slotno;
@@ -393,7 +420,7 @@ SimpleLruWaitIO(SlruCtl ctl, int slotno)
  * Control lock must be held at entry, and will be held at exit.
  */
 int
-SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
+SimpleLruReadPage(SlruCtl ctl, int64 pageno, bool write_ok,
                                  TransactionId xid)
 {
        SlruShared      shared = ctl->shared;
@@ -493,7 +520,7 @@ SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
  * It is unspecified whether the lock will be shared or exclusive.
  */
 int
-SimpleLruReadPage_ReadOnly(SlruCtl ctl, int pageno, TransactionId xid)
+SimpleLruReadPage_ReadOnly(SlruCtl ctl, int64 pageno, TransactionId xid)
 {
        SlruShared      shared = ctl->shared;
        int                     slotno;
@@ -540,7 +567,7 @@ static void
 SlruInternalWritePage(SlruCtl ctl, int slotno, SlruWriteAll fdata)
 {
        SlruShared      shared = ctl->shared;
-       int                     pageno = shared->page_number[slotno];
+       int64           pageno = shared->page_number[slotno];
        bool            ok;
 
        /* If a write is in progress, wait for it to finish */
@@ -624,9 +651,9 @@ SimpleLruWritePage(SlruCtl ctl, int slotno)
  * large enough to contain the given page.
  */
 bool
-SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno)
+SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int64 pageno)
 {
-       int                     segno = pageno / SLRU_PAGES_PER_SEGMENT;
+       int64           segno = pageno / SLRU_PAGES_PER_SEGMENT;
        int                     rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
        int                     offset = rpageno * BLCKSZ;
        char            path[MAXPGPATH];
@@ -682,10 +709,10 @@ SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno)
  * read/write operations.  We could cache one virtual file pointer ...
  */
 static bool
-SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
+SlruPhysicalReadPage(SlruCtl ctl, int64 pageno, int slotno)
 {
        SlruShared      shared = ctl->shared;
-       int                     segno = pageno / SLRU_PAGES_PER_SEGMENT;
+       int64           segno = pageno / SLRU_PAGES_PER_SEGMENT;
        int                     rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
        off_t           offset = rpageno * BLCKSZ;
        char            path[MAXPGPATH];
@@ -754,10 +781,10 @@ SlruPhysicalReadPage(SlruCtl ctl, int pageno, int slotno)
  * SimpleLruWriteAll.
  */
 static bool
-SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno, SlruWriteAll fdata)
+SlruPhysicalWritePage(SlruCtl ctl, int64 pageno, int slotno, SlruWriteAll fdata)
 {
        SlruShared      shared = ctl->shared;
-       int                     segno = pageno / SLRU_PAGES_PER_SEGMENT;
+       int64           segno = pageno / SLRU_PAGES_PER_SEGMENT;
        int                     rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
        off_t           offset = rpageno * BLCKSZ;
        char            path[MAXPGPATH];
@@ -929,9 +956,9 @@ SlruPhysicalWritePage(SlruCtl ctl, int pageno, int slotno, SlruWriteAll fdata)
  * SlruPhysicalWritePage.  Call this after cleaning up shared-memory state.
  */
 static void
-SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
+SlruReportIOError(SlruCtl ctl, int64 pageno, TransactionId xid)
 {
-       int                     segno = pageno / SLRU_PAGES_PER_SEGMENT;
+       int64           segno = pageno / SLRU_PAGES_PER_SEGMENT;
        int                     rpageno = pageno % SLRU_PAGES_PER_SEGMENT;
        int                     offset = rpageno * BLCKSZ;
        char            path[MAXPGPATH];
@@ -1014,7 +1041,7 @@ SlruReportIOError(SlruCtl ctl, int pageno, TransactionId xid)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-SlruSelectLRUPage(SlruCtl ctl, int pageno)
+SlruSelectLRUPage(SlruCtl ctl, int64 pageno)
 {
        SlruShared      shared = ctl->shared;
 
@@ -1025,10 +1052,10 @@ SlruSelectLRUPage(SlruCtl ctl, int pageno)
                int                     cur_count;
                int                     bestvalidslot = 0;      /* keep compiler quiet */
                int                     best_valid_delta = -1;
-               int                     best_valid_page_number = 0; /* keep compiler quiet */
+               int64           best_valid_page_number = 0; /* keep compiler quiet */
                int                     bestinvalidslot = 0;    /* keep compiler quiet */
                int                     best_invalid_delta = -1;
-               int                     best_invalid_page_number = 0;   /* keep compiler quiet */
+               int64           best_invalid_page_number = 0;   /* keep compiler quiet */
 
                /* See if page already has a buffer assigned */
                for (slotno = 0; slotno < shared->num_slots; slotno++)
@@ -1069,7 +1096,7 @@ SlruSelectLRUPage(SlruCtl ctl, int pageno)
                for (slotno = 0; slotno < shared->num_slots; slotno++)
                {
                        int                     this_delta;
-                       int                     this_page_number;
+                       int64           this_page_number;
 
                        if (shared->page_status[slotno] == SLRU_PAGE_EMPTY)
                                return slotno;
@@ -1159,7 +1186,7 @@ SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied)
        SlruShared      shared = ctl->shared;
        SlruWriteAllData fdata;
        int                     slotno;
-       int                     pageno = 0;
+       int64           pageno = 0;
        int                     i;
        bool            ok;
 
@@ -1224,7 +1251,7 @@ SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied)
  * after it has accrued freshly-written data.
  */
 void
-SimpleLruTruncate(SlruCtl ctl, int cutoffPage)
+SimpleLruTruncate(SlruCtl ctl, int64 cutoffPage)
 {
        SlruShared      shared = ctl->shared;
        int                     slotno;
@@ -1302,7 +1329,7 @@ restart:
  * they either can't yet contain anything, or have already been cleaned out.
  */
 static void
-SlruInternalDeleteSegment(SlruCtl ctl, int segno)
+SlruInternalDeleteSegment(SlruCtl ctl, int64 segno)
 {
        char            path[MAXPGPATH];
 
@@ -1325,7 +1352,7 @@ SlruInternalDeleteSegment(SlruCtl ctl, int segno)
  * Delete an individual SLRU segment, identified by the segment number.
  */
 void
-SlruDeleteSegment(SlruCtl ctl, int segno)
+SlruDeleteSegment(SlruCtl ctl, int64 segno)
 {
        SlruShared      shared = ctl->shared;
        int                     slotno;
@@ -1389,9 +1416,9 @@ restart:
  * first>=cutoff && last>=cutoff: no; every page of this segment is too young
  */
 static bool
-SlruMayDeleteSegment(SlruCtl ctl, int segpage, int cutoffPage)
+SlruMayDeleteSegment(SlruCtl ctl, int64 segpage, int64 cutoffPage)
 {
-       int                     seg_last_page = segpage + SLRU_PAGES_PER_SEGMENT - 1;
+       int64           seg_last_page = segpage + SLRU_PAGES_PER_SEGMENT - 1;
 
        Assert(segpage % SLRU_PAGES_PER_SEGMENT == 0);
 
@@ -1405,7 +1432,7 @@ SlruPagePrecedesTestOffset(SlruCtl ctl, int per_page, uint32 offset)
 {
        TransactionId lhs,
                                rhs;
-       int                     newestPage,
+       int64           newestPage,
                                oldestPage;
        TransactionId newestXact,
                                oldestXact;
@@ -1498,9 +1525,10 @@ SlruPagePrecedesUnitTests(SlruCtl ctl, int per_page)
  *             one containing the page passed as "data".
  */
 bool
-SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int64 segpage,
+                                                       void *data)
 {
-       int                     cutoffPage = *(int *) data;
+       int64           cutoffPage = *(int64 *) data;
 
        if (SlruMayDeleteSegment(ctl, segpage, cutoffPage))
                return true;                    /* found one; don't iterate any more */
@@ -1513,9 +1541,10 @@ SlruScanDirCbReportPresence(SlruCtl ctl, char *filename, int segpage, void *data
  *             This callback deletes segments prior to the one passed in as "data".
  */
 static bool
-SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int64 segpage,
+                                                 void *data)
 {
-       int                     cutoffPage = *(int *) data;
+       int64           cutoffPage = *(int64 *) data;
 
        if (SlruMayDeleteSegment(ctl, segpage, cutoffPage))
                SlruInternalDeleteSegment(ctl, segpage / SLRU_PAGES_PER_SEGMENT);
@@ -1528,13 +1557,37 @@ SlruScanDirCbDeleteCutoff(SlruCtl ctl, char *filename, int segpage, void *data)
  *             This callback deletes all segments.
  */
 bool
-SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage, void *data)
+SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
        SlruInternalDeleteSegment(ctl, segpage / SLRU_PAGES_PER_SEGMENT);
 
        return false;                           /* keep going */
 }
 
+/*
+ * An internal function used by SlruScanDirectory().
+ *
+ * Returns true if a file with a name of a given length may be a correct
+ * SLRU segment.
+ */
+static inline bool
+SlruCorrectSegmentFilenameLength(SlruCtl ctl, size_t len)
+{
+       if (ctl->long_segment_names)
+               return (len == 15);             /* see SlruFileName() */
+       else
+
+               /*
+                * Commit 638cf09e76d allowed 5-character lengths. Later commit
+                * 73c986adde5 allowed 6-character length.
+                *
+                * Note: There is an ongoing plan to migrate all SLRUs to 64-bit page
+                * numbers, and the corresponding 15-character file names, which may
+                * eventually deprecate the support for 4, 5, and 6-character names.
+                */
+               return (len == 4 || len == 5 || len == 6);
+}
+
 /*
  * Scan the SimpleLru directory and apply a callback to each file found in it.
  *
@@ -1556,8 +1609,8 @@ SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data)
        bool            retval = false;
        DIR                *cldir;
        struct dirent *clde;
-       int                     segno;
-       int                     segpage;
+       int64           segno;
+       int64           segpage;
 
        cldir = AllocateDir(ctl->Dir);
        while ((clde = ReadDir(cldir, ctl->Dir)) != NULL)
@@ -1566,10 +1619,10 @@ SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data)
 
                len = strlen(clde->d_name);
 
-               if ((len == 4 || len == 5 || len == 6) &&
+               if (SlruCorrectSegmentFilenameLength(ctl, len) &&
                        strspn(clde->d_name, "0123456789ABCDEF") == len)
                {
-                       segno = (int) strtol(clde->d_name, NULL, 16);
+                       segno = strtoi64(clde->d_name, NULL, 16);
                        segpage = segno * SLRU_PAGES_PER_SEGMENT;
 
                        elog(DEBUG2, "SlruScanDirectory invoking callback on %s/%s",

diff --git a/src/backend/access/transam/subtrans.c b/src/backend/access/transam/subtrans.c
index 62bb610167c0a060baaa73e0cdc310e1428db440..64673eaef6bdb1cb8493f0baeaad10ff2c3ae617 100644 (file)
--- a/src/backend/access/transam/subtrans.c
+++ b/src/backend/access/transam/subtrans.c
@@ -51,7 +51,16 @@
 /* We need four bytes per xact */
 #define SUBTRANS_XACTS_PER_PAGE (BLCKSZ / sizeof(TransactionId))
 
-#define TransactionIdToPage(xid) ((xid) / (TransactionId) SUBTRANS_XACTS_PER_PAGE)
+/*
+ * Although we return an int64 the actual value can't currently exceed
+ * 0xFFFFFFFF/SUBTRANS_XACTS_PER_PAGE.
+ */
+static inline int64
+TransactionIdToPage(TransactionId xid)
+{
+       return xid / (int64) SUBTRANS_XACTS_PER_PAGE;
+}
+
 #define TransactionIdToEntry(xid) ((xid) % (TransactionId) SUBTRANS_XACTS_PER_PAGE)
 
 
@@ -63,8 +72,8 @@ static SlruCtlData SubTransCtlData;
 #define SubTransCtl  (&SubTransCtlData)
 
 
-static int     ZeroSUBTRANSPage(int pageno);
-static bool SubTransPagePrecedes(int page1, int page2);
+static int     ZeroSUBTRANSPage(int64 pageno);
+static bool SubTransPagePrecedes(int64 page1, int64 page2);
 
 
 /*
@@ -73,7 +82,7 @@ static bool SubTransPagePrecedes(int page1, int page2);
 void
 SubTransSetParent(TransactionId xid, TransactionId parent)
 {
-       int                     pageno = TransactionIdToPage(xid);
+       int64           pageno = TransactionIdToPage(xid);
        int                     entryno = TransactionIdToEntry(xid);
        int                     slotno;
        TransactionId *ptr;
@@ -108,7 +117,7 @@ SubTransSetParent(TransactionId xid, TransactionId parent)
 TransactionId
 SubTransGetParent(TransactionId xid)
 {
-       int                     pageno = TransactionIdToPage(xid);
+       int64           pageno = TransactionIdToPage(xid);
        int                     entryno = TransactionIdToEntry(xid);
        int                     slotno;
        TransactionId *ptr;
@@ -193,7 +202,8 @@ SUBTRANSShmemInit(void)
        SubTransCtl->PagePrecedes = SubTransPagePrecedes;
        SimpleLruInit(SubTransCtl, "Subtrans", NUM_SUBTRANS_BUFFERS, 0,
                                  SubtransSLRULock, "pg_subtrans",
-                                 LWTRANCHE_SUBTRANS_BUFFER, SYNC_HANDLER_NONE);
+                                 LWTRANCHE_SUBTRANS_BUFFER, SYNC_HANDLER_NONE,
+                                 false);
        SlruPagePrecedesUnitTests(SubTransCtl, SUBTRANS_XACTS_PER_PAGE);
 }
 
@@ -233,7 +243,7 @@ BootStrapSUBTRANS(void)
  * Control lock must be held at entry, and will be held at exit.
  */
 static int
-ZeroSUBTRANSPage(int pageno)
+ZeroSUBTRANSPage(int64 pageno)
 {
        return SimpleLruZeroPage(SubTransCtl, pageno);
 }
@@ -249,8 +259,8 @@ void
 StartupSUBTRANS(TransactionId oldestActiveXID)
 {
        FullTransactionId nextXid;
-       int                     startPage;
-       int                     endPage;
+       int64           startPage;
+       int64           endPage;
 
        /*
         * Since we don't expect pg_subtrans to be valid across crashes, we
@@ -307,7 +317,7 @@ CheckPointSUBTRANS(void)
 void
 ExtendSUBTRANS(TransactionId newestXact)
 {
-       int                     pageno;
+       int64           pageno;
 
        /*
         * No work except at first XID of a page.  But beware: just after
@@ -337,7 +347,7 @@ ExtendSUBTRANS(TransactionId newestXact)
 void
 TruncateSUBTRANS(TransactionId oldestXact)
 {
-       int                     cutoffPage;
+       int64           cutoffPage;
 
        /*
         * The cutoff point is the start of the segment containing oldestXact. We
@@ -359,7 +369,7 @@ TruncateSUBTRANS(TransactionId oldestXact)
  * Analogous to CLOGPagePrecedes().
  */
 static bool
-SubTransPagePrecedes(int page1, int page2)
+SubTransPagePrecedes(int64 page1, int64 page2)
 {
        TransactionId xid1;
        TransactionId xid2;

diff --git a/src/backend/commands/async.c b/src/backend/commands/async.c
index 38ddae08b860e54d7c05d65189d0758cd4c07ddb..346bc28a36faef0770703170f1d087af391ae1d3 100644 (file)
--- a/src/backend/commands/async.c
+++ b/src/backend/commands/async.c
@@ -196,7 +196,7 @@ typedef struct AsyncQueueEntry
  */
 typedef struct QueuePosition
 {
-       int                     page;                   /* SLRU page number */
+       int64           page;                   /* SLRU page number */
        int                     offset;                 /* byte offset within page */
 } QueuePosition;
 
@@ -443,8 +443,8 @@ static bool tryAdvanceTail = false;
 bool           Trace_notify = false;
 
 /* local function prototypes */
-static int     asyncQueuePageDiff(int p, int q);
-static bool asyncQueuePagePrecedes(int p, int q);
+static int64 asyncQueuePageDiff(int64 p, int64 q);
+static bool asyncQueuePagePrecedes(int64 p, int64 q);
 static void queue_listen(ListenActionKind action, const char *channel);
 static void Async_UnlistenOnExit(int code, Datum arg);
 static void Exec_ListenPreCommit(void);
@@ -477,10 +477,10 @@ static void ClearPendingActionsAndNotifies(void);
  * Compute the difference between two queue page numbers (i.e., p - q),
  * accounting for wraparound.
  */
-static int
-asyncQueuePageDiff(int p, int q)
+static int64
+asyncQueuePageDiff(int64 p, int64 q)
 {
-       int                     diff;
+       int64           diff;
 
        /*
         * We have to compare modulo (QUEUE_MAX_PAGE+1)/2.  Both inputs should be
@@ -504,7 +504,7 @@ asyncQueuePageDiff(int p, int q)
  * extant page, we need not assess entries within a page.
  */
 static bool
-asyncQueuePagePrecedes(int p, int q)
+asyncQueuePagePrecedes(int64 p, int64 q)
 {
        return asyncQueuePageDiff(p, q) < 0;
 }
@@ -571,7 +571,7 @@ AsyncShmemInit(void)
        NotifyCtl->PagePrecedes = asyncQueuePagePrecedes;
        SimpleLruInit(NotifyCtl, "Notify", NUM_NOTIFY_BUFFERS, 0,
                                  NotifySLRULock, "pg_notify", LWTRANCHE_NOTIFY_BUFFER,
-                                 SYNC_HANDLER_NONE);
+                                 SYNC_HANDLER_NONE, false);
 
        if (!found)
        {
@@ -1336,7 +1336,7 @@ asyncQueueIsFull(void)
 static bool
 asyncQueueAdvance(volatile QueuePosition *position, int entryLength)
 {
-       int                     pageno = QUEUE_POS_PAGE(*position);
+       int64           pageno = QUEUE_POS_PAGE(*position);
        int                     offset = QUEUE_POS_OFFSET(*position);
        bool            pageJump = false;
 
@@ -1409,7 +1409,7 @@ asyncQueueAddEntries(ListCell *nextNotify)
 {
        AsyncQueueEntry qe;
        QueuePosition queue_head;
-       int                     pageno;
+       int64           pageno;
        int                     offset;
        int                     slotno;
 

diff --git a/src/backend/storage/lmgr/predicate.c b/src/backend/storage/lmgr/predicate.c
index a794546db3e6d1cec6cf20440e691530234c44e0..eb684fa5444bfbcd4344a3e63f2640eafc0d6570 100644 (file)
--- a/src/backend/storage/lmgr/predicate.c
+++ b/src/backend/storage/lmgr/predicate.c
@@ -437,7 +437,7 @@ static void SetPossibleUnsafeConflict(SERIALIZABLEXACT *roXact, SERIALIZABLEXACT
 static void ReleaseRWConflict(RWConflict conflict);
 static void FlagSxactUnsafe(SERIALIZABLEXACT *sxact);
 
-static bool SerialPagePrecedesLogically(int page1, int page2);
+static bool SerialPagePrecedesLogically(int64 page1, int64 page2);
 static void SerialInit(void);
 static void SerialAdd(TransactionId xid, SerCommitSeqNo minConflictCommitSeqNo);
 static SerCommitSeqNo SerialGetMinConflictCommitSeqNo(TransactionId xid);
@@ -724,7 +724,7 @@ FlagSxactUnsafe(SERIALIZABLEXACT *sxact)
  * Analogous to CLOGPagePrecedes().
  */
 static bool
-SerialPagePrecedesLogically(int page1, int page2)
+SerialPagePrecedesLogically(int64 page1, int64 page2)
 {
        TransactionId xid1;
        TransactionId xid2;
@@ -744,7 +744,7 @@ SerialPagePrecedesLogicallyUnitTests(void)
 {
        int                     per_page = SERIAL_ENTRIESPERPAGE,
                                offset = per_page / 2;
-       int                     newestPage,
+       int64           newestPage,
                                oldestPage,
                                headPage,
                                targetPage;
@@ -809,7 +809,8 @@ SerialInit(void)
        SerialSlruCtl->PagePrecedes = SerialPagePrecedesLogically;
        SimpleLruInit(SerialSlruCtl, "Serial",
                                  NUM_SERIAL_BUFFERS, 0, SerialSLRULock, "pg_serial",
-                                 LWTRANCHE_SERIAL_BUFFER, SYNC_HANDLER_NONE);
+                                 LWTRANCHE_SERIAL_BUFFER, SYNC_HANDLER_NONE,
+                                 false);
 #ifdef USE_ASSERT_CHECKING
        SerialPagePrecedesLogicallyUnitTests();
 #endif
@@ -842,9 +843,9 @@ static void
 SerialAdd(TransactionId xid, SerCommitSeqNo minConflictCommitSeqNo)
 {
        TransactionId tailXid;
-       int                     targetPage;
+       int64           targetPage;
        int                     slotno;
-       int                     firstZeroPage;
+       int64           firstZeroPage;
        bool            isNewPage;
 
        Assert(TransactionIdIsValid(xid));

diff --git a/src/include/access/clog.h b/src/include/access/clog.h
index d99444f073f042cabe4e856a741a66f27cf226ad..8fd99ba670b7ab26af3ddfc72b6be8771f33f53c 100644 (file)
--- a/src/include/access/clog.h
+++ b/src/include/access/clog.h
@@ -31,7 +31,7 @@ typedef int XidStatus;
 
 typedef struct xl_clog_truncate
 {
-       int                     pageno;
+       int64           pageno;
        TransactionId oldestXact;
        Oid                     oldestXactDb;
 } xl_clog_truncate;

diff --git a/src/include/access/commit_ts.h b/src/include/access/commit_ts.h
index 5087cdce51e2cca6c3a752c206d4cd77fee6c7d6..11fa3a922d9d8e2737af9b1d7deaed4953f1d6cb 100644 (file)
--- a/src/include/access/commit_ts.h
+++ b/src/include/access/commit_ts.h
@@ -60,7 +60,7 @@ typedef struct xl_commit_ts_set
 
 typedef struct xl_commit_ts_truncate
 {
-       int                     pageno;
+       int64           pageno;
        TransactionId oldestXid;
 } xl_commit_ts_truncate;
 

diff --git a/src/include/access/slru.h b/src/include/access/slru.h
index 552cc19e68592ae40c487c6d95a7708c4f89165c..091e2202c953621ecc9ca26a2352f52a3b8a84a6 100644 (file)
--- a/src/include/access/slru.h
+++ b/src/include/access/slru.h
@@ -64,7 +64,7 @@ typedef struct SlruSharedData
        char      **page_buffer;
        SlruPageStatus *page_status;
        bool       *page_dirty;
-       int                *page_number;
+       int64      *page_number;
        int                *page_lru_count;
        LWLockPadded *buffer_locks;
 
@@ -95,7 +95,7 @@ typedef struct SlruSharedData
         * this is not critical data, since we use it only to avoid swapping out
         * the latest page.
         */
-       int                     latest_page_number;
+       int64           latest_page_number;
 
        /* SLRU's index for statistics purposes (might not be unique) */
        int                     slru_stats_idx;
@@ -127,7 +127,15 @@ typedef struct SlruCtlData
         * the behavior of this callback has no functional implications.)  Use
         * SlruPagePrecedesUnitTests() in SLRUs meeting its criteria.
         */
-       bool            (*PagePrecedes) (int, int);
+       bool            (*PagePrecedes) (int64, int64);
+
+       /*
+        * If true, use long segment filenames formed from lower 48 bits of the
+        * segment number, e.g. pg_xact/000000001234. Otherwise, use short
+        * filenames formed from lower 16 bits of the segment number e.g.
+        * pg_xact/1234.
+        */
+       bool            long_segment_names;
 
        /*
         * Dir is set during SimpleLruInit and does not change thereafter. Since
@@ -142,11 +150,12 @@ typedef SlruCtlData *SlruCtl;
 extern Size SimpleLruShmemSize(int nslots, int nlsns);
 extern void SimpleLruInit(SlruCtl ctl, const char *name, int nslots, int nlsns,
                                                  LWLock *ctllock, const char *subdir, int tranche_id,
-                                                 SyncRequestHandler sync_handler);
-extern int     SimpleLruZeroPage(SlruCtl ctl, int pageno);
-extern int     SimpleLruReadPage(SlruCtl ctl, int pageno, bool write_ok,
+                                                 SyncRequestHandler sync_handler,
+                                                 bool long_segment_names);
+extern int     SimpleLruZeroPage(SlruCtl ctl, int64 pageno);
+extern int     SimpleLruReadPage(SlruCtl ctl, int64 pageno, bool write_ok,
                                                          TransactionId xid);
-extern int     SimpleLruReadPage_ReadOnly(SlruCtl ctl, int pageno,
+extern int     SimpleLruReadPage_ReadOnly(SlruCtl ctl, int64 pageno,
                                                                           TransactionId xid);
 extern void SimpleLruWritePage(SlruCtl ctl, int slotno);
 extern void SimpleLruWriteAll(SlruCtl ctl, bool allow_redirtied);
@@ -155,20 +164,20 @@ extern void SlruPagePrecedesUnitTests(SlruCtl ctl, int per_page);
 #else
 #define SlruPagePrecedesUnitTests(ctl, per_page) do {} while (0)
 #endif
-extern void SimpleLruTruncate(SlruCtl ctl, int cutoffPage);
-extern bool SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int pageno);
+extern void SimpleLruTruncate(SlruCtl ctl, int64 cutoffPage);
+extern bool SimpleLruDoesPhysicalPageExist(SlruCtl ctl, int64 pageno);
 
-typedef bool (*SlruScanCallback) (SlruCtl ctl, char *filename, int segpage,
+typedef bool (*SlruScanCallback) (SlruCtl ctl, char *filename, int64 segpage,
                                                                  void *data);
 extern bool SlruScanDirectory(SlruCtl ctl, SlruScanCallback callback, void *data);
-extern void SlruDeleteSegment(SlruCtl ctl, int segno);
+extern void SlruDeleteSegment(SlruCtl ctl, int64 segno);
 
 extern int     SlruSyncFileTag(SlruCtl ctl, const FileTag *ftag, char *path);
 
 /* SlruScanDirectory public callbacks */
 extern bool SlruScanDirCbReportPresence(SlruCtl ctl, char *filename,
-                                                                               int segpage, void *data);
-extern bool SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int segpage,
+                                                                               int64 segpage, void *data);
+extern bool SlruScanDirCbDeleteAll(SlruCtl ctl, char *filename, int64 segpage,
                                                                   void *data);
 
 #endif                                                 /* SLRU_H */

diff --git a/src/include/storage/proc.h b/src/include/storage/proc.h
index ef74f3269324a40d5a2c23b56ce69cdf885f5462..4b2596124956134abd3d10bac19f397981b97468 100644 (file)
--- a/src/include/storage/proc.h
+++ b/src/include/storage/proc.h
@@ -281,7 +281,7 @@ struct PGPROC
        TransactionId clogGroupMemberXid;       /* transaction id of clog group member */
        XidStatus       clogGroupMemberXidStatus;       /* transaction status of clog
                                                                                         * group member */
-       int                     clogGroupMemberPage;    /* clog page corresponding to
+       int64           clogGroupMemberPage;    /* clog page corresponding to
                                                                                 * transaction id of clog group member */
        XLogRecPtr      clogGroupMemberLsn; /* WAL location of commit record for clog
                                                                         * group member */

diff --git a/src/include/storage/sync.h b/src/include/storage/sync.h
index 963cc82125aa49bc6ce2f68745351e8c0c0fc582..d0af1d902a79c81bd8b16f4e70752479c42c41d3 100644 (file)
--- a/src/include/storage/sync.h
+++ b/src/include/storage/sync.h
@@ -52,7 +52,7 @@ typedef struct FileTag
        int16           handler;                /* SyncRequestHandler value, saving space */
        int16           forknum;                /* ForkNumber, saving space */
        RelFileLocator rlocator;
-       uint32          segno;
+       uint64          segno;
 } FileTag;
 
 extern void InitSync(void);

diff --git a/src/test/modules/test_slru/expected/test_slru.out b/src/test/modules/test_slru/expected/test_slru.out
index 0e66fdc205fc46bd2348ae983c6b6f305326a5fa..4e5e562b43424ce00d86701eeb23586dcff4bcab 100644 (file)
--- a/src/test/modules/test_slru/expected/test_slru.out
+++ b/src/test/modules/test_slru/expected/test_slru.out
@@ -61,7 +61,7 @@ SELECT test_slru_page_writeall();
 
 -- Flush the last page written out.
 SELECT test_slru_page_sync(12393);
-NOTICE:  Called SlruSyncFileTag() for segment 387 on path pg_test_slru/0183
+NOTICE:  Called SlruSyncFileTag() for segment 387 on path pg_test_slru/000000000000183
  test_slru_page_sync 
 ---------------------
  

diff --git a/src/test/modules/test_slru/test_slru--1.0.sql b/src/test/modules/test_slru/test_slru--1.0.sql
index 8635e7df015744e58c62e7935a8c478fbd3d16ff..202e8da3fdeeb6c58634dfef55a42ae1b2c2f65f 100644 (file)
--- a/src/test/modules/test_slru/test_slru--1.0.sql
+++ b/src/test/modules/test_slru/test_slru--1.0.sql
@@ -1,21 +1,21 @@
 -- complain if script is sourced in psql, rather than via CREATE EXTENSION
 \echo Use "CREATE EXTENSION test_slru" to load this file. \quit
 
-CREATE OR REPLACE FUNCTION test_slru_page_write(int, text) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_write(bigint, text) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_write' LANGUAGE C;
 CREATE OR REPLACE FUNCTION test_slru_page_writeall() RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_writeall' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_sync(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_sync(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_sync' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_read(int, bool DEFAULT true) RETURNS text
+CREATE OR REPLACE FUNCTION test_slru_page_read(bigint, bool DEFAULT true) RETURNS text
   AS 'MODULE_PATHNAME', 'test_slru_page_read' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_readonly(int) RETURNS text
+CREATE OR REPLACE FUNCTION test_slru_page_readonly(bigint) RETURNS text
   AS 'MODULE_PATHNAME', 'test_slru_page_readonly' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_exists(int) RETURNS bool
+CREATE OR REPLACE FUNCTION test_slru_page_exists(bigint) RETURNS bool
   AS 'MODULE_PATHNAME', 'test_slru_page_exists' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_delete(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_delete(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_delete' LANGUAGE C;
-CREATE OR REPLACE FUNCTION test_slru_page_truncate(int) RETURNS VOID
+CREATE OR REPLACE FUNCTION test_slru_page_truncate(bigint) RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_page_truncate' LANGUAGE C;
 CREATE OR REPLACE FUNCTION test_slru_delete_all() RETURNS VOID
   AS 'MODULE_PATHNAME', 'test_slru_delete_all' LANGUAGE C;

diff --git a/src/test/modules/test_slru/test_slru.c b/src/test/modules/test_slru/test_slru.c
index ae21444c47637103b3dc73fadc60a23d271ef7c4..d0fb9444e8a9bd17eeaa365aa26b3053c3ac3be2 100644 (file)
--- a/src/test/modules/test_slru/test_slru.c
+++ b/src/test/modules/test_slru/test_slru.c
@@ -51,7 +51,7 @@ static shmem_request_hook_type prev_shmem_request_hook = NULL;
 static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
 
 static bool
-test_slru_scan_cb(SlruCtl ctl, char *filename, int segpage, void *data)
+test_slru_scan_cb(SlruCtl ctl, char *filename, int64 segpage, void *data)
 {
        elog(NOTICE, "Calling test_slru_scan_cb()");
        return SlruScanDirCbDeleteAll(ctl, filename, segpage, data);
@@ -60,7 +60,7 @@ test_slru_scan_cb(SlruCtl ctl, char *filename, int segpage, void *data)
 Datum
 test_slru_page_write(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        char       *data = text_to_cstring(PG_GETARG_TEXT_PP(1));
        int                     slotno;
 
@@ -95,7 +95,7 @@ test_slru_page_writeall(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_read(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        bool            write_ok = PG_GETARG_BOOL(1);
        char       *data = NULL;
        int                     slotno;
@@ -113,7 +113,7 @@ test_slru_page_read(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_readonly(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        char       *data = NULL;
        int                     slotno;
 
@@ -131,7 +131,7 @@ test_slru_page_readonly(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_exists(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        bool            found;
 
        LWLockAcquire(TestSLRULock, LW_EXCLUSIVE);
@@ -144,7 +144,7 @@ test_slru_page_exists(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_sync(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        FileTag         ftag;
        char            path[MAXPGPATH];
 
@@ -152,8 +152,8 @@ test_slru_page_sync(PG_FUNCTION_ARGS)
        ftag.segno = pageno / SLRU_PAGES_PER_SEGMENT;
        SlruSyncFileTag(TestSlruCtl, &ftag, path);
 
-       elog(NOTICE, "Called SlruSyncFileTag() for segment %u on path %s",
-                ftag.segno, path);
+       elog(NOTICE, "Called SlruSyncFileTag() for segment %lld on path %s",
+                (long long) ftag.segno, path);
 
        PG_RETURN_VOID();
 }
@@ -161,13 +161,14 @@ test_slru_page_sync(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_delete(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
        FileTag         ftag;
 
        ftag.segno = pageno / SLRU_PAGES_PER_SEGMENT;
        SlruDeleteSegment(TestSlruCtl, ftag.segno);
 
-       elog(NOTICE, "Called SlruDeleteSegment() for segment %u", ftag.segno);
+       elog(NOTICE, "Called SlruDeleteSegment() for segment %lld",
+                (long long) ftag.segno);
 
        PG_RETURN_VOID();
 }
@@ -175,7 +176,7 @@ test_slru_page_delete(PG_FUNCTION_ARGS)
 Datum
 test_slru_page_truncate(PG_FUNCTION_ARGS)
 {
-       int                     pageno = PG_GETARG_INT32(0);
+       int64           pageno = PG_GETARG_INT64(0);
 
        SimpleLruTruncate(TestSlruCtl, pageno);
        PG_RETURN_VOID();
@@ -205,7 +206,7 @@ test_slru_shmem_request(void)
 }
 
 static bool
-test_slru_page_precedes_logically(int page1, int page2)
+test_slru_page_precedes_logically(int64 page1, int64 page2)
 {
        return page1 < page2;
 }
@@ -213,6 +214,11 @@ test_slru_page_precedes_logically(int page1, int page2)
 static void
 test_slru_shmem_startup(void)
 {
+       /*
+        * Short segments names are well tested elsewhere so in this test we are
+        * focusing on long names.
+        */
+       const bool      long_segment_names = true;
        const char      slru_dir_name[] = "pg_test_slru";
        int                     test_tranche_id;
 
@@ -233,7 +239,7 @@ test_slru_shmem_startup(void)
        TestSlruCtl->PagePrecedes = test_slru_page_precedes_logically;
        SimpleLruInit(TestSlruCtl, "TestSLRU",
                                  NUM_TEST_BUFFERS, 0, TestSLRULock, slru_dir_name,
-                                 test_tranche_id, SYNC_HANDLER_NONE);
+                                 test_tranche_id, SYNC_HANDLER_NONE, long_segment_names);
 }
 
 void