[CF 5616] v4 - Extensible storage manager API

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/5616

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/3217233f-6a86-4792-a055-606134c82d15@proxel.se
Author(s): Andreas Karlsson, Anastasia Lubennikova, Matthias van de Meent, Tristan Partin

Author: Commitfest Bot

contrib/Makefile

@@ -19,6 +19,7 @@ SUBDIRS = \
 		dict_int	\
 		dict_xsyn	\
 		earthdistance	\
+		fsync_checker	\
 		file_fdw	\
 		fuzzystrmatch	\
 		hstore		\

contrib/fsync_checker/fsync_checker.control

@@ -0,0 +1,5 @@
+# fsync_checker extension
+comment = 'SMGR extension for checking volatile writes'
+default_version = '1.0'
+module_pathname = '$libdir/fsync_checker'
+relocatable = true

contrib/fsync_checker/fsync_checker_smgr.c

@@ -0,0 +1,248 @@
+#include "postgres.h"
+
+#include "access/xlog.h"
+#include "fmgr.h"
+#include "miscadmin.h"
+#include "storage/ipc.h"
+#include "storage/lwlock.h"
+#include "storage/shmem.h"
+#include "storage/smgr.h"
+#include "storage/md.h"
+#include "utils/hsearch.h"
+
+PG_MODULE_MAGIC;
+
+typedef struct
+{
+	RelFileLocator locator;
+	ForkNumber	forknum;
+} VolatileRelnKey;
+
+typedef struct
+{
+	VolatileRelnKey key;
+	XLogRecPtr	lsn;
+} VolatileRelnEntry;
+
+void		_PG_init(void);
+
+static void fsync_checker_extend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+								 const void *buffer, bool skipFsync, SmgrChainIndex chain_index);
+static void fsync_checker_immedsync(SMgrRelation reln, ForkNumber forknum, SmgrChainIndex chain_index);
+static void fsync_checker_writev(SMgrRelation reln, ForkNumber forknum,
+								 BlockNumber blocknum, const void **buffers,
+								 BlockNumber nblocks, bool skipFsync, SmgrChainIndex chain_index);
+static void fsync_checker_writeback(SMgrRelation reln, ForkNumber forknum,
+									BlockNumber blocknum, BlockNumber nblocks, SmgrChainIndex chain_index);
+static void fsync_checker_zeroextend(SMgrRelation reln, ForkNumber forknum,
+									 BlockNumber blocknum, int nblocks, bool skipFsync, SmgrChainIndex chain_index);
+
+static void fsync_checker_checkpoint_create(const CheckPoint *checkPoint);
+static void fsync_checker_shmem_request(void);
+static void fsync_checker_shmem_startup(void);
+
+static void add_reln(SMgrRelation reln, ForkNumber forknum);
+static void remove_reln(SMgrRelation reln, ForkNumber forknum);
+
+static SMgrId fsync_checker_smgr_id;
+static const struct f_smgr fsync_checker_smgr = {
+	.name = "fsync_checker",
+	.chain_position = SMGR_CHAIN_MODIFIER,
+	.smgr_init = NULL,
+	.smgr_shutdown = NULL,
+	.smgr_open = NULL,
+	.smgr_close = NULL,
+	.smgr_create = NULL,
+	.smgr_exists = NULL,
+	.smgr_unlink = NULL,
+	.smgr_extend = fsync_checker_extend,
+	.smgr_zeroextend = fsync_checker_zeroextend,
+	.smgr_prefetch = NULL,
+	.smgr_maxcombine = NULL,
+	.smgr_readv = NULL,
+	.smgr_writev = fsync_checker_writev,
+	.smgr_writeback = fsync_checker_writeback,
+	.smgr_nblocks = NULL,
+	.smgr_truncate = NULL,
+	.smgr_immedsync = fsync_checker_immedsync,
+	.smgr_registersync = NULL,
+};
+
+static HTAB *volatile_relns;
+static LWLock *volatile_relns_lock;
+static shmem_request_hook_type prev_shmem_request_hook;
+static shmem_startup_hook_type prev_shmem_startup_hook;
+static checkpoint_create_hook_type prev_checkpoint_create_hook;
+
+void
+_PG_init(void)
+{
+	prev_checkpoint_create_hook = checkpoint_create_hook;
+	checkpoint_create_hook = fsync_checker_checkpoint_create;
+
+	prev_shmem_request_hook = shmem_request_hook;
+	shmem_request_hook = fsync_checker_shmem_request;
+
+	prev_shmem_startup_hook = shmem_startup_hook;
+	shmem_startup_hook = fsync_checker_shmem_startup;
+
+	/*
+	 * Relation size of 0 means we can just defer to md, but it would be nice
+	 * to just expose this functionality, so if I needed my own relation, I
+	 * could use MdSmgrRelation as the parent.
+	 */
+	fsync_checker_smgr_id = smgr_register(&fsync_checker_smgr, 0);
+}
+
+static void
+fsync_checker_checkpoint_create(const CheckPoint *checkPoint)
+{
+	long		num_entries;
+	HASH_SEQ_STATUS status;
+	VolatileRelnEntry *entry;
+
+	if (prev_checkpoint_create_hook)
+		prev_checkpoint_create_hook(checkPoint);
+
+	LWLockAcquire(volatile_relns_lock, LW_EXCLUSIVE);
+
+	hash_seq_init(&status, volatile_relns);
+
+	num_entries = hash_get_num_entries(volatile_relns);
+	elog(INFO, "Analyzing %ld volatile relations", num_entries);
+	while ((entry = hash_seq_search(&status)))
+	{
+		if (entry->lsn < checkPoint->redo)
+		{
+			RelPathStr	path;
+
+			path = relpathperm(entry->key.locator, entry->key.forknum);
+
+			elog(WARNING, "Relation not previously synced: %s", path.str);
+		}
+	}
+
+	LWLockRelease(volatile_relns_lock);
+}
+
+static void
+fsync_checker_shmem_request(void)
+{
+	if (prev_shmem_request_hook)
+		prev_shmem_request_hook();
+
+	RequestAddinShmemSpace(hash_estimate_size(1024, sizeof(VolatileRelnEntry)));
+	RequestNamedLWLockTranche("fsync_checker volatile relns lock", 1);
+}
+
+static void
+fsync_checker_shmem_startup(void)
+{
+	HASHCTL		ctl;
+
+	if (prev_shmem_startup_hook)
+		prev_shmem_startup_hook();
+
+	ctl.keysize = sizeof(VolatileRelnKey);
+	ctl.entrysize = sizeof(VolatileRelnEntry);
+	volatile_relns = NULL;
+	volatile_relns_lock = NULL;
+
+	/*
+	 * Create or attach to the shared memory state, including hash table
+	 */
+	LWLockAcquire(AddinShmemInitLock, LW_EXCLUSIVE);
+
+	volatile_relns = ShmemInitHash("fsync_checker volatile relns",
+								   1024, 1024, &ctl, HASH_BLOBS | HASH_ELEM);
+	volatile_relns_lock = &GetNamedLWLockTranche("fsync_checker volatile relns lock")->lock;
+
+	LWLockRelease(AddinShmemInitLock);
+}
+
+static void
+add_reln(SMgrRelation reln, ForkNumber forknum)
+{
+	bool		found;
+	XLogRecPtr	lsn;
+	VolatileRelnKey key;
+	VolatileRelnEntry *entry;
+
+	key.locator = reln->smgr_rlocator.locator;
+	key.forknum = forknum;
+
+	lsn = GetXLogWriteRecPtr();
+
+	LWLockAcquire(volatile_relns_lock, LW_EXCLUSIVE);
+
+	entry = hash_search(volatile_relns, &key, HASH_ENTER, &found);
+	if (!found)
+		entry->lsn = lsn;
+
+	LWLockRelease(volatile_relns_lock);
+}
+
+static void
+remove_reln(SMgrRelation reln, ForkNumber forknum)
+{
+	VolatileRelnKey key;
+
+	key.locator = reln->smgr_rlocator.locator;
+	key.forknum = forknum;
+
+	LWLockAcquire(volatile_relns_lock, LW_EXCLUSIVE);
+
+	hash_search(volatile_relns, &key, HASH_REMOVE, NULL);
+
+	LWLockRelease(volatile_relns_lock);
+}
+
+static void
+fsync_checker_extend(SMgrRelation reln, ForkNumber forknum, BlockNumber blocknum,
+					 const void *buffer, bool skipFsync, SmgrChainIndex chain_index)
+{
+	if (!SmgrIsTemp(reln) && !skipFsync)
+		add_reln(reln, forknum);
+
+	smgr_extend_next(reln, forknum, blocknum, buffer, skipFsync, chain_index + 1);
+}
+
+static void
+fsync_checker_immedsync(SMgrRelation reln, ForkNumber forknum, SmgrChainIndex chain_index)
+{
+	if (!SmgrIsTemp(reln))
+		remove_reln(reln, forknum);
+
+	smgr_immedsync_next(reln, forknum, chain_index + 1);
+}
+
+static void
+fsync_checker_writev(SMgrRelation reln, ForkNumber forknum,
+					 BlockNumber blocknum, const void **buffers,
+					 BlockNumber nblocks, bool skipFsync, SmgrChainIndex chain_index)
+{
+	if (!SmgrIsTemp(reln) && !skipFsync)
+		add_reln(reln, forknum);
+
+	smgr_writev_next(reln, forknum, blocknum, buffers, nblocks, skipFsync, chain_index + 1);
+}
+
+static void
+fsync_checker_writeback(SMgrRelation reln, ForkNumber forknum,
+						BlockNumber blocknum, BlockNumber nblocks, SmgrChainIndex chain_index)
+{
+	if (!SmgrIsTemp(reln))
+		remove_reln(reln, forknum);
+
+	smgr_writeback_next(reln, forknum, blocknum, nblocks, chain_index + 1);
+}
+
+static void
+fsync_checker_zeroextend(SMgrRelation reln, ForkNumber forknum,
+						 BlockNumber blocknum, int nblocks, bool skipFsync, SmgrChainIndex chain_index)
+{
+	if (!SmgrIsTemp(reln) && !skipFsync)
+		add_reln(reln, forknum);
+
+	smgr_zeroextend_next(reln, forknum, blocknum, nblocks, skipFsync, chain_index + 1);
+}

