Update minimum recovery point on truncation during WAL replay of abort record.

If a file is truncated, we must update minRecoveryPoint. Once a file is
truncated, there's no going back; it would not be safe to stop recovery
at a point earlier than that anymore.

Commit 7bffc9b7bf changed xact_redo_commit() so that it updates
minRecoveryPoint on truncation, but forgot to change xact_redo_abort().

Back-patch to all supported versions.

Reported-by: mengjuan.cmj@alibaba-inc.com
Author: Fujii Masao
Reviewed-by: Heikki Linnakangas
Discussion: https://postgr.es/m/b029fce3-4fac-4265-968e-16f36ff4d075.mengjuan.cmj@alibaba-inc.com

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index 7181718d6e26e5a9e7e93a36646a24a0a77b54a2..2f0c4b1e137dcbcfb3a86433a181a9d5b85c606d 100644 (file)
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -5733,7 +5733,8 @@ xact_redo_commit(xl_xact_parsed_commit *parsed,
  * because subtransaction commit is never WAL logged.
  */
 static void
-xact_redo_abort(xl_xact_parsed_abort *parsed, TransactionId xid)
+xact_redo_abort(xl_xact_parsed_abort *parsed, TransactionId xid,
+                               XLogRecPtr lsn)
 {
        TransactionId max_xid;
 
@@ -5798,7 +5799,16 @@ xact_redo_abort(xl_xact_parsed_abort *parsed, TransactionId xid)
        }
 
        /* Make sure files supposed to be dropped are dropped */
-       DropRelationFiles(parsed->xnodes, parsed->nrels, true);
+       if (parsed->nrels > 0)
+       {
+               /*
+                * See comments about update of minimum recovery point on truncation,
+                * in xact_redo_commit().
+                */
+               XLogFlush(lsn);
+
+               DropRelationFiles(parsed->xnodes, parsed->nrels, true);
+       }
 }
 
 void
@@ -5838,7 +5848,7 @@ xact_redo(XLogReaderState *record)
                xl_xact_parsed_abort parsed;
 
                ParseAbortRecord(XLogRecGetInfo(record), xlrec, &parsed);
-               xact_redo_abort(&parsed, XLogRecGetXid(record));
+               xact_redo_abort(&parsed, XLogRecGetXid(record), record->EndRecPtr);
        }
        else if (info == XLOG_XACT_ABORT_PREPARED)
        {
@@ -5846,7 +5856,7 @@ xact_redo(XLogReaderState *record)
                xl_xact_parsed_abort parsed;
 
                ParseAbortRecord(XLogRecGetInfo(record), xlrec, &parsed);
-               xact_redo_abort(&parsed, parsed.twophase_xid);
+               xact_redo_abort(&parsed, parsed.twophase_xid, record->EndRecPtr);
 
                /* Delete TwoPhaseState gxact entry and/or 2PC file. */
                LWLockAcquire(TwoPhaseStateLock, LW_EXCLUSIVE);