public inbox for [email protected]  
help / color / mirror / Atom feed
[PATCH v1 1/2] WIP: local bulk allocation
2+ messages / 2 participants
[nested] [flat]

* [PATCH v1 1/2] WIP: local bulk allocation
@ 2020-12-31 08:11  Luc Vlaming <[email protected]>
  0 siblings, 0 replies; 2+ messages in thread

From: Luc Vlaming @ 2020-12-31 08:11 UTC (permalink / raw)

---
 src/backend/access/heap/heapam.c      |  12 ++
 src/backend/access/heap/hio.c         | 178 ++++++--------------------
 src/backend/access/heap/rewriteheap.c |   2 +-
 src/backend/storage/buffer/bufmgr.c   |  60 ++++++++-
 src/backend/storage/file/buffile.c    |   3 +-
 src/backend/storage/file/fd.c         |  16 ++-
 src/backend/storage/smgr/md.c         |  56 ++++----
 src/backend/storage/smgr/smgr.c       |  15 ++-
 src/include/access/hio.h              |   4 +
 src/include/storage/bufmgr.h          |   2 +
 src/include/storage/fd.h              |   2 +-
 src/include/storage/md.h              |   2 +-
 src/include/storage/smgr.h            |   2 +
 13 files changed, 183 insertions(+), 171 deletions(-)

diff --git a/src/backend/access/heap/heapam.c b/src/backend/access/heap/heapam.c
index 26c2006f23..0a923c1ffd 100644
--- a/src/backend/access/heap/heapam.c
+++ b/src/backend/access/heap/heapam.c
@@ -1810,6 +1810,10 @@ GetBulkInsertState(void)
 	bistate = (BulkInsertState) palloc(sizeof(BulkInsertStateData));
 	bistate->strategy = GetAccessStrategy(BAS_BULKWRITE);
 	bistate->current_buf = InvalidBuffer;
+	bistate->local_buffers_idx = BULK_INSERT_BATCH_SIZE;
+	for (int i=0; i<BULK_INSERT_BATCH_SIZE; ++i)
+		bistate->local_buffers[i] = InvalidBuffer;
+	bistate->empty_buffer = palloc0(BULK_INSERT_BATCH_SIZE * BLCKSZ);
 	return bistate;
 }
 
@@ -1821,7 +1825,15 @@ FreeBulkInsertState(BulkInsertState bistate)
 {
 	if (bistate->current_buf != InvalidBuffer)
 		ReleaseBuffer(bistate->current_buf);
+	for (int i=bistate->local_buffers_idx; i<BULK_INSERT_BATCH_SIZE; ++i)
+		if (bistate->local_buffers[i] != InvalidBuffer)
+		{
+			// FSM?
+			//LockBuffer(bistate->local_buffers[i], BUFFER_LOCK_UNLOCK);
+			ReleaseBuffer(bistate->local_buffers[i]);
+		}
 	FreeAccessStrategy(bistate->strategy);
+	pfree(bistate->empty_buffer);
 	pfree(bistate);
 }
 
diff --git a/src/backend/access/heap/hio.c b/src/backend/access/heap/hio.c
index ca357410a2..9111badd2f 100644
--- a/src/backend/access/heap/hio.c
+++ b/src/backend/access/heap/hio.c
@@ -24,6 +24,7 @@
 #include "storage/lmgr.h"
 #include "storage/smgr.h"
 
+#include "miscadmin.h"
 
 /*
  * RelationPutHeapTuple - place tuple at specified page
@@ -118,9 +119,19 @@ ReadBufferBI(Relation relation, BlockNumber targetBlock,
 		bistate->current_buf = InvalidBuffer;
 	}
 
-	/* Perform a read using the buffer strategy */
-	buffer = ReadBufferExtended(relation, MAIN_FORKNUM, targetBlock,
-								mode, bistate->strategy);
+	if (targetBlock == P_NEW && mode == RBM_ZERO_AND_LOCK && bistate->local_buffers_idx < BULK_INSERT_BATCH_SIZE)
+	{
+		/* If we have a local buffer remaining, use that */
+		buffer = bistate->local_buffers[bistate->local_buffers_idx++];
+		LockBuffer(buffer, BUFFER_LOCK_EXCLUSIVE);
+		Assert(buffer != InvalidBuffer);
+	}
+	else
+	{
+		/* Perform a read using the buffer strategy */
+		buffer = ReadBufferExtended(relation, MAIN_FORKNUM, targetBlock,
+									mode, bistate->strategy);
+	}
 
 	/* Save the selected block as target for future inserts */
 	IncrBufferRefCount(buffer);
