Fix dsm segments leak in parallel scan code

include/tableam/handler.h

@@ -199,6 +199,7 @@ typedef struct ParallelOScanDescData
 	pg_atomic_uint64 downlinkIndex;
 	int			workersReportedCount;	/* number of workers that reported
 										 * disk downlinks number */
+	int 		dsmSegAttached;			/* number of workers currenly attached to dsm segment during disk scan stage */
 	bits8		flags;
 	int			nworkers;
 	dsm_handle	dsmHandle;

src/btree/scan.c

@@ -424,6 +424,7 @@ switch_to_disk_scan(BTreeSeqScan *scan)
 				LWLockAcquire(&poscan->downlinksPublish, LW_EXCLUSIVE);
 				Assert(!poscan->dsmHandle);
 				scan->dsmSeg = dsm_create(MAXALIGN(poscan->downlinksCount * sizeof(scan->diskDownlinks[0])), 0);
+				poscan->dsmSegAttached = 0;
 				poscan->dsmHandle = dsm_segment_handle(scan->dsmSeg);
 				memcpy((Pointer) dsm_segment_address(scan->dsmSeg), scan->diskDownlinks,
 					   scan->downlinksCount * sizeof(scan->diskDownlinks[0]));
@@ -448,6 +449,7 @@ switch_to_disk_scan(BTreeSeqScan *scan)
 				LWLockAcquire(&poscan->downlinksPublish, LW_EXCLUSIVE);
 				LWLockRelease(&poscan->downlinksPublish);
 
+				Assert(poscan->dsmSegAttached == 0);
 				qsort(dsm_segment_address(scan->dsmSeg), poscan->downlinksCount,
 					  sizeof(scan->diskDownlinks[0]), cmp_downlinks);
 			}
@@ -466,13 +468,16 @@ switch_to_disk_scan(BTreeSeqScan *scan)
 			{
 				Assert(poscan->dsmHandle && !scan->dsmSeg);
 				scan->dsmSeg = dsm_attach(poscan->dsmHandle);
+				poscan->dsmSegAttached++;
 			}
 			if (scan->downlinksCount > 0)
 			{
 				index = pg_atomic_fetch_add_u64(&poscan->downlinkIndex, scan->downlinksCount);
 				memcpy((Pointer) dsm_segment_address(scan->dsmSeg) + index * sizeof(scan->diskDownlinks[0]),
 					   scan->diskDownlinks, scan->downlinksCount * sizeof(scan->diskDownlinks[0]));
 				index += scan->downlinksCount;
+				dsm_detach(scan->dsmSeg);
+				poscan->dsmSegAttached--;
 			}
 			LWLockRelease(&poscan->downlinksPublish);
 
@@ -1620,7 +1625,10 @@ free_btree_seq_scan(BTreeSeqScan *scan)
 	END_CRIT_SECTION();
 
 	if (scan->dsmSeg)
+	{
+		Assert(scan->poscan->dsmSegAttached == 0); /* All workers should have already detached */
 		dsm_detach(scan->dsmSeg);
+	}
 	pfree(scan->diskDownlinks);
 	pfree(scan);
 }
@@ -1648,7 +1656,10 @@ seq_scans_cleanup(void)
 		}
 		dlist_delete(&scan->listNode);
 		if (scan->dsmSeg)
+		{
+			Assert(scan->poscan->dsmSegAttached == 0);
 			dsm_detach(scan->dsmSeg);
+		}
 		pfree(scan);
 	}
 	dlist_init(&listOfScans);

src/orioledb.c

@@ -245,6 +245,7 @@ _PG_init(void)
 	int			i;
 	int			min_pool_size;
 
+
 	if (!process_shared_preload_libraries_in_progress)
 		return;