diff --git a/src/backend/replication/basebackup.c b/src/backend/replication/basebackup.c index 09ecc15..2f688ec 100644 *** a/src/backend/replication/basebackup.c --- b/src/backend/replication/basebackup.c *************** static int64 throttling_counter; *** 95,101 **** static int64 elapsed_min_unit; /* The last check of the transfer rate. */ ! static int64 throttled_last; /* * The contents of these directories are removed or recreated during server --- 95,101 ---- static int64 elapsed_min_unit; /* The last check of the transfer rate. */ ! static IntegerTimestamp throttled_last; /* * The contents of these directories are removed or recreated during server *************** throttle(size_t increment) *** 1346,1352 **** return; /* Time elapsed since the last measurement (and possible wake up). */ ! elapsed = GetCurrentIntegerTimestamp() - throttled_last; /* How much should have elapsed at minimum? */ elapsed_min = elapsed_min_unit * (throttling_counter / throttling_sample); sleep = elapsed_min - elapsed; --- 1346,1352 ---- return; /* Time elapsed since the last measurement (and possible wake up). */ ! elapsed = GetCurrentIntegerTimestamp().integer_ts - throttled_last.integer_ts; /* How much should have elapsed at minimum? */ elapsed_min = elapsed_min_unit * (throttling_counter / throttling_sample); sleep = elapsed_min - elapsed; diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c index 0b19fec..30e2f19 100644 *** a/src/backend/replication/logical/worker.c --- b/src/backend/replication/logical/worker.c *************** ApplyLoop(void) *** 989,995 **** start_lsn = pq_getmsgint64(&s); end_lsn = pq_getmsgint64(&s); send_time = ! IntegerTimestampToTimestampTz(pq_getmsgint64(&s)); if (last_received < start_lsn) last_received = start_lsn; --- 989,995 ---- start_lsn = pq_getmsgint64(&s); end_lsn = pq_getmsgint64(&s); send_time = ! Int64ToTimestampTz(pq_getmsgint64(&s)); if (last_received < start_lsn) last_received = start_lsn; *************** ApplyLoop(void) *** 1009,1015 **** endpos = pq_getmsgint64(&s); timestamp = ! IntegerTimestampToTimestampTz(pq_getmsgint64(&s)); reply_requested = pq_getmsgbyte(&s); send_feedback(endpos, reply_requested, false); --- 1009,1015 ---- endpos = pq_getmsgint64(&s); timestamp = ! Int64ToTimestampTz(pq_getmsgint64(&s)); reply_requested = pq_getmsgbyte(&s); send_feedback(endpos, reply_requested, false); diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c index 5c2e72b..289c806 100644 *** a/src/backend/replication/walreceiver.c --- b/src/backend/replication/walreceiver.c *************** XLogWalRcvProcessMsg(unsigned char type, *** 892,898 **** /* read the fields */ dataStart = pq_getmsgint64(&incoming_message); walEnd = pq_getmsgint64(&incoming_message); ! sendTime = IntegerTimestampToTimestampTz( pq_getmsgint64(&incoming_message)); ProcessWalSndrMessage(walEnd, sendTime); --- 892,898 ---- /* read the fields */ dataStart = pq_getmsgint64(&incoming_message); walEnd = pq_getmsgint64(&incoming_message); ! sendTime = Int64ToTimestampTz( pq_getmsgint64(&incoming_message)); ProcessWalSndrMessage(walEnd, sendTime); *************** XLogWalRcvProcessMsg(unsigned char type, *** 913,919 **** /* read the fields */ walEnd = pq_getmsgint64(&incoming_message); ! sendTime = IntegerTimestampToTimestampTz( pq_getmsgint64(&incoming_message)); replyRequested = pq_getmsgbyte(&incoming_message); --- 913,919 ---- /* read the fields */ walEnd = pq_getmsgint64(&incoming_message); ! sendTime = Int64ToTimestampTz( pq_getmsgint64(&incoming_message)); replyRequested = pq_getmsgbyte(&incoming_message); *************** XLogWalRcvSendReply(bool force, bool req *** 1149,1155 **** pq_sendint64(&reply_message, writePtr); pq_sendint64(&reply_message, flushPtr); pq_sendint64(&reply_message, applyPtr); ! pq_sendint64(&reply_message, GetCurrentIntegerTimestamp()); pq_sendbyte(&reply_message, requestReply ? 1 : 0); /* Send it */ --- 1149,1155 ---- pq_sendint64(&reply_message, writePtr); pq_sendint64(&reply_message, flushPtr); pq_sendint64(&reply_message, applyPtr); ! pq_sendint64(&reply_message, GetCurrentIntegerTimestamp().integer_ts); pq_sendbyte(&reply_message, requestReply ? 1 : 0); /* Send it */ *************** XLogWalRcvSendHSFeedback(bool immed) *** 1241,1247 **** /* Construct the message and send it. */ resetStringInfo(&reply_message); pq_sendbyte(&reply_message, 'h'); ! pq_sendint64(&reply_message, GetCurrentIntegerTimestamp()); pq_sendint(&reply_message, xmin, 4); pq_sendint(&reply_message, nextEpoch, 4); walrcv_send(wrconn, reply_message.data, reply_message.len); --- 1241,1247 ---- /* Construct the message and send it. */ resetStringInfo(&reply_message); pq_sendbyte(&reply_message, 'h'); ! pq_sendint64(&reply_message, GetCurrentIntegerTimestamp().integer_ts); pq_sendint(&reply_message, xmin, 4); pq_sendint(&reply_message, nextEpoch, 4); walrcv_send(wrconn, reply_message.data, reply_message.len); diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index ba506e2..ed0fef8 100644 *** a/src/backend/replication/walsender.c --- b/src/backend/replication/walsender.c *************** WalSndWriteData(LogicalDecodingContext * *** 1016,1022 **** * several releases by streaming physical replication. */ resetStringInfo(&tmpbuf); ! pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp()); memcpy(&ctx->out->data[1 + sizeof(int64) + sizeof(int64)], tmpbuf.data, sizeof(int64)); --- 1016,1022 ---- * several releases by streaming physical replication. */ resetStringInfo(&tmpbuf); ! pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp().integer_ts); memcpy(&ctx->out->data[1 + sizeof(int64) + sizeof(int64)], tmpbuf.data, sizeof(int64)); *************** XLogSendPhysical(void) *** 2336,2342 **** * Fill the send timestamp last, so that it is taken as late as possible. */ resetStringInfo(&tmpbuf); ! pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp()); memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)], tmpbuf.data, sizeof(int64)); --- 2336,2342 ---- * Fill the send timestamp last, so that it is taken as late as possible. */ resetStringInfo(&tmpbuf); ! pq_sendint64(&tmpbuf, GetCurrentIntegerTimestamp().integer_ts); memcpy(&output_message.data[1 + sizeof(int64) + sizeof(int64)], tmpbuf.data, sizeof(int64)); *************** WalSndKeepalive(bool requestReply) *** 2844,2850 **** resetStringInfo(&output_message); pq_sendbyte(&output_message, 'k'); pq_sendint64(&output_message, sentPtr); ! pq_sendint64(&output_message, GetCurrentIntegerTimestamp()); pq_sendbyte(&output_message, requestReply ? 1 : 0); /* ... and send it wrapped in CopyData */ --- 2844,2850 ---- resetStringInfo(&output_message); pq_sendbyte(&output_message, 'k'); pq_sendint64(&output_message, sentPtr); ! pq_sendint64(&output_message, GetCurrentIntegerTimestamp().integer_ts); pq_sendbyte(&output_message, requestReply ? 1 : 0); /* ... and send it wrapped in CopyData */ diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index 3cb5120..9717273 100644 *** a/src/backend/storage/buffer/bufmgr.c --- b/src/backend/storage/buffer/bufmgr.c *************** void *** 4333,4339 **** TestForOldSnapshot_impl(Snapshot snapshot, Relation relation) { if (RelationAllowsEarlyPruning(relation) ! && (snapshot)->whenTaken < GetOldSnapshotThresholdTimestamp()) ereport(ERROR, (errcode(ERRCODE_SNAPSHOT_TOO_OLD), errmsg("snapshot too old"))); --- 4333,4339 ---- TestForOldSnapshot_impl(Snapshot snapshot, Relation relation) { if (RelationAllowsEarlyPruning(relation) ! && (snapshot)->whenTaken.integer_ts < GetOldSnapshotThresholdTimestamp().integer_ts) ereport(ERROR, (errcode(ERRCODE_SNAPSHOT_TOO_OLD), errmsg("snapshot too old"))); diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c index cd14667..fcb4093 100644 *** a/src/backend/storage/ipc/procarray.c --- b/src/backend/storage/ipc/procarray.c *************** GetSnapshotData(Snapshot snapshot) *** 1760,1766 **** * dummy values that don't require any locking. */ snapshot->lsn = InvalidXLogRecPtr; ! snapshot->whenTaken = 0; } else { --- 1760,1766 ---- * dummy values that don't require any locking. */ snapshot->lsn = InvalidXLogRecPtr; ! snapshot->whenTaken.integer_ts = 0; } else { diff --git a/src/backend/utils/adt/timestamp.c b/src/backend/utils/adt/timestamp.c index 9b4c012..9d18875 100644 *** a/src/backend/utils/adt/timestamp.c --- b/src/backend/utils/adt/timestamp.c *************** GetCurrentTimestamp(void) *** 1711,1720 **** * with --enable-integer-datetimes, this is identical to GetCurrentTimestamp(), * and is implemented as a macro. */ ! #ifndef HAVE_INT64_TIMESTAMP ! int64 GetCurrentIntegerTimestamp(void) { int64 result; struct timeval tp; --- 1711,1721 ---- * with --enable-integer-datetimes, this is identical to GetCurrentTimestamp(), * and is implemented as a macro. */ ! //#ifndef HAVE_INT64_TIMESTAMP ! IntegerTimestamp GetCurrentIntegerTimestamp(void) { + IntegerTimestamp ts; int64 result; struct timeval tp; *************** GetCurrentIntegerTimestamp(void) *** 1725,1748 **** result = (result * USECS_PER_SEC) + tp.tv_usec; return result; } #endif /* ! * IntegerTimestampToTimestampTz -- convert an int64 timestamp to native format * * When compiled with --enable-integer-datetimes, this is implemented as a ! * no-op macro. */ #ifndef HAVE_INT64_TIMESTAMP TimestampTz ! IntegerTimestampToTimestampTz(int64 timestamp) { TimestampTz result; ! result = timestamp / USECS_PER_SEC; ! result += (timestamp % USECS_PER_SEC) / 1000000.0; return result; } --- 1726,1761 ---- result = (result * USECS_PER_SEC) + tp.tv_usec; + ts.integer_ts = result; + return ts; + } + //#endif + + #ifndef HAVE_INT64_TIMESTAMP + TimestampTz + Int64ToTimestampTz(int64 timestamp) + { + TimestampTz result; + + result = ((TimestampTz) timestamp) / USECS_PER_SEC; + return result; } #endif /* ! * IntegerTimestampToTimestampTz -- convert integer timestamp to native format * * When compiled with --enable-integer-datetimes, this is implemented as a ! * trivial macro. */ #ifndef HAVE_INT64_TIMESTAMP TimestampTz ! IntegerTimestampToTimestampTz(IntegerTimestamp timestamp) { TimestampTz result; ! result = ((TimestampTz) timestamp.integer_ts) / USECS_PER_SEC; return result; } diff --git a/src/backend/utils/time/snapmgr.c b/src/backend/utils/time/snapmgr.c index 92afc32..338c2e1 100644 *** a/src/backend/utils/time/snapmgr.c --- b/src/backend/utils/time/snapmgr.c *************** typedef struct OldSnapshotControlData *** 83,95 **** * only allowed to move forward. */ slock_t mutex_current; /* protect current_timestamp */ ! int64 current_timestamp; /* latest snapshot timestamp */ slock_t mutex_latest_xmin; /* protect latest_xmin and * next_map_update */ TransactionId latest_xmin; /* latest snapshot xmin */ ! int64 next_map_update; /* latest snapshot valid up to */ slock_t mutex_threshold; /* protect threshold fields */ ! int64 threshold_timestamp; /* earlier snapshot is old */ TransactionId threshold_xid; /* earlier xid may be gone */ /* --- 83,95 ---- * only allowed to move forward. */ slock_t mutex_current; /* protect current_timestamp */ ! IntegerTimestamp current_timestamp; /* latest snapshot timestamp */ slock_t mutex_latest_xmin; /* protect latest_xmin and * next_map_update */ TransactionId latest_xmin; /* latest snapshot xmin */ ! IntegerTimestamp next_map_update; /* latest snapshot valid up to */ slock_t mutex_threshold; /* protect threshold fields */ ! IntegerTimestamp threshold_timestamp; /* earlier snapshot is old */ TransactionId threshold_xid; /* earlier xid may be gone */ /* *************** typedef struct OldSnapshotControlData *** 121,127 **** * Persistence is not needed. */ int head_offset; /* subscript of oldest tracked time */ ! int64 head_timestamp; /* time corresponding to head xid */ int count_used; /* how many slots are in use */ TransactionId xid_by_minute[FLEXIBLE_ARRAY_MEMBER]; } OldSnapshotControlData; --- 121,127 ---- * Persistence is not needed. */ int head_offset; /* subscript of oldest tracked time */ ! IntegerTimestamp head_timestamp; /* time corresponding to head xid */ int count_used; /* how many slots are in use */ TransactionId xid_by_minute[FLEXIBLE_ARRAY_MEMBER]; } OldSnapshotControlData; *************** static Snapshot FirstXactSnapshot = NULL *** 219,225 **** static List *exportedSnapshots = NIL; /* Prototypes for local functions */ ! static int64 AlignTimestampToMinuteBoundary(int64 ts); static Snapshot CopySnapshot(Snapshot snapshot); static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); --- 219,225 ---- static List *exportedSnapshots = NIL; /* Prototypes for local functions */ ! static IntegerTimestamp AlignTimestampToMinuteBoundary(IntegerTimestamp ts); static Snapshot CopySnapshot(Snapshot snapshot); static void FreeSnapshot(Snapshot snapshot); static void SnapshotResetXmin(void); *************** typedef struct SerializedSnapshotData *** 239,245 **** bool suboverflowed; bool takenDuringRecovery; CommandId curcid; ! int64 whenTaken; XLogRecPtr lsn; } SerializedSnapshotData; --- 239,245 ---- bool suboverflowed; bool takenDuringRecovery; CommandId curcid; ! IntegerTimestamp whenTaken; XLogRecPtr lsn; } SerializedSnapshotData; *************** SnapMgrInit(void) *** 274,288 **** if (!found) { SpinLockInit(&oldSnapshotControl->mutex_current); ! oldSnapshotControl->current_timestamp = 0; SpinLockInit(&oldSnapshotControl->mutex_latest_xmin); oldSnapshotControl->latest_xmin = InvalidTransactionId; ! oldSnapshotControl->next_map_update = 0; SpinLockInit(&oldSnapshotControl->mutex_threshold); ! oldSnapshotControl->threshold_timestamp = 0; oldSnapshotControl->threshold_xid = InvalidTransactionId; oldSnapshotControl->head_offset = 0; ! oldSnapshotControl->head_timestamp = 0; oldSnapshotControl->count_used = 0; } } --- 274,288 ---- if (!found) { SpinLockInit(&oldSnapshotControl->mutex_current); ! oldSnapshotControl->current_timestamp.integer_ts = 0; SpinLockInit(&oldSnapshotControl->mutex_latest_xmin); oldSnapshotControl->latest_xmin = InvalidTransactionId; ! oldSnapshotControl->next_map_update.integer_ts = 0; SpinLockInit(&oldSnapshotControl->mutex_threshold); ! oldSnapshotControl->threshold_timestamp.integer_ts = 0; oldSnapshotControl->threshold_xid = InvalidTransactionId; oldSnapshotControl->head_offset = 0; ! oldSnapshotControl->head_timestamp.integer_ts = 0; oldSnapshotControl->count_used = 0; } } *************** ThereAreNoPriorRegisteredSnapshots(void) *** 1611,1642 **** /* ! * Return an int64 timestamp which is exactly on a minute boundary. * * If the argument is already aligned, return that value, otherwise move to * the next minute boundary following the given time. */ ! static int64 ! AlignTimestampToMinuteBoundary(int64 ts) { ! int64 retval = ts + (USECS_PER_MINUTE - 1); ! ! return retval - (retval % USECS_PER_MINUTE); } /* ! * Get current timestamp for snapshots as int64 that never moves backward. */ ! int64 GetSnapshotCurrentTimestamp(void) { ! int64 now = GetCurrentIntegerTimestamp(); /* * Don't let time move backward; if it hasn't advanced, use the old value. */ SpinLockAcquire(&oldSnapshotControl->mutex_current); ! if (now <= oldSnapshotControl->current_timestamp) now = oldSnapshotControl->current_timestamp; else oldSnapshotControl->current_timestamp = now; --- 1611,1642 ---- /* ! * Return an integer timestamp which is exactly on a minute boundary. * * If the argument is already aligned, return that value, otherwise move to * the next minute boundary following the given time. */ ! static IntegerTimestamp ! AlignTimestampToMinuteBoundary(IntegerTimestamp ts) { ! ts.integer_ts += USECS_PER_MINUTE - 1; ! ts.integer_ts -= ts.integer_ts % USECS_PER_MINUTE; ! return ts; } /* ! * Get current timestamp for snapshots as integer that never moves backward. */ ! IntegerTimestamp GetSnapshotCurrentTimestamp(void) { ! IntegerTimestamp now = GetCurrentIntegerTimestamp(); /* * Don't let time move backward; if it hasn't advanced, use the old value. */ SpinLockAcquire(&oldSnapshotControl->mutex_current); ! if (now.integer_ts <= oldSnapshotControl->current_timestamp.integer_ts) now = oldSnapshotControl->current_timestamp; else oldSnapshotControl->current_timestamp = now; *************** GetSnapshotCurrentTimestamp(void) *** 1652,1661 **** * XXX: So far, we never trust that a 64-bit value can be read atomically; if * that ever changes, we could get rid of the spinlock here. */ ! int64 GetOldSnapshotThresholdTimestamp(void) { ! int64 threshold_timestamp; SpinLockAcquire(&oldSnapshotControl->mutex_threshold); threshold_timestamp = oldSnapshotControl->threshold_timestamp; --- 1652,1661 ---- * XXX: So far, we never trust that a 64-bit value can be read atomically; if * that ever changes, we could get rid of the spinlock here. */ ! IntegerTimestamp GetOldSnapshotThresholdTimestamp(void) { ! IntegerTimestamp threshold_timestamp; SpinLockAcquire(&oldSnapshotControl->mutex_threshold); threshold_timestamp = oldSnapshotControl->threshold_timestamp; *************** GetOldSnapshotThresholdTimestamp(void) *** 1665,1671 **** } static void ! SetOldSnapshotThresholdTimestamp(int64 ts, TransactionId xlimit) { SpinLockAcquire(&oldSnapshotControl->mutex_threshold); oldSnapshotControl->threshold_timestamp = ts; --- 1665,1671 ---- } static void ! SetOldSnapshotThresholdTimestamp(IntegerTimestamp ts, TransactionId xlimit) { SpinLockAcquire(&oldSnapshotControl->mutex_threshold); oldSnapshotControl->threshold_timestamp = ts; *************** TransactionIdLimitedForOldSnapshots(Tran *** 1690,1699 **** && old_snapshot_threshold >= 0 && RelationAllowsEarlyPruning(relation)) { ! int64 ts = GetSnapshotCurrentTimestamp(); TransactionId xlimit = recentXmin; TransactionId latest_xmin; ! int64 update_ts; bool same_ts_as_threshold = false; SpinLockAcquire(&oldSnapshotControl->mutex_latest_xmin); --- 1690,1699 ---- && old_snapshot_threshold >= 0 && RelationAllowsEarlyPruning(relation)) { ! IntegerTimestamp ts = GetSnapshotCurrentTimestamp(); TransactionId xlimit = recentXmin; TransactionId latest_xmin; ! IntegerTimestamp update_ts; bool same_ts_as_threshold = false; SpinLockAcquire(&oldSnapshotControl->mutex_latest_xmin); *************** TransactionIdLimitedForOldSnapshots(Tran *** 1715,1732 **** && TransactionIdFollows(latest_xmin, xlimit)) xlimit = latest_xmin; ! ts -= 5 * USECS_PER_SEC; SetOldSnapshotThresholdTimestamp(ts, xlimit); return xlimit; } ! ts = AlignTimestampToMinuteBoundary(ts) ! - (old_snapshot_threshold * USECS_PER_MINUTE); /* Check for fast exit without LW locking. */ SpinLockAcquire(&oldSnapshotControl->mutex_threshold); ! if (ts == oldSnapshotControl->threshold_timestamp) { xlimit = oldSnapshotControl->threshold_xid; same_ts_as_threshold = true; --- 1715,1732 ---- && TransactionIdFollows(latest_xmin, xlimit)) xlimit = latest_xmin; ! ts.integer_ts -= 5 * USECS_PER_SEC; SetOldSnapshotThresholdTimestamp(ts, xlimit); return xlimit; } ! ts = AlignTimestampToMinuteBoundary(ts); ! ts.integer_ts -= old_snapshot_threshold * USECS_PER_MINUTE; /* Check for fast exit without LW locking. */ SpinLockAcquire(&oldSnapshotControl->mutex_threshold); ! if (ts.integer_ts == oldSnapshotControl->threshold_timestamp.integer_ts) { xlimit = oldSnapshotControl->threshold_xid; same_ts_as_threshold = true; *************** TransactionIdLimitedForOldSnapshots(Tran *** 1735,1741 **** if (!same_ts_as_threshold) { ! if (ts == update_ts) { xlimit = latest_xmin; if (NormalTransactionIdFollows(xlimit, recentXmin)) --- 1735,1741 ---- if (!same_ts_as_threshold) { ! if (ts.integer_ts == update_ts.integer_ts) { xlimit = latest_xmin; if (NormalTransactionIdFollows(xlimit, recentXmin)) *************** TransactionIdLimitedForOldSnapshots(Tran *** 1746,1756 **** LWLockAcquire(OldSnapshotTimeMapLock, LW_SHARED); if (oldSnapshotControl->count_used > 0 ! && ts >= oldSnapshotControl->head_timestamp) { int offset; ! offset = ((ts - oldSnapshotControl->head_timestamp) / USECS_PER_MINUTE); if (offset > oldSnapshotControl->count_used - 1) offset = oldSnapshotControl->count_used - 1; --- 1746,1756 ---- LWLockAcquire(OldSnapshotTimeMapLock, LW_SHARED); if (oldSnapshotControl->count_used > 0 ! && ts.integer_ts >= oldSnapshotControl->head_timestamp.integer_ts) { int offset; ! offset = ((ts.integer_ts - oldSnapshotControl->head_timestamp.integer_ts) / USECS_PER_MINUTE); if (offset > oldSnapshotControl->count_used - 1) offset = oldSnapshotControl->count_used - 1; *************** TransactionIdLimitedForOldSnapshots(Tran *** 1790,1800 **** * Take care of the circular buffer that maps time to xid. */ void ! MaintainOldSnapshotTimeMapping(int64 whenTaken, TransactionId xmin) { ! int64 ts; TransactionId latest_xmin; ! int64 update_ts; bool map_update_required = false; /* Never call this function when old snapshot checking is disabled. */ --- 1790,1800 ---- * Take care of the circular buffer that maps time to xid. */ void ! MaintainOldSnapshotTimeMapping(IntegerTimestamp whenTaken, TransactionId xmin) { ! IntegerTimestamp ts; TransactionId latest_xmin; ! IntegerTimestamp update_ts; bool map_update_required = false; /* Never call this function when old snapshot checking is disabled. */ *************** MaintainOldSnapshotTimeMapping(int64 whe *** 1809,1815 **** SpinLockAcquire(&oldSnapshotControl->mutex_latest_xmin); latest_xmin = oldSnapshotControl->latest_xmin; update_ts = oldSnapshotControl->next_map_update; ! if (ts > update_ts) { oldSnapshotControl->next_map_update = ts; map_update_required = true; --- 1809,1815 ---- SpinLockAcquire(&oldSnapshotControl->mutex_latest_xmin); latest_xmin = oldSnapshotControl->latest_xmin; update_ts = oldSnapshotControl->next_map_update; ! if (ts.integer_ts > update_ts.integer_ts) { oldSnapshotControl->next_map_update = ts; map_update_required = true; *************** MaintainOldSnapshotTimeMapping(int64 whe *** 1832,1842 **** * processing for this feature; so if something seems unreasonable, just * log at DEBUG level and return without doing anything. */ ! if (whenTaken < 0) { elog(DEBUG1, "MaintainOldSnapshotTimeMapping called with negative whenTaken = %ld", ! (long) whenTaken); return; } if (!TransactionIdIsNormal(xmin)) --- 1832,1842 ---- * processing for this feature; so if something seems unreasonable, just * log at DEBUG level and return without doing anything. */ ! if (whenTaken.integer_ts < 0) { elog(DEBUG1, "MaintainOldSnapshotTimeMapping called with negative whenTaken = %ld", ! (long) whenTaken.integer_ts); return; } if (!TransactionIdIsNormal(xmin)) *************** MaintainOldSnapshotTimeMapping(int64 whe *** 1851,1857 **** Assert(oldSnapshotControl->head_offset >= 0); Assert(oldSnapshotControl->head_offset < OLD_SNAPSHOT_TIME_MAP_ENTRIES); ! Assert((oldSnapshotControl->head_timestamp % USECS_PER_MINUTE) == 0); Assert(oldSnapshotControl->count_used >= 0); Assert(oldSnapshotControl->count_used <= OLD_SNAPSHOT_TIME_MAP_ENTRIES); --- 1851,1857 ---- Assert(oldSnapshotControl->head_offset >= 0); Assert(oldSnapshotControl->head_offset < OLD_SNAPSHOT_TIME_MAP_ENTRIES); ! Assert((oldSnapshotControl->head_timestamp.integer_ts % USECS_PER_MINUTE) == 0); Assert(oldSnapshotControl->count_used >= 0); Assert(oldSnapshotControl->count_used <= OLD_SNAPSHOT_TIME_MAP_ENTRIES); *************** MaintainOldSnapshotTimeMapping(int64 whe *** 1863,1884 **** oldSnapshotControl->count_used = 1; oldSnapshotControl->xid_by_minute[0] = xmin; } ! else if (ts < oldSnapshotControl->head_timestamp) { /* old ts; log it at DEBUG */ LWLockRelease(OldSnapshotTimeMapLock); elog(DEBUG1, "MaintainOldSnapshotTimeMapping called with old whenTaken = %ld", ! (long) whenTaken); return; } ! else if (ts <= (oldSnapshotControl->head_timestamp + ((oldSnapshotControl->count_used - 1) * USECS_PER_MINUTE))) { /* existing mapping; advance xid if possible */ int bucket = (oldSnapshotControl->head_offset ! + ((ts - oldSnapshotControl->head_timestamp) / USECS_PER_MINUTE)) % OLD_SNAPSHOT_TIME_MAP_ENTRIES; --- 1863,1884 ---- oldSnapshotControl->count_used = 1; oldSnapshotControl->xid_by_minute[0] = xmin; } ! else if (ts.integer_ts < oldSnapshotControl->head_timestamp.integer_ts) { /* old ts; log it at DEBUG */ LWLockRelease(OldSnapshotTimeMapLock); elog(DEBUG1, "MaintainOldSnapshotTimeMapping called with old whenTaken = %ld", ! (long) whenTaken.integer_ts); return; } ! else if (ts.integer_ts <= (oldSnapshotControl->head_timestamp.integer_ts + ((oldSnapshotControl->count_used - 1) * USECS_PER_MINUTE))) { /* existing mapping; advance xid if possible */ int bucket = (oldSnapshotControl->head_offset ! + ((ts.integer_ts - oldSnapshotControl->head_timestamp.integer_ts) / USECS_PER_MINUTE)) % OLD_SNAPSHOT_TIME_MAP_ENTRIES; *************** MaintainOldSnapshotTimeMapping(int64 whe *** 1888,1894 **** else { /* We need a new bucket, but it might not be the very next one. */ ! int advance = ((ts - oldSnapshotControl->head_timestamp) / USECS_PER_MINUTE); oldSnapshotControl->head_timestamp = ts; --- 1888,1894 ---- else { /* We need a new bucket, but it might not be the very next one. */ ! int advance = ((ts.integer_ts - oldSnapshotControl->head_timestamp.integer_ts) / USECS_PER_MINUTE); oldSnapshotControl->head_timestamp = ts; diff --git a/src/bin/pg_basebackup/pg_recvlogical.c b/src/bin/pg_basebackup/pg_recvlogical.c index d16d08b..53a4bbd 100644 *** a/src/bin/pg_basebackup/pg_recvlogical.c --- b/src/bin/pg_basebackup/pg_recvlogical.c *************** static int outfd = -1; *** 57,63 **** static volatile sig_atomic_t time_to_abort = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; ! static int64 output_last_fsync = -1; static bool output_needs_fsync = false; static XLogRecPtr output_written_lsn = InvalidXLogRecPtr; static XLogRecPtr output_fsync_lsn = InvalidXLogRecPtr; --- 57,63 ---- static volatile sig_atomic_t time_to_abort = false; static volatile sig_atomic_t output_reopen = false; static bool output_isfile; ! static IntegerTimestamp output_last_fsync = {-1}; static bool output_needs_fsync = false; static XLogRecPtr output_written_lsn = InvalidXLogRecPtr; static XLogRecPtr output_fsync_lsn = InvalidXLogRecPtr; *************** static XLogRecPtr output_fsync_lsn = Inv *** 65,71 **** static void usage(void); static void StreamLogicalLog(void); static void disconnect_and_exit(int code); ! static bool flushAndSendFeedback(PGconn *conn, TimestampTz *now); static void prepareToTerminate(PGconn *conn, XLogRecPtr endpos, bool keepalive, XLogRecPtr lsn); --- 65,71 ---- static void usage(void); static void StreamLogicalLog(void); static void disconnect_and_exit(int code); ! static bool flushAndSendFeedback(PGconn *conn, IntegerTimestamp *now); static void prepareToTerminate(PGconn *conn, XLogRecPtr endpos, bool keepalive, XLogRecPtr lsn); *************** usage(void) *** 112,118 **** * Send a Standby Status Update message to server. */ static bool ! sendFeedback(PGconn *conn, int64 now, bool force, bool replyRequested) { static XLogRecPtr last_written_lsn = InvalidXLogRecPtr; static XLogRecPtr last_fsync_lsn = InvalidXLogRecPtr; --- 112,118 ---- * Send a Standby Status Update message to server. */ static bool ! sendFeedback(PGconn *conn, IntegerTimestamp now, bool force, bool replyRequested) { static XLogRecPtr last_written_lsn = InvalidXLogRecPtr; static XLogRecPtr last_fsync_lsn = InvalidXLogRecPtr; *************** sendFeedback(PGconn *conn, int64 now, bo *** 146,152 **** len += 8; fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */ len += 8; ! fe_sendint64(now, &replybuf[len]); /* sendTime */ len += 8; replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */ len += 1; --- 146,152 ---- len += 8; fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */ len += 8; ! fe_sendint64(now.integer_ts, &replybuf[len]); /* sendTime */ len += 8; replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */ len += 1; *************** disconnect_and_exit(int code) *** 175,181 **** } static bool ! OutputFsync(int64 now) { output_last_fsync = now; --- 175,181 ---- } static bool ! OutputFsync(IntegerTimestamp now) { output_last_fsync = now; *************** StreamLogicalLog(void) *** 212,218 **** { PGresult *res; char *copybuf = NULL; ! int64 last_status = -1; int i; PQExpBuffer query; --- 212,218 ---- { PGresult *res; char *copybuf = NULL; ! IntegerTimestamp last_status = {-1}; int i; PQExpBuffer query; *************** StreamLogicalLog(void) *** 285,291 **** int r; int bytes_left; int bytes_written; ! int64 now; int hdr_len; XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; --- 285,291 ---- int r; int bytes_left; int bytes_written; ! IntegerTimestamp now; int hdr_len; XLogRecPtr cur_record_lsn = InvalidXLogRecPtr; *************** StreamLogicalLog(void) *** 365,372 **** * response back to the client. */ fd_set input_mask; ! int64 message_target = 0; ! int64 fsync_target = 0; struct timeval timeout; struct timeval *timeoutptr = NULL; --- 365,372 ---- * response back to the client. */ fd_set input_mask; ! IntegerTimestamp message_target = {0}; ! IntegerTimestamp fsync_target = {0}; struct timeval timeout; struct timeval *timeoutptr = NULL; *************** StreamLogicalLog(void) *** 383,406 **** /* Compute when we need to wakeup to send a keepalive message. */ if (standby_message_timeout) ! message_target = last_status + (standby_message_timeout - 1) * ((int64) 1000); /* Compute when we need to wakeup to fsync the output file. */ if (fsync_interval > 0 && output_needs_fsync) ! fsync_target = output_last_fsync + (fsync_interval - 1) * ((int64) 1000); /* Now compute when to wakeup. */ ! if (message_target > 0 || fsync_target > 0) { ! int64 targettime; long secs; int usecs; targettime = message_target; ! if (fsync_target > 0 && fsync_target < targettime) targettime = fsync_target; feTimestampDifference(now, --- 383,406 ---- /* Compute when we need to wakeup to send a keepalive message. */ if (standby_message_timeout) ! message_target.integer_ts = last_status.integer_ts + (standby_message_timeout - 1) * ((int64) 1000); /* Compute when we need to wakeup to fsync the output file. */ if (fsync_interval > 0 && output_needs_fsync) ! fsync_target.integer_ts = output_last_fsync.integer_ts + (fsync_interval - 1) * ((int64) 1000); /* Now compute when to wakeup. */ ! if (message_target.integer_ts > 0 || fsync_target.integer_ts > 0) { ! IntegerTimestamp targettime; long secs; int usecs; targettime = message_target; ! if (fsync_target.integer_ts > 0 && fsync_target.integer_ts < targettime.integer_ts) targettime = fsync_target; feTimestampDifference(now, *************** StreamLogicalLog(void) *** 622,628 **** if (outfd != -1 && strcmp(outfile, "-") != 0) { ! int64 t = feGetCurrentTimestamp(); /* no need to jump to error on failure here, we're finishing anyway */ OutputFsync(t); --- 622,628 ---- if (outfd != -1 && strcmp(outfile, "-") != 0) { ! IntegerTimestamp t = feGetCurrentTimestamp(); /* no need to jump to error on failure here, we're finishing anyway */ OutputFsync(t); *************** main(int argc, char **argv) *** 1024,1030 **** * feedback. */ static bool ! flushAndSendFeedback(PGconn *conn, TimestampTz *now) { /* flush data to disk, so that we send a recent flush pointer */ if (!OutputFsync(*now)) --- 1024,1030 ---- * feedback. */ static bool ! flushAndSendFeedback(PGconn *conn, IntegerTimestamp *now) { /* flush data to disk, so that we send a recent flush pointer */ if (!OutputFsync(*now)) diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c index 5561283..639f739 100644 *** a/src/bin/pg_basebackup/receivelog.c --- b/src/bin/pg_basebackup/receivelog.c *************** static PGresult *HandleCopyStream(PGconn *** 42,56 **** static int CopyStreamPoll(PGconn *conn, long timeout_ms); static int CopyStreamReceive(PGconn *conn, long timeout, char **buffer); static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, ! int len, XLogRecPtr blockpos, int64 *last_status); static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos); static PGresult *HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos); static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos, XLogRecPtr *stoppos); ! static long CalculateCopyStreamSleeptime(int64 now, int standby_message_timeout, ! int64 last_status); static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline); --- 42,56 ---- static int CopyStreamPoll(PGconn *conn, long timeout_ms); static int CopyStreamReceive(PGconn *conn, long timeout, char **buffer); static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, ! int len, XLogRecPtr blockpos, IntegerTimestamp *last_status); static bool ProcessXLogDataMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, XLogRecPtr *blockpos); static PGresult *HandleEndOfCopyStream(PGconn *conn, StreamCtl *stream, char *copybuf, XLogRecPtr blockpos, XLogRecPtr *stoppos); static bool CheckCopyStreamStop(PGconn *conn, StreamCtl *stream, XLogRecPtr blockpos, XLogRecPtr *stoppos); ! static long CalculateCopyStreamSleeptime(IntegerTimestamp now, int standby_message_timeout, ! IntegerTimestamp last_status); static bool ReadEndOfStreamingResult(PGresult *res, XLogRecPtr *startpos, uint32 *timeline); *************** writeTimeLineHistoryFile(StreamCtl *stre *** 319,325 **** * Send a Standby Status Update message to server. */ static bool ! sendFeedback(PGconn *conn, XLogRecPtr blockpos, int64 now, bool replyRequested) { char replybuf[1 + 8 + 8 + 8 + 8 + 1]; int len = 0; --- 319,325 ---- * Send a Standby Status Update message to server. */ static bool ! sendFeedback(PGconn *conn, XLogRecPtr blockpos, IntegerTimestamp now, bool replyRequested) { char replybuf[1 + 8 + 8 + 8 + 8 + 1]; int len = 0; *************** sendFeedback(PGconn *conn, XLogRecPtr bl *** 335,341 **** len += 8; fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */ len += 8; ! fe_sendint64(now, &replybuf[len]); /* sendTime */ len += 8; replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */ len += 1; --- 335,341 ---- len += 8; fe_sendint64(InvalidXLogRecPtr, &replybuf[len]); /* apply */ len += 8; ! fe_sendint64(now.integer_ts, &replybuf[len]); /* sendTime */ len += 8; replybuf[len] = replyRequested ? 1 : 0; /* replyRequested */ len += 1; *************** HandleCopyStream(PGconn *conn, StreamCtl *** 761,767 **** XLogRecPtr *stoppos) { char *copybuf = NULL; ! int64 last_status = -1; XLogRecPtr blockpos = stream->startpos; still_sending = true; --- 761,767 ---- XLogRecPtr *stoppos) { char *copybuf = NULL; ! IntegerTimestamp last_status = {-1}; XLogRecPtr blockpos = stream->startpos; still_sending = true; *************** HandleCopyStream(PGconn *conn, StreamCtl *** 769,775 **** while (1) { int r; ! int64 now; long sleeptime; /* --- 769,775 ---- while (1) { int r; ! IntegerTimestamp now; long sleeptime; /* *************** CopyStreamReceive(PGconn *conn, long tim *** 994,1004 **** */ static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, ! XLogRecPtr blockpos, int64 *last_status) { int pos; bool replyRequested; ! int64 now; /* * Parse the keepalive message, enclosed in the CopyData message. We just --- 994,1004 ---- */ static bool ProcessKeepaliveMsg(PGconn *conn, StreamCtl *stream, char *copybuf, int len, ! XLogRecPtr blockpos, IntegerTimestamp *last_status) { int pos; bool replyRequested; ! IntegerTimestamp now; /* * Parse the keepalive message, enclosed in the CopyData message. We just *************** CheckCopyStreamStop(PGconn *conn, Stream *** 1253,1269 **** * Calculate how long send/receive loops should sleep */ static long ! CalculateCopyStreamSleeptime(int64 now, int standby_message_timeout, ! int64 last_status) { ! int64 status_targettime = 0; long sleeptime; if (standby_message_timeout && still_sending) ! status_targettime = last_status + (standby_message_timeout - 1) * ((int64) 1000); ! if (status_targettime > 0) { long secs; int usecs; --- 1253,1269 ---- * Calculate how long send/receive loops should sleep */ static long ! CalculateCopyStreamSleeptime(IntegerTimestamp now, int standby_message_timeout, ! IntegerTimestamp last_status) { ! IntegerTimestamp status_targettime = {0}; long sleeptime; if (standby_message_timeout && still_sending) ! status_targettime.integer_ts = last_status.integer_ts + (standby_message_timeout - 1) * ((int64) 1000); ! if (status_targettime.integer_ts > 0) { long secs; int usecs; diff --git a/src/bin/pg_basebackup/streamutil.c b/src/bin/pg_basebackup/streamutil.c index 31290d3..81c90d6 100644 *** a/src/bin/pg_basebackup/streamutil.c --- b/src/bin/pg_basebackup/streamutil.c *************** DropReplicationSlot(PGconn *conn, const *** 436,448 **** /* ! * Frontend version of GetCurrentTimestamp(), since we are not linked with ! * backend code. The replication protocol always uses integer timestamps, * regardless of the server setting. */ ! int64 feGetCurrentTimestamp(void) { int64 result; struct timeval tp; --- 436,449 ---- /* ! * Frontend version of GetCurrentIntegerTimestamp(), since we are not linked ! * with backend code. The replication protocol always uses integer timestamps, * regardless of the server setting. */ ! IntegerTimestamp feGetCurrentTimestamp(void) { + IntegerTimestamp ts; int64 result; struct timeval tp; *************** feGetCurrentTimestamp(void) *** 453,459 **** result = (result * USECS_PER_SEC) + tp.tv_usec; ! return result; } /* --- 454,461 ---- result = (result * USECS_PER_SEC) + tp.tv_usec; ! ts.integer_ts = result; ! return ts; } /* *************** feGetCurrentTimestamp(void) *** 461,470 **** * backend code. */ void ! feTimestampDifference(int64 start_time, int64 stop_time, long *secs, int *microsecs) { ! int64 diff = stop_time - start_time; if (diff <= 0) { --- 463,472 ---- * backend code. */ void ! feTimestampDifference(IntegerTimestamp start_time, IntegerTimestamp stop_time, long *secs, int *microsecs) { ! int64 diff = stop_time.integer_ts - start_time.integer_ts; if (diff <= 0) { *************** feTimestampDifference(int64 start_time, *** 483,493 **** * linked with backend code. */ bool ! feTimestampDifferenceExceeds(int64 start_time, ! int64 stop_time, int msec) { ! int64 diff = stop_time - start_time; return (diff >= msec * INT64CONST(1000)); } --- 485,495 ---- * linked with backend code. */ bool ! feTimestampDifferenceExceeds(IntegerTimestamp start_time, ! IntegerTimestamp stop_time, int msec) { ! int64 diff = stop_time.integer_ts - start_time.integer_ts; return (diff >= msec * INT64CONST(1000)); } diff --git a/src/bin/pg_basebackup/streamutil.h b/src/bin/pg_basebackup/streamutil.h index 663bfac..7c1a125 100644 *** a/src/bin/pg_basebackup/streamutil.h --- b/src/bin/pg_basebackup/streamutil.h *************** extern bool RunIdentifySystem(PGconn *co *** 38,48 **** TimeLineID *starttli, XLogRecPtr *startpos, char **db_name); ! extern int64 feGetCurrentTimestamp(void); ! extern void feTimestampDifference(int64 start_time, int64 stop_time, long *secs, int *microsecs); ! extern bool feTimestampDifferenceExceeds(int64 start_time, int64 stop_time, int msec); extern void fe_sendint64(int64 i, char *buf); extern int64 fe_recvint64(char *buf); --- 38,50 ---- TimeLineID *starttli, XLogRecPtr *startpos, char **db_name); ! extern IntegerTimestamp feGetCurrentTimestamp(void); ! extern void feTimestampDifference(IntegerTimestamp start_time, ! IntegerTimestamp stop_time, long *secs, int *microsecs); ! extern bool feTimestampDifferenceExceeds(IntegerTimestamp start_time, ! IntegerTimestamp stop_time, int msec); extern void fe_sendint64(int64 i, char *buf); extern int64 fe_recvint64(char *buf); diff --git a/src/include/c.h b/src/include/c.h index 91e5baa..a7c0774 100644 *** a/src/include/c.h --- b/src/include/c.h *************** typedef unsigned PG_INT128_TYPE uint128; *** 346,351 **** --- 346,362 ---- #endif /* + * In some places we prefer to work with timestamps that are integers + * regardless of the value of HAVE_INT64_TIMESTAMP. To avoid coding errors + * from confusing such timestamps with Timestamp/TimestampTz, we use this + * datatype: + */ + typedef struct IntegerTimestamp + { + int64 integer_ts; + } IntegerTimestamp; + + /* * Size * Size of any memory resident object, as returned by sizeof. */ diff --git a/src/include/utils/snapmgr.h b/src/include/utils/snapmgr.h index 2618cc4..fb345a0 100644 *** a/src/include/utils/snapmgr.h --- b/src/include/utils/snapmgr.h *************** extern PGDLLIMPORT int old_snapshot_thre *** 51,58 **** extern Size SnapMgrShmemSize(void); extern void SnapMgrInit(void); ! extern int64 GetSnapshotCurrentTimestamp(void); ! extern int64 GetOldSnapshotThresholdTimestamp(void); extern bool FirstSnapshotSet; --- 51,58 ---- extern Size SnapMgrShmemSize(void); extern void SnapMgrInit(void); ! extern IntegerTimestamp GetSnapshotCurrentTimestamp(void); ! extern IntegerTimestamp GetOldSnapshotThresholdTimestamp(void); extern bool FirstSnapshotSet; *************** extern void DeleteAllExportedSnapshotFil *** 93,99 **** extern bool ThereAreNoPriorRegisteredSnapshots(void); extern TransactionId TransactionIdLimitedForOldSnapshots(TransactionId recentXmin, Relation relation); ! extern void MaintainOldSnapshotTimeMapping(int64 whenTaken, TransactionId xmin); extern char *ExportSnapshot(Snapshot snapshot); --- 93,99 ---- extern bool ThereAreNoPriorRegisteredSnapshots(void); extern TransactionId TransactionIdLimitedForOldSnapshots(TransactionId recentXmin, Relation relation); ! extern void MaintainOldSnapshotTimeMapping(IntegerTimestamp whenTaken, TransactionId xmin); extern char *ExportSnapshot(Snapshot snapshot); diff --git a/src/include/utils/snapshot.h b/src/include/utils/snapshot.h index ebd5d5a..66d554b 100644 *** a/src/include/utils/snapshot.h --- b/src/include/utils/snapshot.h *************** typedef struct SnapshotData *** 107,113 **** uint32 regd_count; /* refcount on RegisteredSnapshots */ pairingheap_node ph_node; /* link in the RegisteredSnapshots heap */ ! int64 whenTaken; /* timestamp when snapshot was taken */ XLogRecPtr lsn; /* position in the WAL stream when taken */ } SnapshotData; --- 107,113 ---- uint32 regd_count; /* refcount on RegisteredSnapshots */ pairingheap_node ph_node; /* link in the RegisteredSnapshots heap */ ! IntegerTimestamp whenTaken; /* timestamp when snapshot was taken */ XLogRecPtr lsn; /* position in the WAL stream when taken */ } SnapshotData; diff --git a/src/include/utils/timestamp.h b/src/include/utils/timestamp.h index 21651b1..ae9eb1c 100644 *** a/src/include/utils/timestamp.h --- b/src/include/utils/timestamp.h *************** extern bool TimestampDifferenceExceeds(T *** 105,116 **** * Prototypes for functions to deal with integer timestamps, when the native * format is float timestamps. */ #ifndef HAVE_INT64_TIMESTAMP ! extern int64 GetCurrentIntegerTimestamp(void); ! extern TimestampTz IntegerTimestampToTimestampTz(int64 timestamp); #else ! #define GetCurrentIntegerTimestamp() GetCurrentTimestamp() ! #define IntegerTimestampToTimestampTz(timestamp) (timestamp) #endif extern TimestampTz time_t_to_timestamptz(pg_time_t tm); --- 105,118 ---- * Prototypes for functions to deal with integer timestamps, when the native * format is float timestamps. */ + extern IntegerTimestamp GetCurrentIntegerTimestamp(void); #ifndef HAVE_INT64_TIMESTAMP ! extern TimestampTz Int64ToTimestampTz(int64 timestamp); ! extern TimestampTz IntegerTimestampToTimestampTz(IntegerTimestamp timestamp); #else ! //#define GetCurrentIntegerTimestamp() GetCurrentTimestamp() ! #define Int64ToTimestampTz(timestamp) (timestamp) ! #define IntegerTimestampToTimestampTz(timestamp) ((timestamp).integer_ts) #endif extern TimestampTz time_t_to_timestamptz(pg_time_t tm);