bugfix: wrong scan logic recently enbugged.

related to issue #819 fix (54176d2) and 18d5904
heterodb · Oct 20, 2024 · 1a5606e · 1a5606e
1 parent 51c0b0d
commit 1a5606e
Show file tree

Hide file tree

Showing 3 changed files with 15 additions and 23 deletions.
diff --git a/src/cuda_gpuscan.cu b/src/cuda_gpuscan.cu
@@ -268,6 +268,7 @@ __gpuscan_load_source_row(kern_context *kcxt,
 		wp->smx_row_count++;
 		WARP_WRITE_POS(wp,0) += count;
 	}
+	__syncthreads();
 	/* move to the next depth, if more than blockSize tuples were fetched. */
 	return (WARP_WRITE_POS(wp,0) >= WARP_READ_POS(wp,0) + get_local_size() ? 1 : 0);
 }
@@ -472,6 +473,7 @@ __gpuscan_load_source_arrow(kern_context *kcxt,
 		wp->smx_row_count++;
 		WARP_WRITE_POS(wp,0) += count;
 	}
+	__syncthreads();
 	/* move to the next depth, if more than blockSize rows were fetched. */
 	return (WARP_WRITE_POS(wp,0) >= WARP_READ_POS(wp,0) + get_local_size() ? 1 : 0);
 }
@@ -596,6 +598,7 @@ __gpuscan_load_source_column(kern_context *kcxt,
 		wp->smx_row_count++;
 		WARP_WRITE_POS(wp,0) += count;
 	}
+	__syncthreads();
 	/* move to the next depth if more than 32 htuples were fetched */
 	return (WARP_WRITE_POS(wp,0) >= WARP_READ_POS(wp,0) + get_local_size() ? 1 : 0);
 }

diff --git a/src/executor.c b/src/executor.c
@@ -762,20 +762,8 @@ static bool
 pgstromTaskStateBeginScan(pgstromTaskState *pts)
 {
 	pgstromSharedState *ps_state = pts->ps_state;
-	HeapScanDesc	h_scan = (HeapScanDesc)pts->css.ss.ss_currentScanDesc;
 	XpuConnection  *conn = pts->conn;
 	uint32_t		curval, newval;
-	uint32_t		block_nums = 0;
-	uint32_t		block_start = 0;
-
-	/* init heap-scan position, if any */
-	if (h_scan)
-	{
-		block_nums	= h_scan->rs_nblocks;
-		block_start	= h_scan->rs_startblock;
-	}
-	ps_state->scan_block_nums  = block_nums;
-	ps_state->scan_block_start = block_start;
 
 	/* update the parallel_task_control */
 	Assert(conn != NULL);
@@ -2153,6 +2141,8 @@ pgstromSharedStateInitDSM(CustomScanState *node,
 			table_parallelscan_initialize(relation, pdesc, snapshot);
 			scan = table_beginscan_parallel(relation, pdesc);
 			ps_state->parallel_scan_desc_offset = ((char *)pdesc - (char *)ps_state);
+			ps_state->scan_block_nums  = ((HeapScanDesc)scan)->rs_nblocks;
+			ps_state->scan_block_start = ((HeapScanDesc)scan)->rs_startblock;
 		}
 	}
 	else
@@ -2167,7 +2157,11 @@ pgstromSharedStateInitDSM(CustomScanState *node,
 		if (pts->arrow_state)
 			pgstromArrowFdwInitDSM(pts->arrow_state, ps_state);
 		else
+		{
 			scan = table_beginscan(relation, estate->es_snapshot, 0, NULL);
+			ps_state->scan_block_nums  = ((HeapScanDesc)scan)->rs_nblocks;
+			ps_state->scan_block_start = ((HeapScanDesc)scan)->rs_startblock;
+		}
 	}
 	ps_state->query_plan_id = ((uint64_t)MyProcPid) << 32 |
 		(uint64_t)pts->css.ss.ps.plan->plan_node_id;

diff --git a/src/relscan.c b/src/relscan.c
@@ -712,19 +712,14 @@ pgstromRelScanChunkDirect(pgstromTaskState *pts,
 			uint32_t	num_blocks = kds_nrooms - kds->nitems;
 			uint64_t	scan_block_limit = (ps_state->scan_block_nums *
 											pts->num_scan_repeats);
-			uint64_t	scan_block_count;
 
-			scan_block_count = pg_atomic_fetch_add_u64(&ps_state->scan_block_count,
-													   num_blocks);
-			if (scan_block_count >= scan_block_limit)
+			pts->curr_block_num  = pg_atomic_fetch_add_u64(&ps_state->scan_block_count,
+														   num_blocks);
+			pts->curr_block_tail = pts->curr_block_num + num_blocks;
+			if (pts->curr_block_num >= scan_block_limit)
 				pts->scan_done = true;
-			else
-			{
-				pts->curr_block_num = (scan_block_count % ps_state->scan_block_nums);
-				pts->curr_block_tail = pts->curr_block_num + num_blocks;
-				if (pts->curr_block_tail > ps_state->scan_block_nums)
-					pts->curr_block_tail = ps_state->scan_block_nums;
-			}
+			if (pts->curr_block_tail > scan_block_limit)
+				pts->curr_block_tail = scan_block_limit;
 		}
 	}
 out: