Subject: [PATCH 5/7] Add infrastructure to WAL-logging skip feature

From: Kyotaro Horiguchi <[email-masked]>
Date: Tue, 2 Apr 2019 18:05:10 +0900

We used to optimize WAL-logging for truncation of in-transaction
crated tables in minimal mode by just signaling by
HEAP_INSERT_SKIP_WAL option on heap operations. This mechanism can
emit WAL records that results in corrupt state for certain series of
in-transaction operations. This patch provides infrastructure to track
pending at-commit fsyncs for a relation and in-transaction
truncations. table_relation_register_walskip() should be used to start
tracking before batch operations like COPY and CLUSTER, and use
BufferNeedsWAL() instead of RelationNeedsWAL() at the places related
to WAL-logging about heap-modifying operations, then remove
call to table_finish_bulk_insert() and the tableam intaface.
---
 src/backend/access/transam/xact.c   |  12 +-
 src/backend/catalog/storage.c       | 612 +++++++++++++++++++++++++++++++++---
 src/backend/commands/tablecmds.c    |   6 +-
 src/backend/storage/buffer/bufmgr.c |  39 ++-
 src/backend/utils/cache/relcache.c  |   3 +
 src/include/catalog/storage.h       |  17 +-
 src/include/storage/bufmgr.h        |   2 +
 src/include/utils/rel.h             |   7 +
 8 files changed, 631 insertions(+), 67 deletions(-)

diff --git a/src/backend/access/transam/xact.c b/src/backend/access/transam/xact.c
index e9ed92b70b..33a83dc784 100644
--- a/src/backend/access/transam/xact.c
+++ b/src/backend/access/transam/xact.c
@@ -2102,6 +2102,9 @@ CommitTransaction(void)
 	/* close large objects before lower-level cleanup */
 	AtEOXact_LargeObject(true);
 
+	/* Flush updates to relations that we didn't WAL-logged */
+	smgrFinishBulkInsert(true);
+
 	/*
 	 * Mark serializable transaction as complete for predicate locking
 	 * purposes.  This should be done as late as we can put it and still allow
@@ -2334,6 +2337,9 @@ PrepareTransaction(void)
 	/* close large objects before lower-level cleanup */
 	AtEOXact_LargeObject(true);
 
+	/* Flush updates to relations that we didn't WAL-logged */
+	smgrFinishBulkInsert(true);
+
 	/*
 	 * Mark serializable transaction as complete for predicate locking
 	 * purposes.  This should be done as late as we can put it and still allow
@@ -2659,6 +2665,7 @@ AbortTransaction(void)
 	AtAbort_Notify();
 	AtEOXact_RelationMap(false, is_parallel_worker);
 	AtAbort_Twophase();
+	smgrFinishBulkInsert(false);	/* abandon pending syncs */
 
 	/*
 	 * Advertise the fact that we aborted in pg_xact (assuming that we got as
@@ -4792,8 +4799,7 @@ CommitSubTransaction(void)
 	AtEOSubXact_RelationCache(true, s->subTransactionId,
 							  s->parent->subTransactionId);
 	AtEOSubXact_Inval(true);
-	AtSubCommit_smgr();
-
+	AtSubCommit_smgr(s->subTransactionId, s->parent->subTransactionId);
 	/*
 	 * The only lock we actually release here is the subtransaction XID lock.
 	 */
@@ -4970,7 +4976,7 @@ AbortSubTransaction(void)
 		ResourceOwnerRelease(s->curTransactionOwner,
 							 RESOURCE_RELEASE_AFTER_LOCKS,
 							 false, false);
-		AtSubAbort_smgr();
+		AtSubAbort_smgr(s->subTransactionId, s->parent->subTransactionId);
 
 		AtEOXact_GUC(false, s->gucNestLevel);
 		AtEOSubXact_SPI(false, s->subTransactionId);
diff --git a/src/backend/catalog/storage.c b/src/backend/catalog/storage.c
index 72242b2476..4cd112f86c 100644
--- a/src/backend/catalog/storage.c
+++ b/src/backend/catalog/storage.c
@@ -21,6 +21,7 @@
 
 #include "miscadmin.h"
 
+#include "access/tableam.h"
 #include "access/visibilitymap.h"
 #include "access/xact.h"
 #include "access/xlog.h"
@@ -29,10 +30,18 @@
 #include "catalog/storage.h"
 #include "catalog/storage_xlog.h"
 #include "storage/freespace.h"
-#include "storage/smgr.h"
+#include "utils/hsearch.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 
+ /* #define STORAGEDEBUG */	/* turns DEBUG elogs on */
+
+#ifdef STORAGEDEBUG
+#define STORAGE_elog(...)				elog(__VA_ARGS__)
+#else
+#define STORAGE_elog(...)
+#endif
+
 /*
  * We keep a list of all relations (represented as RelFileNode values)
  * that have been created or deleted in the current transaction.  When
@@ -64,6 +73,61 @@ typedef struct PendingRelDelete
 
 static PendingRelDelete *pendingDeletes = NULL; /* head of linked list */
 
