DRAFT: ADBDEV-7181 - Orphaned files removal

.abi-check/6.27.1_arenadata63/postgres.symbols.ignore

@@ -0,0 +1,2 @@
+ConfigureNamesBool_gp
+log_smgrcreate

src/backend/access/rmgrdesc/smgrdesc.c

@@ -24,7 +24,7 @@ smgr_desc(StringInfo buf, XLogRecord *record)
 	uint8           info = record->xl_info & ~XLR_INFO_MASK;
 	char            *rec = XLogRecGetData(record);
 
-	if (info == XLOG_SMGR_CREATE)
+	if ((info == XLOG_SMGR_CREATE) || (info == XLOG_SMGR_CREATE_PDL))
 	{
 		xl_smgr_create *xlrec = (xl_smgr_create *) rec;
 		char	   *path = relpathperm(xlrec->rnode, xlrec->forkNum);

src/backend/access/rmgrdesc/xlogdesc.c

@@ -17,6 +17,7 @@
 #include "access/xlog.h"
 #include "access/xlog_internal.h"
 #include "catalog/pg_control.h"
+#include "catalog/storage_pending_deletes.h"
 #include "utils/guc.h"
 #include "utils/timestamp.h"
 
@@ -233,6 +234,11 @@ xlog_desc(StringInfo buf, XLogRecord *record)
 						 (uint32) xlrec.overwritten_lsn,
 						 timestamptz_to_str(xlrec.overwrite_time));
 	}
+	else if (info == XLOG_PENDING_DELETE)
+	{
+		appendStringInfo(buf, "orphaned relfilenodes to delete: %zu",
+						 ((PendingRelXactDeleteArray *)rec)->count);
+	}
 	else
 		appendStringInfoString(buf, "UNKNOWN");
 }

src/backend/access/transam/twophase.c

@@ -56,6 +56,7 @@
 #include "access/xlogutils.h"
 #include "catalog/pg_type.h"
 #include "catalog/storage.h"
+#include "catalog/storage_pending_deletes_redo.h"
 #include "catalog/storage_tablespace.h"
 #include "catalog/storage_database.h"
 #include "funcapi.h"
@@ -2459,3 +2460,85 @@ getTwoPhaseOldestPreparedTransactionXLogRecPtr(prepared_transaction_agg_state *p
 	return oldest;
 
 }  /* end getTwoPhaseOldestPreparedTransactionXLogRecPtr */
+
+bool
+RemovePendingDeletesForPreparedTransactions()
+{
+	HASH_SEQ_STATUS scan_status;
+	prpt_map   *entry;
+	XLogReaderState *xlogreader;
+	volatile bool result = true;
+	XLogRecord *xlogrec = NULL;
+	MemoryContext oldcontext = CurrentMemoryContext;
+
+	if (NULL == crashRecoverPostCheckpointPreparedTransactions_map_ht)
+		return result;
+
+	xlogreader = XLogReaderAllocate(&read_local_xlog_page, NULL);
+	if (!xlogreader)
+		ereport(ERROR,
+				(errcode(ERRCODE_OUT_OF_MEMORY),
+				 errmsg("out of memory"),
+		   errdetail("Failed while allocating an XLog reading processor.")));
+
+	hash_seq_init(&scan_status,
+				  crashRecoverPostCheckpointPreparedTransactions_map_ht);
+	while ((entry = (prpt_map *) hash_seq_search(&scan_status)) != NULL)
+	{
+		char	   *errormsg = NULL;
+		TwoPhaseFileHeader *hdr;
+
+		if (entry->xlogrecptr == InvalidXLogRecPtr)
+			continue;
+
+		int savedInterruptHoldoffCount = InterruptHoldoffCount;
+		PG_TRY();
+		{
+			xlogrec = XLogReadRecord(xlogreader, entry->xlogrecptr, &errormsg);
+		}
+		PG_CATCH();
+		{
+			MemoryContextSwitchTo(oldcontext);
+			InterruptHoldoffCount = savedInterruptHoldoffCount;
+			FlushErrorState();
+			result = false;
+		}
+		PG_END_TRY();
+
+		if (!result)
+		{
+			elog(LOG, "Failed to read WAL record %X/%X for XID %u in %s",
+				 (uint32) (entry->xlogrecptr >> 32),
+				 (uint32) entry->xlogrecptr,
+				 entry->xid,
+				 __func__);
+			break;
+		}
+
+		if (NULL == xlogrec)
+		{
+			if (errormsg)
+				ereport(ERROR,
+						(errcode(ERRCODE_DATA_CORRUPTED),
+						 errmsg("xlog record is invalid"),
+						 errdetail("%s", errormsg)));
+			else
+				ereport(ERROR,
+						(errcode(ERRCODE_DATA_CORRUPTED),
+						 errmsg("xlog record is invalid")));
+		}
+
+		hdr = (TwoPhaseFileHeader *) XLogRecGetData(xlogrec);
+
+		TransactionId *subxids = (hdr->nsubxacts > 0) ?
+			(TransactionId *)
+				((char *) hdr + MAXALIGN(sizeof(TwoPhaseFileHeader))) :
+			NULL;
+
+		PdlRedoRemoveTree(hdr->xid, subxids, hdr->nsubxacts);
+	}
+
+	XLogReaderFree(xlogreader);
+
+	return result;
+}  /* end RemovePendingDeletesForPreparedTransactions */

