fix-wal-level-minimal-michael-horiguchi-3.patch

*** a/src/backend/access/heap/heapam.c
--- b/src/backend/access/heap/heapam.c
***************
*** 34,39 ****
--- 34,61 ----
   *	  the POSTGRES heap access method used for all POSTGRES
   *	  relations.
   *
+  * WAL CONSIDERATIONS
+  *	  All heap operations are normally WAL-logged. but there are a few
+  *	  exceptions. Temporary and unlogged relations never need to be
+  *	  WAL-logged, but we can also skip WAL-logging for a table that was
+  *	  created in the same transaction, if we don't need WAL for PITR or
+  *	  WAL archival purposes (i.e. if wal_level=minimal), and we fsync()
+  *	  the file to disk at COMMIT instead.
+  *
+  *	  The same-relation optimization is not employed automatically on all
+  *	  updates to a table that was created in the same transacton, because
+  *	  for a small number of changes, it's cheaper to just create the WAL
+  *	  records than fsyncing() the whole relation at COMMIT. It is only
+  *	  worthwhile for (presumably) large operations like COPY, CLUSTER,
+  *	  or VACUUM FULL. Use heap_register_sync() to initiate such an
+  *	  operation; it will cause any subsequent updates to the table to skip
+  *	  WAL-logging, if possible, and cause the heap to be synced to disk at
+  *	  COMMIT.
+  *
+  *	  To make that work, all modifications to heap must use
+  *	  HeapNeedsWAL() to check if WAL-logging is needed in this transaction
+  *	  for the given block.
+  *
   *-------------------------------------------------------------------------
   */
  #include "postgres.h"
***************
*** 56,61 ****
--- 78,84 ----
  #include "access/xlogutils.h"
  #include "catalog/catalog.h"
  #include "catalog/namespace.h"
+ #include "catalog/storage.h"
  #include "miscadmin.h"
  #include "pgstat.h"
  #include "port/atomics.h"
***************
*** 2370,2381 **** ReleaseBulkInsertStatePin(BulkInsertState bistate)
   * The new tuple is stamped with current transaction ID and the specified
   * command ID.
   *
-  * If the HEAP_INSERT_SKIP_WAL option is specified, the new tuple is not
-  * logged in WAL, even for a non-temp relation.  Safe usage of this behavior
-  * requires that we arrange that all new tuples go into new pages not
-  * containing any tuples from other transactions, and that the relation gets
-  * fsync'd before commit.  (See also heap_sync() comments)
-  *
   * The HEAP_INSERT_SKIP_FSM option is passed directly to
   * RelationGetBufferForTuple, which see for more info.
   *
--- 2393,2398 ----
*** a/src/backend/access/heap/pruneheap.c
--- b/src/backend/access/heap/pruneheap.c
***************
*** 20,25 ****
--- 20,26 ----
  #include "access/htup_details.h"
  #include "access/xlog.h"
  #include "catalog/catalog.h"
+ #include "catalog/storage.h"
  #include "miscadmin.h"
  #include "pgstat.h"
  #include "storage/bufmgr.h"