@@ -187,90 +198,6 @@ GetVisibilityMapPins(Relation relation, Buffer buffer1, Buffer buffer2,
 	}
 }
 
-/*
- * Extend a relation by multiple blocks to avoid future contention on the
- * relation extension lock.  Our goal is to pre-extend the relation by an
- * amount which ramps up as the degree of contention ramps up, but limiting
- * the result to some sane overall value.
- */
-static void
-RelationAddExtraBlocks(Relation relation, BulkInsertState bistate)
-{
-	BlockNumber blockNum,
-				firstBlock = InvalidBlockNumber;
-	int			extraBlocks;
-	int			lockWaiters;
-
-	/* Use the length of the lock wait queue to judge how much to extend. */
-	lockWaiters = RelationExtensionLockWaiterCount(relation);
-	if (lockWaiters <= 0)
-		return;
-
-	/*
-	 * It might seem like multiplying the number of lock waiters by as much as
-	 * 20 is too aggressive, but benchmarking revealed that smaller numbers
-	 * were insufficient.  512 is just an arbitrary cap to prevent
-	 * pathological results.
-	 */
-	extraBlocks = Min(512, lockWaiters * 20);
-
-	do
-	{
-		Buffer		buffer;
-		Page		page;
-		Size		freespace;
-
-		/*
-		 * Extend by one page.  This should generally match the main-line
-		 * extension code in RelationGetBufferForTuple, except that we hold
-		 * the relation extension lock throughout, and we don't immediately
-		 * initialize the page (see below).
-		 */
-		buffer = ReadBufferBI(relation, P_NEW, RBM_ZERO_AND_LOCK, bistate);
-		page = BufferGetPage(buffer);
-
-		if (!PageIsNew(page))
-			elog(ERROR, "page %u of relation \"%s\" should be empty but is not",
-				 BufferGetBlockNumber(buffer),
-				 RelationGetRelationName(relation));
-
-		/*
-		 * Add the page to the FSM without initializing. If we were to
-		 * initialize here, the page would potentially get flushed out to disk
-		 * before we add any useful content. There's no guarantee that that'd
-		 * happen before a potential crash, so we need to deal with
-		 * uninitialized pages anyway, thus avoid the potential for
-		 * unnecessary writes.
-		 */
-
-		/* we'll need this info below */
-		blockNum = BufferGetBlockNumber(buffer);
-		freespace = BufferGetPageSize(buffer) - SizeOfPageHeaderData;
-
-		UnlockReleaseBuffer(buffer);
-
-		/* Remember first block number thus added. */
-		if (firstBlock == InvalidBlockNumber)
-			firstBlock = blockNum;
-
-		/*
-		 * Immediately update the bottom level of the FSM.  This has a good
-		 * chance of making this page visible to other concurrently inserting
-		 * backends, and we want that to happen without delay.
-		 */
-		RecordPageWithFreeSpace(relation, blockNum, freespace);
-	}
-	while (--extraBlocks > 0);
-
-	/*
-	 * Updating the upper levels of the free space map is too expensive to do
-	 * for every block, but it's worth doing once at the end to make sure that
-	 * subsequent insertion activity sees all of those nifty free pages we
-	 * just inserted.
-	 */
-	FreeSpaceMapVacuumRange(relation, firstBlock, blockNum + 1);
-}
-
 /*
  * RelationGetBufferForTuple
  *
@@ -333,14 +260,14 @@ RelationGetBufferForTuple(Relation relation, Size len,
 						  BulkInsertState bistate,
 						  Buffer *vmbuffer, Buffer *vmbuffer_other)
 {
-	bool		use_fsm = !(options & HEAP_INSERT_SKIP_FSM);
+	bool		use_fsm = false; //!(options & HEAP_INSERT_SKIP_FSM);
 	Buffer		buffer = InvalidBuffer;
 	Page		page;
 	Size		pageFreeSpace = 0,
 				saveFreeSpace = 0;
 	BlockNumber targetBlock,
 				otherBlock;
-	bool		needLock;
+	bool		needLock = false;
 
 	len = MAXALIGN(len);		/* be conservative */
 