+/*
+ * We also track relation files (RelFileNode values) that have been created
+ * in the same transaction, and that have been modified without WAL-logging
+ * the action (an optimization possible with wal_level=minimal). When we are
+ * about to skip WAL-logging, a RelWalSkip entry is created, and
+ * 'skip_wal_min_blk' is set to the current size of the relation. Any
+ * operations on blocks < skip_wal_min_blk need to be WAL-logged as usual, but
+ * for operations on higher blocks, WAL-logging is skipped.
+
+ *
+ * NB: after WAL-logging has been skipped for a block, we must not WAL-log
+ * any subsequent actions on the same block either. Replaying the WAL record
+ * of the subsequent action might fail otherwise, as the "before" state of
+ * the block might not match, as the earlier actions were not WAL-logged.
+ * Likewise, after we have WAL-logged an operation for a block, we must
+ * WAL-log any subsequent operations on the same page as well. Replaying
+ * a possible full-page-image from the earlier WAL record would otherwise
+ * revert the page to the old state, even if we sync the relation at end
+ * of transaction.
+ *
+ * If a relation is truncated (without creating a new relfilenode), and we
+ * emit a WAL record of the truncation, we can't skip WAL-logging for any
+ * of the truncated blocks anymore, as replaying the truncation record will
+ * destroy all the data inserted after that. But if we have already decided
+ * to skip WAL-logging changes to a relation, and the relation is truncated,
+ * we don't need to WAL-log the truncation either.
+ *
+ * This mechanism is currently only used by heaps. Indexes are always
+ * WAL-logged. Also, this only applies for wal_level=minimal; with higher
+ * WAL levels we need the WAL for PITR/replication anyway.
+ */
+typedef struct RelWalSkip
+{
+	RelFileNode relnode;			/* relation created in same xact */
+	bool		forks[MAX_FORKNUM + 1];	/* target forknums */
+	BlockNumber skip_wal_min_blk;	/* WAL-logging skipped for blocks >=
+									 * skip_wal_min_blk */
+	BlockNumber wal_log_min_blk; 	/* The minimum blk number that requires
+									 * WAL-logging even if skipped by the
+									 * above*/
+	SubTransactionId create_sxid;	/* subxid where this entry is created */
+	SubTransactionId invalidate_sxid; /* subxid where this entry is
+									   * invalidated */
+	const TableAmRoutine *tableam;	/* Table access routine */
+}	RelWalSkip;
+
+/* Relations that need to be fsync'd at commit */
+static HTAB *walSkipHash = NULL;
+
+static RelWalSkip *getWalSkipEntry(Relation rel, bool create);
+static RelWalSkip *getWalSkipEntryRNode(RelFileNode *node,
+													  bool create);
+static void smgrProcessWALSkipInval(bool isCommit, SubTransactionId mySubid,
+						SubTransactionId parentSubid);
+
 /*
  * RelationCreateStorage
  *		Create physical storage for a relation.
@@ -261,31 +325,59 @@ RelationTruncate(Relation rel, BlockNumber nblocks)
 	 */
 	if (RelationNeedsWAL(rel))
 	{
-		/*
-		 * Make an XLOG entry reporting the file truncation.
-		 */
-		XLogRecPtr	lsn;
-		xl_smgr_truncate xlrec;
+		RelWalSkip *walskip;
 
-		xlrec.blkno = nblocks;
-		xlrec.rnode = rel->rd_node;
-		xlrec.flags = SMGR_TRUNCATE_ALL;
-
-		XLogBeginInsert();
-		XLogRegisterData((char *) &xlrec, sizeof(xlrec));
-
-		lsn = XLogInsert(RM_SMGR_ID,
-						 XLOG_SMGR_TRUNCATE | XLR_SPECIAL_REL_UPDATE);
+		/* get pending sync entry, create if not yet */
+		walskip = getWalSkipEntry(rel, true);
 
 		/*
-		 * Flush, because otherwise the truncation of the main relation might
-		 * hit the disk before the WAL record, and the truncation of the FSM
-		 * or visibility map. If we crashed during that window, we'd be left
-		 * with a truncated heap, but the FSM or visibility map would still
-		 * contain entries for the non-existent heap pages.
+		 * walskip is null here if rel doesn't support WAL-logging skip,
+		 * otherwise check for WAL-skipping status.
 		 */
-		if (fsm || vm)
-			XLogFlush(lsn);
+		if (walskip == NULL ||
+			walskip->skip_wal_min_blk == InvalidBlockNumber ||
+			walskip->skip_wal_min_blk < nblocks)
+		{
+			/*
+			 * If WAL-skipping is enabled, this is the first time truncation
+			 * of this relation in this transaction or truncation that leaves
+			 * pages that need at-commit fsync.  Make an XLOG entry reporting
+			 * the file truncation.
+			 */
+			XLogRecPtr		lsn;
+			xl_smgr_truncate xlrec;
+
+			xlrec.blkno = nblocks;
+			xlrec.rnode = rel->rd_node;
+			xlrec.flags = SMGR_TRUNCATE_ALL;
+
+			XLogBeginInsert();
+			XLogRegisterData((char *) &xlrec, sizeof(xlrec));
+
+			lsn = XLogInsert(RM_SMGR_ID,
+							 XLOG_SMGR_TRUNCATE | XLR_SPECIAL_REL_UPDATE);
+
+			STORAGE_elog(DEBUG2,
+						 "WAL-logged truncation of rel %u/%u/%u to %u blocks",
+						 rel->rd_node.spcNode, rel->rd_node.dbNode,
+						 rel->rd_node.relNode, nblocks);
+			/*
+			 * Flush, because otherwise the truncation of the main relation
+			 * might hit the disk before the WAL record, and the truncation of
+			 * the FSM or visibility map. If we crashed during that window,
+			 * we'd be left with a truncated heap, but the FSM or visibility
+			 * map would still contain entries for the non-existent heap
+			 * pages.
+			 */
+			if (fsm || vm)
+				XLogFlush(lsn);
+
+			if (walskip)
+			{
+				/* no longer skip WAL-logging for the blocks */
+				walskip->wal_log_min_blk = nblocks;
+			}
+		}
 	}
 
 	/* Do the real work */