***************
*** 259,265 **** heap_page_prune(Relation relation, Buffer buffer, TransactionId OldestXmin,
  		/*
  		 * Emit a WAL HEAP_CLEAN record showing what we did
  		 */
! 		if (RelationNeedsWAL(relation))
  		{
  			XLogRecPtr	recptr;
  
--- 260,266 ----
  		/*
  		 * Emit a WAL HEAP_CLEAN record showing what we did
  		 */
! 		if (BufferNeedsWAL(relation, buffer))
  		{
  			XLogRecPtr	recptr;
  
*** a/src/backend/access/heap/rewriteheap.c
--- b/src/backend/access/heap/rewriteheap.c
***************
*** 649,657 **** raw_heap_insert(RewriteState state, HeapTuple tup)
  	}
  	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
  		heaptup = toast_insert_or_update(state->rs_new_rel, tup, NULL,
! 										 HEAP_INSERT_SKIP_FSM |
! 										 (state->rs_use_wal ?
! 										  0 : HEAP_INSERT_SKIP_WAL));
  	else
  		heaptup = tup;
  
--- 649,655 ----
  	}
  	else if (HeapTupleHasExternal(tup) || tup->t_len > TOAST_TUPLE_THRESHOLD)
  		heaptup = toast_insert_or_update(state->rs_new_rel, tup, NULL,
! 										 HEAP_INSERT_SKIP_FSM);
  	else
  		heaptup = tup;
  
*** a/src/backend/access/heap/visibilitymap.c
--- b/src/backend/access/heap/visibilitymap.c
***************
*** 88,93 ****
--- 88,94 ----
  #include "access/heapam_xlog.h"
  #include "access/visibilitymap.h"
  #include "access/xlog.h"
+ #include "catalog/storage.h"
  #include "miscadmin.h"
  #include "storage/bufmgr.h"
  #include "storage/lmgr.h"
***************
*** 307,313 **** visibilitymap_set(Relation rel, BlockNumber heapBlk, Buffer heapBuf,
  		map[mapByte] |= (flags << mapOffset);
  		MarkBufferDirty(vmBuf);
  
! 		if (RelationNeedsWAL(rel))
  		{
  			if (XLogRecPtrIsInvalid(recptr))
  			{
--- 308,314 ----
  		map[mapByte] |= (flags << mapOffset);
  		MarkBufferDirty(vmBuf);
  
! 		if (BufferNeedsWAL(rel, heapBuf))
  		{
  			if (XLogRecPtrIsInvalid(recptr))
  			{
*** a/src/backend/access/transam/xact.c
--- b/src/backend/access/transam/xact.c
***************
*** 2007,2012 **** CommitTransaction(void)
--- 2007,2015 ----
  	/* close large objects before lower-level cleanup */
  	AtEOXact_LargeObject(true);
  
+ 	/* Flush updates to relations that we didn't WAL-logged */
+ 	smgrDoPendingSyncs(true);
+ 
  	/*
  	 * Mark serializable transaction as complete for predicate locking
  	 * purposes.  This should be done as late as we can put it and still allow
***************
*** 2235,2240 **** PrepareTransaction(void)
--- 2238,2246 ----
  	/* close large objects before lower-level cleanup */
  	AtEOXact_LargeObject(true);
  
+ 	/* Flush updates to relations that we didn't WAL-logged */
+ 	smgrDoPendingSyncs(true);
+ 
  	/*
  	 * Mark serializable transaction as complete for predicate locking
  	 * purposes.  This should be done as late as we can put it and still allow
***************
*** 2548,2553 **** AbortTransaction(void)
--- 2554,2560 ----
  	AtAbort_Notify();
  	AtEOXact_RelationMap(false);
  	AtAbort_Twophase();
+ 	smgrDoPendingSyncs(false);	/* abandone pending syncs */
  
  	/*
  	 * Advertise the fact that we aborted in pg_xact (assuming that we got as
*** a/src/backend/catalog/storage.c
--- b/src/backend/catalog/storage.c
***************
*** 29,34 ****
--- 29,35 ----
  #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"
  
***************
*** 64,69 **** typedef struct PendingRelDelete
--- 65,113 ----
  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 PendingRelSync entry is created, and
+  * 'sync_above' is set to the current size of the relation. Any operations
+  * on blocks < sync_above 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 PendingRelSync
+ {
+ 	RelFileNode relnode;		/* relation created in same xact */
+ 	BlockNumber sync_above;		/* WAL-logging skipped for blocks >=
+ 								 * sync_above */
+ 	BlockNumber truncated_to;	/* truncation WAL record was written */
+ }	PendingRelSync;
+ 
+ /* Relations that need to be fsync'd at commit */
+ static HTAB *pendingSyncs = NULL;
+ 
+ static void createPendingSyncsHash(void);
+ 
+ /*
   * RelationCreateStorage
   *		Create physical storage for a relation.
   *
***************
*** 226,231 **** RelationPreserveStorage(RelFileNode rnode, bool atCommit)
--- 270,277 ----
  void
  RelationTruncate(Relation rel, BlockNumber nblocks)
  {
+ 	PendingRelSync *pending = NULL;
+ 	bool		found;
  	bool		fsm;
  	bool		vm;
  
***************
*** 260,296 **** RelationTruncate(Relation rel, BlockNumber nblocks)
  	 */
  	if (RelationNeedsWAL(rel))
  	{
! 		/*
! 		 * 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);
! 
! 		/*
! 		 * 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);
  	}
  
  	/* Do the real work */
  	smgrtruncate(rel->rd_smgr, MAIN_FORKNUM, nblocks);
  }
  
  /*
   *	smgrDoPendingDeletes() -- Take care of relation deletes at end of xact.
   *
--- 306,386 ----
  	 */
  	if (RelationNeedsWAL(rel))
  	{
! 		/* no_pending_sync is ignored since new entry is created here */
! 		if (!rel->pending_sync)
! 		{
! 			if (!pendingSyncs)
! 				createPendingSyncsHash();
! 			elog(DEBUG2, "RelationTruncate: accessing hash");
! 			pending = (PendingRelSync *) hash_search(pendingSyncs,
! 												 (void *) &rel->rd_node,
! 												 HASH_ENTER, &found);
! 			if (!found)
! 			{
! 				pending->sync_above = InvalidBlockNumber;
! 				pending->truncated_to = InvalidBlockNumber;
! 			}
! 
! 			rel->no_pending_sync= false;
! 			rel->pending_sync = pending;
! 		}
! 
! 		if (rel->pending_sync->sync_above == InvalidBlockNumber ||
! 			rel->pending_sync->sync_above < nblocks)
! 		{
! 			/*
! 			 * Make an XLOG entry reporting the file truncation.
! 			 */
! 			XLogRecPtr		lsn;
! 			xl_smgr_truncate xlrec;
! 
! 			xlrec.blkno = nblocks;
! 			xlrec.rnode = rel->rd_node;
! 
! 			XLogBeginInsert();
! 			XLogRegisterData((char *) &xlrec, sizeof(xlrec));
! 
! 			lsn = XLogInsert(RM_SMGR_ID,
! 							 XLOG_SMGR_TRUNCATE | XLR_SPECIAL_REL_UPDATE);
! 
! 			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);
! 
! 			rel->pending_sync->truncated_to = nblocks;
! 		}
  	}
  
  	/* Do the real work */
  	smgrtruncate(rel->rd_smgr, MAIN_FORKNUM, nblocks);
  }
  
+ /* create the hash table to track pending at-commit fsyncs */
+ static void
+ createPendingSyncsHash(void)
+ {
+ 	/* First time through: initialize the hash table */
+ 	HASHCTL		ctl;
+ 
+ 	MemSet(&ctl, 0, sizeof(ctl));
+ 	ctl.keysize = sizeof(RelFileNode);
+ 	ctl.entrysize = sizeof(PendingRelSync);
+ 	ctl.hash = tag_hash;
+ 	pendingSyncs = hash_create("pending relation sync table", 5,
+ 							   &ctl, HASH_ELEM | HASH_FUNCTION);
+ }
+ 
  /*
   *	smgrDoPendingDeletes() -- Take care of relation deletes at end of xact.
   *
***************
*** 369,374 **** smgrDoPendingDeletes(bool isCommit)
--- 459,482 ----
  }
  
  /*
+  * RelationRemovePendingSync() -- remove pendingSync entry for a relation
+  */
+ void
+ RelationRemovePendingSync(Relation rel)
+ {
+ 	bool found;
+ 
+ 	rel->pending_sync = NULL;
+ 	rel->no_pending_sync = true;
+ 	if (pendingSyncs)
+ 	{
+ 		elog(DEBUG2, "RelationRemovePendingSync: accessing hash");
+ 		hash_search(pendingSyncs, (void *) &rel->rd_node, HASH_REMOVE, &found);
+ 	}
+ }
+ 
+ 
+ /*
   * smgrGetPendingDeletes() -- Get a list of non-temp relations to be deleted.
   *
   * The return value is the number of relations scheduled for termination.
***************
*** 419,424 **** smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr)
--- 527,696 ----
  	return nrels;
  }
  
+ 
+ /*
+  * Remember that the given relation needs to be sync'd at commit, because we
+  * are going to skip WAL-logging subsequent actions to it.
+  */
+ void
+ RecordPendingSync(Relation rel)
+ {
+ 	bool found = true;
+ 	BlockNumber nblocks;
+ 
+ 	Assert(RelationNeedsWAL(rel));
+ 
+ 	/* ignore no_pending_sync since new entry is created here */
+ 	if (!rel->pending_sync)
+ 	{
+ 		if (!pendingSyncs)
+ 			createPendingSyncsHash();
+ 
+ 		/* Look up or create an entry */
+ 		rel->no_pending_sync = false;
+ 		elog(DEBUG2, "RecordPendingSync: accessing hash");
+ 		rel->pending_sync =
+ 			(PendingRelSync *) hash_search(pendingSyncs,
+ 										   (void *) &rel->rd_node,
+ 										   HASH_ENTER, &found);
+ 	}
+ 
+ 	nblocks = RelationGetNumberOfBlocks(rel);
+ 	if (!found)
+ 	{
+ 		rel->pending_sync->truncated_to = InvalidBlockNumber;
+ 		rel->pending_sync->sync_above = nblocks;
+ 
+ 		elog(DEBUG2,
+ 			 "registering new pending sync for rel %u/%u/%u at block %u",
+ 			 rel->rd_node.spcNode, rel->rd_node.dbNode, rel->rd_node.relNode,
+ 			 nblocks);
+ 
+ 	}
+ 	else if (rel->pending_sync->sync_above == InvalidBlockNumber)
+ 	{
+ 		elog(DEBUG2, "registering pending sync for rel %u/%u/%u at block %u",
+ 			 rel->rd_node.spcNode, rel->rd_node.dbNode, rel->rd_node.relNode,
+ 			 nblocks);
+ 		rel->pending_sync->sync_above = nblocks;
+ 	}
+ 	else
+ 		elog(DEBUG2,
+ 			 "pending sync 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,
+ 			 rel->pending_sync->sync_above, nblocks);
+ }
+ 
+ /*
+  * 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;
+ 
+ 	if (!RelationNeedsWAL(rel))
+ 		return false;
+ 
+ 	elog(DEBUG2, "BufferNeedsWAL(r %d, b %d): hash = %p, ent=%p, neg = %d", rel->rd_id, BufferGetBlockNumber(buf), pendingSyncs, rel->pending_sync, rel->no_pending_sync);
+ 	/* no further work if we know that we don't have pending sync */
+ 	if (!pendingSyncs || rel->no_pending_sync)
+ 		return true;
+ 
+ 	/* do the real work */
+ 	if (!rel->pending_sync)
+ 	{
+ 		bool found = false;
+ 
+ 		/*
+ 		 * Hold the entry in rel. This relies on the fact that hash entry
+ 		 * never moves.
+ 		 */
+ 		rel->pending_sync =
+ 			(PendingRelSync *) hash_search(pendingSyncs,
+ 										   (void *) &rel->rd_node,
+ 										   HASH_FIND, &found);
+ 		elog(DEBUG2, "BufferNeedsWAL: accessing hash : %s", found ? "found" : "not found");
+ 		if (!found)
+ 		{
+ 			/* we don't have no one. don't access the hash no longer */
+ 			rel->no_pending_sync = true;
+ 			return true;
+ 		}
+ 	}
+ 
+ 	blkno = BufferGetBlockNumber(buf);
+ 	if (rel->pending_sync->sync_above == InvalidBlockNumber ||
+ 		rel->pending_sync->sync_above > blkno)
+ 	{
+ 		elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because sync_above is %u",
+ 			 rel->rd_node.spcNode, rel->rd_node.dbNode, rel->rd_node.relNode,
+ 			 blkno, rel->pending_sync->sync_above);
+ 		return true;
+ 	}
+ 
+ 	/*
+ 	 * We have emitted a truncation record for this block.
+ 	 */
+ 	if (rel->pending_sync->truncated_to != InvalidBlockNumber &&
+ 		rel->pending_sync->truncated_to <= blkno)
+ 	{
+ 		elog(DEBUG2, "not skipping WAL-logging for rel %u/%u/%u block %u, because it was truncated earlier in the same xact",
+ 			 rel->rd_node.spcNode, rel->rd_node.dbNode, rel->rd_node.relNode,
+ 			 blkno);
+ 		return true;
+ 	}
+ 
+ 	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;
+ }
+ 
+ /*
+  * Sync to disk any relations that we skipped WAL-logging for earlier.
+  */
+ void
+ smgrDoPendingSyncs(bool isCommit)
+ {
+ 	if (!pendingSyncs)
+ 		return;
+ 
+ 	if (isCommit)
+ 	{
+ 		HASH_SEQ_STATUS status;
+ 		PendingRelSync *pending;
+ 
+ 		hash_seq_init(&status, pendingSyncs);
+ 
+ 		while ((pending = hash_seq_search(&status)) != NULL)
+ 		{
+ 			if (pending->sync_above != InvalidBlockNumber)
+ 			{
+ 				FlushRelationBuffersWithoutRelCache(pending->relnode, false);
+ 				smgrimmedsync(smgropen(pending->relnode, InvalidBackendId), MAIN_FORKNUM);
+ 
+ 				elog(DEBUG2, "syncing rel %u/%u/%u", pending->relnode.spcNode,
+ 					 pending->relnode.dbNode, pending->relnode.relNode);
+ 			}
+ 		}
+ 	}
+ 
+ 	hash_destroy(pendingSyncs);
+ 	pendingSyncs = NULL;
+ }
+ 
  /*
   *	PostPrepare_smgr -- Clean up after a successful PREPARE
   *
*** a/src/backend/commands/copy.c
--- b/src/backend/commands/copy.c
***************
*** 2347,2354 **** CopyFrom(CopyState cstate)
  	 *	- data is being written to relfilenode created in this transaction
  	 * then we can skip writing WAL.  It's safe because if the transaction
  	 * doesn't commit, we'll discard the table (or the new relfilenode file).
! 	 * If it does commit, we'll have done the heap_sync at the bottom of this
! 	 * routine first.
  	 *
  	 * As mentioned in comments in utils/rel.h, the in-same-transaction test
  	 * is not always set correctly, since in rare cases rd_newRelfilenodeSubid
--- 2347,2353 ----
  	 *	- data is being written to relfilenode created in this transaction
  	 * then we can skip writing WAL.  It's safe because if the transaction
  	 * doesn't commit, we'll discard the table (or the new relfilenode file).
! 	 * If it does commit, commit will do heap_sync().
  	 *
  	 * As mentioned in comments in utils/rel.h, the in-same-transaction test
  	 * is not always set correctly, since in rare cases rd_newRelfilenodeSubid
***************
*** 2380,2386 **** CopyFrom(CopyState cstate)
  	{
  		hi_options |= HEAP_INSERT_SKIP_FSM;
  		if (!XLogIsNeeded())
! 			hi_options |= HEAP_INSERT_SKIP_WAL;
  	}
  
  	/*
--- 2379,2385 ----
  	{
  		hi_options |= HEAP_INSERT_SKIP_FSM;
  		if (!XLogIsNeeded())
! 			heap_register_sync(cstate->rel);
  	}
  
  	/*
***************
*** 2862,2872 **** CopyFrom(CopyState cstate)
  	FreeExecutorState(estate);
  
  	/*
! 	 * If we skipped writing WAL, then we need to sync the heap (but not
! 	 * indexes since those use WAL anyway)
  	 */
- 	if (hi_options & HEAP_INSERT_SKIP_WAL)
- 		heap_sync(cstate->rel);
  
  	return processed;
  }
