Skip to content

Commit 7b4ac19

Browse files
robertmhaasrobertmhaas
committed
Extend index AM API for parallel index scans.
This patch doesn't actually make any index AM parallel-aware, but it provides the necessary functions at the AM layer to do so. Rahila Syed, Amit Kapila, Robert Haas
1 parent 587cda3 commit 7b4ac19

File tree

14 files changed

+262
-5
lines changed

14 files changed

+262
-5
lines changed

‎contrib/bloom/blutils.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,9 @@ blhandler(PG_FUNCTION_ARGS)
138138
amroutine->amendscan = blendscan;
139139
amroutine->ammarkpos = NULL;
140140
amroutine->amrestrpos = NULL;
141+
amroutine->amestimateparallelscan = NULL;
142+
amroutine->aminitparallelscan = NULL;
143+
amroutine->amparallelrescan = NULL;
141144

142145
PG_RETURN_POINTER(amroutine);
143146
}

‎doc/src/sgml/indexam.sgml

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,6 +131,11 @@ typedef struct IndexAmRoutine
131131
amendscan_function amendscan;
132132
ammarkpos_function ammarkpos; /* can be NULL */
133133
amrestrpos_function amrestrpos; /* can be NULL */
134+
135+
/* interface functions to support parallel index scans */
136+
amestimateparallelscan_function amestimateparallelscan; /* can be NULL */
137+
aminitparallelscan_function aminitparallelscan; /* can be NULL */
138+
amparallelrescan_function amparallelrescan; /* can be NULL */
134139
} IndexAmRoutine;
135140
</programlisting>
136141
</para>
@@ -624,6 +629,68 @@ amrestrpos (IndexScanDesc scan);
624629
the <structfield>amrestrpos</> field in its <structname>IndexAmRoutine</>
625630
struct may be set to NULL.
626631
</para>
632+
633+
<para>
634+
In addition to supporting ordinary index scans, some types of index
635+
may wish to support <firstterm>parallel index scans</>, which allow
636+
multiple backends to cooperate in performing an index scan. The
637+
index access method should arrange things so that each cooperating
638+
process returns a subset of the tuples that would be performed by
639+
an ordinary, non-parallel index scan, but in such a way that the
640+
union of those subsets is equal to the set of tuples that would be
641+
returned by an ordinary, non-parallel index scan. Furthermore, while
642+
there need not be any global ordering of tuples returned by a parallel
643+
scan, the ordering of that subset of tuples returned within each
644+
cooperating backend must match the requested ordering. The following
645+
functions may be implemented to support parallel index scans:
646+
</para>
647+
648+
<para>
649+
<programlisting>
650+
Size
651+
amestimateparallelscan (void);
652+
</programlisting>
653+
Estimate and return the number of bytes of dynamic shared memory which
654+
the access method will be needed to perform a parallel scan. (This number
655+
is in addition to, not in lieu of, the amount of space needed for
656+
AM-independent data in <structname>ParallelIndexScanDescData</>.)
657+
</para>
658+
659+
<para>
660+
It is not necessary to implement this function for access methods which
661+
do not support parallel scans or for which the number of additional bytes
662+
of storage required is zero.
663+
</para>
664+
665+
<para>
666+
<programlisting>
667+
void
668+
aminitparallelscan (void *target);
669+
</programlisting>
670+
This function will be called to initialize dynamic shared memory at the
671+
beginning of a parallel scan. <parameter>target</> will point to at least
672+
the number of bytes previously returned by
673+
<function>amestimateparallelscan</>, and this function may use that
674+
amount of space to store whatever data it wishes.
675+
</para>
676+
677+
<para>
678+
It is not necessary to implement this function for access methods which
679+
do not support parallel scans or in cases where the shared memory space
680+
required needs no initialization.
681+
</para>
682+
683+
<para>
684+
<programlisting>
685+
void
686+
amparallelrescan (IndexScanDesc scan);
687+
</programlisting>
688+
This function, if implemented, will be called when a parallel index scan
689+
must be restarted. It should reset any shared state set up by
690+
<function>aminitparallelscan</> such that the scan will be restarted from
691+
the beginning.
692+
</para>
693+
627694
</sect1>
628695

629696
<sect1 id="index-scanning">

‎src/backend/access/brin/brin.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,9 @@ brinhandler(PG_FUNCTION_ARGS)
112112
amroutine->amendscan = brinendscan;
113113
amroutine->ammarkpos = NULL;
114114
amroutine->amrestrpos = NULL;
115+
amroutine->amestimateparallelscan = NULL;
116+
amroutine->aminitparallelscan = NULL;
117+
amroutine->amparallelrescan = NULL;
115118

