Tune costs of nodeCustomTempScan: use analogy with Append.

Author: danolivo

contrib/tempscan/expected/basic.out

@@ -28,18 +28,15 @@ SELECT count(*) FROM parallel_test;
                ->  Parallel Seq Scan on parallel_test
 (5 rows)
 
--- Should also utilise parallel workers like scanning of a plain table
+-- Do not use parallel scan of temporary table so far. As a result, shouldn't
+-- find nodeCustomTempScan here
 EXPLAIN (COSTS OFF)
 SELECT count(*) FROM parallel_test_tmp;
-                           QUERY PLAN                           
-----------------------------------------------------------------
- Finalize Aggregate
-   ->  Gather
-         Workers Planned: 1
-         ->  Partial Aggregate
-               ->  Custom Scan (nodeCustomTempScan)
-                     ->  Parallel Seq Scan on parallel_test_tmp
-(6 rows)
+             QUERY PLAN              
+-------------------------------------
+ Aggregate
+   ->  Seq Scan on parallel_test_tmp
+(2 rows)
 
 -- Want to see here partial aggregate over parallel join
 EXPLAIN (COSTS OFF)
@@ -59,18 +56,18 @@ SELECT count(*) FROM parallel_test t1 NATURAL JOIN parallel_test t2;
 
 EXPLAIN (COSTS OFF)
 SELECT count(*) FROM parallel_test_tmp t1 NATURAL JOIN parallel_test t2;
-                               QUERY PLAN                                
--------------------------------------------------------------------------
+                              QUERY PLAN                              
+----------------------------------------------------------------------
  Finalize Aggregate
    ->  Gather
          Workers Planned: 1
          ->  Partial Aggregate
-               ->  Parallel Hash Join
-                     Hash Cond: (t1.x = t2.x)
-                     ->  Custom Scan (nodeCustomTempScan)
-                           ->  Parallel Seq Scan on parallel_test_tmp t1
-                     ->  Parallel Hash
-                           ->  Parallel Seq Scan on parallel_test t2
+               ->  Hash Join
+                     Hash Cond: (t2.x = t1.x)
+                     ->  Parallel Seq Scan on parallel_test t2
+                     ->  Hash
+                           ->  Custom Scan (nodeCustomTempScan)
+                                 ->  Seq Scan on parallel_test_tmp t1
 (10 rows)
 
 -- Just see how merge join manages custom parallel scan path
@@ -95,100 +92,81 @@ SELECT count(*) FROM parallel_test t1 NATURAL JOIN parallel_test t2;
 
 EXPLAIN (COSTS OFF)
 SELECT count(*) FROM parallel_test_tmp t1 NATURAL JOIN parallel_test t2;
-                                  QUERY PLAN                                   
--------------------------------------------------------------------------------
+                              QUERY PLAN                              
+----------------------------------------------------------------------
  Finalize Aggregate
    ->  Gather
          Workers Planned: 1
          ->  Partial Aggregate
                ->  Merge Join
-                     Merge Cond: (t1.x = t2.x)
+                     Merge Cond: (t2.x = t1.x)
+                     ->  Sort
+                           Sort Key: t2.x
+                           ->  Parallel Seq Scan on parallel_test t2
                      ->  Sort
                            Sort Key: t1.x
                            ->  Custom Scan (nodeCustomTempScan)
-                                 ->  Parallel Seq Scan on parallel_test_tmp t1
-                     ->  Sort
-                           Sort Key: t2.x
-                           ->  Seq Scan on parallel_test t2
+                                 ->  Seq Scan on parallel_test_tmp t1
 (13 rows)
 
 RESET enable_hashjoin;
 -- Increase table size and see how indexes work
 ALTER TABLE parallel_test ADD COLUMN y text DEFAULT 'none';
 INSERT INTO parallel_test (x,y) SELECT x, 'data' || x AS y FROM generate_series(1,10000) AS x;
-CREATE INDEX ON parallel_test (x);
-ANALYZE parallel_test;
+CREATE INDEX ON parallel_test_tmp (x);
+VACUUM ANALYZE;
+-- Use IndexScan on temporary table
 EXPLAIN (COSTS OFF)