--- 2861,2871 ----
  	FreeExecutorState(estate);
  
  	/*
! 	 * If we skipped writing WAL, then we will sync the heap at the end of
! 	 * the transaction. (We used to do it here, but it was later found out
! 	 * that to be safe, we must also avoid WAL-logging any subsequent
! 	 * actions on the pages we skipped WAL for). Indexes always use WAL.
  	 */
  
  	return processed;
  }
*** a/src/backend/commands/createas.c
--- b/src/backend/commands/createas.c
***************
*** 567,574 **** intorel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
  	 * We can skip WAL-logging the insertions, unless PITR or streaming
  	 * replication is in use. We can skip the FSM in any case.
  	 */
! 	myState->hi_options = HEAP_INSERT_SKIP_FSM |
! 		(XLogIsNeeded() ? 0 : HEAP_INSERT_SKIP_WAL);
  	myState->bistate = GetBulkInsertState();
  
  	/* Not using WAL requires smgr_targblock be initially invalid */
--- 567,575 ----
  	 * We can skip WAL-logging the insertions, unless PITR or streaming
  	 * replication is in use. We can skip the FSM in any case.
  	 */
! 	if (!XLogIsNeeded())
! 		heap_register_sync(intoRelationDesc);
! 	myState->hi_options = HEAP_INSERT_SKIP_FSM;
  	myState->bistate = GetBulkInsertState();
  
  	/* Not using WAL requires smgr_targblock be initially invalid */
