commit 8d6ad4e1551252e17b7d7609f42f7a24921a2a31
Author: Robert Haas <rhaas@postgresql.org>
Date:   Fri Jan 30 08:39:22 2015 -0500

    contrib/parallel_count, now using heap_parallel_beginscan

diff --git a/contrib/parallel_count/Makefile b/contrib/parallel_count/Makefile
new file mode 100644
index 0000000..221c569
--- /dev/null
+++ b/contrib/parallel_count/Makefile
@@ -0,0 +1,19 @@
+MODULE_big = parallel_count
+OBJS = parallel_count.o $(WIN32RES)
+PGFILEDESC = "parallel_count - simple parallel tuple counter"
+
+EXTENSION = parallel_count
+DATA = parallel_count--1.0.sql
+
+REGRESS = parallel_count
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = contrib/parallel_count
+top_builddir = ../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/contrib/parallel_count/parallel_count--1.0.sql b/contrib/parallel_count/parallel_count--1.0.sql
new file mode 100644
index 0000000..a8a6266
--- /dev/null
+++ b/contrib/parallel_count/parallel_count--1.0.sql
@@ -0,0 +1,7 @@
+-- complain if script is sourced in psql, rather than via CREATE EXTENSION
+\echo Use "CREATE EXTENSION parallel_count" to load this file. \quit
+
+CREATE FUNCTION parallel_count(rel pg_catalog.regclass,
+							  nworkers pg_catalog.int4)
+    RETURNS pg_catalog.int8 STRICT
+	AS 'MODULE_PATHNAME' LANGUAGE C;
diff --git a/contrib/parallel_count/parallel_count.c b/contrib/parallel_count/parallel_count.c
new file mode 100644
index 0000000..06a5ec3
--- /dev/null
+++ b/contrib/parallel_count/parallel_count.c
@@ -0,0 +1,154 @@
+/*--------------------------------------------------------------------------
+ *
+ * parallel_count.c
+ *		simple parallel tuple counter
+ *
+ * Copyright (C) 2013-2014, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *		contrib/parallel_count/parallel_count.c
+ *
+ * -------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/heapam.h"
+#include "access/parallel.h"
+#include "access/relscan.h"
+#include "access/xact.h"
+#include "fmgr.h"
+#include "miscadmin.h"
+#include "storage/bufmgr.h"
+#include "storage/spin.h"
+#include "utils/builtins.h"
+#include "utils/guc.h"
+#include "utils/snapmgr.h"
+#include "utils/tqual.h"
+
+PG_MODULE_MAGIC;
+
+PG_FUNCTION_INFO_V1(parallel_count);
+
+#define		KEY_SCAN			1
+#define		KEY_RESULT			2
+
+void		_PG_init(void);
+void		count_worker_main(dsm_segment *seg, shm_toc *toc);
+
+static void count_helper(Relation rel, ParallelHeapScanDesc scan,
+						 int64 *result);
+
+Datum
+parallel_count(PG_FUNCTION_ARGS)
+{
+	Oid			relid = PG_GETARG_OID(0);
+	int32		nworkers = PG_GETARG_INT32(1);
+	int32		i;
+	bool		already_in_parallel_mode = IsInParallelMode();
+	ParallelContext *pcxt;
+	Snapshot	snapshot;
+	Size		pscan_size;
+	ParallelHeapScanDesc pscan;
+	Relation	rel;
+	int64	   *result = NULL;
+	int64		total = 0;
+
+	if (nworkers < 0)
+		ereport(ERROR,
+				(errmsg("number of parallel workers must be non-negative")));
+
+	rel = relation_open(relid, AccessShareLock);
+
+	if (!already_in_parallel_mode)
+		EnterParallelMode();
+
+	pcxt = CreateParallelContextForExtension("parallel_count",
+											 "count_worker_main",
+											 nworkers);
+
+	snapshot = GetActiveSnapshot();
+	pscan_size = heap_parallelscan_estimate(snapshot);
+	shm_toc_estimate_chunk(&pcxt->estimator, pscan_size);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+	if (nworkers > 0)
+	{
+		shm_toc_estimate_chunk(&pcxt->estimator, nworkers * sizeof(int64));
+		shm_toc_estimate_keys(&pcxt->estimator, 1);
+	}
+
+	InitializeParallelDSM(pcxt);
+
+	pscan = shm_toc_allocate(pcxt->toc, pscan_size);
+	heap_parallelscan_initialize(pscan, rel, snapshot);
+	shm_toc_insert(pcxt->toc, KEY_SCAN, pscan);
+	if (nworkers > 0)
+	{
+		result = shm_toc_allocate(pcxt->toc, nworkers * sizeof(int64));
+		shm_toc_insert(pcxt->toc, KEY_RESULT, result);
+	}
+
+	LaunchParallelWorkers(pcxt);
+
+	/* here's where we do the "real work" ... */
+	count_helper(rel, pscan, &total);
+
+	WaitForParallelWorkersToFinish(pcxt);
+
+	for (i = 0; i < nworkers; ++i)
+		total += result[i];
+
+	DestroyParallelContext(pcxt);
+
+	relation_close(rel, AccessShareLock);
+
+	if (!already_in_parallel_mode)
+		ExitParallelMode();
+
+	PG_RETURN_INT64(total);
+}
+
+void
+count_worker_main(dsm_segment *seg, shm_toc *toc)
+{
+	ParallelHeapScanDesc	pscan;
+	int64	   *result;
+	Relation	rel;
+
+	pscan = shm_toc_lookup(toc, KEY_SCAN);
+	Assert(pscan != NULL);
+
+	result = shm_toc_lookup(toc, KEY_RESULT);
+	Assert(result != NULL);
+
+	rel = relation_open(pscan->phs_relid, AccessShareLock);
+	count_helper(rel, pscan, &result[ParallelWorkerNumber]);
+	relation_close(rel, AccessShareLock);
+}
+
+static void
+count_helper(Relation rel, ParallelHeapScanDesc pscan, int64 *result)
+{
+	int64		ntuples = 0;
+	HeapScanDesc	scan;
+
+	scan = heap_beginscan_parallel(rel, pscan);
+
+	for (;;)
+	{
+		HeapTuple	tuple;
+
+		CHECK_FOR_INTERRUPTS();
+
+		tuple = heap_getnext(scan, ForwardScanDirection);
+		if (!HeapTupleIsValid(tuple))
+			break;
+
+		++ntuples;
+	}
+
+	heap_endscan(scan);
+
+	*result = ntuples;
+	elog(NOTICE, "PID %d counted " INT64_FORMAT " tuples", MyProcPid, ntuples);
+}
diff --git a/contrib/parallel_count/parallel_count.control b/contrib/parallel_count/parallel_count.control
new file mode 100644
index 0000000..76f332d
--- /dev/null
+++ b/contrib/parallel_count/parallel_count.control
@@ -0,0 +1,4 @@
+comment = 'simple parallel tuple counter'
+default_version = '1.0'
+module_pathname = '$libdir/parallel_count'
+relocatable = true