-SELECT count(*) FROM parallel_test t1 NATURAL JOIN parallel_test t2
+SELECT count(*) FROM parallel_test t1 NATURAL JOIN parallel_test_tmp t2
 WHERE t1.x < 10;
-                                     QUERY PLAN                                      
--------------------------------------------------------------------------------------
+                                          QUERY PLAN                                           
+-----------------------------------------------------------------------------------------------
  Aggregate
    ->  Gather
-         Workers Planned: 1
+         Workers Planned: 3
          ->  Nested Loop
-               ->  Parallel Index Scan using parallel_test_x_idx on parallel_test t1
-                     Index Cond: (x < 10)
-               ->  Index Scan using parallel_test_x_idx on parallel_test t2
-                     Index Cond: (x = t1.x)
-                     Filter: (y = t1.y)
+               ->  Parallel Seq Scan on parallel_test t1
+                     Filter: (x < 10)
+               ->  Custom Scan (nodeCustomTempScan)
+                     ->  Index Only Scan using parallel_test_tmp_x_idx on parallel_test_tmp t2
+                           Index Cond: (x = t1.x)
 (9 rows)
 
-EXPLAIN (COSTS OFF)
-SELECT count(*) FROM parallel_test t1 NATURAL JOIN parallel_test_tmp t2
-WHERE t1.x < 10;
-                                   QUERY PLAN                                    
----------------------------------------------------------------------------------
- Aggregate
-   ->  Gather
-         Workers Planned: 1
-         ->  Merge Join
-               Merge Cond: (t2.x = t1.x)
-               ->  Sort
-                     Sort Key: t2.x
-                     ->  Custom Scan (nodeCustomTempScan)
-                           ->  Parallel Seq Scan on parallel_test_tmp t2
-               ->  Index Only Scan using parallel_test_x_idx on parallel_test t1
-                     Index Cond: (x < 10)
-(11 rows)
-
 CREATE TEMP TABLE parallel_test_tmp_2 AS (SELECT * FROM parallel_test);
 CREATE INDEX ON parallel_test_tmp_2 (x);
-ANALYZE parallel_test_tmp_2;
+VACUUM ANALYZE;
+-- Can't use parallel workers here: for now temp tables can be used as parallel
+-- safe source but not be read concurrently.
 EXPLAIN (COSTS OFF)
 SELECT count(*) FROM parallel_test_tmp t1 NATURAL JOIN parallel_test_tmp_2 t2
 WHERE t2.x < 10;
-                                         QUERY PLAN                                          
----------------------------------------------------------------------------------------------
+                                      QUERY PLAN                                       
+---------------------------------------------------------------------------------------
  Aggregate
-   ->  Gather
-         Workers Planned: 1
-         ->  Merge Join
-               Merge Cond: (t1.x = t2.x)
-               ->  Sort
-                     Sort Key: t1.x
-                     ->  Custom Scan (nodeCustomTempScan)
-                           ->  Parallel Seq Scan on parallel_test_tmp t1
-               ->  Index Only Scan using parallel_test_tmp_2_x_idx on parallel_test_tmp_2 t2
-                     Index Cond: (x < 10)
-(11 rows)
+   ->  Merge Join
+         Merge Cond: (t1.x = t2.x)
+         ->  Sort
+               Sort Key: t1.x
+               ->  Seq Scan on parallel_test_tmp t1
+         ->  Index Only Scan using parallel_test_tmp_2_x_idx on parallel_test_tmp_2 t2
+               Index Cond: (x < 10)
+(8 rows)
 
 EXPLAIN (COSTS OFF)
 SELECT count(*) FROM parallel_test_tmp_2 t1 NATURAL JOIN parallel_test_tmp_2 t2
 WHERE t1.x < 10;
-                                              QUERY PLAN                                               
--------------------------------------------------------------------------------------------------------
+                                       QUERY PLAN                                       
+----------------------------------------------------------------------------------------
  Aggregate