***************
*** 617,625 **** intorel_shutdown(DestReceiver *self)
  
  	FreeBulkInsertState(myState->bistate);
  
! 	/* If we skipped using WAL, must heap_sync before commit */
! 	if (myState->hi_options & HEAP_INSERT_SKIP_WAL)
! 		heap_sync(myState->rel);
  
  	/* close rel, but keep lock until commit */
  	heap_close(myState->rel, NoLock);
--- 618,624 ----
  
  	FreeBulkInsertState(myState->bistate);
  
! 	/* If we skipped using WAL, we will sync the relation at commit */
  
  	/* close rel, but keep lock until commit */
  	heap_close(myState->rel, NoLock);
*** a/src/backend/commands/matview.c
--- b/src/backend/commands/matview.c
***************
*** 477,483 **** transientrel_startup(DestReceiver *self, int operation, TupleDesc typeinfo)
  	 */
  	myState->hi_options = HEAP_INSERT_SKIP_FSM | HEAP_INSERT_FROZEN;
  	if (!XLogIsNeeded())
! 		myState->hi_options |= HEAP_INSERT_SKIP_WAL;
  	myState->bistate = GetBulkInsertState();
  
  	/* Not using WAL requires smgr_targblock be initially invalid */
--- 477,483 ----
  	 */
  	myState->hi_options = HEAP_INSERT_SKIP_FSM | HEAP_INSERT_FROZEN;
  	if (!XLogIsNeeded())
