Support unlogged tables.

The contents of an unlogged table are WAL-logged; thus, they are not
available on standby servers and are truncated whenever the database
system enters recovery.  Indexes on unlogged tables are also unlogged.
Unlogged GiST indexes are not currently supported.

5 files changed, 434 insertions, 22 deletions

diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 860e736ff05..34e54536692 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -82,7 +82,7 @@ static bool IsForInput;
 static volatile BufferDesc *PinCountWaitBuf = NULL;
 
 
-static Buffer ReadBuffer_common(SMgrRelation reln,
+static Buffer ReadBuffer_common(SMgrRelation reln, char relpersistence,
 				  ForkNumber forkNum, BlockNumber blockNum,
 				  ReadBufferMode mode, BufferAccessStrategy strategy,
 				  bool *hit);
@@ -97,7 +97,9 @@ static void TerminateBufferIO(volatile BufferDesc *buf, bool clear_dirty,
 				  int set_flag_bits);
 static void shared_buffer_write_error_callback(void *arg);
 static void local_buffer_write_error_callback(void *arg);
-static volatile BufferDesc *BufferAlloc(SMgrRelation smgr, ForkNumber forkNum,
+static volatile BufferDesc *BufferAlloc(SMgrRelation smgr,
+			char relpersistence,
+			ForkNumber forkNum,
 			BlockNumber blockNum,
 			BufferAccessStrategy strategy,
 			bool *foundPtr);
@@ -241,8 +243,8 @@ ReadBufferExtended(Relation reln, ForkNumber forkNum, BlockNumber blockNum,
 	 * miss.
 	 */
 	pgstat_count_buffer_read(reln);
-	buf = ReadBuffer_common(reln->rd_smgr, forkNum, blockNum,
-							mode, strategy, &hit);
+	buf = ReadBuffer_common(reln->rd_smgr, reln->rd_rel->relpersistence,
+							forkNum, blockNum, mode, strategy, &hit);
 	if (hit)
 		pgstat_count_buffer_hit(reln);
 	return buf;
@@ -253,10 +255,10 @@ ReadBufferExtended(Relation reln, ForkNumber forkNum, BlockNumber blockNum,
  * ReadBufferWithoutRelcache -- like ReadBufferExtended, but doesn't require
  *		a relcache entry for the relation.
  *
- * NB: At present, this function may not be used on temporary relations, which
+ * NB: At present, this function may only be used on permanent relations, which
  * is OK, because we only use it during XLOG replay.  If in the future we
- * want to use it on temporary relations, we could pass the backend ID as an
- * additional parameter.
+ * want to use it on temporary or unlogged relations, we could pass additional
+ * parameters.
  */
 Buffer
 ReadBufferWithoutRelcache(RelFileNode rnode, ForkNumber forkNum,
@@ -267,7 +269,8 @@ ReadBufferWithoutRelcache(RelFileNode rnode, ForkNumber forkNum,
 
 	SMgrRelation smgr = smgropen(rnode, InvalidBackendId);
 
-	return ReadBuffer_common(smgr, forkNum, blockNum, mode, strategy, &hit);
+	return ReadBuffer_common(smgr, RELPERSISTENCE_PERMANENT, forkNum, blockNum,
+							 mode, strategy, &hit);
 }
 
 
@@ -277,7 +280,7 @@ ReadBufferWithoutRelcache(RelFileNode rnode, ForkNumber forkNum,
  * *hit is set to true if the request was satisfied from shared buffer cache.
  */
 static Buffer
-ReadBuffer_common(SMgrRelation smgr, ForkNumber forkNum,
+ReadBuffer_common(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 				  BlockNumber blockNum, ReadBufferMode mode,
 				  BufferAccessStrategy strategy, bool *hit)
 {
@@ -319,7 +322,8 @@ ReadBuffer_common(SMgrRelation smgr, ForkNumber forkNum,
 		 * lookup the buffer.  IO_IN_PROGRESS is set if the requested block is
 		 * not currently in memory.
 		 */
-		bufHdr = BufferAlloc(smgr, forkNum, blockNum, strategy, &found);
+		bufHdr = BufferAlloc(smgr, relpersistence, forkNum, blockNum,
+							 strategy, &found);
 		if (found)
 			pgBufferUsage.shared_blks_hit++;
 		else
@@ -500,7 +504,7 @@ ReadBuffer_common(SMgrRelation smgr, ForkNumber forkNum,
  * No locks are held either at entry or exit.
  */
 static volatile BufferDesc *
-BufferAlloc(SMgrRelation smgr, ForkNumber forkNum,
+BufferAlloc(SMgrRelation smgr, char relpersistence, ForkNumber forkNum,
 			BlockNumber blockNum,
 			BufferAccessStrategy strategy,
 			bool *foundPtr)
@@ -797,8 +801,11 @@ BufferAlloc(SMgrRelation smgr, ForkNumber forkNum,
 	 * 1 so that the buffer can survive one clock-sweep pass.)
 	 */
 	buf->tag = newTag;
-	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR);
-	buf->flags |= BM_TAG_VALID;
+	buf->flags &= ~(BM_VALID | BM_DIRTY | BM_JUST_DIRTIED | BM_CHECKPOINT_NEEDED | BM_IO_ERROR | BM_PERMANENT);
+	if (relpersistence == RELPERSISTENCE_PERMANENT)
+		buf->flags |= BM_TAG_VALID | BM_PERMANENT;
+	else
+		buf->flags |= BM_TAG_VALID;
 	buf->usage_count = 1;
 
 	UnlockBufHdr(buf);
@@ -1155,8 +1162,10 @@ UnpinBuffer(volatile BufferDesc *buf, bool fixOwner)
  * BufferSync -- Write out all dirty buffers in the pool.
  *
  * This is called at checkpoint time to write out all dirty shared buffers.
- * The checkpoint request flags should be passed in; currently the only one
- * examined is CHECKPOINT_IMMEDIATE, which disables delays between writes.
+ * The checkpoint request flags should be passed in.  If CHECKPOINT_IMMEDIATE
+ * is set, we disable delays between writes; if CHECKPOINT_IS_SHUTDOWN is
+ * set, we write even unlogged buffers, which are otherwise skipped.  The
+ * remaining flags currently have no effect here.
  */
 static void
 BufferSync(int flags)
@@ -1165,11 +1174,19 @@ BufferSync(int flags)
 	int			num_to_scan;
 	int			num_to_write;
 	int			num_written;
+	int			mask = BM_DIRTY;
 
 	/* Make sure we can handle the pin inside SyncOneBuffer */
 	ResourceOwnerEnlargeBuffers(CurrentResourceOwner);
 
 	/*
+	 * Unless this is a shutdown checkpoint, we write only permanent, dirty
+	 * buffers.  But at shutdown time, we write all dirty buffers.
+	 */
+	if (!(flags & CHECKPOINT_IS_SHUTDOWN))
+		flags |= BM_PERMANENT;
+
+	/*
 	 * Loop over all buffers, and mark the ones that need to be written with
 	 * BM_CHECKPOINT_NEEDED.  Count them as we go (num_to_write), so that we
 	 * can estimate how much work needs to be done.
@@ -1196,7 +1213,7 @@ BufferSync(int flags)
 		 */
 		LockBufHdr(bufHdr);
 
-		if (bufHdr->flags & BM_DIRTY)
+		if ((bufHdr->flags & mask) == mask)
 		{
 			bufHdr->flags |= BM_CHECKPOINT_NEEDED;
 			num_to_write++;
@@ -1897,12 +1914,12 @@ FlushBuffer(volatile BufferDesc *buf, SMgrRelation reln)
  *		Determines the current number of pages in the relation.
  */
 BlockNumber
-RelationGetNumberOfBlocks(Relation relation)
+RelationGetNumberOfBlocksInFork(Relation relation, ForkNumber forkNum)
 {
 	/* Open it at the smgr level if not already done */
 	RelationOpenSmgr(relation);
 
-	return smgrnblocks(relation->rd_smgr, MAIN_FORKNUM);
+	return smgrnblocks(relation->rd_smgr, forkNum);
 }
 
 /* ---------------------------------------------------------------------
diff --git a/src/backend/storage/file/Makefile b/src/backend/storage/file/Makefile
index 3b93aa1b45d..d2198f2b93e 100644
--- a/src/backend/storage/file/Makefile
+++ b/src/backend/storage/file/Makefile
@@ -12,6 +12,6 @@ subdir = src/backend/storage/file
 top_builddir = ../../../..
 include $(top_builddir)/src/Makefile.global
 
-OBJS = fd.o buffile.o copydir.o
+OBJS = fd.o buffile.o copydir.o reinit.o
 
 include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/storage/file/copydir.c b/src/backend/storage/file/copydir.c
index f7dc509b500..587fb9260c0 100644
--- a/src/backend/storage/file/copydir.c
+++ b/src/backend/storage/file/copydir.c
@@ -38,7 +38,6 @@
 #endif
 
 
-static void copy_file(char *fromfile, char *tofile);
 static void fsync_fname(char *fname, bool isdir);
 
 
@@ -142,7 +141,7 @@ copydir(char *fromdir, char *todir, bool recurse)
 /*
  * copy one file
  */
-static void
+void
 copy_file(char *fromfile, char *tofile)
 {
 	char	   *buffer;
diff --git a/src/backend/storage/file/fd.c b/src/backend/storage/file/fd.c
index 4f7dc39d638..a1dc18be44b 100644
--- a/src/backend/storage/file/fd.c
+++ b/src/backend/storage/file/fd.c
@@ -2055,7 +2055,7 @@ looks_like_temp_rel_name(const char *name)
 	/* We might have _forkname or .segment or both. */
 	if (name[pos] == '_')
 	{
-		int		forkchar = forkname_chars(&name[pos+1]);
+		int		forkchar = forkname_chars(&name[pos+1], NULL);
 		if (forkchar <= 0)
 			return false;
 		pos += forkchar + 1;
diff --git a/src/backend/storage/file/reinit.c b/src/backend/storage/file/reinit.c
new file mode 100644
index 00000000000..b75178b8045
--- /dev/null
+++ b/src/backend/storage/file/reinit.c
@@ -0,0 +1,396 @@
+/*-------------------------------------------------------------------------
+ *
+ * reinit.c
+ *	  Reinitialization of unlogged relations
+ *
+ * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
+ * Portions Copyright (c) 1994, Regents of the University of California
+ *
+ * IDENTIFICATION
+ *	  src/backend/storage/file/reinit.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include <unistd.h>
+
+#include "catalog/catalog.h"
+#include "storage/copydir.h"
+#include "storage/fd.h"
+#include "storage/reinit.h"
+#include "utils/hsearch.h"
+#include "utils/memutils.h"
+
+static void ResetUnloggedRelationsInTablespaceDir(const char *tsdirname,
+									 int op);
+static void ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname,
+								   int op);
+static bool parse_filename_for_nontemp_relation(const char *name,
+									int *oidchars, ForkNumber *fork);
+
+typedef struct {
+	char oid[OIDCHARS+1];
+} unlogged_relation_entry;
+
+/*
+ * Reset unlogged relations from before the last restart.
+ *
+ * If op includes UNLOGGED_RELATION_CLEANUP, we remove all forks of any
+ * relation with an "init" fork, except for the "init" fork itself.
+ *
+ * If op includes UNLOGGED_RELATION_INIT, we copy the "init" fork to the main
+ * fork.
+ */
+void
+ResetUnloggedRelations(int op)
+{
+	char		temp_path[MAXPGPATH];
+	DIR		   *spc_dir;
+	struct dirent *spc_de;
+	MemoryContext tmpctx, oldctx;
+
+	/* Log it. */
+	ereport(DEBUG1,
+			(errmsg("resetting unlogged relations: cleanup %d init %d",
+			 (op & UNLOGGED_RELATION_CLEANUP) != 0,
+			 (op & UNLOGGED_RELATION_INIT) != 0)));
+
+	/*
+	 * Just to be sure we don't leak any memory, let's create a temporary
+	 * memory context for this operation.
+	 */
+	tmpctx = AllocSetContextCreate(CurrentMemoryContext,
+								   "ResetUnloggedRelations",
+								   ALLOCSET_DEFAULT_MINSIZE,
+								   ALLOCSET_DEFAULT_INITSIZE,
+								   ALLOCSET_DEFAULT_MAXSIZE);
+	oldctx = MemoryContextSwitchTo(tmpctx);
+
+	/*
+	 * First process unlogged files in pg_default ($PGDATA/base)
+	 */
+	ResetUnloggedRelationsInTablespaceDir("base", op);
+
+	/*
+	 * Cycle through directories for all non-default tablespaces.
+	 */
+	spc_dir = AllocateDir("pg_tblspc");
+
+	while ((spc_de = ReadDir(spc_dir, "pg_tblspc")) != NULL)
+	{
+		if (strcmp(spc_de->d_name, ".") == 0 ||
+			strcmp(spc_de->d_name, "..") == 0)
+			continue;
+
+		snprintf(temp_path, sizeof(temp_path), "pg_tblspc/%s/%s",
+			spc_de->d_name, TABLESPACE_VERSION_DIRECTORY);
+		ResetUnloggedRelationsInTablespaceDir(temp_path, op);
+	}
+
+	FreeDir(spc_dir);
+
+	/*
+	 * Restore memory context.
+	 */
+	MemoryContextSwitchTo(oldctx);
+	MemoryContextDelete(tmpctx);
+}
+
+/* Process one tablespace directory for ResetUnloggedRelations */
+static void
+ResetUnloggedRelationsInTablespaceDir(const char *tsdirname, int op)
+{
+	DIR		   *ts_dir;
+	struct dirent *de;
+	char		dbspace_path[MAXPGPATH];
+
+	ts_dir = AllocateDir(tsdirname);
+	if (ts_dir == NULL)
+	{
+		/* anything except ENOENT is fishy */
+		if (errno != ENOENT)
+			elog(LOG,
+				 "could not open tablespace directory \"%s\": %m",
+				 tsdirname);
+		return;
+	}
+
+	while ((de = ReadDir(ts_dir, tsdirname)) != NULL)
+	{
+		int		i = 0;
+
+		/*
+		 * We're only interested in the per-database directories, which have
+		 * numeric names.  Note that this code will also (properly) ignore "."
+		 * and "..".
+		 */
+		while (isdigit((unsigned char) de->d_name[i]))
+			++i;
+		if (de->d_name[i] != '\0' || i == 0)
+			continue;
+
+		snprintf(dbspace_path, sizeof(dbspace_path), "%s/%s",
+				 tsdirname, de->d_name);
+		ResetUnloggedRelationsInDbspaceDir(dbspace_path, op);
+	}
+
+	FreeDir(ts_dir);
+}
+
+/* Process one per-dbspace directory for ResetUnloggedRelations */
+static void
+ResetUnloggedRelationsInDbspaceDir(const char *dbspacedirname, int op)
+{
+	DIR		   *dbspace_dir;
+	struct dirent *de;
+	char		rm_path[MAXPGPATH];
+
+	/* Caller must specify at least one operation. */
+	Assert((op & (UNLOGGED_RELATION_CLEANUP | UNLOGGED_RELATION_INIT)) != 0);
+
+	/*
+	 * Cleanup is a two-pass operation.  First, we go through and identify all
+	 * the files with init forks.  Then, we go through again and nuke
+	 * everything with the same OID except the init fork.
+	 */
+	if ((op & UNLOGGED_RELATION_CLEANUP) != 0)
+	{
+		HTAB	   *hash = NULL;
+		HASHCTL		ctl;
+
+		/* Open the directory. */
+		dbspace_dir = AllocateDir(dbspacedirname);
+		if (dbspace_dir == NULL)
+		{
+			elog(LOG,
+				 "could not open dbspace directory \"%s\": %m",
+				 dbspacedirname);
+			return;
+		}
+
+		/*
+		 * It's possible that someone could create a ton of unlogged relations
+		 * in the same database & tablespace, so we'd better use a hash table
+		 * rather than an array or linked list to keep track of which files
+		 * need to be reset.  Otherwise, this cleanup operation would be
+		 * O(n^2).
+		 */
+		ctl.keysize = sizeof(unlogged_relation_entry);
+		ctl.entrysize = sizeof(unlogged_relation_entry);
+		hash = hash_create("unlogged hash", 32, &ctl, HASH_ELEM);
+
+		/* Scan the directory. */
+		while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL)
+		{
+			ForkNumber forkNum;
+			int		oidchars;
+			unlogged_relation_entry ent;
+
+			/* Skip anything that doesn't look like a relation data file. */
+			if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars,
+													 &forkNum))
+				continue;
+
+			/* Also skip it unless this is the init fork. */
+			if (forkNum != INIT_FORKNUM)
+				continue;
+
+			/*
+			 * Put the OID portion of the name into the hash table, if it isn't
+			 * already.
+			 */
+			memset(ent.oid, 0, sizeof(ent.oid));
+			memcpy(ent.oid, de->d_name, oidchars);
+			hash_search(hash, &ent, HASH_ENTER, NULL);
+		}
+
+		/* Done with the first pass. */
+		FreeDir(dbspace_dir);
+
+		/*
+		 * If we didn't find any init forks, there's no point in continuing;
+		 * we can bail out now.
+		 */
+		if (hash_get_num_entries(hash) == 0)
+		{
+			hash_destroy(hash);
+			return;
+		}
+
+		/*
+		 * Now, make a second pass and remove anything that matches. First,
+		 * reopen the directory.
+		 */
+		dbspace_dir = AllocateDir(dbspacedirname);
+		if (dbspace_dir == NULL)
+		{
+			elog(LOG,
+				 "could not open dbspace directory \"%s\": %m",
+				 dbspacedirname);
+			hash_destroy(hash);
+			return;
+		}
+
+		/* Scan the directory. */
+		while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL)
+		{
+			ForkNumber forkNum;
+			int		oidchars;
+			bool	found;
+			unlogged_relation_entry ent;
+
+			/* Skip anything that doesn't look like a relation data file. */
+			if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars,
+													 &forkNum))
+				continue;
+
+			/* We never remove the init fork. */
+			if (forkNum == INIT_FORKNUM)
+				continue;
+
+			/*
+			 * See whether the OID portion of the name shows up in the hash
+			 * table.
+			 */
+			memset(ent.oid, 0, sizeof(ent.oid));
+			memcpy(ent.oid, de->d_name, oidchars);
+			hash_search(hash, &ent, HASH_FIND, &found);
+
+			/* If so, nuke it! */
+			if (found)
+			{
+				snprintf(rm_path, sizeof(rm_path), "%s/%s",
+					dbspacedirname, de->d_name);
+				/*
+				 * It's tempting to actually throw an error here, but since
+				 * this code gets run during database startup, that could
+				 * result in the database failing to start.  (XXX Should we do
+				 * it anyway?)
+				 */
+				if (unlink(rm_path))
+					elog(LOG, "could not unlink file \"%s\": %m", rm_path);
+				else
+					elog(DEBUG2, "unlinked file \"%s\"", rm_path);
+			}
+		}
+
+		/* Cleanup is complete. */
+		FreeDir(dbspace_dir);
+		hash_destroy(hash);
+	}
+
+	/*
+	 * Initialization happens after cleanup is complete: we copy each init
+	 * fork file to the corresponding main fork file.  Note that if we are
+	 * asked to do both cleanup and init, we may never get here: if the cleanup
+	 * code determines that there are no init forks in this dbspace, it will
+	 * return before we get to this point.
+	 */
+	if ((op & UNLOGGED_RELATION_INIT) != 0)
+	{
+		/* Open the directory. */
+		dbspace_dir = AllocateDir(dbspacedirname);
+		if (dbspace_dir == NULL)
+		{
+			/* we just saw this directory, so it really ought to be there */
+			elog(LOG,
+				 "could not open dbspace directory \"%s\": %m",
+				 dbspacedirname);
+			return;
+		}
+
+		/* Scan the directory. */
+		while ((de = ReadDir(dbspace_dir, dbspacedirname)) != NULL)
+		{
+			ForkNumber forkNum;
+			int		oidchars;
+			char	oidbuf[OIDCHARS+1];
+			char	srcpath[MAXPGPATH];
+			char	dstpath[MAXPGPATH];
+
+			/* Skip anything that doesn't look like a relation data file. */
+			if (!parse_filename_for_nontemp_relation(de->d_name, &oidchars,
+													 &forkNum))
+				continue;
+
+			/* Also skip it unless this is the init fork. */
+			if (forkNum != INIT_FORKNUM)
+				continue;
+
+			/* Construct source pathname. */
+			snprintf(srcpath, sizeof(srcpath), "%s/%s",
+					 dbspacedirname, de->d_name);
+
+			/* Construct destination pathname. */
+			memcpy(oidbuf, de->d_name, oidchars);
+			oidbuf[oidchars] = '\0';
+			snprintf(dstpath, sizeof(dstpath), "%s/%s%s",
+					 dbspacedirname, oidbuf, de->d_name + oidchars + 1 +
+					 strlen(forkNames[INIT_FORKNUM]));
+
+			/* OK, we're ready to perform the actual copy. */
+			elog(DEBUG2, "copying %s to %s", srcpath, dstpath);
+			copy_file(srcpath, dstpath);
+		}
+
+		/* Done with the first pass. */
+		FreeDir(dbspace_dir);
+	}
+}
+
+/*
+ * Basic parsing of putative relation filenames.
+ *
+ * This funtion returns true if the file appears to be in the correct format
+ * for a non-temporary relation and false otherwise.
+ *
+ * NB: If this function returns true, the caller is entitled to assume that
+ * *oidchars has been set to the a value no more than OIDCHARS, and thus
+ * that a buffer of OIDCHARS+1 characters is sufficient to hold the OID
+ * portion of the filename.  This is critical to protect against a possible
+ * buffer overrun.
+ */
+static bool
+parse_filename_for_nontemp_relation(const char *name, int *oidchars,
+									ForkNumber *fork)
+{
+	int			pos;
+
+	/* Look for a non-empty string of digits (that isn't too long). */
+	for (pos = 0; isdigit((unsigned char) name[pos]); ++pos)
+		;
+	if (pos == 0 || pos > OIDCHARS)
+		return false;
+	*oidchars = pos;
+
+	/* Check for a fork name. */
+	if (name[pos] != '_')
+		*fork = MAIN_FORKNUM;
+	else
+	{
+		int		forkchar;
+
+		forkchar = forkname_chars(&name[pos+1], fork);
+		if (forkchar <= 0)
+			return false;
+		pos += forkchar + 1;
+	}
+
+	/* Check for a segment number. */
+	if (name[pos] == '.')
+	{
+		int		segchar;
+		for (segchar = 1; isdigit((unsigned char) name[pos+segchar]); ++segchar)
+			;
+		if (segchar <= 1)
+			return false;
+		pos += segchar;
+	}
+
+	/* Now we should be at the end. */
+	if (name[pos] != '\0')
+		return false;
+	return true;
+}