diff --git a/contrib/bloom/blutils.c b/contrib/bloom/blutils.c index 06077af..f5ee7bc 100644 --- a/contrib/bloom/blutils.c +++ b/contrib/bloom/blutils.c @@ -131,8 +131,11 @@ blhandler(PG_FUNCTION_ARGS) amroutine->amoptions = bloptions; amroutine->amproperty = NULL; amroutine->amvalidate = blvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = blbeginscan; amroutine->amrescan = blrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = NULL; amroutine->amgetbitmap = blgetbitmap; amroutine->amendscan = blendscan; diff --git a/doc/src/sgml/indexam.sgml b/doc/src/sgml/indexam.sgml index 40f201b..df71c06 100644 --- a/doc/src/sgml/indexam.sgml +++ b/doc/src/sgml/indexam.sgml @@ -124,8 +124,11 @@ typedef struct IndexAmRoutine amoptions_function amoptions; amproperty_function amproperty; /* can be NULL */ amvalidate_function amvalidate; + amestimateparallelscan_function amestimateparallelscan; /* can be NULL */ + aminitparallelscan_function aminitparallelscan; /* can be NULL */ ambeginscan_function ambeginscan; amrescan_function amrescan; + amparallelrescan_function amparallelrescan; /* can be NULL */ amgettuple_function amgettuple; /* can be NULL */ amgetbitmap_function amgetbitmap; /* can be NULL */ amendscan_function amendscan; @@ -459,6 +462,35 @@ amvalidate (Oid opclassoid); invalid. Problems should be reported with ereport messages. + + +Size +amestimateparallelscan (void); + + Estimate and return the storage space required for parallel index scan. + The size of index-type-specific parallel information will be returned back + to caller. + + + + +void +aminitparallelscan (void *target); + + Initialize the parallel index scan state. This function will be used to + initialize index-type-specific parallel information which will be stored + immediately after generic parallel information required for parallel index + scans. The required state information will be set in target. + + + + The aminitparallelscan and amestimateparallelscan + functions need only be provided if the access method supports parallel + index scans. If it doesn't, the aminitparallelscan and + amestimateparallelscan fields in its IndexAmRoutine + struct must be set to NULL. Note that these fields can be NULL even for + parallel index scans. + The purpose of an index, of course, is to support scans for tuples matching @@ -511,6 +543,24 @@ amrescan (IndexScanDesc scan, +void +amparallelrescan (IndexScanDesc scan); + + Restart the parallel index scan. This function resets the parallel index + scan state. It must be called only during restart of scan which will be + typically required for the inner side of nest-loop join. + + + + The amparallelrescan function need only be provided if the + access method supports parallel index scans. If it doesn't, + the amparallelrescan field in its IndexAmRoutine + struct must be set to NULL. Note that this field can be NULL even for + parallel index scans. + + + + boolean amgettuple (IndexScanDesc scan, ScanDirection direction); diff --git a/doc/src/sgml/monitoring.sgml b/doc/src/sgml/monitoring.sgml index 1545f03..8451eb4 100644 --- a/doc/src/sgml/monitoring.sgml +++ b/doc/src/sgml/monitoring.sgml @@ -1199,7 +1199,7 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser Waiting in an extension. - IPC + IPC BgWorkerShutdown Waiting for background worker to shut down. @@ -1232,6 +1232,11 @@ postgres 27093 0.0 0.0 30096 2752 ? Ss 11:34 0:00 postgres: ser Waiting for parallel workers to finish computing. + ParallelBtreePage + Waiting for the page number needed to continue a parallel btree scan + to become available. + + SafeSnapshot Waiting for a snapshot for a READ ONLY DEFERRABLE transaction. diff --git a/src/backend/access/brin/brin.c b/src/backend/access/brin/brin.c index d60ddd2..826f3cc 100644 --- a/src/backend/access/brin/brin.c +++ b/src/backend/access/brin/brin.c @@ -105,8 +105,11 @@ brinhandler(PG_FUNCTION_ARGS) amroutine->amoptions = brinoptions; amroutine->amproperty = NULL; amroutine->amvalidate = brinvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = brinbeginscan; amroutine->amrescan = brinrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = NULL; amroutine->amgetbitmap = bringetbitmap; amroutine->amendscan = brinendscan; diff --git a/src/backend/access/gin/ginutil.c b/src/backend/access/gin/ginutil.c index 3909638..a80039f 100644 --- a/src/backend/access/gin/ginutil.c +++ b/src/backend/access/gin/ginutil.c @@ -61,8 +61,11 @@ ginhandler(PG_FUNCTION_ARGS) amroutine->amoptions = ginoptions; amroutine->amproperty = NULL; amroutine->amvalidate = ginvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = ginbeginscan; amroutine->amrescan = ginrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = NULL; amroutine->amgetbitmap = gingetbitmap; amroutine->amendscan = ginendscan; diff --git a/src/backend/access/gist/gist.c b/src/backend/access/gist/gist.c index 597056a..54f9dc4 100644 --- a/src/backend/access/gist/gist.c +++ b/src/backend/access/gist/gist.c @@ -82,8 +82,11 @@ gisthandler(PG_FUNCTION_ARGS) amroutine->amoptions = gistoptions; amroutine->amproperty = gistproperty; amroutine->amvalidate = gistvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = gistbeginscan; amroutine->amrescan = gistrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = gistgettuple; amroutine->amgetbitmap = gistgetbitmap; amroutine->amendscan = gistendscan; diff --git a/src/backend/access/hash/hash.c b/src/backend/access/hash/hash.c index a64a9b9..17fd125 100644 --- a/src/backend/access/hash/hash.c +++ b/src/backend/access/hash/hash.c @@ -79,8 +79,11 @@ hashhandler(PG_FUNCTION_ARGS) amroutine->amoptions = hashoptions; amroutine->amproperty = NULL; amroutine->amvalidate = hashvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = hashbeginscan; amroutine->amrescan = hashrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = hashgettuple; amroutine->amgetbitmap = hashgetbitmap; amroutine->amendscan = hashendscan; diff --git a/src/backend/access/index/indexam.c b/src/backend/access/index/indexam.c index 4822af9..a19bb49 100644 --- a/src/backend/access/index/indexam.c +++ b/src/backend/access/index/indexam.c @@ -120,7 +120,8 @@ do { \ } while(0) static IndexScanDesc index_beginscan_internal(Relation indexRelation, - int nkeys, int norderbys, Snapshot snapshot); + int nkeys, int norderbys, Snapshot snapshot, + ParallelIndexScanDesc pscan, bool temp_snap); /* ---------------------------------------------------------------- @@ -207,6 +208,89 @@ index_insert(Relation indexRelation, } /* + * index_parallelscan_estimate - estimate storage for ParallelIndexScanDesc + * + * This function calls am specific routine to obtain size of am specific + * shared information. + */ +Size +index_parallelscan_estimate(Relation indexrel, Snapshot snapshot) +{ + Size index_size; + Size amindex_size; + + index_size = offsetof(ParallelIndexScanDescData, ps_snapshot_data); + index_size = add_size(index_size, EstimateSnapshotSpace(snapshot)); + index_size = MAXALIGN(index_size); + + /* amestimateparallelscan is optional; assume no-op if not provided by AM */ + if (indexrel->rd_amroutine->amestimateparallelscan == NULL) + return index_size; + + amindex_size = indexrel->rd_amroutine->amestimateparallelscan(); + return add_size(index_size, amindex_size); +} + +/* + * index_parallelscan_initialize - initialize ParallelIndexScanDesc + * + * This function calls access method specific initialization routine to + * initialize am specific information. Call this just once in the leader + * process; then, individual workers attach via index_beginscan_parallel. + */ +void +index_parallelscan_initialize(Relation heaprel, Relation indexrel, + Snapshot snapshot, ParallelIndexScanDesc target) +{ + Size offset; + void *amtarget; + + offset = add_size(offsetof(ParallelIndexScanDescData, ps_snapshot_data), + EstimateSnapshotSpace(snapshot)); + offset = MAXALIGN(offset); + + target->ps_relid = RelationGetRelid(heaprel); + target->ps_indexid = RelationGetRelid(indexrel); + target->ps_offset = offset; + SerializeSnapshot(snapshot, target->ps_snapshot_data); + + /* aminitparallelscan is optional; assume no-op if not provided by AM */ + if (indexrel->rd_amroutine->aminitparallelscan == NULL) + return; + + amtarget = (char *) ((void *) target) + offset; + indexrel->rd_amroutine->aminitparallelscan(amtarget); +} + +/* + * index_beginscan_parallel - join parallel index scan + * + * Caller must be holding suitable locks on the heap and the index. + */ +IndexScanDesc +index_beginscan_parallel(Relation heaprel, Relation indexrel, int nkeys, + int norderbys, ParallelIndexScanDesc pscan) +{ + Snapshot snapshot; + IndexScanDesc scan; + + Assert(RelationGetRelid(heaprel) == pscan->ps_relid); + snapshot = RestoreSnapshot(pscan->ps_snapshot_data); + RegisterSnapshot(snapshot); + scan = index_beginscan_internal(indexrel, nkeys, norderbys, snapshot, + pscan, true); + + /* + * Save additional parameters into the scandesc. Everything else was set + * up by RelationGetIndexScan. + */ + scan->heapRelation = heaprel; + scan->xs_snapshot = snapshot; + + return scan; +} + +/* * index_beginscan - start a scan of an index with amgettuple * * Caller must be holding suitable locks on the heap and the index. @@ -219,7 +303,7 @@ index_beginscan(Relation heapRelation, { IndexScanDesc scan; - scan = index_beginscan_internal(indexRelation, nkeys, norderbys, snapshot); + scan = index_beginscan_internal(indexRelation, nkeys, norderbys, snapshot, NULL, false); /* * Save additional parameters into the scandesc. Everything else was set @@ -244,7 +328,7 @@ index_beginscan_bitmap(Relation indexRelation, { IndexScanDesc scan; - scan = index_beginscan_internal(indexRelation, nkeys, 0, snapshot); + scan = index_beginscan_internal(indexRelation, nkeys, 0, snapshot, NULL, false); /* * Save additional parameters into the scandesc. Everything else was set @@ -260,8 +344,11 @@ index_beginscan_bitmap(Relation indexRelation, */ static IndexScanDesc index_beginscan_internal(Relation indexRelation, - int nkeys, int norderbys, Snapshot snapshot) + int nkeys, int norderbys, Snapshot snapshot, + ParallelIndexScanDesc pscan, bool temp_snap) { + IndexScanDesc scan; + RELATION_CHECKS; CHECK_REL_PROCEDURE(ambeginscan); @@ -276,8 +363,13 @@ index_beginscan_internal(Relation indexRelation, /* * Tell the AM to open a scan. */ - return indexRelation->rd_amroutine->ambeginscan(indexRelation, nkeys, + scan = indexRelation->rd_amroutine->ambeginscan(indexRelation, nkeys, norderbys); + /* Initialize information for parallel scan. */ + scan->parallel_scan = pscan; + scan->xs_temp_snap = temp_snap; + + return scan; } /* ---------------- @@ -319,6 +411,22 @@ index_rescan(IndexScanDesc scan, } /* ---------------- + * index_parallelrescan - (re)start a parallel scan of an index + * ---------------- + */ +void +index_parallelrescan(IndexScanDesc scan) +{ + SCAN_CHECKS; + + /* amparallelrescan is optional; assume no-op if not provided by AM */ + if (scan->indexRelation->rd_amroutine->amparallelrescan == NULL) + return; + + scan->indexRelation->rd_amroutine->amparallelrescan(scan); +} + +/* ---------------- * index_endscan - end a scan * ---------------- */ @@ -341,6 +449,9 @@ index_endscan(IndexScanDesc scan) /* Release index refcount acquired by index_beginscan */ RelationDecrementReferenceCount(scan->indexRelation); + if (scan->xs_temp_snap) + UnregisterSnapshot(scan->xs_snapshot); + /* Release the scan data structure itself */ IndexScanEnd(scan); } diff --git a/src/backend/access/nbtree/nbtree.c b/src/backend/access/nbtree/nbtree.c index 1bb1acf..fad5f1b 100644 --- a/src/backend/access/nbtree/nbtree.c +++ b/src/backend/access/nbtree/nbtree.c @@ -23,6 +23,8 @@ #include "access/xlog.h" #include "catalog/index.h" #include "commands/vacuum.h" +#include "pgstat.h" +#include "storage/condition_variable.h" #include "storage/indexfsm.h" #include "storage/ipc.h" #include "storage/lmgr.h" @@ -63,6 +65,46 @@ typedef struct MemoryContext pagedelcontext; } BTVacState; +/* + * Below flags are used to indicate the state of parallel scan. + * + * BTPARALLEL_NOT_INITIALIZED implies that the scan is not started + * + * BTPARALLEL_ADVANCING implies one of the worker or backend is advancing the + * scan to a new page; others must wait. + * + * BTPARALLEL_IDLE implies that no backend is advancing the scan; someone can + * start doing it + * + * BTPARALLEL_DONE implies that the scan is complete (including error exit) + */ +typedef enum +{ + BTPARALLEL_NOT_INITIALIZED, + BTPARALLEL_ADVANCING, + BTPARALLEL_IDLE, + BTPARALLEL_DONE +} BTPS_State; + +/* + * BTParallelScanDescData contains btree specific shared information required + * for parallel scan. + */ +typedef struct BTParallelScanDescData +{ + BlockNumber btps_scanPage; /* latest or next page to be scanned */ + BTPS_State btps_pageStatus;/* indicates whether next page is available + * for scan. see above for possible states of + * parallel scan. */ + int btps_arrayKeyCount; /* count indicating number of array + * scan keys processed by parallel + * scan */ + slock_t btps_mutex; /* protects above variables */ + ConditionVariable btps_cv; /* used to synchronize parallel scan */ +} BTParallelScanDescData; + +typedef struct BTParallelScanDescData *BTParallelScanDesc; + static void btbuildCallback(Relation index, HeapTuple htup, @@ -111,8 +153,11 @@ bthandler(PG_FUNCTION_ARGS) amroutine->amoptions = btoptions; amroutine->amproperty = btproperty; amroutine->amvalidate = btvalidate; + amroutine->amestimateparallelscan = btestimateparallelscan; + amroutine->aminitparallelscan = btinitparallelscan; amroutine->ambeginscan = btbeginscan; amroutine->amrescan = btrescan; + amroutine->amparallelrescan = btparallelrescan; amroutine->amgettuple = btgettuple; amroutine->amgetbitmap = btgetbitmap; amroutine->amendscan = btendscan; @@ -468,6 +513,192 @@ btbeginscan(Relation rel, int nkeys, int norderbys) } /* + * btestimateparallelscan - estimate storage for BTParallelScanDescData + */ +Size +btestimateparallelscan(void) +{ + return sizeof(BTParallelScanDescData); +} + +/* + * btinitparallelscan - Initializing BTParallelScanDesc for parallel btree scan + */ +void +btinitparallelscan(void *target) +{ + BTParallelScanDesc bt_target = (BTParallelScanDesc) target; + + SpinLockInit(&bt_target->btps_mutex); + bt_target->btps_scanPage = InvalidBlockNumber; + bt_target->btps_pageStatus = BTPARALLEL_NOT_INITIALIZED; + bt_target->btps_arrayKeyCount = 0; + ConditionVariableInit(&bt_target->btps_cv); +} + +/* + * _bt_parallel_seize() -- returns the next block to be scanned for forward + * scans and latest block scanned for backward scans. + * + * status - The value of status tells caller whether to continue the scan or + * not. The true value of status indicates either one of the following (a) + * the block number returned is valid and the scan can be continued (b) the + * block number is invalid and the scan has just begun (c) the block number + * is P_NONE and the scan is finished. The false value indicates that we + * have reached the end of scan for current scankeys and for that we return + * block number as P_NONE. + * + * The first time master backend or worker hits last page, it will return + * P_NONE and status as 'True', after that any worker tries to fetch next + * page, it will return status as 'False'. + * + * Callers ignore the return value, if the status is false. + */ +BlockNumber +_bt_parallel_seize(IndexScanDesc scan, bool *status) +{ + BTScanOpaque so = (BTScanOpaque) scan->opaque; + BTPS_State pageStatus; + bool exit_loop = false; + BlockNumber nextPage = InvalidBlockNumber; + ParallelIndexScanDesc parallel_scan = scan->parallel_scan; + BTParallelScanDesc btscan; + + btscan = (BTParallelScanDesc) OffsetToPointer((void *) parallel_scan, + parallel_scan->ps_offset); + + *status = true; + while (1) + { + /* + * Fetch the next block to scan and update the page status so that + * other participants of parallel scan can wait till next page is + * available for scan. Set the status as false, if scan is finished. + */ + SpinLockAcquire(&btscan->btps_mutex); + pageStatus = btscan->btps_pageStatus; + + /* Check if the scan for current scan keys is finished */ + if (so->arrayKeyCount < btscan->btps_arrayKeyCount) + *status = false; + else if (pageStatus == BTPARALLEL_DONE) + *status = false; + else if (pageStatus != BTPARALLEL_ADVANCING) + { + btscan->btps_pageStatus = BTPARALLEL_ADVANCING; + nextPage = btscan->btps_scanPage; + exit_loop = true; + } + SpinLockRelease(&btscan->btps_mutex); + if (exit_loop || !*status) + break; + ConditionVariableSleep(&btscan->btps_cv, WAIT_EVENT_BTREE_PAGE); + } + ConditionVariableCancelSleep(); + + /* no more pages to scan */ + if (!*status) + return P_NONE; + + *status = true; + return nextPage; +} + +/* + * _bt_parallel_release() -- Advances the parallel scan to allow scan of next + * page + * + * Save information about scan position and wake up next worker to continue + * scan. + * + * For backward scan, scan_page holds the latest page being scanned. + * For forward scan, scan_page holds the next page to be scanned. + */ +void +_bt_parallel_release(IndexScanDesc scan, BlockNumber scan_page) +{ + ParallelIndexScanDesc parallel_scan = scan->parallel_scan; + BTParallelScanDesc btscan; + + btscan = (BTParallelScanDesc) OffsetToPointer((void *) parallel_scan, + parallel_scan->ps_offset); + + SpinLockAcquire(&btscan->btps_mutex); + btscan->btps_scanPage = scan_page; + btscan->btps_pageStatus = BTPARALLEL_IDLE; + SpinLockRelease(&btscan->btps_mutex); + ConditionVariableSignal(&btscan->btps_cv); +} + +/* + * _bt_parallel_done() -- Finishes the parallel scan + * + * When there are no pages left to scan, this function should be called to + * notify other workers. Otherwise, they might wait forever for the scan to + * advance to the next page. + */ +void +_bt_parallel_done(IndexScanDesc scan) +{ + BTScanOpaque so = (BTScanOpaque) scan->opaque; + ParallelIndexScanDesc parallel_scan = scan->parallel_scan; + BTParallelScanDesc btscan; + bool status_changed = false; + + /* Do nothing, for non-parallel scans */ + if (parallel_scan == NULL) + return; + + btscan = (BTParallelScanDesc) OffsetToPointer((void *) parallel_scan, + parallel_scan->ps_offset); + + /* + * Ensure to mark parallel scan as done no more than once for single scan. + * We rely on this state to initiate the next scan for multiple array + * keys, see _bt_advance_array_keys. + */ + SpinLockAcquire(&btscan->btps_mutex); + if (so->arrayKeyCount >= btscan->btps_arrayKeyCount && + btscan->btps_pageStatus != BTPARALLEL_DONE) + { + btscan->btps_pageStatus = BTPARALLEL_DONE; + status_changed = true; + } + SpinLockRelease(&btscan->btps_mutex); + + /* wake up all the workers associated with this parallel scan */ + if (status_changed) + ConditionVariableBroadcast(&btscan->btps_cv); +} + +/* + * _bt_parallel_advance_scan() -- Advances the parallel scan + * + * Updates the count of array keys processed for both local and parallel + * scans. + */ +void +_bt_parallel_advance_scan(IndexScanDesc scan) +{ + BTScanOpaque so = (BTScanOpaque) scan->opaque; + ParallelIndexScanDesc parallel_scan = scan->parallel_scan; + BTParallelScanDesc btscan; + + btscan = (BTParallelScanDesc) OffsetToPointer((void *) parallel_scan, + parallel_scan->ps_offset); + + so->arrayKeyCount++; + SpinLockAcquire(&btscan->btps_mutex); + if (btscan->btps_pageStatus == BTPARALLEL_DONE) + { + btscan->btps_scanPage = InvalidBlockNumber; + btscan->btps_pageStatus = BTPARALLEL_NOT_INITIALIZED; + btscan->btps_arrayKeyCount++; + } + SpinLockRelease(&btscan->btps_mutex); +} + +/* * btrescan() -- rescan an index relation */ void @@ -487,6 +718,7 @@ btrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys, } so->markItemIndex = -1; + so->arrayKeyCount = 0; BTScanPosUnpinIfPinned(so->markPos); BTScanPosInvalidate(so->markPos); @@ -527,6 +759,33 @@ btrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys, } /* + * btparallelrescan() -- reset parallel scan + */ +void +btparallelrescan(IndexScanDesc scan) +{ + ParallelIndexScanDesc parallel_scan = scan->parallel_scan; + + if (parallel_scan) + { + + BTParallelScanDesc btscan; + + btscan = (BTParallelScanDesc) OffsetToPointer((void *) parallel_scan, + parallel_scan->ps_offset); + + /* + * Ideally, we don't need to acquire spinlock here, but being + * consistent with heap_rescan seems to be a good idea. + */ + SpinLockAcquire(&btscan->btps_mutex); + btscan->btps_scanPage = InvalidBlockNumber; + btscan->btps_pageStatus = BTPARALLEL_NOT_INITIALIZED; + SpinLockRelease(&btscan->btps_mutex); + } +} + +/* * btendscan() -- close down a scan */ void diff --git a/src/backend/access/nbtree/nbtsearch.c b/src/backend/access/nbtree/nbtsearch.c index 4fba75a..b565e09 100644 --- a/src/backend/access/nbtree/nbtsearch.c +++ b/src/backend/access/nbtree/nbtsearch.c @@ -30,9 +30,12 @@ static bool _bt_readpage(IndexScanDesc scan, ScanDirection dir, static void _bt_saveitem(BTScanOpaque so, int itemIndex, OffsetNumber offnum, IndexTuple itup); static bool _bt_steppage(IndexScanDesc scan, ScanDirection dir); +static bool _bt_readnextpage(IndexScanDesc scan, BlockNumber blkno, ScanDirection dir); +static bool _bt_parallel_readpage(IndexScanDesc scan, BlockNumber blkno, ScanDirection dir); static Buffer _bt_walk_left(Relation rel, Buffer buf, Snapshot snapshot); static bool _bt_endpoint(IndexScanDesc scan, ScanDirection dir); static void _bt_drop_lock_and_maybe_pin(IndexScanDesc scan, BTScanPos sp); +static inline void _bt_initialize_more_data(BTScanOpaque so, ScanDirection dir); /* @@ -544,8 +547,10 @@ _bt_first(IndexScanDesc scan, ScanDirection dir) ScanKeyData notnullkeys[INDEX_MAX_KEYS]; int keysCount = 0; int i; + bool status = true; StrategyNumber strat_total; BTScanPosItem *currItem; + BlockNumber blkno; Assert(!BTScanPosIsValid(so->currPos)); @@ -564,6 +569,38 @@ _bt_first(IndexScanDesc scan, ScanDirection dir) if (!so->qual_ok) return false; + /* + * For parallel scans, get the page from shared state. If scan has not + * started, proceed to find out first leaf page to scan by keeping other + * workers waiting until we have descended to appropriate leaf page to be + * scanned for matching tuples. + * + * If the scan has already begun, skip finding the first leaf page and + * directly scanning the page stored in shared structure or the page to + * its left in case of backward scan. + */ + if (scan->parallel_scan != NULL) + { + blkno = _bt_parallel_seize(scan, &status); + if (!status) + { + BTScanPosInvalidate(so->currPos); + return false; + } + else if (blkno == P_NONE) + { + _bt_parallel_done(scan); + BTScanPosInvalidate(so->currPos); + return false; + } + else if (blkno != InvalidBlockNumber) + { + if (!_bt_parallel_readpage(scan, blkno, dir)) + return false; + goto readcomplete; + } + } + /*---------- * Examine the scan keys to discover where we need to start the scan. * @@ -743,7 +780,19 @@ _bt_first(IndexScanDesc scan, ScanDirection dir) * there. */ if (keysCount == 0) - return _bt_endpoint(scan, dir); + { + bool match; + + match = _bt_endpoint(scan, dir); + if (!match) + { + /* No match , indicate (parallel) scan finished */ + _bt_parallel_done(scan); + BTScanPosInvalidate(so->currPos); + } + + return match; + } /* * We want to start the scan somewhere within the index. Set up an @@ -993,25 +1042,21 @@ _bt_first(IndexScanDesc scan, ScanDirection dir) * because nothing finer to lock exists. */ PredicateLockRelation(rel, scan->xs_snapshot); + + /* + * mark parallel scan as done, so that all the workers can finish + * their scan + */ + _bt_parallel_done(scan); + BTScanPosInvalidate(so->currPos); + return false; } else PredicateLockPage(rel, BufferGetBlockNumber(buf), scan->xs_snapshot); - /* initialize moreLeft/moreRight appropriately for scan direction */ - if (ScanDirectionIsForward(dir)) - { - so->currPos.moreLeft = false; - so->currPos.moreRight = true; - } - else - { - so->currPos.moreLeft = true; - so->currPos.moreRight = false; - } - so->numKilled = 0; /* just paranoia */ - Assert(so->markItemIndex == -1); + _bt_initialize_more_data(so, dir); /* position to the precise item on the page */ offnum = _bt_binsrch(rel, buf, keysCount, scankeys, nextkey); @@ -1060,6 +1105,7 @@ _bt_first(IndexScanDesc scan, ScanDirection dir) _bt_drop_lock_and_maybe_pin(scan, &so->currPos); } +readcomplete: /* OK, itemIndex says what to return */ currItem = &so->currPos.items[so->currPos.itemIndex]; scan->xs_ctup.t_self = currItem->heapTid; @@ -1154,6 +1200,16 @@ _bt_readpage(IndexScanDesc scan, ScanDirection dir, OffsetNumber offnum) page = BufferGetPage(so->currPos.buf); opaque = (BTPageOpaque) PageGetSpecialPointer(page); + + /* allow next page be processed by parallel worker */ + if (scan->parallel_scan) + { + if (ScanDirectionIsForward(dir)) + _bt_parallel_release(scan, opaque->btpo_next); + else + _bt_parallel_release(scan, BufferGetBlockNumber(so->currPos.buf)); + } + minoff = P_FIRSTDATAKEY(opaque); maxoff = PageGetMaxOffsetNumber(page); @@ -1278,21 +1334,16 @@ _bt_saveitem(BTScanOpaque so, int itemIndex, * if pinned, we'll drop the pin before moving to next page. The buffer is * not locked on entry. * - * On success exit, so->currPos is updated to contain data from the next - * interesting page. For success on a scan using a non-MVCC snapshot we hold - * a pin, but not a read lock, on that page. If we do not hold the pin, we - * set so->currPos.buf to InvalidBuffer. We return TRUE to indicate success. - * - * If there are no more matching records in the given direction, we drop all - * locks and pins, set so->currPos.buf to InvalidBuffer, and return FALSE. + * For success on a scan using a non-MVCC snapshot we hold a pin, but not a + * read lock, on that page. If we do not hold the pin, we set so->currPos.buf + * to InvalidBuffer. We return TRUE to indicate success. */ static bool _bt_steppage(IndexScanDesc scan, ScanDirection dir) { BTScanOpaque so = (BTScanOpaque) scan->opaque; - Relation rel; - Page page; - BTPageOpaque opaque; + BlockNumber blkno = InvalidBlockNumber; + bool status = true; Assert(BTScanPosIsValid(so->currPos)); @@ -1319,13 +1370,27 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) so->markItemIndex = -1; } - rel = scan->indexRelation; - if (ScanDirectionIsForward(dir)) { /* Walk right to the next page with data */ - /* We must rely on the previously saved nextPage link! */ - BlockNumber blkno = so->currPos.nextPage; + + /* + * We must rely on the previously saved nextPage link for non-parallel + * scans! + */ + if (scan->parallel_scan != NULL) + { + blkno = _bt_parallel_seize(scan, &status); + if (!status) + { + /* release the previous buffer, if pinned */ + BTScanPosUnpinIfPinned(so->currPos); + BTScanPosInvalidate(so->currPos); + return false; + } + } + else + blkno = so->currPos.nextPage; /* Remember we left a page with data */ so->currPos.moreLeft = true; @@ -1333,11 +1398,68 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) /* release the previous buffer, if pinned */ BTScanPosUnpinIfPinned(so->currPos); + if (!_bt_readnextpage(scan, blkno, dir)) + return false; + } + else + { + /* Remember we left a page with data */ + so->currPos.moreRight = true; + + /* For parallel scans, get the last page scanned */ + if (scan->parallel_scan != NULL) + { + blkno = _bt_parallel_seize(scan, &status); + BTScanPosUnpinIfPinned(so->currPos); + if (!status) + { + BTScanPosInvalidate(so->currPos); + return false; + } + } + + if (!_bt_readnextpage(scan, blkno, dir)) + return false; + } + + /* Drop the lock, and maybe the pin, on the current page */ + _bt_drop_lock_and_maybe_pin(scan, &so->currPos); + + return true; +} + +/* + * _bt_readnextpage() -- Read next page containing valid data for scan + * + * On success exit, so->currPos is updated to contain data from the next + * interesting page. Caller is responsible to release lock and pin on + * buffer on success. We return TRUE to indicate success. + * + * If there are no more matching records in the given direction, we drop all + * locks and pins, set so->currPos.buf to InvalidBuffer, and return FALSE. + */ +static bool +_bt_readnextpage(IndexScanDesc scan, BlockNumber blkno, ScanDirection dir) +{ + BTScanOpaque so = (BTScanOpaque) scan->opaque; + Relation rel; + Page page; + BTPageOpaque opaque; + bool status = true; + + rel = scan->indexRelation; + + if (ScanDirectionIsForward(dir)) + { for (;;) { - /* if we're at end of scan, give up */ + /* + * if we're at end of scan, give up and mark parallel scan as + * done, so that all the workers can finish their scan + */ if (blkno == P_NONE || !so->currPos.moreRight) { + _bt_parallel_done(scan); BTScanPosInvalidate(so->currPos); return false; } @@ -1345,10 +1467,10 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) CHECK_FOR_INTERRUPTS(); /* step right one page */ so->currPos.buf = _bt_getbuf(rel, blkno, BT_READ); - /* check for deleted page */ page = BufferGetPage(so->currPos.buf); TestForOldSnapshot(scan->xs_snapshot, rel, page); opaque = (BTPageOpaque) PageGetSpecialPointer(page); + /* check for deleted page */ if (!P_IGNORE(opaque)) { PredicateLockPage(rel, blkno, scan->xs_snapshot); @@ -1359,14 +1481,30 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) } /* nope, keep going */ - blkno = opaque->btpo_next; + if (scan->parallel_scan != NULL) + { + blkno = _bt_parallel_seize(scan, &status); + if (!status) + { + _bt_relbuf(rel, so->currPos.buf); + BTScanPosInvalidate(so->currPos); + return false; + } + } + else + blkno = opaque->btpo_next; _bt_relbuf(rel, so->currPos.buf); } } else { - /* Remember we left a page with data */ - so->currPos.moreRight = true; + /* + * for parallel scans, current block number needs to be retrieved from + * shared state and it is the responsibility of caller to pass the + * correct block number. + */ + if (blkno != InvalidBlockNumber) + so->currPos.currPage = blkno; /* * Walk left to the next page with data. This is much more complex @@ -1401,6 +1539,12 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) if (!so->currPos.moreLeft) { _bt_relbuf(rel, so->currPos.buf); + + /* + * mark parallel scan as done, so that all the workers can + * finish their scan + */ + _bt_parallel_done(scan); BTScanPosInvalidate(so->currPos); return false; } @@ -1412,6 +1556,7 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) /* if we're physically at end of index, return failure */ if (so->currPos.buf == InvalidBuffer) { + _bt_parallel_done(scan); BTScanPosInvalidate(so->currPos); return false; } @@ -1432,9 +1577,48 @@ _bt_steppage(IndexScanDesc scan, ScanDirection dir) if (_bt_readpage(scan, dir, PageGetMaxOffsetNumber(page))) break; } + + /* + * For parallel scans, get the last page scanned as it is quite + * possible that by the time we try to fetch previous page, other + * worker has also decided to scan that previous page. We could + * avoid that by doing _bt_parallel_release once we have read the + * current page, but it is bad to make other workers wait till we + * read the page. + */ + if (scan->parallel_scan != NULL) + { + _bt_relbuf(rel, so->currPos.buf); + blkno = _bt_parallel_seize(scan, &status); + if (!status) + { + BTScanPosInvalidate(so->currPos); + return false; + } + so->currPos.buf = _bt_getbuf(rel, blkno, BT_READ); + } } } + return true; +} + +/* + * _bt_parallel_readpage() -- Read current page containing valid data for scan + * + * On success, release lock and pin on buffer. We return TRUE to indicate + * success. + */ +static bool +_bt_parallel_readpage(IndexScanDesc scan, BlockNumber blkno, ScanDirection dir) +{ + BTScanOpaque so = (BTScanOpaque) scan->opaque; + + _bt_initialize_more_data(so, dir); + + if (!_bt_readnextpage(scan, blkno, dir)) + return false; + /* Drop the lock, and maybe the pin, on the current page */ _bt_drop_lock_and_maybe_pin(scan, &so->currPos); @@ -1712,19 +1896,7 @@ _bt_endpoint(IndexScanDesc scan, ScanDirection dir) /* remember which buffer we have pinned */ so->currPos.buf = buf; - /* initialize moreLeft/moreRight appropriately for scan direction */ - if (ScanDirectionIsForward(dir)) - { - so->currPos.moreLeft = false; - so->currPos.moreRight = true; - } - else - { - so->currPos.moreLeft = true; - so->currPos.moreRight = false; - } - so->numKilled = 0; /* just paranoia */ - so->markItemIndex = -1; /* ditto */ + _bt_initialize_more_data(so, dir); /* * Now load data from the first page of the scan. @@ -1753,3 +1925,25 @@ _bt_endpoint(IndexScanDesc scan, ScanDirection dir) return true; } + +/* + * _bt_initialize_more_data() -- initialize moreLeft/moreRight appropriately + * for scan direction + */ +static inline void +_bt_initialize_more_data(BTScanOpaque so, ScanDirection dir) +{ + /* initialize moreLeft/moreRight appropriately for scan direction */ + if (ScanDirectionIsForward(dir)) + { + so->currPos.moreLeft = false; + so->currPos.moreRight = true; + } + else + { + so->currPos.moreLeft = true; + so->currPos.moreRight = false; + } + so->numKilled = 0; /* just paranoia */ + so->markItemIndex = -1; /* ditto */ +} diff --git a/src/backend/access/nbtree/nbtutils.c b/src/backend/access/nbtree/nbtutils.c index da0f330..692ced4 100644 --- a/src/backend/access/nbtree/nbtutils.c +++ b/src/backend/access/nbtree/nbtutils.c @@ -590,6 +590,10 @@ _bt_advance_array_keys(IndexScanDesc scan, ScanDirection dir) break; } + /* advance parallel scan */ + if (scan->parallel_scan != NULL) + _bt_parallel_advance_scan(scan); + return found; } diff --git a/src/backend/access/spgist/spgutils.c b/src/backend/access/spgist/spgutils.c index ca4b0bd..1729797 100644 --- a/src/backend/access/spgist/spgutils.c +++ b/src/backend/access/spgist/spgutils.c @@ -61,8 +61,11 @@ spghandler(PG_FUNCTION_ARGS) amroutine->amoptions = spgoptions; amroutine->amproperty = NULL; amroutine->amvalidate = spgvalidate; + amroutine->amestimateparallelscan = NULL; + amroutine->aminitparallelscan = NULL; amroutine->ambeginscan = spgbeginscan; amroutine->amrescan = spgrescan; + amroutine->amparallelrescan = NULL; amroutine->amgettuple = spggettuple; amroutine->amgetbitmap = spggetbitmap; amroutine->amendscan = spgendscan; diff --git a/src/backend/postmaster/pgstat.c b/src/backend/postmaster/pgstat.c index f37a0bf..090711a 100644 --- a/src/backend/postmaster/pgstat.c +++ b/src/backend/postmaster/pgstat.c @@ -3386,6 +3386,9 @@ pgstat_get_wait_ipc(WaitEventIPC w) case WAIT_EVENT_PARALLEL_FINISH: event_name = "ParallelFinish"; break; + case WAIT_EVENT_BTREE_PAGE: + event_name = "ParallelBtreePage"; + break; case WAIT_EVENT_SAFE_SNAPSHOT: event_name = "SafeSnapshot"; break; diff --git a/src/include/access/amapi.h b/src/include/access/amapi.h index 6a5f279..18259ad 100644 --- a/src/include/access/amapi.h +++ b/src/include/access/amapi.h @@ -108,6 +108,12 @@ typedef bool (*amproperty_function) (Oid index_oid, int attno, /* validate definition of an opclass for this AM */ typedef bool (*amvalidate_function) (Oid opclassoid); +/* estimate size of parallel scan descriptor */ +typedef Size (*amestimateparallelscan_function) (void); + +/* prepare for parallel index scan */ +typedef void (*aminitparallelscan_function) (void *target); + /* prepare for index scan */ typedef IndexScanDesc (*ambeginscan_function) (Relation indexRelation, int nkeys, @@ -120,6 +126,9 @@ typedef void (*amrescan_function) (IndexScanDesc scan, ScanKey orderbys, int norderbys); +/* (re)start parallel index scan */ +typedef void (*amparallelrescan_function) (IndexScanDesc scan); + /* next valid tuple */ typedef bool (*amgettuple_function) (IndexScanDesc scan, ScanDirection direction); @@ -189,8 +198,11 @@ typedef struct IndexAmRoutine amoptions_function amoptions; amproperty_function amproperty; /* can be NULL */ amvalidate_function amvalidate; + amestimateparallelscan_function amestimateparallelscan; /* can be NULL */ + aminitparallelscan_function aminitparallelscan; /* can be NULL */ ambeginscan_function ambeginscan; amrescan_function amrescan; + amparallelrescan_function amparallelrescan; /* can be NULL */ amgettuple_function amgettuple; /* can be NULL */ amgetbitmap_function amgetbitmap; /* can be NULL */ amendscan_function amendscan; diff --git a/src/include/access/genam.h b/src/include/access/genam.h index b2e078a..d2258f6 100644 --- a/src/include/access/genam.h +++ b/src/include/access/genam.h @@ -83,6 +83,8 @@ typedef bool (*IndexBulkDeleteCallback) (ItemPointer itemptr, void *state); typedef struct IndexScanDescData *IndexScanDesc; typedef struct SysScanDescData *SysScanDesc; +typedef struct ParallelIndexScanDescData *ParallelIndexScanDesc; + /* * Enumeration specifying the type of uniqueness check to perform in * index_insert(). @@ -131,6 +133,13 @@ extern bool index_insert(Relation indexRelation, Relation heapRelation, IndexUniqueCheck checkUnique); +extern Size index_parallelscan_estimate(Relation indexrel, Snapshot snapshot); +extern void index_parallelscan_initialize(Relation heaprel, Relation indexrel, + Snapshot snapshot, ParallelIndexScanDesc target); +extern IndexScanDesc index_beginscan_parallel(Relation heaprel, + Relation indexrel, int nkeys, int norderbys, + ParallelIndexScanDesc pscan); + extern IndexScanDesc index_beginscan(Relation heapRelation, Relation indexRelation, Snapshot snapshot, @@ -141,6 +150,7 @@ extern IndexScanDesc index_beginscan_bitmap(Relation indexRelation, extern void index_rescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys); +extern void index_parallelrescan(IndexScanDesc scan); extern void index_endscan(IndexScanDesc scan); extern void index_markpos(IndexScanDesc scan); extern void index_restrpos(IndexScanDesc scan); diff --git a/src/include/access/nbtree.h b/src/include/access/nbtree.h index 011a72e..547c1cf 100644 --- a/src/include/access/nbtree.h +++ b/src/include/access/nbtree.h @@ -609,6 +609,8 @@ typedef struct BTScanOpaqueData ScanKey arrayKeyData; /* modified copy of scan->keyData */ int numArrayKeys; /* number of equality-type array keys (-1 if * there are any unsatisfiable array keys) */ + int arrayKeyCount; /* count indicating number of array scan keys + * processed */ BTArrayKeyInfo *arrayKeys; /* info about each equality-type array key */ MemoryContext arrayContext; /* scan-lifespan context for array data */ @@ -652,7 +654,8 @@ typedef BTScanOpaqueData *BTScanOpaque; #define SK_BT_NULLS_FIRST (INDOPTION_NULLS_FIRST << SK_BT_INDOPTION_SHIFT) /* - * prototypes for functions in nbtree.c (external entry points for btree) + * prototypes for functions in nbtree.c (external entry points for btree and + * functions to maintain state of parallel scan) */ extern IndexBuildResult *btbuild(Relation heap, Relation index, struct IndexInfo *indexInfo); @@ -661,10 +664,17 @@ extern bool btinsert(Relation rel, Datum *values, bool *isnull, ItemPointer ht_ctid, Relation heapRel, IndexUniqueCheck checkUnique); extern IndexScanDesc btbeginscan(Relation rel, int nkeys, int norderbys); +extern Size btestimateparallelscan(void); +extern void btinitparallelscan(void *target); +extern BlockNumber _bt_parallel_seize(IndexScanDesc scan, bool *status); +extern void _bt_parallel_release(IndexScanDesc scan, BlockNumber scan_page); +extern void _bt_parallel_done(IndexScanDesc scan); +extern void _bt_parallel_advance_scan(IndexScanDesc scan); extern bool btgettuple(IndexScanDesc scan, ScanDirection dir); extern int64 btgetbitmap(IndexScanDesc scan, TIDBitmap *tbm); extern void btrescan(IndexScanDesc scan, ScanKey scankey, int nscankeys, ScanKey orderbys, int norderbys); +extern void btparallelrescan(IndexScanDesc scan); extern void btendscan(IndexScanDesc scan); extern void btmarkpos(IndexScanDesc scan); extern void btrestrpos(IndexScanDesc scan); diff --git a/src/include/access/relscan.h b/src/include/access/relscan.h index 8746045..793c46b 100644 --- a/src/include/access/relscan.h +++ b/src/include/access/relscan.h @@ -126,8 +126,20 @@ typedef struct IndexScanDescData /* state data for traversing HOT chains in index_getnext */ bool xs_continue_hot; /* T if must keep walking HOT chain */ + bool xs_temp_snap; /* unregister snapshot at scan end? */ + ParallelIndexScanDesc parallel_scan; /* parallel index scan + * information */ } IndexScanDescData; +/* Generic structure for parallel scans */ +typedef struct ParallelIndexScanDescData +{ + Oid ps_relid; + Oid ps_indexid; + Size ps_offset; /* Offset in bytes of am specific structure */ + char ps_snapshot_data[FLEXIBLE_ARRAY_MEMBER]; +} ParallelIndexScanDescData; + /* Struct for heap-or-index scans of system tables */ typedef struct SysScanDescData { diff --git a/src/include/c.h b/src/include/c.h index efbb77f..a2c043a 100644 --- a/src/include/c.h +++ b/src/include/c.h @@ -527,6 +527,9 @@ typedef NameData *Name; #define PointerIsAligned(pointer, type) \ (((uintptr_t)(pointer) % (sizeof (type))) == 0) +#define OffsetToPointer(base, offset) \ + ((void *)((char *) base + offset)) + #define OidIsValid(objectId) ((bool) ((objectId) != InvalidOid)) #define RegProcedureIsValid(p) OidIsValid(p) diff --git a/src/include/pgstat.h b/src/include/pgstat.h index 5b37894..dfd095f 100644 --- a/src/include/pgstat.h +++ b/src/include/pgstat.h @@ -784,6 +784,7 @@ typedef enum WAIT_EVENT_MQ_RECEIVE, WAIT_EVENT_MQ_SEND, WAIT_EVENT_PARALLEL_FINISH, + WAIT_EVENT_BTREE_PAGE, WAIT_EVENT_SAFE_SNAPSHOT, WAIT_EVENT_SYNC_REP } WaitEventIPC; diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list index 993880d..9f876ae 100644 --- a/src/tools/pgindent/typedefs.list +++ b/src/tools/pgindent/typedefs.list @@ -161,6 +161,9 @@ BTPageOpaque BTPageOpaqueData BTPageStat BTPageState +BTParallelScanDesc +BTParallelScanDescData +BTPS_State BTScanOpaque BTScanOpaqueData BTScanPos @@ -1264,6 +1267,8 @@ OverrideSearchPath OverrideStackEntry PACE_HEADER PACL +ParallelIndexScanDesc +ParallelIndexScanDescData PATH PBOOL PCtxtHandle