116119
PG_RETURN_POINTER(amroutine);
117120
}

‎src/backend/access/gin/ginutil.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ ginhandler(PG_FUNCTION_ARGS)
6868
amroutine->amendscan = ginendscan;
6969
amroutine->ammarkpos = NULL;
7070
amroutine->amrestrpos = NULL;
71+
amroutine->amestimateparallelscan = NULL;
72+
amroutine->aminitparallelscan = NULL;
73+
amroutine->amparallelrescan = NULL;
7174

7275
PG_RETURN_POINTER(amroutine);
7376
}

‎src/backend/access/gist/gist.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,9 @@ gisthandler(PG_FUNCTION_ARGS)
8989
amroutine->amendscan = gistendscan;
9090
amroutine->ammarkpos = NULL;
9191
amroutine->amrestrpos = NULL;
92+
amroutine->amestimateparallelscan = NULL;
93+
amroutine->aminitparallelscan = NULL;
94+
amroutine->amparallelrescan = NULL;
9295

9396
PG_RETURN_POINTER(amroutine);
9497
}

‎src/backend/access/hash/hash.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,9 @@ hashhandler(PG_FUNCTION_ARGS)
8686
amroutine->amendscan = hashendscan;
8787
amroutine->ammarkpos = NULL;
8888
amroutine->amrestrpos = NULL;
89+
amroutine->amestimateparallelscan = NULL;
90+
amroutine->aminitparallelscan = NULL;
91+
amroutine->amparallelrescan = NULL;
8992

9093
PG_RETURN_POINTER(amroutine);
9194
}

‎src/backend/access/index/indexam.c

Lines changed: 130 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,10 @@
2020
* index_insert - insert an index tuple into a relation
2121
* index_markpos - mark a scan position
2222
* index_restrpos - restore a scan position
23+
* index_parallelscan_estimate - estimate shared memory for parallel scan
24+
* index_parallelscan_initialize - initialize parallel scan
25+
* index_parallelrescan - (re)start a parallel scan of an index
26+
* index_beginscan_parallel - join parallel index scan
2327
* index_getnext_tid - get the next TID from a scan
2428
* index_fetch_heap - get the scan's next heap tuple
2529
* index_getnext - get the next heap tuple from a scan
@@ -120,7 +124,8 @@ do { \
120124
} while(0)
121125

122126
static IndexScanDesc index_beginscan_internal(Relation indexRelation,
123-
int nkeys, int norderbys, Snapshot snapshot);
127+
int nkeys, int norderbys, Snapshot snapshot,
128+
ParallelIndexScanDesc pscan, bool temp_snap);
124129

125130

126131
/* ----------------------------------------------------------------
@@ -219,7 +224,7 @@ index_beginscan(Relation heapRelation,
219224
{
220225
IndexScanDesc scan;
221226

222-
scan = index_beginscan_internal(indexRelation, nkeys, norderbys, snapshot);
227+
scan = index_beginscan_internal(indexRelation, nkeys, norderbys, snapshot, NULL, false);
223228

224229
/*
225230
* Save additional parameters into the scandesc. Everything else was set
@@ -244,7 +249,7 @@ index_beginscan_bitmap(Relation indexRelation,
244249
{
245250
IndexScanDesc scan;
246251

247-
scan = index_beginscan_internal(indexRelation, nkeys, 0, snapshot);
252+
scan = index_beginscan_internal(indexRelation, nkeys, 0, snapshot, NULL, false);
248253

249254
/*
250255
* Save additional parameters into the scandesc. Everything else was set
@@ -260,8 +265,11 @@ index_beginscan_bitmap(Relation indexRelation,
260265
*/
261266
static IndexScanDesc
262267
index_beginscan_internal(Relation indexRelation,
263-
int nkeys, int norderbys, Snapshot snapshot)
268+
int nkeys, int norderbys, Snapshot snapshot,
269+
ParallelIndexScanDesc pscan, bool temp_snap)
264270
{
271+
IndexScanDesc scan;
272+
265273
RELATION_CHECKS;
266274
CHECK_REL_PROCEDURE(ambeginscan);
267275

@@ -276,8 +284,13 @@ index_beginscan_internal(Relation indexRelation,
276284
/*
277285
* Tell the AM to open a scan.
278286
*/
279-
return indexRelation->rd_amroutine->ambeginscan(indexRelation, nkeys,
287+
scan = indexRelation->rd_amroutine->ambeginscan(indexRelation, nkeys,
280288
norderbys);
289+
/* Initialize information for parallel scan. */
290+
scan->parallel_scan = pscan;
291+
scan->xs_temp_snap = temp_snap;
292+
293+
return scan;
281294
}
282295

283296
/* ----------------
@@ -341,6 +354,9 @@ index_endscan(IndexScanDesc scan)
341354
/* Release index refcount acquired by index_beginscan */
342355
RelationDecrementReferenceCount(scan->indexRelation);
343356

357+
if (scan->xs_temp_snap)
358+
UnregisterSnapshot(scan->xs_snapshot);
359+
344360
/* Release the scan data structure itself */
345361
IndexScanEnd(scan);
346362
}
@@ -389,6 +405,115 @@ index_restrpos(IndexScanDesc scan)
389405
scan->indexRelation->rd_amroutine->amrestrpos(scan);
390406
}
391407