@@ -296,8 +388,7 @@ RelationTruncate(Relation rel, BlockNumber nblocks)
  * Copy a fork's data, block by block.
  */
 void
-RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
-					ForkNumber forkNum, char relpersistence)
+RelationCopyStorage(Relation srcrel, SMgrRelation dst, ForkNumber forkNum)
 {
 	PGAlignedBlock buf;
 	Page		page;
@@ -305,6 +396,8 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 	bool		copying_initfork;
 	BlockNumber nblocks;
 	BlockNumber blkno;
+	SMgrRelation src = srcrel->rd_smgr;
+	char 		relpersistence = srcrel->rd_rel->relpersistence;
 
 	page = (Page) buf.data;
 
@@ -316,12 +409,33 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 	copying_initfork = relpersistence == RELPERSISTENCE_UNLOGGED &&
 		forkNum == INIT_FORKNUM;
 
-	/*
-	 * We need to log the copied data in WAL iff WAL archiving/streaming is
-	 * enabled AND it's a permanent relation.
-	 */
-	use_wal = XLogIsNeeded() &&
-		(relpersistence == RELPERSISTENCE_PERMANENT || copying_initfork);
+	if (relpersistence == RELPERSISTENCE_PERMANENT || copying_initfork)
+	{
+		/*
+		 * We need to log the copied data in WAL iff WAL archiving/streaming
+		 * is enabled AND it's a permanent relation.
+		 */
+		if (XLogIsNeeded())
+			use_wal = true;
+
+		/*
+		 * If the rel is WAL-logged, must fsync before commit.  We use
+		 * heap_sync to ensure that the toast table gets fsync'd too.  (For a
+		 * temp or unlogged rel we don't care since the data will be gone
+		 * after a crash anyway.)
+		 *
+		 * It's obvious that we must do this when not WAL-logging the
+		 * copy. It's less obvious that we have to do it even if we did
+		 * WAL-log the copied pages. The reason is that since we're copying
+		 * outside shared buffers, a CHECKPOINT occurring during the copy has
+		 * no way to flush the previously written data to disk (indeed it
+		 * won't know the new rel even exists).  A crash later on would replay
+		 * WAL from the checkpoint, therefore it wouldn't replay our earlier
+		 * WAL entries. If we do not fsync those pages here, they might still
+		 * not be on disk when the crash occurs.
+		 */
+		RecordPendingSync(srcrel, dst, forkNum);
+	}
 
 	nblocks = smgrnblocks(src, forkNum);
 
@@ -358,24 +472,321 @@ RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
 		 */
 		smgrextend(dst, forkNum, blkno, buf.data, true);
 	}