@@ -396,19 +323,6 @@ RelationGetBufferForTuple(Relation relation, Size len,
 		 * target.
 		 */
 		targetBlock = GetPageWithFreeSpace(relation, len + saveFreeSpace);
-
-		/*
-		 * If the FSM knows nothing of the rel, try the last page before we
-		 * give up and extend.  This avoids one-tuple-per-page syndrome during
-		 * bootstrapping or in a recently-started system.
-		 */
-		if (targetBlock == InvalidBlockNumber)
-		{
-			BlockNumber nblocks = RelationGetNumberOfBlocks(relation);
-
-			if (nblocks > 0)
-				targetBlock = nblocks - 1;
-		}
 	}
 
 loop:
@@ -541,37 +455,36 @@ loop:
 		 * Update FSM as to condition of this page, and ask for another page
 		 * to try.
 		 */
-		targetBlock = RecordAndGetPageWithFreeSpace(relation,
+		targetBlock = InvalidBlockNumber; 
+		/*RecordAndGetPageWithFreeSpace(relation,
 													targetBlock,
 													pageFreeSpace,
-													len + saveFreeSpace);
+													len + saveFreeSpace);*/
 	}
 
-	/*
-	 * Have to extend the relation.
-	 *
-	 * We have to use a lock to ensure no one else is extending the rel at the
-	 * same time, else we will both try to initialize the same new page.  We
-	 * can skip locking for new or temp relations, however, since no one else
-	 * could be accessing them.
-	 */
-	needLock = !RELATION_IS_LOCAL(relation);
 
-	/*
-	 * If we need the lock but are not able to acquire it immediately, we'll
-	 * consider extending the relation by multiple blocks at a time to manage
-	 * contention on the relation extension lock.  However, this only makes
-	 * sense if we're using the FSM; otherwise, there's no point.
-	 */
-	if (needLock)
+	if (bistate)
 	{
-		if (!use_fsm)
+		if (bistate->local_buffers_idx == BULK_INSERT_BATCH_SIZE)
+			/* Ran out of local buffers for the bulk state, allocate some more */
+			ReadBufferExtendBulk(relation, bistate);
+	}
+	else
+	{
+		/*
+		 * Have to extend the relation.
+		 *
+		 * We have to use a lock to ensure no one else is extending the rel at the
+		 * same time, else we will both try to initialize the same new page.  We
+		 * can skip locking for new or temp relations, however, since no one else
+		 * could be accessing them.
+		 */
+		needLock = !RELATION_IS_LOCAL(relation);
+
+		if (needLock)
 			LockRelationForExtension(relation, ExclusiveLock);
-		else if (!ConditionalLockRelationForExtension(relation, ExclusiveLock))
+		if (use_fsm)
 		{
-			/* Couldn't get the lock immediately; wait for it. */
-			LockRelationForExtension(relation, ExclusiveLock);
-
 			/*
 			 * Check if some other backend has extended a block for us while
 			 * we were waiting on the lock.
@@ -587,21 +500,9 @@ loop:
 				UnlockRelationForExtension(relation, ExclusiveLock);
 				goto loop;
 			}
-
-			/* Time to bulk-extend. */
-			RelationAddExtraBlocks(relation, bistate);
 		}
 	}
 