408+
/*
409+
* index_parallelscan_estimate - estimate shared memory for parallel scan
410+
*
411+
* Currently, we don't pass any information to the AM-specific estimator,
412+
* so it can probably only return a constant. In the future, we might need
413+
* to pass more information.
414+
*/
415+
Size
416+
index_parallelscan_estimate(Relation indexRelation, Snapshot snapshot)
417+
{
418+
Size nbytes;
419+
420+
RELATION_CHECKS;
421+
422+
nbytes = offsetof(ParallelIndexScanDescData, ps_snapshot_data);
423+
nbytes = add_size(nbytes, EstimateSnapshotSpace(snapshot));
424+
nbytes = MAXALIGN(nbytes);
425+
426+
/*
427+
* If amestimateparallelscan is not provided, assume there is no
428+
* AM-specific data needed. (It's hard to believe that could work, but
429+
* it's easy enough to cater to it here.)
430+
*/
431+
if (indexRelation->rd_amroutine->amestimateparallelscan != NULL)
432+
nbytes = add_size(nbytes,
433+
indexRelation->rd_amroutine->amestimateparallelscan());
434+
435+
return nbytes;
436+
}
437+
438+
/*
439+
* index_parallelscan_initialize - initialize parallel scan
440+
*
441+
* We initialize both the ParallelIndexScanDesc proper and the AM-specific
442+
* information which follows it.
443+
*
444+
* This function calls access method specific initialization routine to
445+
* initialize am specific information. Call this just once in the leader
446+
* process; then, individual workers attach via index_beginscan_parallel.
447+
*/
448+
void
449+
index_parallelscan_initialize(Relation heapRelation, Relation indexRelation,
450+
Snapshot snapshot, ParallelIndexScanDesc target)
451+
{
452+
Size offset;
453+
454+
RELATION_CHECKS;
455+
456+
offset = add_size(offsetof(ParallelIndexScanDescData, ps_snapshot_data),
457+
EstimateSnapshotSpace(snapshot));
458+
offset = MAXALIGN(offset);
459+
460+
target->ps_relid = RelationGetRelid(heapRelation);
461+
target->ps_indexid = RelationGetRelid(indexRelation);
462+
target->ps_offset = offset;
463+
SerializeSnapshot(snapshot, target->ps_snapshot_data);
464+
465+
/* aminitparallelscan is optional; assume no-op if not provided by AM */
466+
if (indexRelation->rd_amroutine->aminitparallelscan != NULL)
467+
{
468+
void *amtarget;
469+
470+
amtarget = OffsetToPointer(target, offset);
471+
indexRelation->rd_amroutine->aminitparallelscan(amtarget);
472+
}
473+
}
474+
475+
/* ----------------
476+
* index_parallelrescan - (re)start a parallel scan of an index
477+
* ----------------
478+
*/
479+
void
480+
index_parallelrescan(IndexScanDesc scan)
481+
{
482+
SCAN_CHECKS;
483+
484+
/* amparallelrescan is optional; assume no-op if not provided by AM */
485+
if (scan->indexRelation->rd_amroutine->amparallelrescan != NULL)
486+
scan->indexRelation->rd_amroutine->amparallelrescan(scan);
487+
}
488+
489+
/*
490+
* index_beginscan_parallel - join parallel index scan
491+
*
492+
* Caller must be holding suitable locks on the heap and the index.
493+
*/
494+
IndexScanDesc
495+
index_beginscan_parallel(Relation heaprel, Relation indexrel, int nkeys,
496+
int norderbys, ParallelIndexScanDesc pscan)
497+
{
498+
Snapshot snapshot;
499+
IndexScanDesc scan;
500+
501+
Assert(RelationGetRelid(heaprel) == pscan->ps_relid);
502+
snapshot = RestoreSnapshot(pscan->ps_snapshot_data);
503+
RegisterSnapshot(snapshot);
504+
scan = index_beginscan_internal(indexrel, nkeys, norderbys, snapshot,
505+
pscan, true);
506+
507+
/*
508+
* Save additional parameters into the scandesc. Everything else was set
509+
* up by index_beginscan_internal.
510+
*/
511+
scan->heapRelation = heaprel;
512+
scan->xs_snapshot = snapshot;
513+
514+
return scan;
515+
}
516+
392517
/* ----------------
393518
* index_getnext_tid - get the next TID from a scan
394519
*

‎src/backend/access/nbtree/nbtree.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -118,6 +118,9 @@ bthandler(PG_FUNCTION_ARGS)
118118
amroutine->amendscan = btendscan;
119119
amroutine->ammarkpos = btmarkpos;
120120
amroutine->amrestrpos = btrestrpos;
121+
amroutine->amestimateparallelscan = NULL;
122+
amroutine->aminitparallelscan = NULL;
123+
amroutine->amparallelrescan = NULL;
121124

122125
PG_RETURN_POINTER(amroutine);
123126
}

‎src/backend/access/spgist/spgutils.c

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -68,6 +68,9 @@ spghandler(PG_FUNCTION_ARGS)
6868
amroutine->amendscan = spgendscan;
6969
amroutine->ammarkpos = NULL;
7070
amroutine->amrestrpos = NULL;
71+
amroutine->amestimateparallelscan = NULL;
72+
amroutine->aminitparallelscan = NULL;
73+
amroutine->amparallelrescan = NULL;
7174

7275
PG_RETURN_POINTER(amroutine);
7376
}

‎src/include/access/amapi.h

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -137,6 +137,18 @@ typedef void (*ammarkpos_function) (IndexScanDesc scan);
137137
/* restore marked scan position */
138138
typedef void (*amrestrpos_function) (IndexScanDesc scan);
139139