-   ->  Gather
-         Workers Planned: 1
-         ->  Nested Loop
-               ->  Custom Scan (nodeCustomTempScan)
-                     ->  Parallel Index Scan using parallel_test_tmp_2_x_idx on parallel_test_tmp_2 t1
-                           Index Cond: (x < 10)
+   ->  Nested Loop
+         ->  Custom Scan (nodeCustomTempScan)
+               ->  Index Scan using parallel_test_tmp_2_x_idx on parallel_test_tmp_2 t1
+                     Index Cond: (x < 10)
+         ->  Custom Scan (nodeCustomTempScan)
                ->  Index Scan using parallel_test_tmp_2_x_idx on parallel_test_tmp_2 t2
                      Index Cond: (x = t1.x)
                      Filter: (y = t1.y)
-(10 rows)
+(9 rows)
 
 RESET tempscan.enable;
 DROP TABLE parallel_test, parallel_test_tmp;

contrib/tempscan/nodeCustomTempScan.c

@@ -16,6 +16,7 @@
 #include "nodes/extensible.h"
 #include "optimizer/clauses.h"
 #include "optimizer/cost.h"
+#include "optimizer/optimizer.h"
 #include "optimizer/pathnode.h"
 #include "optimizer/paths.h"
 #include "utils/guc.h"
@@ -27,6 +28,9 @@ PG_MODULE_MAGIC;
 #define MODULENAME	"tempscan"
 #define NODENAME	"nodeCustomTempScan"
 