-	/*
-	 * In addition to whatever extension we performed above, we always add at
-	 * least one block to satisfy our own request.
-	 *
-	 * XXX This does an lseek - rather expensive - but at the moment it is the
-	 * only way to accurately determine how many blocks are in a relation.  Is
-	 * it worth keeping an accurate file length in shared memory someplace,
-	 * rather than relying on the kernel to do it for us?
-	 */
 	buffer = ReadBufferBI(relation, P_NEW, RBM_ZERO_AND_LOCK, bistate);
 
 	/*
@@ -612,7 +513,8 @@ loop:
 	page = BufferGetPage(buffer);
 
 	if (!PageIsNew(page))
-		elog(ERROR, "page %u of relation \"%s\" should be empty but is not",
+		elog(ERROR, "%d page %u of relation \"%s\" should be empty but is not",
+			 MyProcPid,
 			 BufferGetBlockNumber(buffer),
 			 RelationGetRelationName(relation));
 
diff --git a/src/backend/access/heap/rewriteheap.c b/src/backend/access/heap/rewriteheap.c
index 65942cc428..e1f2ac8faa 100644
--- a/src/backend/access/heap/rewriteheap.c
+++ b/src/backend/access/heap/rewriteheap.c
@@ -912,7 +912,7 @@ logical_heap_rewrite_flush_mappings(RewriteState state)
 		 * check the above "Logical rewrite support" comment for reasoning.
 		 */
 		written = FileWrite(src->vfd, waldata_start, len, src->off,
-							WAIT_EVENT_LOGICAL_REWRITE_WRITE);
+							WAIT_EVENT_LOGICAL_REWRITE_WRITE, false);
 		if (written != len)
 			ereport(ERROR,
 					(errcode_for_file_access(),
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index c5e8707151..d1ce88fee5 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -34,6 +34,7 @@
 #include <unistd.h>
 
 #include "access/tableam.h"
+#include "access/hio.h"
 #include "access/xlog.h"
 #include "catalog/catalog.h"
 #include "catalog/storage.h"
@@ -47,6 +48,7 @@
 #include "storage/bufmgr.h"
 #include "storage/ipc.h"
 #include "storage/proc.h"
+#include "storage/lmgr.h"
 #include "storage/smgr.h"
 #include "storage/standby.h"
 #include "utils/memdebug.h"
@@ -824,8 +826,8 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 		bufBlock = isLocalBuf ? LocalBufHdrGetBlock(bufHdr) : BufHdrGetBlock(bufHdr);
 		if (!PageIsNew((Page) bufBlock))
 			ereport(ERROR,
-					(errmsg("unexpected data beyond EOF in block %u of relation %s",
-							blockNum, relpath(smgr->smgr_rnode, forkNum)),
+					(errmsg("%d unexpected data beyond EOF in block %u of relation %s",
+							MyProcPid, blockNum, relpath(smgr->smgr_rnode, forkNum)),
 					 errhint("This has been seen to occur with buggy kernels; consider updating your system.")));
 
 		/*
@@ -985,6 +987,60 @@ ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 	return BufferDescriptorGetBuffer(bufHdr);
 }
 
+void
+ReadBufferExtendBulk(Relation reln, struct BulkInsertStateData* bistate)
+{
+	char relpersistence = reln->rd_rel->relpersistence;
+	SMgrRelation smgr;
+	BufferDesc *bufHdr;
+	BlockNumber	firstBlock,
+				blockNum;
+	Block		bufBlock;
+	bool		found;
+
+	/* Open it at the smgr level if not already done */
+	RelationOpenSmgr(reln);
+	smgr = reln->rd_smgr;
+
+	if (SmgrIsTemp(smgr))
+		elog(ERROR, "Bulk extend for temporary tables not supported");
+
+	LockRelationForExtension(reln, ExclusiveLock);
+	firstBlock = smgrnblocks(smgr, MAIN_FORKNUM);
+	smgrextend_count(smgr, MAIN_FORKNUM, firstBlock, bistate->empty_buffer, false, BULK_INSERT_BATCH_SIZE);
+	UnlockRelationForExtension(reln, ExclusiveLock);
+
+	for (int i=0; i<BULK_INSERT_BATCH_SIZE; ++i)
+	{
+		blockNum = firstBlock + i;
+
+		pgstat_count_buffer_read(reln);
+		/* Make sure we will have room to remember the buffer pin */
+		ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
+		bufHdr = BufferAlloc(smgr, relpersistence, MAIN_FORKNUM, blockNum,
+							 bistate->strategy, &found);
+		pgBufferUsage.shared_blks_written++;
+
+		Assert(!found);
+
+		bufBlock = BufHdrGetBlock(bufHdr);
+
+		/* new buffers are zero-filled */
+		MemSet((char *) bufBlock, 0, BLCKSZ);
+
+		/* Set BM_VALID, terminate IO, and wake up any waiters */
+		TerminateBufferIO(bufHdr, false, BM_VALID);
+		bistate->local_buffers[i] = BufferDescriptorGetBuffer(bufHdr);
+	}
+
+	bistate->local_buffers_idx = 0;
+
+	VacuumPageMiss += BULK_INSERT_BATCH_SIZE;
+	if (VacuumCostActive)
+		VacuumCostBalance += VacuumCostPageMiss * BULK_INSERT_BATCH_SIZE;
+}
+
+
 /*
  * BufferAlloc -- subroutine for ReadBuffer.  Handles lookup of a shared
  *		buffer.  If no buffer exists already, selects a replacement
diff --git a/src/backend/storage/file/buffile.c b/src/backend/storage/file/buffile.c
index d581f96eda..5055674090 100644
--- a/src/backend/storage/file/buffile.c
+++ b/src/backend/storage/file/buffile.c
@@ -499,7 +499,8 @@ BufFileDumpBuffer(BufFile *file)
 								 file->buffer.data + wpos,
 								 bytestowrite,
 								 file->curOffset,
-								 WAIT_EVENT_BUFFILE_WRITE);
+								 WAIT_EVENT_BUFFILE_WRITE,
+								 false);
 		if (bytestowrite <= 0)
 			ereport(ERROR,
 					(errcode_for_file_access(),
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index f07b5325aa..0f11bc98c4 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2058,7 +2058,7 @@ retry:
 
 int
 FileWrite(File file, char *buffer, int amount, off_t offset,
-		  uint32 wait_event_info)
+		  uint32 wait_event_info, bool is_empty)
 {
 	int			returnCode;
 	Vfd		   *vfdP;
@@ -2104,7 +2104,21 @@ FileWrite(File file, char *buffer, int amount, off_t offset,
 retry:
 	errno = 0;
 	pgstat_report_wait_start(wait_event_info);
+#if defined(HAVE_POSIX_FALLOCATE) && defined(__linux__)
+	if (is_empty)
+	{
+		do
+		{
+			returnCode = ftruncate(VfdCache[file].fd, offset + amount);
+			//returnCode = fallocate(VfdCache[file].fd, 0, offset, amount);
+		} while (unlikely(returnCode == EINTR && !(ProcDiePending || QueryCancelPending)));
+		returnCode = amount;
+	}
+	else
+		returnCode = pg_pwrite(VfdCache[file].fd, buffer, amount, offset);
+#else
 	returnCode = pg_pwrite(VfdCache[file].fd, buffer, amount, offset);
+#endif
 	pgstat_report_wait_end();
 
 	/* if write didn't set errno, assume problem is no disk space */
diff --git a/src/backend/storage/smgr/md.c b/src/backend/storage/smgr/md.c
index 9889ad6ad8..741da34656 100644
--- a/src/backend/storage/smgr/md.c
+++ b/src/backend/storage/smgr/md.c
@@ -412,10 +412,13 @@ mdunlinkfork(RelFileNodeBackend rnode, ForkNumber forkNum, bool isRedo)
  */
 void
 mdextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
-		 char *buffer, bool skipFsync)
+		 char *buffer, bool skipFsync, int count)
 {
 	off_t		seekpos;
+	BlockNumber block_in_seg;
+	BlockNumber count_in_seg;
 	int			nbytes;
+	int			write_size;
 	MdfdVec    *v;
 
 	/* This assert is too expensive to have on normally ... */
@@ -435,31 +438,40 @@ mdextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 						relpath(reln->smgr_rnode, forknum),
 						InvalidBlockNumber)));
 
