Fix contrib/bloom to work for unlogged indexes.

blbuildempty did not do even approximately the right thing: it tried
to add a metapage to the relation's regular data fork, which already
has one at that point.  It should look like the ambuildempty methods
for all the standard index types, ie, initialize a metapage image in
some transient storage and then write it directly to the init fork.
To support that, refactor BloomInitMetapage into two functions.

In passing, fix BloomInitMetapage so it doesn't leave the rd_options
field of the index's relcache entry pointing at transient storage.
I'm not sure this had any visible consequence, since nothing much
else is likely to look at a bloom index's rd_options, but it's
certainly poor practice.

Per bug #14155 from Zhou Digoal.

Report: <20160524144146.22598.42558@wrigleys.postgresql.org>

contrib/bloom/blinsert.c

@@ -18,6 +18,7 @@
 #include "miscadmin.h"
 #include "storage/bufmgr.h"
 #include "storage/indexfsm.h"
+#include "storage/smgr.h"
 #include "utils/memutils.h"
 #include "utils/rel.h"
 
@@ -159,12 +160,26 @@ blbuild(Relation heap, Relation index, IndexInfo *indexInfo)
 void
 blbuildempty(Relation index)
 {
-	if (RelationGetNumberOfBlocks(index) != 0)
+	Page		metapage;
-		elog(ERROR, "index \"%s\" already contains data",
-			 RelationGetRelationName(index));
 
-	/* Initialize the meta page */
+	/* Construct metapage. */
-	BloomInitMetapage(index);
+	metapage = (Page) palloc(BLCKSZ);
+	BloomFillMetapage(index, metapage);
+
+	/* Write the page.  If archiving/streaming, XLOG it. */
+	PageSetChecksumInplace(metapage, BLOOM_METAPAGE_BLKNO);
+	smgrwrite(index->rd_smgr, INIT_FORKNUM, BLOOM_METAPAGE_BLKNO,
+			  (char *) metapage, true);
+	if (XLogIsNeeded())
+		log_newpage(&index->rd_smgr->smgr_rnode.node, INIT_FORKNUM,
+					BLOOM_METAPAGE_BLKNO, metapage, false);
+
+	/*
+	 * An immediate sync is required even if we xlog'd the page, because the
+	 * write did not go through shared_buffers and therefore a concurrent
+	 * checkpoint may have moved the redo pointer past our xlog record.
+	 */
+	smgrimmedsync(index->rd_smgr, INIT_FORKNUM);
 }
 
 /*

contrib/bloom/bloom.h

@@ -166,6 +166,7 @@ typedef BloomScanOpaqueData *BloomScanOpaque;
 extern void _PG_init(void);
 extern Datum blhandler(PG_FUNCTION_ARGS);
 extern void initBloomState(BloomState * state, Relation index);
+extern void BloomFillMetapage(Relation index, Page metaPage);
 extern void BloomInitMetapage(Relation index);
 extern void BloomInitPage(Page page, uint16 flags);
 extern Buffer BloomNewBuffer(Relation index);

contrib/bloom/blutils.c

@@ -346,7 +346,7 @@ BloomNewBuffer(Relation index)
 }
 
 /*
- * Initialize bloom page.
+ * Initialize any page of a bloom index.
  */
 void
 BloomInitPage(Page page, uint16 flags)
@@ -363,6 +363,8 @@ BloomInitPage(Page page, uint16 flags)
 
 /*
  * Adjust options of bloom index.
+ *
+ * This must produce default options when *opts is initially all-zero.
  */
 static void
 adjustBloomOptions(BloomOptions *opts)
@@ -378,7 +380,7 @@ adjustBloomOptions(BloomOptions *opts)
 				 errmsg("length of bloom signature (%d) is greater than maximum %d",
 						opts->bloomLength, MAX_BLOOM_LENGTH)));
 