+}
+
+/*
+ * Do changes to given heap page need to be WAL-logged?
+ *
+ * This takes into account any previous RecordPendingSync() requests.
+ *
+ * Note that it is required to check this before creating any WAL records for
+ * heap pages - it is not merely an optimization! WAL-logging a record, when
+ * we have already skipped a previous WAL record for the same page could lead
+ * to failure at WAL replay, as the "before" state expected by the record
+ * might not match what's on disk. Also, if the heap was truncated earlier, we
+ * must WAL-log any changes to the once-truncated blocks, because replaying
+ * the truncation record will destroy them.
+ */
+bool
+BufferNeedsWAL(Relation rel, Buffer buf)
+{
+	BlockNumber		blkno = InvalidBlockNumber;
+	RelWalSkip *walskip;
+
+	if (!RelationNeedsWAL(rel))
+		return false;
+
+	/* fetch existing pending sync entry */
+	walskip = getWalSkipEntry(rel, false);
 
 	/*
-	 * If the rel is WAL-logged, must fsync before commit.  We use heap_sync
-	 * to ensure that the toast table gets fsync'd too.  (For a temp or
-	 * unlogged rel we don't care since the data will be gone after a crash
-	 * anyway.)
-	 *
-	 * It's obvious that we must do this when not WAL-logging the copy. It's
-	 * less obvious that we have to do it even if we did WAL-log the copied
-	 * pages. The reason is that since we're copying outside shared buffers, a
-	 * CHECKPOINT occurring during the copy has no way to flush the previously
-	 * written data to disk (indeed it won't know the new rel even exists).  A
-	 * crash later on would replay WAL from the checkpoint, therefore it
-	 * wouldn't replay our earlier WAL entries. If we do not fsync those pages
-	 * here, they might still not be on disk when the crash occurs.
+	 * no point in doing further work if we know that we don't skip
+	 * WAL-logging.
 	 */
-	if (relpersistence == RELPERSISTENCE_PERMANENT || copying_initfork)
-		smgrimmedsync(dst, forkNum);
+	if (!walskip)
+	{
+		STORAGE_elog(DEBUG2,
+					 "not skipping WAL-logging for rel %u/%u/%u block %u",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, BufferGetBlockNumber(buf));
+		return true;
+	}
+
+	Assert(BufferIsValid(buf));
+
+	blkno = BufferGetBlockNumber(buf);
+
+	/*
+	 * We don't skip WAL-logging for pages that once done.
+	 */
+	if (walskip->skip_wal_min_blk == InvalidBlockNumber ||
+		walskip->skip_wal_min_blk > blkno)
+	{
+		STORAGE_elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because skip_wal_min_blk is %u",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, blkno, walskip->skip_wal_min_blk);
+		return true;
+	}
+
+	/*
+	 * we don't skip WAL-logging for blocks that have got WAL-logged
+	 * truncation
+	 */
+	if (walskip->wal_log_min_blk != InvalidBlockNumber &&
+		walskip->wal_log_min_blk <= blkno)
+	{
+		STORAGE_elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because wal_log_min_blk is %u",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, blkno, walskip->wal_log_min_blk);
+		return true;
+	}
+
+	STORAGE_elog(DEBUG2, "skipping WAL-logging for rel %u/%u/%u block %u",
+				 rel->rd_node.spcNode, rel->rd_node.dbNode,
+				 rel->rd_node.relNode, blkno);
+
+	return false;
+}
+
+bool
+BlockNeedsWAL(Relation rel, BlockNumber blkno)
+{
+	RelWalSkip *walskip;
+
+	if (!RelationNeedsWAL(rel))
+		return false;
+
+	/* fetch exising pending sync entry */
+	walskip = getWalSkipEntry(rel, false);
+
+	/*
+	 * no point in doing further work if we know that we don't skip
+	 * WAL-logging.
+	 */
+	if (!walskip)
+		return true;
+
+	/*
+	 * We don't skip WAL-logging for pages that once done.
+	 */
+	if (walskip->skip_wal_min_blk == InvalidBlockNumber ||
+		walskip->skip_wal_min_blk > blkno)
+	{
+		STORAGE_elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because skip_wal_min_blk is %u",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, blkno, walskip->skip_wal_min_blk);
+		return true;
+	}
+
+	/*
+	 * we don't skip WAL-logging for blocks that have got WAL-logged
+	 * truncation
+	 */
+	if (walskip->wal_log_min_blk != InvalidBlockNumber &&
+		walskip->wal_log_min_blk <= blkno)
+	{
+		STORAGE_elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because wal_log_min_blk is %u",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, blkno, walskip->wal_log_min_blk);
+
+		return true;
+	}
+
+	STORAGE_elog(DEBUG2, "skipping WAL-logging for rel %u/%u/%u block %u",
+				 rel->rd_node.spcNode, rel->rd_node.dbNode,
+				 rel->rd_node.relNode, blkno);
+
+	return false;
+}
+
+/*
+ * Remember that the given relation doesn't need WAL-logging for the blocks
+ * after the current block size and for the blocks that are going to be synced
+ * at commit.
+ */
+void
+RecordWALSkipping(Relation rel)
+{
+	RelWalSkip *walskip;
+
+	Assert(RelationNeedsWAL(rel));
+
+	/* get pending sync entry, create if not yet  */
+	walskip = getWalSkipEntry(rel, true);
+
+	if (walskip == NULL)
+		return;
+
+	/*
+	 *  Record only the first registration.
+	 */
+	if (walskip->skip_wal_min_blk != InvalidBlockNumber)
+	{
+		STORAGE_elog(DEBUG2, "WAL skipping for rel %u/%u/%u was already registered at block %u (new %u)",
+					 rel->rd_node.spcNode, rel->rd_node.dbNode,
+					 rel->rd_node.relNode, walskip->skip_wal_min_blk,
+					 RelationGetNumberOfBlocks(rel));
+		return;
+	}
+
+	STORAGE_elog(DEBUG2, "registering new WAL skipping rel %u/%u/%u at block %u",
+				 rel->rd_node.spcNode, rel->rd_node.dbNode,
+				 rel->rd_node.relNode, RelationGetNumberOfBlocks(rel));
+
+	walskip->skip_wal_min_blk = RelationGetNumberOfBlocks(rel);
+}
+
+/*
+ * Record commit-time file sync. This shouldn't be used mixing with
+ * RecordWALSkipping.
+ */
+void
+RecordPendingSync(Relation rel, SMgrRelation targetsrel, ForkNumber forknum)
+{
+	RelWalSkip *walskip;
+
+	Assert(RelationNeedsWAL(rel));
+
+	/* check for support for this feature */
+	if (rel->rd_tableam == NULL ||
+		rel->rd_tableam->relation_register_walskip == NULL)
+		return;
+
+	walskip = getWalSkipEntryRNode(&targetsrel->smgr_rnode.node, true);
+	walskip->forks[forknum] = true;
+	walskip->skip_wal_min_blk = 0;
+	walskip->tableam = rel->rd_tableam;
+
+	STORAGE_elog(DEBUG2,
+				 "registering new pending sync for rel %u/%u/%u at block %u",
+				 walskip->relnode.spcNode, walskip->relnode.dbNode,
+				 walskip->relnode.relNode, 0);
+}
+
+/*
+ * RelationInvalidateWALSkip() -- invalidate WAL-skip entry
+ */
+void
+RelationInvalidateWALSkip(Relation rel)
+{
+	RelWalSkip *walskip;
+
+	/* we know we don't have one */
+	if (rel->rd_nowalskip)
+		return;
+
+	walskip = getWalSkipEntry(rel, false);
+
+	if (!walskip)
+		return;
+
+	/*
+	 * The state is reset at subtransaction commit/abort. No invalidation
+	 * request must not come for the same relation in the same subtransaction.
+	 */
+	Assert(walskip->invalidate_sxid == InvalidSubTransactionId);
+
+	walskip->invalidate_sxid = GetCurrentSubTransactionId();
+
+	STORAGE_elog(DEBUG2,
+				 "WAL skip of rel %u/%u/%u invalidated by sxid %d",
+				 walskip->relnode.spcNode, walskip->relnode.dbNode,
+				 walskip->relnode.relNode, walskip->invalidate_sxid);
+}
+
+/*
+ * getWalSkipEntry: get WAL skip entry.
+ *
+ * Returns WAL skip entry for the relation. The entry tracks WAL-skipping
+ * blocks for the relation.  The WAL-skipped blocks need fsync at commit time.
+ * Creates one if needed when create is true. If rel doesn't support this
+ * feature, returns true even if create is true.
+ */
+static inline RelWalSkip *
+getWalSkipEntry(Relation rel, bool create)
+{
+	RelWalSkip *walskip_entry = NULL;
+
+	if (rel->rd_walskip)
+		return rel->rd_walskip;
+
+	/* we know we don't have pending sync entry */
+	if (!create && rel->rd_nowalskip)
+		return NULL;
+
+	/* check for support for this feature */
+	if (rel->rd_tableam == NULL ||
+		rel->rd_tableam->relation_register_walskip == NULL)
+	{
+		rel->rd_nowalskip = true;
+		return NULL;
+	}
+
+	walskip_entry = getWalSkipEntryRNode(&rel->rd_node, create);
+
+	if (!walskip_entry)
+	{
+		/* prevent further hash lookup */
+		rel->rd_nowalskip = true;
+		return NULL;
+	}
+
+	walskip_entry->forks[MAIN_FORKNUM] = true;
+	walskip_entry->tableam = rel->rd_tableam;
+
+	/* hold shortcut in Relation */
+	rel->rd_nowalskip = false;
+	rel->rd_walskip = walskip_entry;
+
+	return walskip_entry;
+}
+
+/*
+ * getWalSkipEntryRNode: get WAL skip entry by rnode
+ *
+ * Returns a WAL skip entry for the RelFileNode.
+ */
+static RelWalSkip *
+getWalSkipEntryRNode(RelFileNode *rnode, bool create)
+{
+	RelWalSkip *walskip_entry = NULL;
+	bool			found;
+
+	if (!walSkipHash)
+	{
+		/* First time through: initialize the hash table */
+		HASHCTL		ctl;
+
+		if (!create)
+			return NULL;
+
+		MemSet(&ctl, 0, sizeof(ctl));
+		ctl.keysize = sizeof(RelFileNode);
+		ctl.entrysize = sizeof(RelWalSkip);
+		ctl.hash = tag_hash;
+		walSkipHash = hash_create("pending relation sync table", 5,
+								   &ctl, HASH_ELEM | HASH_FUNCTION);
+	}
+
+	walskip_entry = (RelWalSkip *)
+		hash_search(walSkipHash, (void *) rnode,
+					create ? HASH_ENTER: HASH_FIND,	&found);
+
+	if (!walskip_entry)
+		return NULL;
+
+	/* new entry created */
+	if (!found)
+	{
+		memset(&walskip_entry->forks, 0, sizeof(walskip_entry->forks));
+		walskip_entry->wal_log_min_blk = InvalidBlockNumber;
+		walskip_entry->skip_wal_min_blk = InvalidBlockNumber;
+		walskip_entry->create_sxid = GetCurrentSubTransactionId();
+		walskip_entry->invalidate_sxid = InvalidSubTransactionId;
+		walskip_entry->tableam = NULL;
+	}
+
+	return walskip_entry;
 }
 
 /*
@@ -506,6 +917,107 @@ smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr)
 	return nrels;
 }
 
+/*
+ * Finish bulk insert of files.
+ */
+void
+smgrFinishBulkInsert(bool isCommit)
+{
+	if (!walSkipHash)
+		return;
+
+	if (isCommit)
+	{
+		HASH_SEQ_STATUS status;
+		RelWalSkip *walskip;
+
+		hash_seq_init(&status, walSkipHash);
+
+		while ((walskip = hash_seq_search(&status)) != NULL)
+		{
+			/*
+			 * On commit, process valid entreis. Rollback doesn't need sync on
+			 * all changes during the transaction.
+			 */
+			if (walskip->skip_wal_min_blk != InvalidBlockNumber &&
+				walskip->invalidate_sxid == InvalidSubTransactionId)
+			{
+				int f;
+
+				FlushRelationBuffersWithoutRelCache(walskip->relnode, false);
+
+				/*
+				 * We mustn't create an entry when the table AM doesn't
+				 * support WAL-skipping.
+				 */
+				Assert (walskip->tableam->finish_bulk_insert);
+
+				/* flush all requested forks  */
+				for (f = MAIN_FORKNUM ; f <= MAX_FORKNUM ; f++)
+				{
+					if (walskip->forks[f])
+					{
+						walskip->tableam->finish_bulk_insert(walskip->relnode, f);
+						STORAGE_elog(DEBUG2, "finishing bulk insert to rel %u/%u/%u fork %d",
+									 walskip->relnode.spcNode,
+									 walskip->relnode.dbNode,
+									 walskip->relnode.relNode, f);
+					}
+				}
+			}
+		}
+	}
+
+	hash_destroy(walSkipHash);
+	walSkipHash = NULL;
+}
+
+/*
+ * Process pending invalidation of WAL skip happened in the subtransaction
+ */
+void
+smgrProcessWALSkipInval(bool isCommit, SubTransactionId mySubid,
+						SubTransactionId parentSubid)
+{
+	HASH_SEQ_STATUS status;
+	RelWalSkip *walskip;
+
+	if (!walSkipHash)
+		return;
+
+	/* We expect that we don't have walSkipHash in almost all cases */
+	hash_seq_init(&status, walSkipHash);
+
+	while ((walskip = hash_seq_search(&status)) != NULL)
+	{
+		if (walskip->create_sxid == mySubid)
+		{
+			/*
+			 * The entry was created in this subxact. Remove it on abort, or
+			 * on commit after invalidation.
+			 */
+			if (!isCommit || walskip->invalidate_sxid == mySubid)
+				hash_search(walSkipHash, &walskip->relnode,
+							HASH_REMOVE, NULL);
+			/* Treat committing valid entry as creation by the parent. */
+			else if (walskip->invalidate_sxid == InvalidSubTransactionId)
+				walskip->create_sxid = parentSubid;
+		}
+		else if (walskip->invalidate_sxid == mySubid)
+		{
+			/*
+			 * This entry was created elsewhere then invalidated by this
+			 * subxact. Treat commit as invalidation by the parent. Otherwise
+			 * cancel invalidation.
+			 */
+			if (isCommit)
+				walskip->invalidate_sxid = parentSubid;
+			else
+				walskip->invalidate_sxid = InvalidSubTransactionId;
+		}
+	}
+}
+
 /*
  *	PostPrepare_smgr -- Clean up after a successful PREPARE
  *
@@ -535,7 +1047,7 @@ PostPrepare_smgr(void)
  * Reassign all items in the pending-deletes list to the parent transaction.
  */
 void