-	v = _mdfd_getseg(reln, forknum, blocknum, skipFsync, EXTENSION_CREATE);
+	do {
+		v = _mdfd_getseg(reln, forknum, blocknum, skipFsync, EXTENSION_CREATE);
 
-	seekpos = (off_t) BLCKSZ * (blocknum % ((BlockNumber) RELSEG_SIZE));
+		block_in_seg = (blocknum % ((BlockNumber) RELSEG_SIZE));
+		count_in_seg = Min((RELSEG_SIZE - block_in_seg), count);
+		seekpos = (off_t) BLCKSZ * block_in_seg;
+		write_size = BLCKSZ * count_in_seg;
 
-	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
+		Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
 
-	if ((nbytes = FileWrite(v->mdfd_vfd, buffer, BLCKSZ, seekpos, WAIT_EVENT_DATA_FILE_EXTEND)) != BLCKSZ)
-	{
-		if (nbytes < 0)
+		if ((nbytes = FileWrite(v->mdfd_vfd, buffer, write_size, seekpos, WAIT_EVENT_DATA_FILE_EXTEND, true)) != write_size)
+		{
+			if (nbytes < 0)
+				ereport(ERROR,
+						(errcode_for_file_access(),
+						 errmsg("could not extend file \"%s\": %m",
+								FilePathName(v->mdfd_vfd)),
+						 errhint("Check free disk space.")));
+			/* short write: complain appropriately */
 			ereport(ERROR,
-					(errcode_for_file_access(),
-					 errmsg("could not extend file \"%s\": %m",
-							FilePathName(v->mdfd_vfd)),
+					(errcode(ERRCODE_DISK_FULL),
+					 errmsg("could not extend file \"%s\": wrote only %d of %d bytes at block %u",
+							FilePathName(v->mdfd_vfd),
+							nbytes, write_size, blocknum),
 					 errhint("Check free disk space.")));
-		/* short write: complain appropriately */
-		ereport(ERROR,
-				(errcode(ERRCODE_DISK_FULL),
-				 errmsg("could not extend file \"%s\": wrote only %d of %d bytes at block %u",
-						FilePathName(v->mdfd_vfd),
-						nbytes, BLCKSZ, blocknum),
-				 errhint("Check free disk space.")));
-	}
+		}
 
-	if (!skipFsync && !SmgrIsTemp(reln))
-		register_dirty_segment(reln, forknum, v);
+		if (!skipFsync && !SmgrIsTemp(reln))
+			register_dirty_segment(reln, forknum, v);
+
+		count -= count_in_seg;
+		buffer += write_size;
+		blocknum += count_in_seg;
+	} while (count > 0);
 
 	Assert(_mdnblocks(reln, forknum, v) <= ((BlockNumber) RELSEG_SIZE));
 }
