fixup! v10 parallel tid range scan

Author: david-rowley

doc/src/sgml/parallel.sgml

@@ -299,6 +299,15 @@ EXPLAIN SELECT * FROM pgbench_accounts WHERE filler LIKE '%x%';
         within each worker process.
       </para>
     </listitem>
+    <listitem>
+      <para>
+        In a <emphasis>parallel tid range scan</emphasis>, the range of blocks
+        will be subdivided into smaller ranges which are shared among the
+        cooperating processes.  Each worker process will complete the scanning
+        of its given range of blocks before requesting an additional range of
+        blocks.
+      </para>
+    </listitem>
   </itemizedlist>
 
     Other scan types, such as scans of non-btree indexes, may support

src/backend/access/heap/heapam.c

@@ -258,7 +258,9 @@ heap_scan_stream_read_next_parallel(ReadStream *stream,
 		/* parallel scan */
 		table_block_parallelscan_startblock_init(scan->rs_base.rs_rd,
 												 scan->rs_parallelworkerdata,
-												 (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel);
+												 (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel,
+												 scan->rs_startblock,
+												 scan->rs_numblocks);
 
 		/* may return InvalidBlockNumber if there are no more blocks */
 		scan->rs_prefetch_block = table_block_parallelscan_nextpage(scan->rs_base.rs_rd,
@@ -490,16 +492,6 @@ heap_setscanlimits(TableScanDesc sscan, BlockNumber startBlk, BlockNumber numBlk
 
 	scan->rs_startblock = startBlk;
 	scan->rs_numblocks = numBlks;
-
-	/* set the limits in the ParallelBlockTableScanDesc, when present as leader */
-	if (scan->rs_base.rs_parallel != NULL && !IsParallelWorker())
-	{
-		ParallelBlockTableScanDesc bpscan;
-
-		bpscan = (ParallelBlockTableScanDesc) scan->rs_base.rs_parallel;
-		bpscan->phs_startblock = startBlk;
-		bpscan->phs_numblock = numBlks;
-	}
 }
 
 /*

src/backend/access/table/tableam.c

@@ -189,8 +189,8 @@ table_beginscan_parallel(Relation relation, ParallelTableScanDesc pscan)
 }
 
 TableScanDesc
-table_beginscan_parallel_tidrange(Relation relation, ParallelTableScanDesc pscan,
-		ItemPointerData * mintid, ItemPointerData * maxtid)
+table_beginscan_parallel_tidrange(Relation relation,
+								  ParallelTableScanDesc pscan)
 {
 	Snapshot	snapshot;
 	uint32		flags = SO_TYPE_TIDRANGESCAN | SO_ALLOW_PAGEMODE;
@@ -216,11 +216,6 @@ table_beginscan_parallel_tidrange(Relation relation, ParallelTableScanDesc pscan
 
 	sscan = relation->rd_tableam->scan_begin(relation, snapshot, 0, NULL,
 											 pscan, flags);
-
-	/* Set the TID range if needed */
-	if (mintid && maxtid)
-		relation->rd_tableam->scan_set_tidrange(sscan, mintid, maxtid);
-
 	return sscan;
 }
 
@@ -453,57 +448,59 @@ table_block_parallelscan_reinitialize(Relation rel, ParallelTableScanDesc pscan)
  *
  * Determine where the parallel seq scan should start.  This function may be
  * called many times, once by each parallel worker.  We must be careful only
- * to set the startblock once.
+ * to set the phs_startblock and phs_numblock fields once.
+ *
+ * Callers may optionally specify a non-InvalidBlockNumber value for
+ * 'startblock' to force the scan to start at the given page.  Likewise,
+ * 'numblocks' can be specified as a non-InvalidBlockNumber to limit the
+ * number of blocks to scan to that many blocks.
  */
 void
 table_block_parallelscan_startblock_init(Relation rel,
 										 ParallelBlockTableScanWorker pbscanwork,
-										 ParallelBlockTableScanDesc pbscan)
+										 ParallelBlockTableScanDesc pbscan,
+										 BlockNumber startblock,
+										 BlockNumber numblocks)
 {
 	BlockNumber sync_startpage = InvalidBlockNumber;
+	BlockNumber scan_nblocks;
 
 	/* Reset the state we use for controlling allocation size. */
 	memset(pbscanwork, 0, sizeof(*pbscanwork));
 
 	StaticAssertStmt(MaxBlockNumber <= 0xFFFFFFFE,
 					 "pg_nextpower2_32 may be too small for non-standard BlockNumber width");
 
-	/*
-	 * We determine the chunk size based on the size of the relation. First we
-	 * split the relation into PARALLEL_SEQSCAN_NCHUNKS chunks but we then
-	 * take the next highest power of 2 number of the chunk size.  This means
-	 * we split the relation into somewhere between PARALLEL_SEQSCAN_NCHUNKS
-	 * and PARALLEL_SEQSCAN_NCHUNKS / 2 chunks.
-	 */
-	pbscanwork->phsw_chunk_size = pg_nextpower2_32(Max(pbscan->phs_nblocks /
-													   PARALLEL_SEQSCAN_NCHUNKS, 1));
-
-	/*
-	 * Ensure we don't go over the maximum chunk size with larger tables. This
-	 * means we may get much more than PARALLEL_SEQSCAN_NCHUNKS for larger
-	 * tables.  Too large a chunk size has been shown to be detrimental to
-	 * synchronous scan performance.
-	 */
-	pbscanwork->phsw_chunk_size = Min(pbscanwork->phsw_chunk_size,
-									  PARALLEL_SEQSCAN_MAX_CHUNK_SIZE);
-
 retry:
 	/* Grab the spinlock. */
 	SpinLockAcquire(&pbscan->phs_mutex);
 
 	/*
-	 * If the scan's startblock has not yet been initialized, we must do so
-	 * now.  If this is not a synchronized scan, we just start at block 0, but
-	 * if it is a synchronized scan, we must get the starting position from
-	 * the synchronized scan machinery.  We can't hold the spinlock while
-	 * doing that, though, so release the spinlock, get the information we
-	 * need, and retry.  If nobody else has initialized the scan in the
-	 * meantime, we'll fill in the value we fetched on the second time
-	 * through.
+	 * When the caller specified a limit on the number of blocks to scan, set
+	 * that in the ParallelBlockTableScanDesc, if it's not been done by
+	 * another worker already.
+	 */
+	if (numblocks != InvalidBlockNumber &&
+		pbscan->phs_numblock == InvalidBlockNumber)
+	{
+		pbscan->phs_numblock = numblocks;
+	}
+
+	/*
+	 * If the scan's phs_startblock has not yet been initialized, we must do
+	 * so now.  If a startblock was specified, start there, otherwise if this
+	 * is not a synchronized scan, we just start at block 0, but if it is a
+	 * synchronized scan, we must get the starting position from the
+	 * synchronized scan machinery.  We can't hold the spinlock while doing
+	 * that, though, so release the spinlock, get the information we need, and
+	 * retry.  If nobody else has initialized the scan in the meantime, we'll
+	 * fill in the value we fetched on the second time through.
 	 */
 	if (pbscan->phs_startblock == InvalidBlockNumber)
 	{
-		if (!pbscan->base.phs_syncscan)
+		if (startblock != InvalidBlockNumber)
+			pbscan->phs_startblock = startblock;
+		else if (!pbscan->base.phs_syncscan)
 			pbscan->phs_startblock = 0;
 		else if (sync_startpage != InvalidBlockNumber)
 			pbscan->phs_startblock = sync_startpage;
@@ -515,6 +512,34 @@ table_block_parallelscan_startblock_init(Relation rel,
 		}
 	}
 	SpinLockRelease(&pbscan->phs_mutex);
+
+	/*
+	 * Figure out how many blocks we're going to scan; either all of them, or
+	 * just phs_numblock's worth, if a limit has been imposed.
+	 */
+	if (pbscan->phs_numblock == InvalidBlockNumber)
+		scan_nblocks = pbscan->phs_nblocks;
+	else
+		scan_nblocks = pbscan->phs_numblock;
+
+	/*
+	 * We determine the chunk size based on scan_nblocks.  First we split
+	 * scan_nblocks into PARALLEL_SEQSCAN_NCHUNKS chunks then we calculate the
+	 * next highest power of 2 number of the result.  This means we split the
+	 * blocks we're scanning into somewhere between PARALLEL_SEQSCAN_NCHUNKS
+	 * and PARALLEL_SEQSCAN_NCHUNKS / 2 chunks.
+	 */
+	pbscanwork->phsw_chunk_size = pg_nextpower2_32(Max(scan_nblocks /
+													   PARALLEL_SEQSCAN_NCHUNKS, 1));
+
+	/*
+	 * Ensure we don't go over the maximum chunk size with larger tables. This
+	 * means we may get much more than PARALLEL_SEQSCAN_NCHUNKS for larger
+	 * tables.  Too large a chunk size has been shown to be detrimental to
+	 * synchronous scan performance.
+	 */
+	pbscanwork->phsw_chunk_size = Min(pbscanwork->phsw_chunk_size,
+									  PARALLEL_SEQSCAN_MAX_CHUNK_SIZE);
 }
 
 /*
@@ -530,6 +555,7 @@ table_block_parallelscan_nextpage(Relation rel,
 								  ParallelBlockTableScanWorker pbscanwork,
 								  ParallelBlockTableScanDesc pbscan)
 {
+	BlockNumber scan_nblocks;
 	BlockNumber page;
 	uint64		nallocated;
 
@@ -550,7 +576,7 @@ table_block_parallelscan_nextpage(Relation rel,
 	 *
 	 * Here we name these ranges of blocks "chunks".  The initial size of
 	 * these chunks is determined in table_block_parallelscan_startblock_init
-	 * based on the size of the relation.  Towards the end of the scan, we
+	 * based on the number of blocks to scan.  Towards the end of the scan, we
 	 * start making reductions in the size of the chunks in order to attempt
 	 * to divide the remaining work over all the workers as evenly as
 	 * possible.
@@ -567,17 +593,23 @@ table_block_parallelscan_nextpage(Relation rel,
 	 * phs_nallocated counter will exceed rs_nblocks, because workers will
 	 * still increment the value, when they try to allocate the next block but
 	 * all blocks have been allocated already. The counter must be 64 bits
-	 * wide because of that, to avoid wrapping around when rs_nblocks is close
-	 * to 2^32.
+	 * wide because of that, to avoid wrapping around when scan_nblocks is
+	 * close to 2^32.
 	 *
 	 * The actual block to return is calculated by adding the counter to the
-	 * starting block number, modulo nblocks.
+	 * starting block number, modulo phs_nblocks.
 	 */
 
+	/* First, figure out how many blocks we're planning on scanning */
+	if (pbscan->phs_numblock == InvalidBlockNumber)
+		scan_nblocks = pbscan->phs_nblocks;
+	else
+		scan_nblocks = pbscan->phs_numblock;
+
 	/*
-	 * First check if we have any remaining blocks in a previous chunk for
-	 * this worker.  We must consume all of the blocks from that before we
-	 * allocate a new chunk to the worker.
+	 * Now check if we have any remaining blocks in a previous chunk for this
+	 * worker.  We must consume all of the blocks from that before we allocate
+	 * a new chunk to the worker.
 	 */
 	if (pbscanwork->phsw_chunk_remaining > 0)
 	{
@@ -599,7 +631,7 @@ table_block_parallelscan_nextpage(Relation rel,
 		 * chunk size set to 1.
 		 */
 		if (pbscanwork->phsw_chunk_size > 1 &&
-			pbscanwork->phsw_nallocated > pbscan->phs_nblocks -
+			pbscanwork->phsw_nallocated > scan_nblocks -
 			(pbscanwork->phsw_chunk_size * PARALLEL_SEQSCAN_RAMPDOWN_CHUNKS))
 			pbscanwork->phsw_chunk_size >>= 1;
 
@@ -614,15 +646,9 @@ table_block_parallelscan_nextpage(Relation rel,
 		pbscanwork->phsw_chunk_remaining = pbscanwork->phsw_chunk_size - 1;
 	}
 
-	/*
-	 * Check if we've allocated every block in the relation, or if we've
-	 * reached the limit imposed by pbscan->phs_numblock (if set).
-	 */
-	if (nallocated >= pbscan->phs_nblocks)
-		page = InvalidBlockNumber; /* all blocks have been allocated */
-	else if (pbscan->phs_numblock != InvalidBlockNumber &&
-			 nallocated >= pbscan->phs_numblock)
-		page = InvalidBlockNumber; /* upper scan limit reached */
+	/* Check if we've run out of blocks to scan */
+	if (nallocated >= scan_nblocks)
+		page = InvalidBlockNumber;	/* all blocks have been allocated */
 	else
 		page = (nallocated + pbscan->phs_startblock) % pbscan->phs_nblocks;
 

src/backend/executor/execParallel.c

@@ -40,8 +40,8 @@
 #include "executor/nodeSeqscan.h"
 #include "executor/nodeSort.h"
 #include "executor/nodeSubplan.h"
-#include "executor/tqueue.h"
 #include "executor/nodeTidrangescan.h"
+#include "executor/tqueue.h"
 #include "jit/jit.h"
 #include "nodes/nodeFuncs.h"
 #include "pgstat.h"
@@ -502,7 +502,7 @@ ExecParallelInitializeDSM(PlanState *planstate,
 		case T_TidRangeScanState:
 			if (planstate->plan->parallel_aware)
 				ExecTidRangeScanInitializeDSM((TidRangeScanState *) planstate,
-										 	  d->pcxt);
+											  d->pcxt);
 			break;
 		case T_AppendState:
 			if (planstate->plan->parallel_aware)
@@ -1008,7 +1008,7 @@ ExecParallelReInitializeDSM(PlanState *planstate,
 		case T_TidRangeScanState:
 			if (planstate->plan->parallel_aware)
 				ExecTidRangeScanReInitializeDSM((TidRangeScanState *) planstate,
-										   	    pcxt);
+												pcxt);
 			break;
 		case T_AppendState:
 			if (planstate->plan->parallel_aware)

src/backend/executor/nodeTidrangescan.c

@@ -250,13 +250,9 @@ TidRangeNext(TidRangeScanState *node)
 		}
 		else
 		{
-			/* rescan with the updated TID range only in non-parallel mode */
-			if (scandesc->rs_parallel == NULL)
-			{
-				/* rescan with the updated TID range */
-				table_rescan_tidrange(scandesc, &node->trss_mintid,
-									  &node->trss_maxtid);
-			}
+			/* rescan with the updated TID range */
+			table_rescan_tidrange(scandesc, &node->trss_mintid,
+								  &node->trss_maxtid);
 		}
 
 		node->trss_inScan = true;
@@ -419,6 +415,7 @@ ExecInitTidRangeScan(TidRangeScan *node, EState *estate, int eflags)
 	 */
 	return tidrangestate;
 }
+
 /* ----------------------------------------------------------------
  *						Parallel Scan Support
  * ----------------------------------------------------------------
@@ -446,7 +443,7 @@ ExecTidRangeScanEstimate(TidRangeScanState *node, ParallelContext *pcxt)
 /* ----------------------------------------------------------------
  *		ExecTidRangeScanInitializeDSM
  *
- *		Set up a parallel TID scan descriptor.
+ *		Set up a parallel TID range scan descriptor.
  * ----------------------------------------------------------------
  */
 void
@@ -460,19 +457,9 @@ ExecTidRangeScanInitializeDSM(TidRangeScanState *node, ParallelContext *pcxt)
 								  pscan,
 								  estate->es_snapshot);
 	shm_toc_insert(pcxt->toc, node->ss.ps.plan->plan_node_id, pscan);