contrib/fsync_checker/meson.build

@@ -0,0 +1,22 @@
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+fsync_checker_sources = files(
+  'fsync_checker_smgr.c',
+)
+
+if host_system == 'windows'
+  fsync_checker_sources += rc_lib_gen.process(win32ver_rc, extra_args: [
+    '--NAME', 'fsync_checker',
+    '--FILEDESC', 'fsync_checker - SMGR extension for checking volatile relations',])
+endif
+
+fsync_checker = shared_module('fsync_checker',
+  fsync_checker_sources,
+  kwargs: contrib_mod_args,
+)
+contrib_targets += fsync_checker
+
+install_data(
+  'fsync_checker.control',
+  kwargs: contrib_data_args,
+)

contrib/meson.build

@@ -28,6 +28,7 @@ subdir('dict_int')
 subdir('dict_xsyn')
 subdir('earthdistance')
 subdir('file_fdw')
+subdir('fsync_checker')
 subdir('fuzzystrmatch')
 subdir('hstore')
 subdir('hstore_plperl')

src/backend/access/heap/heapam_handler.c

@@ -588,6 +588,8 @@ heapam_relation_set_new_filelocator(Relation rel,
 {
 	SMgrRelation srel;
 
+	RelFileLocator oldlocator = rel->rd_locator;
+
 	/*
 	 * Initialize to the minimum XID that could put tuples in the table. We
 	 * know that no xacts older than RecentXmin are still running, so that
@@ -605,7 +607,7 @@ heapam_relation_set_new_filelocator(Relation rel,
 	 */
 	*minmulti = GetOldestMultiXactId();
 
-	srel = RelationCreateStorage(*newrlocator, persistence, true);
+	srel = RelationCreateStorage(oldlocator, *newrlocator, persistence, true);
 
 	/*
 	 * If required, set up an init fork for an unlogged table so that it can
@@ -615,7 +617,7 @@ heapam_relation_set_new_filelocator(Relation rel,
 	{
 		Assert(rel->rd_rel->relkind == RELKIND_RELATION ||
 			   rel->rd_rel->relkind == RELKIND_TOASTVALUE);
-		smgrcreate(srel, INIT_FORKNUM, false);
+		smgrcreate(oldlocator, srel, INIT_FORKNUM, false);
 		log_smgrcreate(newrlocator, INIT_FORKNUM);
 	}
 
@@ -648,7 +650,7 @@ heapam_relation_copy_data(Relation rel, const RelFileLocator *newrlocator)
 	 * NOTE: any conflict in relfilenumber value will be caught in
 	 * RelationCreateStorage().
 	 */
-	dstrel = RelationCreateStorage(*newrlocator, rel->rd_rel->relpersistence, true);
+	dstrel = RelationCreateStorage(rel->rd_locator, *newrlocator, rel->rd_rel->relpersistence, true);
 
 	/* copy main fork */
 	RelationCopyStorage(RelationGetSmgr(rel), dstrel, MAIN_FORKNUM,
@@ -660,7 +662,7 @@ heapam_relation_copy_data(Relation rel, const RelFileLocator *newrlocator)
 	{
 		if (smgrexists(RelationGetSmgr(rel), forkNum))
 		{
-			smgrcreate(dstrel, forkNum, false);
+			smgrcreate(rel->rd_locator, dstrel, forkNum, false);
 
 			/*
 			 * WAL log creation if the relation is persistent, or this is the

src/backend/access/transam/xlog.c

@@ -208,6 +208,8 @@ const struct config_enum_entry archive_mode_options[] = {
  */
 CheckpointStatsData CheckpointStats;
 
+checkpoint_create_hook_type checkpoint_create_hook = NULL;
+
 /*
  * During recovery, lastFullPageWrites keeps track of full_page_writes that
  * the replayed WAL records indicate. It's initialized with full_page_writes
@@ -7173,6 +7175,9 @@ CreateCheckPoint(int flags)
 	 */
 	END_CRIT_SECTION();
 
+	if (checkpoint_create_hook != NULL)
+		checkpoint_create_hook(&checkPoint);
+
 	/*
 	 * In some cases there are groups of actions that must all occur on one
 	 * side or the other of a checkpoint record. Before flushing the

src/backend/access/transam/xlogutils.c

@@ -487,7 +487,7 @@ XLogReadBufferExtended(RelFileLocator rlocator, ForkNumber forknum,
 	 * filesystem loses an inode during a crash.  Better to write the data
 	 * until we are actually told to delete the file.)
 	 */
-	smgrcreate(smgr, forknum, true);
+	smgrcreate(rlocator, smgr, forknum, true);
 
 	lastblock = smgrnblocks(smgr, forknum);
 

src/backend/catalog/heap.c

@@ -386,7 +386,7 @@ heap_create(const char *relname,
 											   relpersistence,
 											   relfrozenxid, relminmxid);
 		else if (RELKIND_HAS_STORAGE(rel->rd_rel->relkind))
-			RelationCreateStorage(rel->rd_locator, relpersistence, true);
+			RelationCreateStorage(rel->rd_locator, rel->rd_locator, relpersistence, true);
 		else
 			Assert(false);
 	}

src/backend/catalog/index.c

@@ -3089,7 +3089,7 @@ index_build(Relation heapRelation,
 	if (indexRelation->rd_rel->relpersistence == RELPERSISTENCE_UNLOGGED &&
 		!smgrexists(RelationGetSmgr(indexRelation), INIT_FORKNUM))
 	{
-		smgrcreate(RelationGetSmgr(indexRelation), INIT_FORKNUM, false);
+		smgrcreate(indexRelation->rd_locator, RelationGetSmgr(indexRelation), INIT_FORKNUM, false);
 		log_smgrcreate(&indexRelation->rd_locator, INIT_FORKNUM);
 		indexRelation->rd_indam->ambuildempty(indexRelation);
 	}