@@ -719,7 +731,7 @@ mdwrite(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 
 	Assert(seekpos < (off_t) BLCKSZ * RELSEG_SIZE);
 
-	nbytes = FileWrite(v->mdfd_vfd, buffer, BLCKSZ, seekpos, WAIT_EVENT_DATA_FILE_WRITE);
+	nbytes = FileWrite(v->mdfd_vfd, buffer, BLCKSZ, seekpos, WAIT_EVENT_DATA_FILE_WRITE, false);
 
 	TRACE_POSTGRESQL_SMGR_MD_WRITE_DONE(forknum, blocknum,
 										reln->smgr_rnode.node.spcNode,
@@ -1254,7 +1266,7 @@ _mdfd_getseg(SMgrRelation reln, ForkNumber forknum, BlockNumber blkno,
 
 				mdextend(reln, forknum,
 						 nextsegno * ((BlockNumber) RELSEG_SIZE) - 1,
-						 zerobuf, skipFsync);
+						 zerobuf, skipFsync, 1);
 				pfree(zerobuf);
 			}
 			flags = O_CREAT;
diff --git a/src/backend/storage/smgr/smgr.c b/src/backend/storage/smgr/smgr.c
index 072bdd118f..1ad6f5d7de 100644
--- a/src/backend/storage/smgr/smgr.c
+++ b/src/backend/storage/smgr/smgr.c
@@ -49,7 +49,7 @@ typedef struct f_smgr
 	void		(*smgr_unlink) (RelFileNodeBackend rnode, ForkNumber forknum,
 								bool isRedo);
 	void		(*smgr_extend) (SMgrRelation reln, ForkNumber forknum,
-								BlockNumber blocknum, char *buffer, bool skipFsync);
+								BlockNumber blocknum, char *buffer, bool skipFsync, int count);
 	bool		(*smgr_prefetch) (SMgrRelation reln, ForkNumber forknum,
 								  BlockNumber blocknum);
 	void		(*smgr_read) (SMgrRelation reln, ForkNumber forknum,
@@ -461,17 +461,24 @@ smgrdounlinkall(SMgrRelation *rels, int nrels, bool isRedo)
 void
 smgrextend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
 		   char *buffer, bool skipFsync)
+{
+	return smgrextend_count(reln, forknum, blocknum, buffer, skipFsync, 1);
+}
+
+void
+smgrextend_count(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+		   char *buffer, bool skipFsync, int count)
 {
 	smgrsw[reln->smgr_which].smgr_extend(reln, forknum, blocknum,
-										 buffer, skipFsync);
+										 buffer, skipFsync, count);
 
 	/*
-	 * Normally we expect this to increase nblocks by one, but if the cached
+	 * Normally we expect this to increase nblocks by count, but if the cached
 	 * value isn't as expected, just invalidate it so the next call asks the
 	 * kernel.
 	 */
 	if (reln->smgr_cached_nblocks[forknum] == blocknum)
-		reln->smgr_cached_nblocks[forknum] = blocknum + 1;
+		reln->smgr_cached_nblocks[forknum] = blocknum + count;
 	else
 		reln->smgr_cached_nblocks[forknum] = InvalidBlockNumber;
 }
