-> 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)

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

->ParallelHash Join

Hash Cond: (t1.x =t2.x)

->CustomScan(nodeCustomTempScan)

->Parallel Seq Scan on parallel_test_tmp t1

->Parallel Hash

->ParallelSeq Scan onparallel_test t2

-> Hash Join

Hash Cond: (t2.x =t1.x)

->Parallel SeqScanon parallel_test t2

->Hash

->Custom Scan (nodeCustomTempScan)

-> Seq Scan onparallel_test_tmp t1

(10 rows)

-- Just see how merge join manages custom parallel scan path

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 JOINparallel_test t2

SELECT count(*) FROM parallel_test t1 NATURAL JOINparallel_test_tmp t2

WHERE t1.x < 10;

QUERY PLAN

-------------------------------------------------------------------------------------

QUERY PLAN

-----------------------------------------------------------------------------------------------

Aggregate

-> Gather

Workers Planned:1

Workers Planned:3

-> Nested Loop

-> ParallelIndex Scan using parallel_test_x_idx on parallel_test t1

Index Cond: (x < 10)

->Index Scanusing parallel_test_x_idx on parallel_test t2

IndexCond: (x = t1.x)

Filter: (y = t1.y)

-> ParallelSeq Scan on parallel_test t1

Filter: (x < 10)

->Custom Scan(nodeCustomTempScan)

->IndexOnly 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;

#defineMODULENAME"tempscan"

#defineNODENAME"nodeCustomTempScan"

#defineTEMPSCAN_CPU_COST_MULTIPLIER(0.5)

staticPlan*create_partial_tempscan_plan(PlannerInfo*root,

RelOptInfo*rel,

CustomPath*best_path,

staticTupleTableSlot*ExecTempScan(CustomScanState*node);

staticvoidEndTempScan(CustomScanState*node);

staticvoidReScanTempScan(CustomScanState*node);

staticSizeEstimateDSMTempScan(CustomScanState*node,ParallelContext*pcxt);

{

.ReScanCustomScan=ReScanTempScan,

.MarkPosCustomScan=NULL,

.RestrPosCustomScan=NULL,

.EstimateDSMCustomScan=NULL,

.EstimateDSMCustomScan=EstimateDSMTempScan,

.InitializeDSMCustomScan=NULL,

.ReInitializeDSMCustomScan=NULL,

.InitializeWorkerCustomScan=NULL,

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 */

pathnode->param_info=NULL;

pathnode->param_info=path->param_info;

pathnode->parallel_safe= true;

pathnode->parallel_aware= false;

pathnode->parallel_workers=path->parallel_workers;

pathnode->startup_cost=path->startup_cost;

pathnode->total_cost=path->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;

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;

try_partial_tempscan(PlannerInfo*root,RelOptInfo*rel,Indexrti,

RangeTblEntry*rte)

{

intparallel_workers;

ListCell*lc;

List*partial_pathlist_new=NIL;

List*parallel_safe_lst=NIL;

List*tmplst=rel->pathlist;

get_rel_persistence(rte->relid)!=RELPERSISTENCE_TEMP)

return;

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 (parallel_workers <=0)

return;

Assert(rel->partial_pathlist==NIL);

rel->consider_parallel= true;

add_partial_path(rel, (Path*)

create_seqscan_path(root,rel,NULL,parallel_workers));

rel->pathlist=NIL;

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);

lappend(partial_pathlist_new,

(void*)create_partial_tempscan_path(root,rel,path));

}

list_free(rel->partial_pathlist);

rel->partial_pathlist=NIL;

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);

}

MarkGUCPrefixReserved(MODULENAME);

}

#definePARALLEL_TUPLE_QUEUE_SIZE65536

EstimateDSMTempScan(CustomScanState*node,ParallelContext*pcxt)

{

Sizesize;

size=mul_size(PARALLEL_TUPLE_QUEUE_SIZE,pcxt->nworkers);

returnsize;

}