! 		heap_register_sync(transientrel);
  	myState->bistate = GetBulkInsertState();
  
  	/* Not using WAL requires smgr_targblock be initially invalid */
***************
*** 520,528 **** transientrel_shutdown(DestReceiver *self)
  
  	FreeBulkInsertState(myState->bistate);
  
! 	/* If we skipped using WAL, must heap_sync before commit */
! 	if (myState->hi_options & HEAP_INSERT_SKIP_WAL)
! 		heap_sync(myState->transientrel);
  
  	/* close transientrel, but keep lock until commit */
  	heap_close(myState->transientrel, NoLock);
--- 520,526 ----
  
  	FreeBulkInsertState(myState->bistate);
  
! 	/* If we skipped using WAL, we will sync the relation at commit */
  
  	/* close transientrel, but keep lock until commit */
  	heap_close(myState->transientrel, NoLock);
*** a/src/backend/commands/tablecmds.c
--- b/src/backend/commands/tablecmds.c
***************
*** 4357,4364 **** ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
  		bistate = GetBulkInsertState();
  
  		hi_options = HEAP_INSERT_SKIP_FSM;
  		if (!XLogIsNeeded())
! 			hi_options |= HEAP_INSERT_SKIP_WAL;
  	}
  	else
  	{
--- 4357,4365 ----
  		bistate = GetBulkInsertState();
  
  		hi_options = HEAP_INSERT_SKIP_FSM;
+ 
  		if (!XLogIsNeeded())
! 			heap_register_sync(newrel);
  	}
  	else
  	{
***************
*** 4624,4631 **** ATRewriteTable(AlteredTableInfo *tab, Oid OIDNewHeap, LOCKMODE lockmode)
  		FreeBulkInsertState(bistate);
  
  		/* If we skipped writing WAL, then we need to sync the heap. */
- 		if (hi_options & HEAP_INSERT_SKIP_WAL)
- 			heap_sync(newrel);
  
  		heap_close(newrel, NoLock);
  	}