-AtSubCommit_smgr(void)
+AtSubCommit_smgr(SubTransactionId mySubid, SubTransactionId parentSubid)
 {
 	int			nestLevel = GetCurrentTransactionNestLevel();
 	PendingRelDelete *pending;
@@ -545,6 +1057,9 @@ AtSubCommit_smgr(void)
 		if (pending->nestLevel >= nestLevel)
 			pending->nestLevel = nestLevel - 1;
 	}
+
+	/* Remove invalidated WAL skip in this subtransaction */
+	smgrProcessWALSkipInval(true, mySubid, parentSubid);
 }
 
 /*
@@ -555,9 +1070,12 @@ AtSubCommit_smgr(void)
  * subtransaction will not commit.
  */
 void
-AtSubAbort_smgr(void)
+AtSubAbort_smgr(SubTransactionId mySubid, SubTransactionId parentSubid)
 {
 	smgrDoPendingDeletes(false);
+
+	/* Remove invalidated WAL skip in this subtransaction */
+	smgrProcessWALSkipInval(false, mySubid, parentSubid);
 }
 
 void
diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c
index 654179297c..8908b77d98 100644
--- a/src/backend/commands/tablecmds.c
+++ b/src/backend/commands/tablecmds.c
@@ -11983,8 +11983,7 @@ index_copy_data(Relation rel, RelFileNode newrnode)
 	RelationCreateStorage(newrnode, rel->rd_rel->relpersistence);
 
 	/* copy main fork */