diff --git a/src/include/access/hio.h b/src/include/access/hio.h
index f69a92521b..1d7b2181cb 100644
--- a/src/include/access/hio.h
+++ b/src/include/access/hio.h
@@ -26,10 +26,14 @@
  *
  * "typedef struct BulkInsertStateData *BulkInsertState" is in heapam.h
  */
+#define BULK_INSERT_BATCH_SIZE 128
 typedef struct BulkInsertStateData
 {
 	BufferAccessStrategy strategy;	/* our BULKWRITE strategy object */
 	Buffer		current_buf;	/* current insertion target page */
+	int 		local_buffers_idx;
+	Buffer 		local_buffers[BULK_INSERT_BATCH_SIZE];
+	char* 		empty_buffer;
 } BulkInsertStateData;
 
 
diff --git a/src/include/storage/bufmgr.h b/src/include/storage/bufmgr.h
index ee91b8fa26..8a37dd3ec0 100644
--- a/src/include/storage/bufmgr.h
+++ b/src/include/storage/bufmgr.h
@@ -60,6 +60,7 @@ struct WritebackContext;
 
 /* forward declared, to avoid including smgr.h here */
 struct SMgrRelationData;
+struct BulkInsertStateData;
 
 /* in globals.c ... this duplicates miscadmin.h */
 extern PGDLLIMPORT int NBuffers;
@@ -180,6 +181,7 @@ extern Buffer ReadBuffer(Relation reln, BlockNumber blockNum);
 extern Buffer ReadBufferExtended(Relation reln, ForkNumber forkNum,
 								 BlockNumber blockNum, ReadBufferMode mode,
 								 BufferAccessStrategy strategy);
+void ReadBufferExtendBulk(Relation reln, struct BulkInsertStateData* bistate);
 extern Buffer ReadBufferWithoutRelcache(RelFileNode rnode,
 										ForkNumber forkNum, BlockNumber blockNum,
 										ReadBufferMode mode, BufferAccessStrategy strategy);
diff --git a/src/include/storage/fd.h b/src/include/storage/fd.h
index 4e1cc12e23..cba2139de3 100644
--- a/src/include/storage/fd.h
+++ b/src/include/storage/fd.h
@@ -82,7 +82,7 @@ extern File OpenTemporaryFile(bool interXact);
 extern void FileClose(File file);
 extern int	FilePrefetch(File file, off_t offset, int amount, uint32 wait_event_info);
 extern int	FileRead(File file, char *buffer, int amount, off_t offset, uint32 wait_event_info);
-extern int	FileWrite(File file, char *buffer, int amount, off_t offset, uint32 wait_event_info);
+extern int	FileWrite(File file, char *buffer, int amount, off_t offset, uint32 wait_event_info, bool is_empty);
 extern int	FileSync(File file, uint32 wait_event_info);
 extern off_t FileSize(File file);
 extern int	FileTruncate(File file, off_t offset, uint32 wait_event_info);