-	/* Check singnature length */
+	/* Check signature length */
 	for (i = 0; i < INDEX_MAX_KEYS; i++)
 	{
 		/*
@@ -392,46 +394,67 @@ adjustBloomOptions(BloomOptions *opts)
 	}
 }
 
+/*
+ * Fill in metapage for bloom index.
+ */
+void
+BloomFillMetapage(Relation index, Page metaPage)
+{
+	BloomOptions *opts;
+	BloomMetaPageData *metadata;
+
+	/*
+	 * Choose the index's options.  If reloptions have been assigned, use
+	 * those, otherwise create default options by applying adjustBloomOptions
+	 * to a zeroed chunk of memory.  We apply adjustBloomOptions to existing
+	 * reloptions too, just out of paranoia; they should be valid already.
+	 */
+	opts = (BloomOptions *) index->rd_options;
+	if (!opts)
+		opts = (BloomOptions *) palloc0(sizeof(BloomOptions));
+	adjustBloomOptions(opts);
+
+	/*
+	 * Initialize contents of meta page, including a copy of the options,
+	 * which are now frozen for the life of the index.
+	 */
+	BloomInitPage(metaPage, BLOOM_META);
+	metadata = BloomPageGetMeta(metaPage);
+	memset(metadata, 0, sizeof(BloomMetaPageData));
+	metadata->magickNumber = BLOOM_MAGICK_NUMBER;
+	metadata->opts = *opts;
+	((PageHeader) metaPage)->pd_lower += sizeof(BloomMetaPageData);
+}
+
 /*
  * Initialize metapage for bloom index.
  */
 void
 BloomInitMetapage(Relation index)
 {
-	Page		metaPage;
 	Buffer		metaBuffer;
-	BloomMetaPageData *metadata;
+	Page		metaPage;
 	GenericXLogState *state;
 
 	/*
-	 * Make a new buffer, since it first buffer it should be associated with
+	 * Make a new page; since it is first page it should be associated with
 	 * block number 0 (BLOOM_METAPAGE_BLKNO).
 	 */
 	metaBuffer = BloomNewBuffer(index);
 	Assert(BufferGetBlockNumber(metaBuffer) == BLOOM_METAPAGE_BLKNO);
 
-	/* Initialize bloom index options */
-	if (!index->rd_options)
-		index->rd_options = palloc0(sizeof(BloomOptions));
-	adjustBloomOptions((BloomOptions *) index->rd_options);
-
 	/* Initialize contents of meta page */
 	state = GenericXLogStart(index);
-	metaPage = GenericXLogRegisterBuffer(state, metaBuffer, GENERIC_XLOG_FULL_IMAGE);
+	metaPage = GenericXLogRegisterBuffer(state, metaBuffer,
-
+										 GENERIC_XLOG_FULL_IMAGE);
-	BloomInitPage(metaPage, BLOOM_META);
+	BloomFillMetapage(index, metaPage);
-	metadata = BloomPageGetMeta(metaPage);
-	memset(metadata, 0, sizeof(BloomMetaPageData));
-	metadata->magickNumber = BLOOM_MAGICK_NUMBER;
-	metadata->opts = *((BloomOptions *) index->rd_options);
-	((PageHeader) metaPage)->pd_lower += sizeof(BloomMetaPageData);
-
 	GenericXLogFinish(state);
+
 	UnlockReleaseBuffer(metaBuffer);
 }
 
 /*
- * Initialize options for bloom index.
+ * Parse reloptions for bloom index, producing a BloomOptions struct.
  */
 bytea *
 bloptions(Datum reloptions, bool validate)

contrib/bloom/expected/bloom.out

@@ -138,6 +138,64 @@ SELECT count(*) FROM tst WHERE i = 7 AND t = '5';
     13
 (1 row)
 
+-- Try an unlogged table too
+CREATE UNLOGGED TABLE tstu (
+	i	int4,
+	t	text
+);
+INSERT INTO tstu SELECT i%10, substr(md5(i::text), 1, 1) FROM generate_series(1,2000) i;
+CREATE INDEX bloomidxu ON tstu USING bloom (i, t) WITH (col2 = 4);
+SET enable_seqscan=off;
+SET enable_bitmapscan=on;
+SET enable_indexscan=on;
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE i = 7;
+                 QUERY PLAN                 
+--------------------------------------------
+ Aggregate
+   ->  Bitmap Heap Scan on tstu
+         Recheck Cond: (i = 7)
+         ->  Bitmap Index Scan on bloomidxu
+               Index Cond: (i = 7)
+(5 rows)
+
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE t = '5';
+                 QUERY PLAN                 
+--------------------------------------------
+ Aggregate
+   ->  Bitmap Heap Scan on tstu
+         Recheck Cond: (t = '5'::text)
+         ->  Bitmap Index Scan on bloomidxu
+               Index Cond: (t = '5'::text)
+(5 rows)
+
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE i = 7 AND t = '5';
+                       QUERY PLAN                        
+---------------------------------------------------------
+ Aggregate
+   ->  Bitmap Heap Scan on tstu
+         Recheck Cond: ((i = 7) AND (t = '5'::text))
+         ->  Bitmap Index Scan on bloomidxu
+               Index Cond: ((i = 7) AND (t = '5'::text))
+(5 rows)
+
+SELECT count(*) FROM tstu WHERE i = 7;
+ count 
+-------
+   200
+(1 row)
+
+SELECT count(*) FROM tstu WHERE t = '5';
+ count 
+-------
+   112
+(1 row)
+
+SELECT count(*) FROM tstu WHERE i = 7 AND t = '5';
+ count 
+-------
+    13
+(1 row)
+
 RESET enable_seqscan;
 RESET enable_bitmapscan;
 RESET enable_indexscan;

contrib/bloom/sql/bloom.sql

@@ -50,6 +50,28 @@ SELECT count(*) FROM tst WHERE i = 7;
 SELECT count(*) FROM tst WHERE t = '5';
 SELECT count(*) FROM tst WHERE i = 7 AND t = '5';
 
+-- Try an unlogged table too
+
+CREATE UNLOGGED TABLE tstu (
+	i	int4,
+	t	text
+);
+
+INSERT INTO tstu SELECT i%10, substr(md5(i::text), 1, 1) FROM generate_series(1,2000) i;
+CREATE INDEX bloomidxu ON tstu USING bloom (i, t) WITH (col2 = 4);
+
+SET enable_seqscan=off;
+SET enable_bitmapscan=on;
+SET enable_indexscan=on;
+
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE i = 7;
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE t = '5';
+EXPLAIN (COSTS OFF) SELECT count(*) FROM tstu WHERE i = 7 AND t = '5';
+
+SELECT count(*) FROM tstu WHERE i = 7;
+SELECT count(*) FROM tstu WHERE t = '5';
+SELECT count(*) FROM tstu WHERE i = 7 AND t = '5';
+
 RESET enable_seqscan;
 RESET enable_bitmapscan;
 RESET enable_indexscan;