src/backend/access/transam/xact.c

@@ -32,6 +32,7 @@
 #include "catalog/namespace.h"
 #include "catalog/oid_dispatch.h"
 #include "catalog/storage.h"
+#include "catalog/storage_pending_deletes_redo.h"
 #include "catalog/storage_tablespace.h"
 #include "catalog/storage_database.h"
 #include "commands/async.h"
@@ -6223,6 +6224,8 @@ xact_redo_commit_internal(TransactionId xid, XLogRecPtr lsn,
 
 	DoTablespaceDeletionForRedoXlog(tablespace_oid_to_delete);
 
+	PdlRedoRemoveTree(xid, sub_xids, nsubxacts);
+
 	/*
 	 * We issue an XLogFlush() for the same reason we emit ForceSyncCommit()
 	 * in normal operation. For example, in CREATE DATABASE, we copy all files
@@ -6380,6 +6383,8 @@ xact_redo_distributed_commit(xl_xact_commit *xlrec, TransactionId xid)
 		DropRelationFiles(xlrec->xnodes, xlrec->nrels, true);
 		DropDatabaseDirectories(deldbs, xlrec->ndeldbs, true);
 		DoTablespaceDeletionForRedoXlog(xlrec->tablespace_oid_to_delete_on_commit);
+
+		PdlRedoRemoveTree(xid, sub_xids, xlrec->nsubxacts);
 	}
 
 	/*
@@ -6455,6 +6460,8 @@ xact_redo_abort(xl_xact_abort *xlrec, TransactionId xid)
 	DropRelationFiles(xlrec->xnodes, xlrec->nrels, true);
 	DropDatabaseDirectories(deldbs, xlrec->ndeldbs, true);
 	DoTablespaceDeletionForRedoXlog(xlrec->tablespace_oid_to_delete_on_abort);
+
+	PdlRedoRemoveTree(xid, sub_xids, xlrec->nsubxacts);
 }
 
 static void

src/backend/access/transam/xlog.c

@@ -36,6 +36,7 @@
 #include "catalog/catversion.h"
 #include "catalog/pg_control.h"
 #include "catalog/pg_database.h"
+#include "catalog/storage_pending_deletes_redo.h"
 #include "miscadmin.h"
 #include "pgstat.h"
 #include "postmaster/bgwriter.h"
@@ -7451,6 +7452,40 @@ StartupXLOG(void)
 					TimeLineID	newTLI = ThisTimeLineID;
 					TimeLineID	prevTLI = ThisTimeLineID;
 
+					if ((info == XLOG_CHECKPOINT_SHUTDOWN) ||
+						(info == XLOG_END_OF_RECOVERY))
+					{
+						/*
+						 * At this point we may encounter a situation, when some
+						 * prepared transaction is yet not committed/aborted,
+						 * but the respective WAL segment file is already
+						 * recycled. It may happen is some corner cases, like:
+						 * 1. Primary successfully performs Prepare for a
+						 * transaction;
+						 * 2. Primary stops responding and Mirror is promoted;
+						 * 3. New Primary (ex Mirror) commits the transaction;
+						 * 4. New Primary (ex Mirror) recycles WAL segment with
+						 * the Prepare record (because both Primary and Mirror
+						 * has done the Prepare);
+						 * 5. Ex Primary is recovered as new Mirror, it has the
+						 * the transaction in the list of prepared transactions,
+						 * but doesn't have the WAL segment. And the new Mirror
+						 * should soon see the commit REDO record from the new
+						 * Primary (and remove the transaction from the list of
+						 * prepared transactions).
+						 *
+						 * In such a case
+						 * RemovePendingDeletesForPreparedTransactions() will
+						 * return FALSE. And we postpone the removal of orphaned
+						 * files until all such prepared transactions without
+						 * WAL segment files are wiped out from the list of
+						 * prepared transactions.
+						 */
+						if (RemovePendingDeletesForPreparedTransactions())
+							/* Clean up orphaned files */
+							PdlRedoDropFiles();
+					}
+
 					if (info == XLOG_CHECKPOINT_SHUTDOWN)
 					{
 						CheckPoint	checkPoint;
@@ -7969,6 +8004,21 @@ StartupXLOG(void)
 
 		UtilityModeCloseDtmRedoFile();
 
+		/*
+		 * By this moment, there shouldn't be any prepared transaction with
+		 * missing respective WAL segment file, meaning
+		 * RemovePendingDeletesForPreparedTransactions() should return TRUE.
+		 * If not, most likely the respective WAL segment file is recycled
+		 * illegally, and we do not perform orphaned files removal (as we might
+		 * remove smth that is already committed). Instead, we emit a warning.
+		 */
+		if (RemovePendingDeletesForPreparedTransactions())
+			/* Clean up orphaned files */
+			PdlRedoDropFiles();
+		else
+			ereport(WARNING, (errmsg(
+					"Couldn't drop orphaned files")));
+
 		/*
 		 * And finally, execute the recovery_end_command, if any.
 		 */
@@ -9316,6 +9366,9 @@ CreateCheckPoint(int flags)
 	 */
 	getDtxCheckPointInfo(&dtxCheckPointInfo, &dtxCheckPointInfoSize);
 
+	if (!shutdown)
+		PdlXLogInsert();
+
 	CheckPointGuts(checkPoint.redo, flags);
 
 	/*
@@ -10782,6 +10835,10 @@ xlog_redo(XLogRecPtr beginLoc __attribute__((unused)), XLogRecPtr lsn __attribut
 		/* Keep track of full_page_writes */
 		lastFullPageWrites = fpw;
 	}
