Fix pageinspect failures on hash indexes.

Make every page in a hash index which isn't all-zeroes have a valid
special space, so that tools like pageinspect don't error out.

Also, make pageinspect cope with all-zeroes pages, because
_hash_alloc_buckets can leave behind large numbers of those until
they're consumed by splits.

Ashutosh Sharma and Robert Haas, reviewed by Amit Kapila.
Original trouble report from Jeff Janes.

Discussion: http://postgr.es/m/CAMkU=1y6NjKmqbJ8wLMhr=F74WzcMALYWcVFhEpm7i=mV=XsOg@mail.gmail.com

contrib/pageinspect/hashfuncs.c

@@ -56,31 +56,33 @@ static Page
 verify_hash_page(bytea *raw_page, int flags)
 {
 	Page		page = get_page_from_raw(raw_page);
-	int			pagetype;
+	int			pagetype = LH_UNUSED_PAGE;
-	HashPageOpaque pageopaque;
 
-	if (PageIsNew(page))
+	/* Treat new pages as unused. */
-		ereport(ERROR,
+	if (!PageIsNew(page))
-				(errcode(ERRCODE_INDEX_CORRUPTED),
+	{
-				 errmsg("index table contains zero page")));
+		HashPageOpaque pageopaque;
 
-	if (PageGetSpecialSize(page) != MAXALIGN(sizeof(HashPageOpaqueData)))
+		if (PageGetSpecialSize(page) != MAXALIGN(sizeof(HashPageOpaqueData)))
-		ereport(ERROR,
+			ereport(ERROR,
-				(errcode(ERRCODE_INDEX_CORRUPTED),
+					(errcode(ERRCODE_INDEX_CORRUPTED),
-				 errmsg("index table contains corrupted page")));
+					 errmsg("index table contains corrupted page")));
 
-	pageopaque = (HashPageOpaque) PageGetSpecialPointer(page);
+		pageopaque = (HashPageOpaque) PageGetSpecialPointer(page);
-	if (pageopaque->hasho_page_id != HASHO_PAGE_ID)
+		if (pageopaque->hasho_page_id != HASHO_PAGE_ID)
-		ereport(ERROR,
+			ereport(ERROR,
-				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
-				 errmsg("page is not a hash page"),
+					 errmsg("page is not a hash page"),
-				 errdetail("Expected %08x, got %08x.",
+					 errdetail("Expected %08x, got %08x.",
-						   HASHO_PAGE_ID, pageopaque->hasho_page_id)));
+							   HASHO_PAGE_ID, pageopaque->hasho_page_id)));
+
+		pagetype = pageopaque->hasho_flag & LH_PAGE_TYPE;
+	}
 
 	/* Check that page type is sane. */
-	pagetype = pageopaque->hasho_flag & LH_PAGE_TYPE;
 	if (pagetype != LH_OVERFLOW_PAGE && pagetype != LH_BUCKET_PAGE &&
-		pagetype != LH_BITMAP_PAGE && pagetype != LH_META_PAGE)
+		pagetype != LH_BITMAP_PAGE && pagetype != LH_META_PAGE &&
+		pagetype != LH_UNUSED_PAGE)
 		ereport(ERROR,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("invalid hash page type %08x", pagetype)));
@@ -190,19 +192,25 @@ hash_page_type(PG_FUNCTION_ARGS)
 				 (errmsg("must be superuser to use raw page functions"))));
 
 	page = verify_hash_page(raw_page, 0);
-	opaque = (HashPageOpaque) PageGetSpecialPointer(page);
 
-	/* page type (flags) */
+	if (PageIsNew(page))
-	if (opaque->hasho_flag & LH_META_PAGE)
-		type = "metapage";
-	else if (opaque->hasho_flag & LH_OVERFLOW_PAGE)
-		type = "overflow";
-	else if (opaque->hasho_flag & LH_BUCKET_PAGE)
-		type = "bucket";
-	else if (opaque->hasho_flag & LH_BITMAP_PAGE)
-		type = "bitmap";
-	else
 		type = "unused";