-	RelationCopyStorage(rel->rd_smgr, dstrel, MAIN_FORKNUM,
-						rel->rd_rel->relpersistence);
+	RelationCopyStorage(rel, dstrel, MAIN_FORKNUM);
 
 	/* copy those extra forks that exist */
 	for (ForkNumber forkNum = MAIN_FORKNUM + 1;
@@ -12002,8 +12001,7 @@ index_copy_data(Relation rel, RelFileNode newrnode)
 				(rel->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED &&
 				 forkNum == INIT_FORKNUM))
 				log_smgrcreate(&newrnode, forkNum);
-			RelationCopyStorage(rel->rd_smgr, dstrel, forkNum,
-								rel->rd_rel->relpersistence);
+			RelationCopyStorage(rel, dstrel, forkNum);
 		}
 	}
 
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 273e2f385f..f00826712a 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -451,6 +451,7 @@ static BufferDesc *BufferAlloc(SMgrRelation smgr,
 			BufferAccessStrategy strategy,
 			bool *foundPtr);
 static void FlushBuffer(BufferDesc *buf, SMgrRelation reln);
+static void FlushRelationBuffers_common(SMgrRelation smgr, bool islocal);
 static void AtProcExit_Buffers(int code, Datum arg);
 static void CheckForBufferLeaks(void);
 static int	rnode_comparator(const void *p1, const void *p2);
