Add parallel nodes support

Author: Maksim Milyutin

src/hooks.c

@@ -324,11 +324,22 @@ pathman_rel_pathlist_hook(PlannerInfo *root,
 
 		/* Clear old path list */
 		list_free(rel->pathlist);
-
 		rel->pathlist = NIL;
+
+#if PG_VERSION_NUM >= 90600
+		/* Clear old partial path list */
+		list_free(rel->partial_pathlist);
+		rel->partial_pathlist = NIL;
+#endif
+
 		set_append_rel_pathlist(root, rel, rti, rte, pathkeyAsc, pathkeyDesc);
 		set_append_rel_size_compat(root, rel, rti, rte);
 
+#if PG_VERSION_NUM >= 90600
+		/* consider gathering partial paths for the parent appendrel */
+		generate_gather_paths(root, rel);
+#endif
+
 		/* No need to go further (both nodes are disabled), return */
 		if (!(pg_pathman_enable_runtimeappend ||
 			  pg_pathman_enable_runtime_merge_append))

src/pg_pathman.c

@@ -26,6 +26,7 @@
 #include "access/transam.h"
 #include "access/xact.h"
 #include "catalog/pg_cast.h"
+#include "catalog/pg_proc.h"
 #include "catalog/pg_type.h"
 #include "executor/spi.h"
 #include "foreign/fdwapi.h"
@@ -98,6 +99,11 @@ static void generate_mergeappend_paths(PlannerInfo *root, RelOptInfo *rel,
 						   PathKey *pathkeyAsc,
 						   PathKey *pathkeyDesc);
 static Path *get_cheapest_parameterized_child_path(PlannerInfo *root, RelOptInfo *rel, Relids required_outer);
+#if PG_VERSION_NUM >= 90600
+static void create_plain_partial_paths(PlannerInfo *root, RelOptInfo *rel);
+static void set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
+						  RangeTblEntry *rte);
+#endif
 
 
 /*
@@ -1756,6 +1762,12 @@ set_plain_rel_pathlist(PlannerInfo *root, RelOptInfo *rel, RangeTblEntry *rte)
 #endif
 	add_path(rel, path);
 
+#if PG_VERSION_NUM >= 90600
+	/* If appropriate, consider parallel sequential scan */
+	if (rel->consider_parallel && required_outer == NULL)
+		create_plain_partial_paths(root, rel);
+#endif
+
 	/* Consider index scans */
 	create_index_paths(root, rel);
 
@@ -1804,6 +1816,10 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	List	   *live_childrels = NIL;
 	List	   *subpaths = NIL;
 	bool		subpaths_valid = true;
+#if PG_VERSION_NUM >= 90600
+	List	   *partial_subpaths = NIL;
+	bool		partial_subpaths_valid = true;
+#endif
 	List	   *all_child_pathkeys = NIL;
 	List	   *all_child_outers = NIL;
 	ListCell   *l;
@@ -1831,6 +1847,18 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 		childRTE = root->simple_rte_array[childRTindex];
 		childrel = root->simple_rel_array[childRTindex];
 
+#if PG_VERSION_NUM >= 90600
+		/*
+		 * If parallelism is allowable for this query in general and for parent
+		 * appendrel, see whether it's allowable for this childrel in
+		 * particular.
+		 *
+		 * For consistency, do this before calling set_rel_size() for the child.
+		 */
+		if (root->glob->parallelModeOK && rel->consider_parallel)
+			set_rel_consider_parallel(root, childrel, childRTE);
+#endif
+
 		/*
 		 * Compute the child's access paths.
 		 */
@@ -1857,6 +1885,18 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 		 */
 		live_childrels = lappend(live_childrels, childrel);
 