+	else
+	{
+		opaque = (HashPageOpaque) PageGetSpecialPointer(page);
+
+		/* page type (flags) */
+		if (opaque->hasho_flag & LH_META_PAGE)
+			type = "metapage";
+		else if (opaque->hasho_flag & LH_OVERFLOW_PAGE)
+			type = "overflow";
+		else if (opaque->hasho_flag & LH_BUCKET_PAGE)
+			type = "bucket";
+		else if (opaque->hasho_flag & LH_BITMAP_PAGE)
+			type = "bitmap";
+		else
+			type = "unused";
+	}
 
 	PG_RETURN_TEXT_P(cstring_to_text(type));
 }

src/backend/access/hash/hash_xlog.c

@@ -697,11 +697,20 @@ hash_xlog_squeeze_page(XLogReaderState *record)
 	if (XLogReadBufferForRedo(record, 2, &ovflbuf) == BLK_NEEDS_REDO)
 	{
 		Page		ovflpage;
+		HashPageOpaque ovflopaque;
 
 		ovflpage = BufferGetPage(ovflbuf);
 
 		_hash_pageinit(ovflpage, BufferGetPageSize(ovflbuf));
 
+		ovflopaque = (HashPageOpaque) PageGetSpecialPointer(ovflpage);
+
+		ovflopaque->hasho_prevblkno = InvalidBlockNumber;
+		ovflopaque->hasho_nextblkno = InvalidBlockNumber;
+		ovflopaque->hasho_bucket = -1;
+		ovflopaque->hasho_flag = LH_UNUSED_PAGE;
+		ovflopaque->hasho_page_id = HASHO_PAGE_ID;
+
 		PageSetLSN(ovflpage, lsn);
 		MarkBufferDirty(ovflbuf);
 	}

src/backend/access/hash/hashovfl.c

@@ -590,11 +590,22 @@ _hash_freeovflpage(Relation rel, Buffer bucketbuf, Buffer ovflbuf,
 	}
 
 	/*
-	 * Initialize the freed overflow page.  Just zeroing the page won't work,
+	 * Reinitialize the freed overflow page.  Just zeroing the page won't
-	 * because WAL replay routines expect pages to be initialized. See
+	 * work, because WAL replay routines expect pages to be initialized. See
-	 * explanation of RBM_NORMAL mode atop XLogReadBufferExtended.
+	 * explanation of RBM_NORMAL mode atop XLogReadBufferExtended.  We are
+	 * careful to make the special space valid here so that tools like
+	 * pageinspect won't get confused.
 	 */
 	_hash_pageinit(ovflpage, BufferGetPageSize(ovflbuf));
+
+	ovflopaque = (HashPageOpaque) PageGetSpecialPointer(ovflpage);
+
+	ovflopaque->hasho_prevblkno = InvalidBlockNumber;
+	ovflopaque->hasho_nextblkno = InvalidBlockNumber;
+	ovflopaque->hasho_bucket = -1;
+	ovflopaque->hasho_flag = LH_UNUSED_PAGE;
+	ovflopaque->hasho_page_id = HASHO_PAGE_ID;
+
 	MarkBufferDirty(ovflbuf);
 
 	if (BufferIsValid(prevbuf))

src/backend/access/hash/hashpage.c

@@ -993,6 +993,7 @@ _hash_alloc_buckets(Relation rel, BlockNumber firstblock, uint32 nblocks)
 	BlockNumber lastblock;
 	char		zerobuf[BLCKSZ];
 	Page		page;
+	HashPageOpaque ovflopaque;
 
 	lastblock = firstblock + nblocks - 1;
 
@@ -1007,10 +1008,19 @@ _hash_alloc_buckets(Relation rel, BlockNumber firstblock, uint32 nblocks)
 
 	/*
 	 * Initialize the page.  Just zeroing the page won't work; see
-	 * _hash_freeovflpage for similar usage.
+	 * _hash_freeovflpage for similar usage.  We take care to make the
+	 * special space valid for the benefit of tools such as pageinspect.
 	 */
 	_hash_pageinit(page, BLCKSZ);
 
+	ovflopaque = (HashPageOpaque) PageGetSpecialPointer(page);
+
+	ovflopaque->hasho_prevblkno = InvalidBlockNumber;
+	ovflopaque->hasho_nextblkno = InvalidBlockNumber;
+	ovflopaque->hasho_bucket = -1;
+	ovflopaque->hasho_flag = LH_UNUSED_PAGE;
+	ovflopaque->hasho_page_id = HASHO_PAGE_ID;
+
 	if (RelationNeedsWAL(rel))
 		log_newpage(&rel->rd_node,
 					MAIN_FORKNUM,