@@ -3153,20 +3154,40 @@ PrintPinnedBufs(void)
 void
 FlushRelationBuffers(Relation rel)
 {
-	int			i;
-	BufferDesc *bufHdr;
-
 	/* Open rel at the smgr level if not already done */
 	RelationOpenSmgr(rel);
 
-	if (RelationUsesLocalBuffers(rel))
+	FlushRelationBuffers_common(rel->rd_smgr, RelationUsesLocalBuffers(rel));
+}
+
+/*
+ * Like FlushRelationBuffers(), but the relation is specified by RelFileNode
+ */
+void
+FlushRelationBuffersWithoutRelCache(RelFileNode rnode, bool islocal)
+{
+	FlushRelationBuffers_common(smgropen(rnode, InvalidBackendId), islocal);
+}
+
+/*
+ * Code shared between functions FlushRelationBuffers() and
+ * FlushRelationBuffersWithoutRelCache().
+ */
+static void
+FlushRelationBuffers_common(SMgrRelation smgr, bool islocal)
+{
+	RelFileNode rnode = smgr->smgr_rnode.node;
+	int			i;
+	BufferDesc *bufHdr;
+
+	if (islocal)
 	{
 		for (i = 0; i < NLocBuffer; i++)
 		{
 			uint32		buf_state;
 
 			bufHdr = GetLocalBufferDescriptor(i);
-			if (RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node) &&
+			if (RelFileNodeEquals(bufHdr->tag.rnode, rnode) &&
 				((buf_state = pg_atomic_read_u32(&bufHdr->state)) &
 				 (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY))
 			{
@@ -3183,7 +3204,7 @@ FlushRelationBuffers(Relation rel)
 
 				PageSetChecksumInplace(localpage, bufHdr->tag.blockNum);
 
-				smgrwrite(rel->rd_smgr,
+				smgrwrite(smgr,
 						  bufHdr->tag.forkNum,
 						  bufHdr->tag.blockNum,
 						  localpage,
@@ -3213,18 +3234,18 @@ FlushRelationBuffers(Relation rel)
 		 * As in DropRelFileNodeBuffers, an unlocked precheck should be safe
 		 * and saves some cycles.
 		 */
-		if (!RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node))
+		if (!RelFileNodeEquals(bufHdr->tag.rnode, rnode))
 			continue;
 
 		ReservePrivateRefCountEntry();
 
 		buf_state = LockBufHdr(bufHdr);
-		if (RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node) &&
+		if (RelFileNodeEquals(bufHdr->tag.rnode, rnode) &&
 			(buf_state & (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY))
 		{
 			PinBuffer_Locked(bufHdr);
 			LWLockAcquire(BufferDescriptorGetContentLock(bufHdr), LW_SHARED);
-			FlushBuffer(bufHdr, rel->rd_smgr);
+			FlushBuffer(bufHdr, smgr);
 			LWLockRelease(BufferDescriptorGetContentLock(bufHdr));
 			UnpinBuffer(bufHdr, true);
 		}
diff --git a/src/backend/utils/cache/relcache.c b/src/backend/utils/cache/relcache.c
index 64f3c2e887..f06d55a8fe 100644
--- a/src/backend/utils/cache/relcache.c
+++ b/src/backend/utils/cache/relcache.c
@@ -75,6 +75,7 @@
 #include "partitioning/partdesc.h"
 #include "rewrite/rewriteDefine.h"
 #include "rewrite/rowsecurity.h"
+#include "storage/bufmgr.h"
 #include "storage/lmgr.h"
 #include "storage/smgr.h"
 #include "utils/array.h"
@@ -5644,6 +5645,8 @@ load_relcache_init_file(bool shared)
 		rel->rd_newRelfilenodeSubid = InvalidSubTransactionId;
 		rel->rd_amcache = NULL;
 		MemSet(&rel->pgstat_info, 0, sizeof(rel->pgstat_info));
+		rel->rd_nowalskip = false;
+		rel->rd_walskip = NULL;
 
 		/*
 		 * Recompute lock and physical addressing info.  This is needed in
diff --git a/src/include/catalog/storage.h b/src/include/catalog/storage.h
index 882dc65c89..83fee7dbfe 100644
--- a/src/include/catalog/storage.h
+++ b/src/include/catalog/storage.h
@@ -23,8 +23,14 @@ extern void RelationCreateStorage(RelFileNode rnode, char relpersistence);
 extern void RelationDropStorage(Relation rel);
 extern void RelationPreserveStorage(RelFileNode rnode, bool atCommit);
 extern void RelationTruncate(Relation rel, BlockNumber nblocks);
-extern void RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
-								ForkNumber forkNum, char relpersistence);
+extern void RelationCopyStorage(Relation srcrel, SMgrRelation dst,
+								ForkNumber forkNum);
+extern bool BufferNeedsWAL(Relation rel, Buffer buf);
+extern bool BlockNeedsWAL(Relation rel, BlockNumber blkno);
+extern void RecordWALSkipping(Relation rel);
+extern void RecordPendingSync(Relation rel, SMgrRelation srel,
+							  ForkNumber forknum);
+extern void RelationInvalidateWALSkip(Relation rel);
 
 /*
  * These functions used to be in storage/smgr/smgr.c, which explains the
@@ -32,8 +38,11 @@ extern void RelationCopyStorage(SMgrRelation src, SMgrRelation dst,
  */
 extern void smgrDoPendingDeletes(bool isCommit);
 extern int	smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr);
-extern void AtSubCommit_smgr(void);
-extern void AtSubAbort_smgr(void);
+extern void smgrFinishBulkInsert(bool isCommit);
+extern void AtSubCommit_smgr(SubTransactionId mySubid,
+							 SubTransactionId parentSubid);
+extern void AtSubAbort_smgr(SubTransactionId mySubid,
+							 SubTransactionId parentSubid);
 extern void PostPrepare_smgr(void);
 
 #endif							/* STORAGE_H */
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index c5826f691d..8a9ea041dd 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -189,6 +189,8 @@ extern BlockNumber RelationGetNumberOfBlocksInFork(Relation relation,
 								ForkNumber forkNum);
 extern void FlushOneBuffer(Buffer buffer);
 extern void FlushRelationBuffers(Relation rel);
+extern void FlushRelationBuffersWithoutRelCache(RelFileNode rnode,
+									bool islocal);
 extern void FlushDatabaseBuffers(Oid dbid);
 extern void DropRelFileNodeBuffers(RelFileNodeBackend rnode,
 					   ForkNumber forkNum, BlockNumber firstDelBlock);
diff --git a/src/include/utils/rel.h b/src/include/utils/rel.h
index 54028515a7..b2b46322b2 100644
--- a/src/include/utils/rel.h
+++ b/src/include/utils/rel.h
@@ -198,6 +198,13 @@ typedef struct RelationData
 
 	/* use "struct" here to avoid needing to include pgstat.h: */
 	struct PgStat_TableStatus *pgstat_info; /* statistics collection area */
+
+	/*
+	 * rd_nowalskip is true if this relation is known not to skip WAL.
+	 * Otherwise we need to ask smgr for an entry if rd_walskip is NULL.
+	 */
+	bool				rd_nowalskip;
+	struct RelWalSkip   *rd_walskip;
 } RelationData;

-- 
2.16.3

----Next_Part(Tue_Apr_02_19_54_06_2019_801)--
Content-Type: Text/X-Patch; charset=us-ascii
Content-Transfer-Encoding: 7bit
Content-Disposition: inline;
 filename="v10-0006-Fix-WAL-skipping-feature.patch"