+	else if (info == XLOG_PENDING_DELETE)
+	{
+		PdlRedoXLogRecord(record);
+	}
 }
 
 /*

src/backend/catalog/Makefile

@@ -23,7 +23,8 @@ OBJS += pg_exttable.o pg_extprotocol.o \
        pg_attribute_encoding.o pg_compression.o aovisimap.o \
        pg_appendonly.o \
        oid_dispatch.o aocatalog.o storage_tablespace.o storage_database.o \
-       storage_tablespace_twophase.o storage_tablespace_xact.o
+       storage_tablespace_twophase.o storage_tablespace_xact.o \
+       storage_pending_deletes_redo.o storage_pending_deletes.o
 
 
 BKIFILES = postgres.bki postgres.description postgres.shdescription

src/backend/catalog/heap.c

@@ -1895,7 +1895,9 @@ heap_create_init_fork(Relation rel)
 {
 	RelationOpenSmgr(rel);
 	smgrcreate(rel->rd_smgr, INIT_FORKNUM, false);
-	log_smgrcreate(&rel->rd_smgr->smgr_rnode.node, INIT_FORKNUM);
+	log_smgrcreate(&rel->rd_smgr->smgr_rnode.node,
+				   INIT_FORKNUM,
+				   rel->rd_rel->relstorage);
 	smgrimmedsync(rel->rd_smgr, INIT_FORKNUM);
 }