-
-	/*
-	 * Initialize parallel scan descriptor with given TID range if it can be
-	 * evaluated successfully.
-	 */
-	if (TidRangeEval(node))
-		node->ss.ss_currentScanDesc =
-			table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, pscan,
-					&node->trss_mintid, &node->trss_maxtid);
-	else
-		node->ss.ss_currentScanDesc =
-			table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, pscan,
-					NULL, NULL);
+	node->ss.ss_currentScanDesc =
+		table_beginscan_parallel_tidrange(node->ss.ss_currentRelation,
+										  pscan);
 }
 
 /* ----------------------------------------------------------------
@@ -483,21 +470,12 @@ ExecTidRangeScanInitializeDSM(TidRangeScanState *node, ParallelContext *pcxt)
  */
 void
 ExecTidRangeScanReInitializeDSM(TidRangeScanState *node,
-						   ParallelContext *pcxt)
+								ParallelContext *pcxt)
 {
 	ParallelTableScanDesc pscan;
 
 	pscan = node->ss.ss_currentScanDesc->rs_parallel;
 	table_parallelscan_reinitialize(node->ss.ss_currentRelation, pscan);
-
-	/* Set the new TID range if it can be evaluated successfully */
-	if (TidRangeEval(node))
-		node->ss.ss_currentRelation->rd_tableam->scan_set_tidrange(
-				node->ss.ss_currentScanDesc, &node->trss_mintid,
-				&node->trss_maxtid);
-	else
-		node->ss.ss_currentRelation->rd_tableam->scan_set_tidrange(
-					node->ss.ss_currentScanDesc, NULL, NULL);
 }
 
 /* ----------------------------------------------------------------
@@ -508,18 +486,12 @@ ExecTidRangeScanReInitializeDSM(TidRangeScanState *node,
  */
 void
 ExecTidRangeScanInitializeWorker(TidRangeScanState *node,
-							ParallelWorkerContext *pwcxt)
+								 ParallelWorkerContext *pwcxt)
 {
 	ParallelTableScanDesc pscan;
 
 	pscan = shm_toc_lookup(pwcxt->toc, node->ss.ps.plan->plan_node_id, false);
-
-	if (TidRangeEval(node))
-		node->ss.ss_currentScanDesc =
-			table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, pscan,
-					&node->trss_mintid, &node->trss_maxtid);
-	else
-		node->ss.ss_currentScanDesc =
-			table_beginscan_parallel_tidrange(node->ss.ss_currentRelation, pscan,
-					NULL, NULL);
+	node->ss.ss_currentScanDesc =
+		table_beginscan_parallel_tidrange(node->ss.ss_currentRelation,
+										  pscan);
 }