diff --git a/src/include/storage/md.h b/src/include/storage/md.h
index 07fd1bb7d0..7df8847737 100644
--- a/src/include/storage/md.h
+++ b/src/include/storage/md.h
@@ -27,7 +27,7 @@ extern void mdcreate(SMgrRelation reln, ForkNumber forknum, bool isRedo);
 extern bool mdexists(SMgrRelation reln, ForkNumber forknum);
 extern void mdunlink(RelFileNodeBackend rnode, ForkNumber forknum, bool isRedo);
 extern void mdextend(SMgrRelation reln, ForkNumber forknum,
-					 BlockNumber blocknum, char *buffer, bool skipFsync);
+					 BlockNumber blocknum, char *buffer, bool skipFsync, int count);
 extern bool mdprefetch(SMgrRelation reln, ForkNumber forknum,
 					   BlockNumber blocknum);
 extern void mdread(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
diff --git a/src/include/storage/smgr.h b/src/include/storage/smgr.h
index f28a842401..5a6c619cf6 100644
--- a/src/include/storage/smgr.h
+++ b/src/include/storage/smgr.h
@@ -90,6 +90,8 @@ extern void smgrdosyncall(SMgrRelation *rels, int nrels);
 extern void smgrdounlinkall(SMgrRelation *rels, int nrels, bool isRedo);
 extern void smgrextend(SMgrRelation reln, ForkNumber forknum,
 					   BlockNumber blocknum, char *buffer, bool skipFsync);
+extern void smgrextend_count(SMgrRelation reln, ForkNumber forknum,
+					   BlockNumber blocknum, char *buffer, bool skipFsync, int count);
 extern bool smgrprefetch(SMgrRelation reln, ForkNumber forknum,
 						 BlockNumber blocknum);
 extern void smgrread(SMgrRelation reln, ForkNumber forknum,
-- 
2.25.1


--------------065E078E0383D7C686F94091
Content-Type: text/x-patch; charset=UTF-8;
 name="v1-0002-WIP-buffer-alloc-specialized-for-relation-extensi.patch"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
 filename*0="v1-0002-WIP-buffer-alloc-specialized-for-relation-extensi.pa";
 filename*1="tch"



^ permalink  raw  reply  [nested|flat] 2+ messages in thread

* Re: ALTER TABLE SET ACCESS METHOD on partitioned tables
@ 2024-03-31 09:00  Alexander Lakhin <[email protected]>
  0 siblings, 0 replies; 2+ messages in thread

From: Alexander Lakhin @ 2024-03-31 09:00 UTC (permalink / raw)
  To: Alvaro Herrera <[email protected]>; Justin Pryzby <[email protected]>; +Cc: Michael Paquier <[email protected]>; Peter Eisentraut <[email protected]>; Soumyadeep Chakraborty <[email protected]>; Zhihong Yu <[email protected]>; pgsql-hackers; Ashwin Agrawal <[email protected]>; [email protected]

Hello Alvaro,

28.03.2024 18:58, Alvaro Herrera wrote:
> Grumble.  I don't like initialization at declare time, so far from the
> code that depends on the value.  But the alternative would have been to
> assign right where this blocks starts, an additional line.  I pushed it
> like you had it.

I've stumbled upon a test failure caused by the test query added in that
commit:
--- .../src/test/regress/expected/create_am.out      2024-03-28 12:14:11.700764888 -0400
+++ .../src/test/recovery/tmp_check/results/create_am.out 2024-03-31 03:10:28.172244122 -0400
@@ -549,7 +549,10 @@
  ERROR:  access method "btree" is not of type TABLE
  -- Other weird invalid cases that cause problems
  CREATE FOREIGN TABLE fp PARTITION OF pg_am DEFAULT SERVER x;
-ERROR:  "pg_am" is not partitioned
+ERROR:  deadlock detected
+DETAIL:  Process 3076180 waits for AccessShareLock on relation 1259 of database 16386; blocked by process 3076181.
+Process 3076181 waits for AccessShareLock on relation 2601 of database 16386; blocked by process 3076180.
+HINT:  See server log for query details.
  -- Drop table access method, which fails as objects depends on it
  DROP ACCESS METHOD heap2;
  ERROR:  cannot drop access method heap2 because other objects depend on it

027_stream_regress_primary.log contains:
2024-03-31 03:10:26.728 EDT [3076181] pg_regress/vacuum LOG: statement: VACUUM FULL pg_class;
...
2024-03-31 03:10:26.797 EDT [3076180] pg_regress/create_am LOG: statement: CREATE FOREIGN TABLE fp PARTITION OF pg_am 
DEFAULT SERVER x;
...
2024-03-31 03:10:28.183 EDT [3076181] pg_regress/vacuum LOG: statement: VACUUM FULL pg_database;

This simple demo confirms the issue:
for ((i=1;i<=20;i++)); do
echo "iteration $i"
echo "VACUUM FULL pg_class;" | psql >psql-1.log &
echo "CREATE FOREIGN TABLE fp PARTITION OF pg_am DEFAULT SERVER x;" | psql >psql-2.log &
wait
done

...
iteration 15
ERROR:  "pg_am" is not partitioned
iteration 16
ERROR:  deadlock detected
DETAIL:  Process 2556377 waits for AccessShareLock on relation 1259 of database 16384; blocked by process 2556378.
Process 2556378 waits for AccessShareLock on relation 2601 of database 16384; blocked by process 2556377.
HINT:  See server log for query details.
...

Best regards,
Alexander






^ permalink  raw  reply  [nested|flat] 2+ messages in thread


end of thread, other threads:[~2024-03-31 09:00 UTC | newest]

Thread overview: 2+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2020-12-31 08:11 [PATCH v1 1/2] WIP: local bulk allocation Luc Vlaming <[email protected]>
2024-03-31 09:00 Re: ALTER TABLE SET ACCESS METHOD on partitioned tables Alexander Lakhin <[email protected]>

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox