Skip to content

Commit 920933d

Browse files
roseduanroseduan
roseduan
authored and
roseduan
committed
Feature: dispatch T_CustomScanState in CBDB parallel walkers
Three GP-side walkers used during parallel setup — EstimateGpParallelDSMEntrySize, InitializeGpParallelWorkers, InitializeGpParallelDSMEntry — cased on other scan/join states but skipped T_CustomScanState. A parallel_aware CustomScan therefore silently under-sized its DSM and failed to attach in workers, with no diagnostic. Add T_CustomScanState arms that dispatch to ExecCustomScan{Estimate,InitializeDSM,InitializeWorker}, gated on parallel_aware to match the other node types in these switches. Also handle T_CustomScanState in planstate_walk_kids. Upstream planstate_tree_walker walks only css->custom_ps for a CustomScanState; the CBDB walker previously fell through to default (lefttree/righttree), skipping custom_ps children under stateful walkers (cdbexplain_*, getMotionState, ...). The new case walks custom_ps first, then lefttree/righttree if set, with an Assert that a CustomScanState never populates both — otherwise the child would be walked twice.
1 parent 18d2ba7 commit 920933d

9 files changed

Lines changed: 768 additions & 1 deletion

File tree

‎src/backend/executor/execParallel.c‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1551,6 +1551,10 @@ EstimateGpParallelDSMEntrySize(PlanState *planstate, ParallelContext *pctx)
15511551
case T_SortState:
15521552
ExecSortEstimate((SortState *) planstate, pctx);
15531553
break;
1554+
case T_CustomScanState:
1555+
if (planstate->plan->parallel_aware)
1556+
ExecCustomScanEstimate((CustomScanState *) planstate, pctx);
1557+
break;
15541558
default:
15551559
break;
15561560

@@ -1604,6 +1608,10 @@ InitializeGpParallelWorkers(PlanState *planstate, ParallelWorkerContext *pwcxt)
16041608
if (planstate->plan->parallel_aware)
16051609
ExecHashJoinInitializeWorker((HashJoinState *) planstate, pwcxt);
16061610
break;
1611+
case T_CustomScanState:
1612+
if (planstate->plan->parallel_aware)
1613+
ExecCustomScanInitializeWorker((CustomScanState *) planstate, pwcxt);
1614+
break;
16071615
default:
16081616
break;
16091617
}
@@ -1662,6 +1670,10 @@ InitializeGpParallelDSMEntry(PlanState *planstate, ParallelContext *pctx)
16621670
case T_SortState:
16631671
ExecSortInitializeDSM((SortState *) planstate, pctx);
16641672
break;
1673+
case T_CustomScanState:
1674+
if (planstate->plan->parallel_aware)
1675+
ExecCustomScanInitializeDSM((CustomScanState *) planstate, pctx);
1676+
break;
16651677
default:
16661678
break;
16671679
}

‎src/backend/executor/execProcnode.c‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1295,6 +1295,28 @@ planstate_walk_kids(PlanState *planstate,
12951295
Assert(!planstate->lefttree && !planstate->righttree);
12961296
break;
12971297

1298+
case T_CustomScanState:
1299+
{
1300+
CustomScanState *css = (CustomScanState *) planstate;
1301+
ListCell *lc;
1302+
1303+
Assert(!(css->custom_ps != NIL &&
1304+
(planstate->lefttree || planstate->righttree)));
1305+
1306+
v = CdbVisit_Walk;
1307+
foreach(lc, css->custom_ps)
1308+
{
1309+
v = planstate_walk_node_extended((PlanState *) lfirst(lc), walker, context, flags);
1310+
if (v != CdbVisit_Walk)
1311+
break;
1312+
}
1313+
if (v == CdbVisit_Walk && planstate->lefttree)
1314+
v = planstate_walk_node_extended(planstate->lefttree, walker, context, flags);
1315+
if (v == CdbVisit_Walk && planstate->righttree)
1316+
v = planstate_walk_node_extended(planstate->righttree, walker, context, flags);
1317+
break;
1318+
}
1319+
12981320
default:
12991321
/* Left subtree */
13001322
v = planstate_walk_node_extended(planstate->lefttree, walker, context, flags);

‎src/test/modules/Makefile‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ SUBDIRS = \
4343
# special check for DML on system relations in GPDB
4444

4545
# GPDB subdirs
46-
SUBDIRS += test_planner
46+
SUBDIRS += test_planner parallel_customscan
4747
ifeq ($(with_ssl),openssl)
4848
SUBDIRS += ssl_passphrase_callback
4949
else

‎src/test/modules/parallel_customscan/Makefile‎

Lines changed: 29 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,29 @@
1+
# src/test/modules/parallel_customscan/Makefile
2+
3+
MODULE_big = parallel_customscan
4+
OBJS = \
5+
$(WIN32RES) \
6+
parallel_customscan.o
7+
8+
EXTENSION = parallel_customscan
9+
DATA = parallel_customscan--1.0.sql
10+
PGFILEDESC = "parallel_customscan - exercise parallel CustomScan dispatch"
11+
12+
REGRESS = parallel_customscan
13+
14+
# Run against an existing cluster (gpdemo). The cluster must have
15+
# 'parallel_customscan' in shared_preload_libraries so segment backends
16+
# and parallel workers have the CustomScan methods registered:
17+
# gpconfig -c shared_preload_libraries -v "'time_series,parallel_customscan'"
18+
# gpstop -ra
19+
20+
ifdef USE_PGXS
21+
PG_CONFIG = pg_config
22+
PGXS := $(shell $(PG_CONFIG) --pgxs)
23+
include $(PGXS)
24+
else
25+
subdir = src/test/modules/parallel_customscan
26+
top_builddir = ../../../..
27+
include $(top_builddir)/src/Makefile.global
28+
include $(top_srcdir)/contrib/contrib-global.mk
29+
endif

‎src/test/modules/parallel_customscan/expected/parallel_customscan.out‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
1+
-- start_matchsubs
2+
-- m/\(actual rows=[^)]*\)/
3+
-- s/\(actual rows=[^)]*\)/(actual rows=...)/
4+
-- end_matchsubs
5+
-- start_matchignore
6+
-- m/^\s*Buckets: \d+ Batches: \d+ Memory Usage: \d+kB/
7+
-- end_matchignore
8+
CREATE EXTENSION parallel_customscan;
9+
-- Set up data BEFORE enabling our hook so ANALYZE doesn't traverse it.
10+
CREATE TABLE pcs_t (a int)
11+
WITH (parallel_workers = 2)
12+
DISTRIBUTED BY (a);
13+
INSERT INTO pcs_t SELECT generate_series(1, 1000);
14+
ANALYZE pcs_t;
15+
SET optimizer = off;
16+
SET enable_parallel = on;
17+
SET enable_seqscan = off;
18+
SET min_parallel_table_scan_size = 0;
19+
SET parallel_setup_cost = 0;
20+
SET parallel_tuple_cost = 0;
21+
SET max_parallel_workers_per_gather = 2;
22+
-- (1) Enabling the hook replaces the Parallel Seq Scan with a Custom Scan.
23+
EXPLAIN (COSTS OFF) SELECT count(*) FROM pcs_t;
24+
QUERY PLAN
25+
------------------------------------------------
26+
Finalize Aggregate
27+
-> Gather Motion 6:1 (slice1; segments: 6)
28+
-> Partial Aggregate
29+
-> Parallel Seq Scan on pcs_t
30+
Optimizer: Postgres query optimizer
31+
(5 rows)
32+
33+
SET parallel_customscan.enabled = on;
34+
EXPLAIN (COSTS OFF) SELECT count(*) FROM pcs_t;
35+
QUERY PLAN
36+
-------------------------------------------------------------
37+
Finalize Aggregate
38+
-> Gather Motion 6:1 (slice1; segments: 6)
39+
-> Partial Aggregate
40+
-> Parallel Custom Scan (ParallelCustomScan)
41+
-> Parallel Seq Scan on pcs_t
42+
Optimizer: Postgres query optimizer
43+
(6 rows)
44+
45+
-- (2) Correctness: results match only if every parallel worker runs our scan.
46+
SELECT count(*) FROM pcs_t;
47+
count
48+
-------
49+
1000
50+
(1 row)
51+
52+
SELECT sum(a) FROM pcs_t;
53+
sum
54+
--------
55+
500500
56+
(1 row)
57+
58+
-- (3) Scan-level qualifier and projection through the custom scan.
59+
EXPLAIN (COSTS OFF) SELECT a FROM pcs_t WHERE a > 990;
60+
QUERY PLAN
61+
-------------------------------------------------
62+
Gather Motion 6:1 (slice1; segments: 6)
63+
-> Parallel Custom Scan (ParallelCustomScan)
64+
-> Parallel Seq Scan on pcs_t
65+
Filter: (a > 990)
66+
Optimizer: Postgres query optimizer
67+
(5 rows)
68+
69+
SELECT a FROM pcs_t WHERE a > 990 ORDER BY a;
70+
a
71+
------
72+
991
73+
992
74+
993
75+
994
76+
995
77+
996
78+
997
79+
998
80+
999
81+
1000
82+
(10 rows)
83+
84+
SELECT count(*) FROM pcs_t WHERE a % 2 = 0;
85+
count
86+
-------
87+
500
88+
(1 row)
89+
90+
-- (4) Join with both inputs scanned by the custom scan, plus a scan-level qual.
91+
CREATE TABLE pcs_t2 (a int)
92+
WITH (parallel_workers = 2)
93+
DISTRIBUTED BY (a);
94+
INSERT INTO pcs_t2 SELECT generate_series(1, 500);
95+
ANALYZE pcs_t2;
96+
EXPLAIN (COSTS OFF)
97+
SELECT count(*) FROM pcs_t x JOIN pcs_t2 y ON x.a = y.a WHERE x.a <= 100;
98+
QUERY PLAN
99+
-------------------------------------------------------------------------
100+
Finalize Aggregate
101+
-> Gather Motion 6:1 (slice1; segments: 6)
102+
-> Partial Aggregate
103+
-> Parallel Hash Join
104+
Hash Cond: (x.a = y.a)
105+
-> Parallel Custom Scan (ParallelCustomScan)
106+
-> Parallel Seq Scan on pcs_t x
107+
Filter: (a <= 100)
108+
-> Parallel Hash
109+
-> Parallel Custom Scan (ParallelCustomScan)
110+
-> Parallel Seq Scan on pcs_t2 y
111+
Filter: (a <= 100)
112+
Optimizer: Postgres query optimizer
113+
(13 rows)
114+
115+
SELECT count(*) FROM pcs_t x JOIN pcs_t2 y ON x.a = y.a WHERE x.a <= 100;
116+
count
117+
-------
118+
100
119+
(1 row)
120+
121+
-- (5) Empty relation: the custom scan must handle an immediate end-of-scan.
122+
CREATE TABLE pcs_empty (a int)
123+
WITH (parallel_workers = 2)
124+
DISTRIBUTED BY (a);
125+
ANALYZE pcs_empty;
126+
EXPLAIN (COSTS OFF) SELECT count(*) FROM pcs_empty;
127+
QUERY PLAN
128+
------------------------------------------------
129+
Aggregate
130+
-> Gather Motion 3:1 (slice1; segments: 3)
131+
-> Custom Scan (ParallelCustomScan)
132+
-> Seq Scan on pcs_empty
133+
Optimizer: Postgres query optimizer
134+
(5 rows)
135+
136+
SELECT count(*) FROM pcs_empty;
137+
count
138+
-------
139+
0
140+
(1 row)
141+
142+
SELECT * FROM pcs_empty;
143+
a
144+
---
145+
(0 rows)
146+
147+
-- (6) Serial path: with no workers, a non-parallel Custom Scan is used.
148+
SET max_parallel_workers_per_gather = 0;
149+
EXPLAIN (COSTS OFF) SELECT count(*) FROM pcs_t;
150+
QUERY PLAN
151+
----------------------------------------------------
152+
Finalize Aggregate
153+
-> Gather Motion 3:1 (slice1; segments: 3)
154+
-> Partial Aggregate
155+
-> Custom Scan (ParallelCustomScan)
156+
-> Seq Scan on pcs_t
157+
Optimizer: Postgres query optimizer
158+
(6 rows)
159+
160+
SELECT count(*) FROM pcs_t;
161+
count
162+
-------
163+
1000
164+
(1 row)
165+
166+
SELECT a FROM pcs_t WHERE a > 995 ORDER BY a;
167+
a
168+
------
169+
996
170+
997
171+
998
172+
999
173+
1000
174+
(5 rows)
175+
176+
SET max_parallel_workers_per_gather = 2;
177+
-- (7) EXPLAIN ANALYZE: exercises planstate_walk_kids' custom_ps recursion.
178+
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF, SUMMARY OFF)
179+
SELECT count(*) FROM pcs_t;
180+
QUERY PLAN
181+
---------------------------------------------------------------------------------------
182+
Finalize Aggregate (actual rows=1 loops=1)
183+
-> Gather Motion 6:1 (slice1; segments: 6) (actual rows=6 loops=1)
184+
-> Partial Aggregate (actual rows=1 loops=1)
185+
-> Parallel Custom Scan (ParallelCustomScan) (actual rows=340 loops=1)
186+
-> Parallel Seq Scan on pcs_t (actual rows=340 loops=1)
187+
Optimizer: Postgres query optimizer
188+
(6 rows)
189+
190+
EXPLAIN (ANALYZE, TIMING OFF, COSTS OFF, SUMMARY OFF)
191+
SELECT count(*) FROM pcs_t x JOIN pcs_t2 y ON x.a = y.a;
192+
QUERY PLAN
193+
-------------------------------------------------------------------------------------------------
194+
Finalize Aggregate (actual rows=1 loops=1)
195+
-> Gather Motion 6:1 (slice1; segments: 6) (actual rows=6 loops=1)
196+
-> Partial Aggregate (actual rows=1 loops=1)
197+
-> Parallel Hash Join (actual rows=0 loops=1)
198+
Hash Cond: (x.a = y.a)
199+
-> Parallel Custom Scan (ParallelCustomScan) (actual rows=340 loops=1)
200+
-> Parallel Seq Scan on pcs_t x (actual rows=340 loops=1)
201+
-> Parallel Hash (actual rows=0 loops=1)
202+
Buckets: 524288 Batches: 1 Memory Usage: 4128kB
203+
-> Parallel Custom Scan (ParallelCustomScan) (actual rows=0 loops=1)
204+
-> Parallel Seq Scan on pcs_t2 y (actual rows=0 loops=1)
205+
Optimizer: Postgres query optimizer
206+
(12 rows)
207+
208+
-- cleanup
209+
DROP TABLE pcs_t2;
210+
DROP TABLE pcs_empty;
211+
DROP TABLE pcs_t;
212+
DROP EXTENSION parallel_customscan;

‎src/test/modules/parallel_customscan/parallel_customscan--1.0.sql‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,15 @@
1+
/* src/test/modules/parallel_customscan/parallel_customscan--1.0.sql */
2+
3+
-- Complain if script is sourced in psql, rather than via CREATE EXTENSION
4+
\echo Use "CREATE EXTENSION parallel_customscan" to load this file. \quit
5+
6+
CREATE FUNCTION pcs_get_hook_calls(
7+
OUT estimate_calls bigint,
8+
OUT init_dsm_calls bigint,
9+
OUT reinit_dsm_calls bigint,
10+
OUT init_worker_calls bigint,
11+
OUT shutdown_calls bigint
12+
)
13+
RETURNS record
14+
AS 'MODULE_PATHNAME', 'pcs_get_hook_calls'
15+
LANGUAGE C STRICT;

0 commit comments

Comments
 (0)