+/* By analogy with Append */
+#define TEMPSCAN_CPU_COST_MULTIPLIER	(0.5)
+
 static Plan *create_partial_tempscan_plan(PlannerInfo *root,
 										  RelOptInfo *rel,
 										  CustomPath *best_path,
@@ -38,6 +42,7 @@ static void BeginTempScan(CustomScanState *node, EState *estate, int eflags);
 static TupleTableSlot *ExecTempScan(CustomScanState *node);
 static void EndTempScan(CustomScanState *node);
 static void ReScanTempScan(CustomScanState *node);
+static Size EstimateDSMTempScan(CustomScanState *node, ParallelContext *pcxt);
 
 static CustomPathMethods path_methods =
 {
@@ -62,7 +67,7 @@ static CustomExecMethods exec_methods =
 	.ReScanCustomScan = ReScanTempScan,
 	.MarkPosCustomScan = NULL,
 	.RestrPosCustomScan = NULL,
-	.EstimateDSMCustomScan = NULL,
+	.EstimateDSMCustomScan = EstimateDSMTempScan,
 	.InitializeDSMCustomScan = NULL,
 	.ReInitializeDSMCustomScan = NULL,
 	.InitializeWorkerCustomScan = NULL,
@@ -93,18 +98,22 @@ create_partial_tempscan_path(PlannerInfo *root, RelOptInfo *rel,
 	pathnode->parent = rel;
 	pathnode->pathtarget = rel->reltarget;
 	pathnode->rows = path->rows; /* Don't use rel->rows! Remember semantics of this field in the parallel case */
-
-	/* XXX: Just for now */
-	pathnode->param_info = NULL;
+	pathnode->param_info = path->param_info;
 
 	pathnode->parallel_safe = true;
 	pathnode->parallel_aware = false;
 	pathnode->parallel_workers = path->parallel_workers;
 
-	/* DEBUGGING purposes only */
 	pathnode->startup_cost = path->startup_cost;
 	pathnode->total_cost = path->total_cost;
 
+	/*
+	 * Although TempScan does not do any selection or projection, it's not free;
+	 * add a small per-tuple overhead.
+	 */
+	pathnode->total_cost +=
+					cpu_tuple_cost * TEMPSCAN_CPU_COST_MULTIPLIER * path->rows;
+
 	cpath->custom_paths = list_make1(path);
 	cpath->custom_private = NIL;
 	cpath->custom_restrictinfo = NIL;
@@ -121,8 +130,7 @@ create_partial_tempscan_plan(PlannerInfo *root, RelOptInfo *rel,
 	CustomScan *cscan = makeNode(CustomScan);
 
 	Assert(list_length(custom_plans) == 1);
-	Assert(best_path->path.parallel_safe = true &&
-		   best_path->path.parallel_workers > 0);
+	Assert(best_path->path.parallel_safe = true);
 
 
 	cscan->scan.plan.targetlist = cscan->custom_scan_tlist = tlist;
@@ -205,9 +213,9 @@ static void
 try_partial_tempscan(PlannerInfo *root, RelOptInfo *rel, Index rti,
 					 RangeTblEntry *rte)
 {
-	int			parallel_workers;
 	ListCell   *lc;
-	List	   *partial_pathlist_new = NIL;
+	List	   *parallel_safe_lst = NIL;
+	List	   *tmplst = rel->pathlist;
 
 	/*
 	 * Some extension intercept this hook earlier. Allow it to do a work
@@ -223,46 +231,56 @@ try_partial_tempscan(PlannerInfo *root, RelOptInfo *rel, Index rti,
 		get_rel_persistence(rte->relid) != RELPERSISTENCE_TEMP)
 		return;
 
-	/* HACK */
 	if (!is_parallel_safe(root, (Node *) rel->baserestrictinfo) ||
 		!is_parallel_safe(root, (Node *) rel->reltarget->exprs))
 		return;
 
-	parallel_workers = compute_parallel_worker(rel, rel->pages, -1,
-											   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;
-
-	/* Enable parallel paths generation for this relation */
+	/* Enable parallel safe paths generation for this relation */
 	Assert(rel->partial_pathlist == NIL);
 	rel->consider_parallel = true;
 
-	/* Add partial sequental scan path. */
-	add_partial_path(rel, (Path *)
-						create_seqscan_path(root, rel, NULL, parallel_workers));
+	/*
+	 * Now we have a problem:
+	 * should generate parallel safe paths. But they will have the same cost as
+	 * previously added non-parallel ones and, being safe, will definitely crowd
+	 * out non-safe ones.
+	 * So, we need a HACK: add new safe paths with cost of custom node.
+	 */
 
-	/* Add there more specific paths too */
+	rel->pathlist = NIL;
+
+	/*
+	 * Build possibly parallel paths other temporary table
+	 */
+	add_path(rel, create_seqscan_path(root, rel, NULL, 0));
 	create_index_paths(root, rel);
 	create_tidscan_paths(root, rel);
 
-	foreach(lc, rel->partial_pathlist)
-	{
-		Path   *path = lfirst(lc);
+	/*
+	 * Dangerous zone. But we assume it is strictly local. What about extension
+	 * which could call ours and may have desire to add some partial paths after
+	 * us?
+	 */
 
-		partial_pathlist_new =
-			lappend(partial_pathlist_new,
-					(void *) create_partial_tempscan_path(root, rel, path));
-	}
+	list_free(rel->partial_pathlist);
+	rel->partial_pathlist = NIL;
 
 	/*
-	 * Dangerous zone. But we assume it is strictly local. What about extension
-	 * which could call ours and add some paths after us?
+	 * Set guard over each parallel_safe path
 	 */
-	rel->partial_pathlist = partial_pathlist_new;
+	parallel_safe_lst = rel->pathlist;
+	rel->pathlist = tmplst;
+	foreach(lc, parallel_safe_lst)
+	{
+		Path   *path = lfirst(lc);
+
+		if (!path->parallel_safe)
+			continue;
+
+		add_path(rel, (Path *) create_partial_tempscan_path(root, rel, path));
+	}
 
-	Assert(IsA(linitial(rel->partial_pathlist), CustomPath));
+	list_free(parallel_safe_lst);
 }
 
 void
@@ -287,3 +305,21 @@ _PG_init(void)
 
 	MarkGUCPrefixReserved(MODULENAME);
 }
+
+/* *****************************************************************************
+ *
+ * Parallel transport stuff
+ *
+ **************************************************************************** */
+
+/* copy from execParallel.c */
+#define PARALLEL_TUPLE_QUEUE_SIZE		65536
+
+static Size
+EstimateDSMTempScan(CustomScanState *node, ParallelContext *pcxt)
+{
+	Size size;
+
+	size = mul_size(PARALLEL_TUPLE_QUEUE_SIZE, pcxt->nworkers);
+	return size;
+}