--- 4625,4630 ----
*** a/src/backend/commands/vacuumlazy.c
--- b/src/backend/commands/vacuumlazy.c
***************
*** 891,897 **** lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
  				 * page has been previously WAL-logged, and if not, do that
  				 * now.
  				 */
! 				if (RelationNeedsWAL(onerel) &&
  					PageGetLSN(page) == InvalidXLogRecPtr)
  					log_newpage_buffer(buf, true);
  
--- 891,897 ----
  				 * page has been previously WAL-logged, and if not, do that
  				 * now.
  				 */
! 				if (BufferNeedsWAL(onerel, buf) &&
  					PageGetLSN(page) == InvalidXLogRecPtr)
  					log_newpage_buffer(buf, true);
  
***************
*** 1118,1124 **** lazy_scan_heap(Relation onerel, int options, LVRelStats *vacrelstats,
  			}
  
  			/* Now WAL-log freezing if necessary */
! 			if (RelationNeedsWAL(onerel))
  			{
  				XLogRecPtr	recptr;
  
--- 1118,1124 ----
  			}
  
  			/* Now WAL-log freezing if necessary */
! 			if (BufferNeedsWAL(onerel, buf))
  			{
  				XLogRecPtr	recptr;
  
***************
*** 1476,1482 **** lazy_vacuum_page(Relation onerel, BlockNumber blkno, Buffer buffer,
  	MarkBufferDirty(buffer);
  
  	/* XLOG stuff */
! 	if (RelationNeedsWAL(onerel))
  	{
  		XLogRecPtr	recptr;
  
--- 1476,1482 ----
  	MarkBufferDirty(buffer);
  
  	/* XLOG stuff */
! 	if (BufferNeedsWAL(onerel, buffer))
  	{
  		XLogRecPtr	recptr;
  
*** a/src/backend/storage/buffer/bufmgr.c
--- b/src/backend/storage/buffer/bufmgr.c
***************
*** 451,456 **** static BufferDesc *BufferAlloc(SMgrRelation smgr,
--- 451,457 ----
  			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);
***************
*** 3147,3166 **** 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))
  	{
  		for (i = 0; i < NLocBuffer; i++)
  		{
  			uint32		buf_state;
  
  			bufHdr = GetLocalBufferDescriptor(i);
! 			if (RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node) &&
  				((buf_state = pg_atomic_read_u32(&bufHdr->state)) &
  				 (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY))
  			{
--- 3148,3188 ----
  void
  FlushRelationBuffers(Relation rel)
  {
  	/* Open rel at the smgr level if not already done */
  	RelationOpenSmgr(rel);
  
! 	FlushRelationBuffers_common(rel->rd_smgr, RelationUsesLocalBuffers(rel));
! }
! 
! /*
!  * Like FlushRelationBuffers(), but the relation is specified by a
!  * 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, rnode) &&
  				((buf_state = pg_atomic_read_u32(&bufHdr->state)) &
  				 (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY))
  			{
***************
*** 3177,3183 **** FlushRelationBuffers(Relation rel)
  
  				PageSetChecksumInplace(localpage, bufHdr->tag.blockNum);
  
! 				smgrwrite(rel->rd_smgr,
  						  bufHdr->tag.forkNum,
  						  bufHdr->tag.blockNum,
  						  localpage,
--- 3199,3205 ----
  
  				PageSetChecksumInplace(localpage, bufHdr->tag.blockNum);
  
! 				smgrwrite(smgr,
  						  bufHdr->tag.forkNum,
  						  bufHdr->tag.blockNum,
  						  localpage,
***************
*** 3207,3224 **** FlushRelationBuffers(Relation rel)
  		 * As in DropRelFileNodeBuffers, an unlocked precheck should be safe
  		 * and saves some cycles.
  		 */
! 		if (!RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node))
  			continue;
  
  		ReservePrivateRefCountEntry();
  
  		buf_state = LockBufHdr(bufHdr);
! 		if (RelFileNodeEquals(bufHdr->tag.rnode, rel->rd_node) &&
  			(buf_state & (BM_VALID | BM_DIRTY)) == (BM_VALID | BM_DIRTY))
  		{
  			PinBuffer_Locked(bufHdr);
  			LWLockAcquire(BufferDescriptorGetContentLock(bufHdr), LW_SHARED);
! 			FlushBuffer(bufHdr, rel->rd_smgr);
  			LWLockRelease(BufferDescriptorGetContentLock(bufHdr));
  			UnpinBuffer(bufHdr, true);
  		}
--- 3229,3246 ----
  		 * As in DropRelFileNodeBuffers, an unlocked precheck should be safe
  		 * and saves some cycles.
  		 */
! 		if (!RelFileNodeEquals(bufHdr->tag.rnode, rnode))
  			continue;
  
  		ReservePrivateRefCountEntry();
  
  		buf_state = LockBufHdr(bufHdr);
! 		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, smgr);
  			LWLockRelease(BufferDescriptorGetContentLock(bufHdr));
  			UnpinBuffer(bufHdr, true);
  		}
*** a/src/backend/utils/cache/relcache.c
--- b/src/backend/utils/cache/relcache.c
***************
*** 72,77 ****
--- 72,78 ----
  #include "optimizer/var.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"
***************
*** 418,423 **** AllocateRelationDesc(Form_pg_class relp)
--- 419,428 ----
  	/* which we mark as a reference-counted tupdesc */
  	relation->rd_att->tdrefcount = 1;
  
+ 	/* We don't know if pending sync for this relation exists so far */
+ 	relation->pending_sync = NULL;
+ 	relation->no_pending_sync = false;
+ 
  	MemoryContextSwitchTo(oldcxt);
  
  	return relation;
***************
*** 2040,2045 **** formrdesc(const char *relationName, Oid relationReltype,
--- 2045,2054 ----
  		relation->rd_rel->relhasindex = true;
  	}
  
+ 	/* We don't know if pending sync for this relation exists so far */
+ 	relation->pending_sync = NULL;
+ 	relation->no_pending_sync = false;
+ 
  	/*
  	 * add new reldesc to relcache
  	 */
***************
*** 3364,3369 **** RelationBuildLocalRelation(const char *relname,
--- 3373,3382 ----
  	else
  		rel->rd_rel->relfilenode = relfilenode;
  
+ 	/* newly built relation has no pending sync */
+ 	rel->no_pending_sync = true;
+ 	rel->pending_sync = NULL;
+ 
  	RelationInitLockInfo(rel);	/* see lmgr.c */
  
  	RelationInitPhysicalAddr(rel);
*** a/src/include/access/heapam.h
--- b/src/include/access/heapam.h
***************
*** 25,34 ****
  
  
  /* "options" flag bits for heap_insert */
! #define HEAP_INSERT_SKIP_WAL	0x0001
! #define HEAP_INSERT_SKIP_FSM	0x0002
! #define HEAP_INSERT_FROZEN		0x0004
! #define HEAP_INSERT_SPECULATIVE 0x0008
  
  typedef struct BulkInsertStateData *BulkInsertState;
  
--- 25,33 ----
  
  
  /* "options" flag bits for heap_insert */
! #define HEAP_INSERT_SKIP_FSM	0x0001
! #define HEAP_INSERT_FROZEN		0x0002
! #define HEAP_INSERT_SPECULATIVE 0x0004
  
  typedef struct BulkInsertStateData *BulkInsertState;
  
***************
*** 179,184 **** extern void simple_heap_delete(Relation relation, ItemPointer tid);
--- 178,184 ----
  extern void simple_heap_update(Relation relation, ItemPointer otid,
  				   HeapTuple tup);
  
+ extern void heap_register_sync(Relation relation);
  extern void heap_sync(Relation relation);
  extern void heap_update_snapshot(HeapScanDesc scan, Snapshot snapshot);
  
*** a/src/include/catalog/storage.h
--- b/src/include/catalog/storage.h
***************
*** 22,34 **** 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);
! 
  /*
   * These functions used to be in storage/smgr/smgr.c, which explains the
   * naming
   */
  extern void smgrDoPendingDeletes(bool isCommit);
  extern int	smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr);
  extern void AtSubCommit_smgr(void);
  extern void AtSubAbort_smgr(void);
  extern void PostPrepare_smgr(void);
--- 22,37 ----
  extern void RelationDropStorage(Relation rel);
  extern void RelationPreserveStorage(RelFileNode rnode, bool atCommit);
  extern void RelationTruncate(Relation rel, BlockNumber nblocks);
! extern void RelationRemovePendingSync(Relation rel);
  /*
   * These functions used to be in storage/smgr/smgr.c, which explains the
   * naming
   */
  extern void smgrDoPendingDeletes(bool isCommit);
  extern int	smgrGetPendingDeletes(bool forCommit, RelFileNode **ptr);
+ extern void smgrDoPendingSyncs(bool isCommit);
+ extern void RecordPendingSync(Relation rel);
+ bool BufferNeedsWAL(Relation rel, Buffer buf);
  extern void AtSubCommit_smgr(void);
  extern void AtSubAbort_smgr(void);
  extern void PostPrepare_smgr(void);
*** a/src/include/storage/bufmgr.h
--- b/src/include/storage/bufmgr.h
***************
*** 190,195 **** extern BlockNumber RelationGetNumberOfBlocksInFork(Relation relation,
--- 190,197 ----
  								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);
*** a/src/include/utils/rel.h
--- b/src/include/utils/rel.h
***************
*** 216,221 **** typedef struct RelationData
--- 216,229 ----
  
  	/* use "struct" here to avoid needing to include pgstat.h: */
  	struct PgStat_TableStatus *pgstat_info; /* statistics collection area */
+ 
+ 	/*
+ 	 * no_pending_sync is true if this relation is known not to have pending
+ 	 * syncs.  Elsewise searching for registered sync is required if
+ 	 * pending_sync is NULL.
+ 	 */
+ 	bool				   no_pending_sync;
+ 	struct PendingRelSync *pending_sync;
  } RelationData;