WIP place some more close_pathman_relation_info()

Author: funbringer

src/hooks.c

@@ -323,9 +323,6 @@ void invalidate_pathman_relation_info_cache(void);
 void close_pathman_relation_info(PartRelationInfo *prel);
 bool has_pathman_relation_info(Oid relid);
 PartRelationInfo *get_pathman_relation_info(Oid relid);
-PartRelationInfo *get_pathman_relation_info_after_lock(Oid relid,
-													   bool unlock_if_not_found,
-													   LockAcquireResult *lock_result);
 
 void shout_if_prel_is_invalid(const Oid parent_oid,
 							  const PartRelationInfo *prel,

src/include/relation_info.h

@@ -334,26 +334,36 @@ create_partitions_for_value_internal(Oid relid, Datum value, Oid value_type,
 
 	PG_TRY();
 	{
-		LockAcquireResult	lock_result; /* could we lock the parent? */
 		Datum				values[Natts_pathman_config];
 		bool				isnull[Natts_pathman_config];
 
 		/* Get both PartRelationInfo & PATHMAN_CONFIG contents for this relation */
 		if (pathman_config_contains_relation(relid, values, isnull, NULL, NULL))
 		{
 			PartRelationInfo   *prel;
+			LockAcquireResult	lock_result;		/* could we lock the parent? */
 			Oid					base_bound_type;	/* base type of prel->ev_type */
 			Oid					base_value_type;	/* base type of value_type */
 
+			/* Prevent modifications of partitioning scheme */
+			lock_result = xact_lock_rel(relid, ShareUpdateExclusiveLock, false);
+
 			/* Fetch PartRelationInfo by 'relid' */
-			prel = get_pathman_relation_info_after_lock(relid, true, &lock_result);
+			prel = get_pathman_relation_info(relid);
 			shout_if_prel_is_invalid(relid, prel, PT_RANGE);
 
 			/* Fetch base types of prel->ev_type & value_type */
 			base_bound_type = getBaseType(prel->ev_type);
 			base_value_type = getBaseType(value_type);
 
-			/* Search for a suitable partition if we didn't hold it */
+			/*
+			 * Search for a suitable partition if we didn't hold it,
+			 * since somebody might have just created it for us.
+			 *
+			 * If the table is locked, it means that we've
+			 * already failed to find a suitable partition
+			 * and called this function to do the job.
+			 */
 			Assert(lock_result != LOCKACQUIRE_NOT_AVAIL);
 			if (lock_result == LOCKACQUIRE_OK)
 			{

src/partition_creation.c

@@ -691,6 +691,7 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 	int				empty_slot_idx = -1,		/* do we have a slot for BGWorker? */
 					i;
 	TransactionId	rel_xmin;
+	LOCKMODE		lockmode = ShareUpdateExclusiveLock;
 
 	/* Check batch_size */
 	if (batch_size < 1 || batch_size > 10000)
@@ -703,12 +704,12 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 		ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 						errmsg("'sleep_time' should not be less than 0.5")));
 
+	/* Prevent concurrent function calls */
+	LockRelationOid(relid, lockmode);
+
 	/* Check if relation is a partitioned table */
-	shout_if_prel_is_invalid(relid,
-							 /* We also lock the parent relation */
-							 get_pathman_relation_info_after_lock(relid, true, NULL),
-							 /* Partitioning type does not matter here */
-							 PT_ANY);
+	if (!has_pathman_relation_info(relid))
+		shout_if_prel_is_invalid(relid, NULL, PT_ANY);
 
 	/* Check that partitioning operation result is visible */
 	if (pathman_config_contains_relation(relid, NULL, NULL, &rel_xmin, NULL))
@@ -723,7 +724,7 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 
 	/*
 	 * Look for an empty slot and also check that a concurrent
-	 * partitioning operation for this table hasn't been started yet
+	 * partitioning operation for this table hasn't started yet.
 	 */
 	for (i = 0; i < PART_WORKER_SLOTS; i++)
 	{
@@ -797,6 +798,9 @@ partition_table_concurrently(PG_FUNCTION_ARGS)
 		 CppAsString(stop_concurrent_part_task),
 		 get_rel_name(relid));
 
+	/* We don't need this lock anymore */
+	UnlockRelationOid(relid, lockmode);
+
 	PG_RETURN_VOID();
 }
 

src/pathman_workers.c

@@ -75,22 +75,22 @@ PG_FUNCTION_INFO_V1( pathman_version );
 /* User context for function show_partition_list_internal() */
 typedef struct
 {
-	Relation				pathman_config;
-	HeapScanDesc			pathman_config_scan;
-	Snapshot				snapshot;
+	Relation			pathman_config;
+	HeapScanDesc		pathman_config_scan;
+	Snapshot			snapshot;
 
-	const PartRelationInfo *current_prel;	/* selected PartRelationInfo */
+	PartRelationInfo   *current_prel;	/* selected PartRelationInfo */
 
-	Size					child_number;	/* child we're looking at */
-	SPITupleTable		   *tuptable;		/* buffer for tuples */
+	Size				child_number;	/* child we're looking at */
+	SPITupleTable	   *tuptable;		/* buffer for tuples */
 } show_partition_list_cxt;
 
 /* User context for function show_pathman_cache_stats_internal() */
 typedef struct
 {
-	MemoryContext			pathman_contexts[PATHMAN_MCXT_COUNT];
-	HTAB				   *pathman_htables[PATHMAN_MCXT_COUNT];
-	int						current_item;
+	MemoryContext		pathman_contexts[PATHMAN_MCXT_COUNT];
+	HTAB			   *pathman_htables[PATHMAN_MCXT_COUNT];
+	int					current_item;
 } show_cache_stats_cxt;
 
 /*
@@ -362,10 +362,10 @@ show_partition_list_internal(PG_FUNCTION_ARGS)
 		/* Iterate through pathman cache */
 		for (;;)
 		{
-			const PartRelationInfo *prel;
-			HeapTuple				htup;
-			Datum					values[Natts_pathman_partition_list];
-			bool					isnull[Natts_pathman_partition_list] = { 0 };
+			HeapTuple			htup;
+			Datum				values[Natts_pathman_partition_list];
+			bool				isnull[Natts_pathman_partition_list] = { 0 };
+			PartRelationInfo   *prel;
 
 			/* Fetch next PartRelationInfo if needed */
 			if (usercxt->current_prel == NULL)
@@ -401,6 +401,9 @@ show_partition_list_internal(PG_FUNCTION_ARGS)
 			/* If we've run out of partitions, switch to the next 'prel' */
 			if (usercxt->child_number >= PrelChildrenCount(prel))
 			{
+				/* Don't forget to close 'prel'! */
+				close_pathman_relation_info(prel);
+
 				usercxt->current_prel = NULL;
 				usercxt->child_number = 0;
 
@@ -787,13 +790,13 @@ add_to_pathman_config(PG_FUNCTION_ARGS)
 	{
 		pfree(children);
 
-		/* Now try to create a PartRelationInfo */
 		PG_TRY();
 		{
 			/* Some flags might change during refresh attempt */
 			save_pathman_init_state(&init_state);
 
-			get_pathman_relation_info(relid);
+			/* Now try to create a PartRelationInfo */
+			has_pathman_relation_info(relid);
 		}
 		PG_CATCH();
 		{