140+
/*
141+
* Callback function signatures - for parallel index scans.
142+
*/
143+
144+
/* estimate size of parallel scan descriptor */
145+
typedef Size (*amestimateparallelscan_function) (void);
146+
147+
/* prepare for parallel index scan */
148+
typedef void (*aminitparallelscan_function) (void *target);
149+
150+
/* (re)start parallel index scan */
151+
typedef void (*amparallelrescan_function) (IndexScanDesc scan);
140152

141153
/*
142154
* API struct for an index AM. Note this must be stored in a single palloc'd
@@ -196,6 +208,11 @@ typedef struct IndexAmRoutine
196208
amendscan_function amendscan;
197209
ammarkpos_function ammarkpos; /* can be NULL */
198210
amrestrpos_function amrestrpos; /* can be NULL */
211+
212+
/* interface functions to support parallel index scans */
213+
amestimateparallelscan_function amestimateparallelscan; /* can be NULL */
214+
aminitparallelscan_function aminitparallelscan; /* can be NULL */
215+
amparallelrescan_function amparallelrescan; /* can be NULL */
199216
} IndexAmRoutine;
200217

201218

‎src/include/access/genam.h

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,8 @@ typedef bool (*IndexBulkDeleteCallback) (ItemPointer itemptr, void *state);
8383
typedef struct IndexScanDescData *IndexScanDesc;
8484
typedef struct SysScanDescData *SysScanDesc;
8585

86+
typedef struct ParallelIndexScanDescData *ParallelIndexScanDesc;
87+
8688
/*
8789
* Enumeration specifying the type of uniqueness check to perform in
8890
* index_insert().
@@ -144,6 +146,13 @@ extern void index_rescan(IndexScanDesc scan,
144146
extern void index_endscan(IndexScanDesc scan);
145147
extern void index_markpos(IndexScanDesc scan);
146148
extern void index_restrpos(IndexScanDesc scan);
149+
extern Size index_parallelscan_estimate(Relation indexrel, Snapshot snapshot);
150+
extern void index_parallelscan_initialize(Relation heaprel, Relation indexrel,
151+
Snapshot snapshot, ParallelIndexScanDesc target);
152+
extern void index_parallelrescan(IndexScanDesc scan);
153+
extern IndexScanDesc index_beginscan_parallel(Relation heaprel,
154+
Relation indexrel, int nkeys, int norderbys,
155+
ParallelIndexScanDesc pscan);
147156
extern ItemPointer index_getnext_tid(IndexScanDesc scan,
148157
ScanDirection direction);
149158
extern HeapTuple index_fetch_heap(IndexScanDesc scan);

0 commit comments

Comments
 (0)