+#if PG_VERSION_NUM >= 90600
+		/*
+		 * If any live child is not parallel-safe, treat the whole appendrel
+		 * as not parallel-safe.  In future we might be able to generate plans
+		 * in which some children are farmed out to workers while others are
+		 * not; but we don't have that today, so it's a waste to consider
+		 * partial paths anywhere in the appendrel unless it's all safe.
+		 */
+		if (!childrel->consider_parallel)
+			rel->consider_parallel = false;
+#endif
+
 		/*
 		 * If child has an unparameterized cheapest-total path, add that to
 		 * the unparameterized Append path we are constructing for the parent.
@@ -1868,6 +1908,15 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 		else
 			subpaths_valid = false;
 
+#if PG_VERSION_NUM >= 90600
+		/* Same idea, but for a partial plan. */
+		if (childrel->partial_pathlist != NIL)
+			partial_subpaths = accumulate_append_subpath(partial_subpaths,
+									   linitial(childrel->partial_pathlist));
+		else
+			partial_subpaths_valid = false;
+#endif
+
 		/*
 		 * Collect lists of all the available path orderings and
 		 * parameterizations for all the children.  We use these as a
@@ -1942,6 +1991,37 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 		add_path(rel,
 				 (Path *) create_append_path_compat(rel, subpaths, NULL, 0));
 
+#if PG_VERSION_NUM >= 90600
+	/*
+	 * Consider an append of partial unordered, unparameterized partial paths.
+	 */
+	if (partial_subpaths_valid)
+	{
+		AppendPath *appendpath;
+		ListCell   *lc;
+		int			parallel_workers = 0;
+
+		/*
+		 * Decide on the number of workers to request for this append path.
+		 * For now, we just use the maximum value from among the members.  It
+		 * might be useful to use a higher number if the Append node were
+		 * smart enough to spread out the workers, but it currently isn't.
+		 */
+		foreach(lc, partial_subpaths)
+		{
+			Path	   *path = lfirst(lc);
+
+			parallel_workers = Max(parallel_workers, path->parallel_workers);
+		}
+		Assert(parallel_workers > 0);
+
+		/* Generate a partial append path. */
+		appendpath = create_append_path(rel, partial_subpaths, NULL,
+										parallel_workers);
+		add_partial_path(rel, (Path *) appendpath);
+	}
+#endif
+
 	/*
 	 * Also build unparameterized MergeAppend paths based on the collected
 	 * list of child pathkeys.
@@ -1995,6 +2075,217 @@ set_append_rel_pathlist(PlannerInfo *root, RelOptInfo *rel,
 	}
 }
 
+#if PG_VERSION_NUM >= 90600
+/*
+ * create_plain_partial_paths
+ *	  Build partial access paths for parallel scan of a plain relation
+ */
+static void
+create_plain_partial_paths(PlannerInfo *root, RelOptInfo *rel)
+{
+	int			parallel_workers;
+
+	/*
+	 * If the user has set the parallel_workers reloption, use that; otherwise
+	 * select a default number of workers.
+	 */
+	if (rel->rel_parallel_workers != -1)
+		parallel_workers = rel->rel_parallel_workers;
+	else
+	{
+		int			parallel_threshold;
+
+		/*
+		 * If this relation is too small to be worth a parallel scan, just
+		 * return without doing anything ... unless it's an inheritance child.
+		 * In that case, we want to generate a parallel path here anyway.  It
+		 * might not be worthwhile just for this relation, but when combined
+		 * with all of its inheritance siblings it may well pay off.
+		 */
+		if (rel->pages < (BlockNumber) min_parallel_relation_size &&
+			rel->reloptkind == RELOPT_BASEREL)
+			return;
+
+		/*
+		 * Select the number of workers based on the log of the size of the
+		 * relation.  This probably needs to be a good deal more
+		 * sophisticated, but we need something here for now.  Note that the
+		 * upper limit of the min_parallel_relation_size GUC is chosen to
+		 * prevent overflow here.
+		 */
+		parallel_workers = 1;
+		parallel_threshold = Max(min_parallel_relation_size, 1);
+		while (rel->pages >= (BlockNumber) (parallel_threshold * 3))
+		{
+			parallel_workers++;
+			parallel_threshold *= 3;
+			if (parallel_threshold > INT_MAX / 3)
+				break;			/* avoid overflow */
+		}
+	}
+
+	/*
+	 * In no case use more than max_parallel_workers_per_gather workers.
+	 */
+	parallel_workers = Min(parallel_workers, max_parallel_workers_per_gather);
+
+	/* If any limit was set to zero, the user doesn't want a parallel scan. */
+	if (parallel_workers <= 0)
+		return;
+
+	/* Add an unordered partial path based on a parallel sequential scan. */
+	add_partial_path(rel, create_seqscan_path(root, rel, NULL, parallel_workers));
+}
+
+/*
+ * If this relation could possibly be scanned from within a worker, then set
+ * its consider_parallel flag.
+ */
+static void
+set_rel_consider_parallel(PlannerInfo *root, RelOptInfo *rel,
+						  RangeTblEntry *rte)
+{
+	/*
+	 * The flag has previously been initialized to false, so we can just
+	 * return if it becomes clear that we can't safely set it.
+	 */
+	Assert(!rel->consider_parallel);
+
+	/* Don't call this if parallelism is disallowed for the entire query. */
+	Assert(root->glob->parallelModeOK);
+
+	/* This should only be called for baserels and appendrel children. */
+	Assert(rel->reloptkind == RELOPT_BASEREL ||
+		   rel->reloptkind == RELOPT_OTHER_MEMBER_REL);
+
+	/* Assorted checks based on rtekind. */
+	switch (rte->rtekind)
+	{
+		case RTE_RELATION:
+
+			/*
+			 * Currently, parallel workers can't access the leader's temporary
+			 * tables.  We could possibly relax this if the wrote all of its
+			 * local buffers at the start of the query and made no changes
+			 * thereafter (maybe we could allow hint bit changes), and if we
+			 * taught the workers to read them.  Writing a large number of
+			 * temporary buffers could be expensive, though, and we don't have
+			 * the rest of the necessary infrastructure right now anyway.  So
+			 * for now, bail out if we see a temporary table.
+			 */
+			if (get_rel_persistence(rte->relid) == RELPERSISTENCE_TEMP)
+				return;
+
+			/*
+			 * Table sampling can be pushed down to workers if the sample
+			 * function and its arguments are safe.
+			 */
+			if (rte->tablesample != NULL)
+			{
+				Oid			proparallel = func_parallel(rte->tablesample->tsmhandler);
+
+				if (proparallel != PROPARALLEL_SAFE)
+					return;
+				if (has_parallel_hazard((Node *) rte->tablesample->args,
+										false))
+					return;
+			}
+
+			/*
+			 * Ask FDWs whether they can support performing a ForeignScan
+			 * within a worker.  Most often, the answer will be no.  For
+			 * example, if the nature of the FDW is such that it opens a TCP
+			 * connection with a remote server, each parallel worker would end
+			 * up with a separate connection, and these connections might not
+			 * be appropriately coordinated between workers and the leader.
+			 */
+			if (rte->relkind == RELKIND_FOREIGN_TABLE)
+			{
+				Assert(rel->fdwroutine);
+				if (!rel->fdwroutine->IsForeignScanParallelSafe)
+					return;
+				if (!rel->fdwroutine->IsForeignScanParallelSafe(root, rel, rte))
+					return;
+			}
+
+			/*
+			 * There are additional considerations for appendrels, which we'll
+			 * deal with in set_append_rel_size and set_append_rel_pathlist.
+			 * For now, just set consider_parallel based on the rel's own
+			 * quals and targetlist.
+			 */
+			break;
+
+		case RTE_SUBQUERY:
+
+			/*
+			 * There's no intrinsic problem with scanning a subquery-in-FROM
+			 * (as distinct from a SubPlan or InitPlan) in a parallel worker.
+			 * If the subquery doesn't happen to have any parallel-safe paths,
+			 * then flagging it as consider_parallel won't change anything,
+			 * but that's true for plain tables, too.  We must set
+			 * consider_parallel based on the rel's own quals and targetlist,
+			 * so that if a subquery path is parallel-safe but the quals and
+			 * projection we're sticking onto it are not, we correctly mark
+			 * the SubqueryScanPath as not parallel-safe.  (Note that
+			 * set_subquery_pathlist() might push some of these quals down
+			 * into the subquery itself, but that doesn't change anything.)
+			 */
+			break;
+
+		case RTE_JOIN:
+			/* Shouldn't happen; we're only considering baserels here. */
+			Assert(false);
+			return;
+
+		case RTE_FUNCTION:
+			/* Check for parallel-restricted functions. */
+			if (has_parallel_hazard((Node *) rte->functions, false))
+				return;
+			break;
+
+		case RTE_VALUES:
+			/* Check for parallel-restricted functions. */
+			if (has_parallel_hazard((Node *) rte->values_lists, false))
+				return;
+			break;
+
+		case RTE_CTE:
+
+			/*
+			 * CTE tuplestores aren't shared among parallel workers, so we
+			 * force all CTE scans to happen in the leader.  Also, populating
+			 * the CTE would require executing a subplan that's not available
+			 * in the worker, might be parallel-restricted, and must get
+			 * executed only once.
+			 */
+			return;
+	}
+
+	/*
+	 * If there's anything in baserestrictinfo that's parallel-restricted, we
+	 * give up on parallelizing access to this relation.  We could consider
+	 * instead postponing application of the restricted quals until we're
+	 * above all the parallelism in the plan tree, but it's not clear that
+	 * that would be a win in very many cases, and it might be tricky to make
+	 * outer join clauses work correctly.  It would likely break equivalence
+	 * classes, too.
+	 */
+	if (has_parallel_hazard((Node *) rel->baserestrictinfo, false))
+		return;
+
+	/*
+	 * Likewise, if the relation's outputs are not parallel-safe, give up.
+	 * (Usually, they're just Vars, but sometimes they're not.)
+	 */
+	if (has_parallel_hazard((Node *) rel->reltarget->exprs, false))
+		return;
+
+	/* We have a winner. */
+	rel->consider_parallel = true;
+}
+#endif
+
 static List *
 accumulate_append_subpath(List *subpaths, Path *path)
 {