resolve conflict in relation_info.c

funbringer · funbringer · commit b94b2c114340 · 2016-10-30T01:01:37.000+03:00
diff --git a/src/init.c b/src/init.c
@@ -466,8 +466,12 @@ fill_prel_with_partitions(const Oid *partitions,
  *
  * borrowed from pg_inherits.c
  */
-Oid *
-find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size)
+find_children_status
+find_inheritance_children_array(Oid parentrelId,
+								LOCKMODE lockmode,
+								bool nowait,
+								uint32 *children_size,	/* ret value #1 */
+								Oid **children)			/* ret value #2 */
 {
 	Relation	relation;
 	SysScanDesc scan;
@@ -485,8 +489,12 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 	 */
 	if (!has_subclass(parentrelId))
 	{
-		*size = 0;
-		return NULL;
+		/* Init return values */
+		*children_size = 0;
+		children = NULL;
+
+		/* Ok, could not find any children */
+		return FCS_NO_CHILDREN;
 	}
 
 	/*
@@ -540,7 +548,25 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 		if (lockmode != NoLock)
 		{
 			/* Get the lock to synchronize against concurrent drop */
-			LockRelationOid(inhrelid, lockmode);
+			if (nowait)
+			{
+				if (!ConditionalLockRelationOid(inhrelid, lockmode))
+				{
+					uint32 j;
+
+					/* Unlock all previously locked children */
+					for (j = 0; j < i; j++)
+						UnlockRelationOid(oidarr[j], lockmode);
+
+					/* Init return values */
+					*children_size = numoids;
+					*children = oidarr;
+
+					/* We couldn't lock this child, retreat! */
+					return FCS_COULD_NOT_LOCK;
+				}
+			}
+			else LockRelationOid(inhrelid, lockmode);
 
 			/*
 			 * Now that we have the lock, double-check to see if the relation
@@ -557,8 +583,12 @@ find_inheritance_children_array(Oid parentrelId, LOCKMODE lockmode, uint32 *size
 		}
 	}
 
-	*size = numoids;
-	return oidarr;
+	/* Init return values */
+	*children_size = numoids;
+	*children = oidarr;
+
+	/* Ok, we have children */
+	return FCS_FOUND;
 }
 
 /*
@@ -752,7 +782,9 @@ read_pathman_config(void)
 		}
 
 		/* Create or update PartRelationInfo for this partitioned table */
-		refresh_pathman_relation_info(relid, parttype, text_to_cstring(attname));
+		refresh_pathman_relation_info(relid, parttype,
+									  text_to_cstring(attname),
+									  true); /* allow lazy prel loading */
 	}
 
 	/* Clean resources */
diff --git a/src/init.h b/src/init.h
@@ -109,9 +109,19 @@ void fill_prel_with_partitions(const Oid *partitions,
 							   const uint32 parts_count,
 							   PartRelationInfo *prel);
 
-Oid *find_inheritance_children_array(Oid parentrelId,
-									 LOCKMODE lockmode,
-									 uint32 *size);
+/* Result of find_inheritance_children_array() */
+typedef enum
+{
+	FCS_NO_CHILDREN = 0,	/* could not find any children (GOOD) */
+	FCS_COULD_NOT_LOCK,		/* could not lock one of the children */
+	FCS_FOUND				/* found some children (GOOD) */
+} find_children_status;
+
+find_children_status find_inheritance_children_array(Oid parentrelId,
+													 LOCKMODE lockmode,
+													 bool nowait,
+													 uint32 *children_size,
+													 Oid **children);
 
 char *build_check_constraint_name_internal(Oid relid,
 										   AttrNumber attno);
diff --git a/src/pl_funcs.c b/src/pl_funcs.c
@@ -623,7 +623,9 @@ add_to_pathman_config(PG_FUNCTION_ARGS)
 		/* Some flags might change during refresh attempt */
 		save_pathman_init_state(&init_state);
 
-		refresh_pathman_relation_info(relid, parttype, text_to_cstring(attname));
+		refresh_pathman_relation_info(relid, parttype,
+									  text_to_cstring(attname),
+									  false); /* initialize immediately */
 	}
 	PG_CATCH();
 	{
diff --git a/src/relation_info.c b/src/relation_info.c
@@ -23,9 +23,10 @@
 #include "utils/builtins.h"
 #include "utils/fmgroids.h"
 #include "utils/hsearch.h"
-#include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/snapmgr.h"
+#include "utils/syscache.h"
+#include "utils/lsyscache.h"
 #include "utils/typcache.h"
 
 
@@ -67,23 +68,24 @@ static Oid get_parent_of_partition_internal(Oid partition,
 const PartRelationInfo *
 refresh_pathman_relation_info(Oid relid,
 							  PartType partitioning_type,
-							  const char *part_column_name)
+							  const char *part_column_name,
+							  bool allow_incomplete)
 {
 	const LOCKMODE			lockmode = AccessShareLock;
 	const TypeCacheEntry   *typcache;
 	Oid					   *prel_children;
 	uint32					prel_children_count = 0,
 							i;
-	bool					found;
+	bool					found_entry;
 	PartRelationInfo	   *prel;
 	Datum					param_values[Natts_pathman_config_params];
 	bool					param_isnull[Natts_pathman_config_params];
 
 	prel = (PartRelationInfo *) hash_search(partitioned_rels,
 											(const void *) &relid,
-											HASH_ENTER, &found);
+											HASH_ENTER, &found_entry);
 	elog(DEBUG2,
-		 found ?
+		 found_entry ?
 			 "Refreshing record for relation %u in pg_pathman's cache [%u]" :
 			 "Creating new record for relation %u in pg_pathman's cache [%u]",
 		 relid, MyProcPid);
@@ -92,25 +94,42 @@ refresh_pathman_relation_info(Oid relid,
 	 * NOTE: Trick clang analyzer (first access without NULL pointer check).
 	 * Access to field 'valid' results in a dereference of a null pointer.
 	 */
-	prel->cmp_proc = InvalidOid;
+	prel->cmp_proc	= InvalidOid;
 
 	/* Clear outdated resources */
-	if (found && PrelIsValid(prel))
+	if (found_entry && PrelIsValid(prel))
 	{
 		/* Free these arrays iff they're not NULL */
 		FreeChildrenArray(prel);
 		FreeRangesArray(prel);
 	}
 
 	/* First we assume that this entry is invalid */
-	prel->valid = false;
+	prel->valid		= false;
+
+	/* Try locking parent, exit fast if 'allow_incomplete' */
+	if (allow_incomplete)
+	{
+		if (!ConditionalLockRelationOid(relid, lockmode))
+			return NULL; /* leave an invalid entry */
+	}
+	else LockRelationOid(relid, lockmode);
+
+	/* Check if parent exists */
+	if (!SearchSysCacheExists1(RELOID, ObjectIdGetDatum(relid)))
+	{
+		/* Nope, it doesn't, remove this entry and exit */
+		UnlockRelationOid(relid, lockmode);
+		remove_pathman_relation_info(relid);
+		return NULL; /* exit */
+	}
 
 	/* Make both arrays point to NULL */
-	prel->children = NULL;
-	prel->ranges = NULL;
+	prel->children	= NULL;
+	prel->ranges	= NULL;
 
 	/* Set partitioning type */
-	prel->parttype = partitioning_type;
+	prel->parttype	= partitioning_type;
 
 	/* Initialize PartRelationInfo using syscache & typcache */
 	prel->attnum	= get_attnum(relid, part_column_name);
@@ -135,16 +154,31 @@ refresh_pathman_relation_info(Oid relid,
 	prel->cmp_proc	= typcache->cmp_proc;
 	prel->hash_proc	= typcache->hash_proc;
 
-	LockRelationOid(relid, lockmode);
-	prel_children = find_inheritance_children_array(relid, lockmode,
-													&prel_children_count);
-	UnlockRelationOid(relid, lockmode);
-
-	/* If there's no children at all, remove this entry */
-	if (prel_children_count == 0)
+	/* Try searching for children (don't wait if we can't lock) */
+	switch (find_inheritance_children_array(relid, lockmode,
+											allow_incomplete,
+											&prel_children_count,
+											&prel_children))
 	{
-		remove_pathman_relation_info(relid);
-		return NULL;
+		/* If there's no children at all, remove this entry */
+		case FCS_NO_CHILDREN:
+			UnlockRelationOid(relid, lockmode);
+			remove_pathman_relation_info(relid);
+			return NULL; /* exit */
+
+		/* If can't lock children, leave an invalid entry */
+		case FCS_COULD_NOT_LOCK:
+			UnlockRelationOid(relid, lockmode);
+			return NULL; /* exit */
+
+		/* Found some children, just unlock parent */
+		case FCS_FOUND:
+			UnlockRelationOid(relid, lockmode);
+			break; /* continue */
+
+		/* Error: unknown result code */
+		default:
+			elog(ERROR, "error in " CppAsString(find_inheritance_children_array));
 	}
 
 	/*
@@ -155,10 +189,16 @@ refresh_pathman_relation_info(Oid relid,
 	 */
 	fill_prel_with_partitions(prel_children, prel_children_count, prel);
 
-	/* Add "partition+parent" tuple to cache */
+	/* Peform some actions for each child */
 	for (i = 0; i < prel_children_count; i++)
+	{
+		/* Add "partition+parent" pair to cache */
 		cache_parent_of_partition(prel_children[i], relid);
 
+		/* Now it's time to unlock this child */
+		UnlockRelationOid(prel_children[i], lockmode);
+	}
+
 	pfree(prel_children);
 
 	/* Read additional parameters ('enable_parent' and 'auto' at the moment) */
@@ -245,7 +285,7 @@ get_pathman_relation_info(Oid relid)
 
 			/* Refresh partitioned table cache entry (might turn NULL) */
 			/* TODO: possible refactoring, pass found 'prel' instead of searching */
-			prel = refresh_pathman_relation_info(relid, part_type, attname);
+			prel = refresh_pathman_relation_info(relid, part_type, attname, false);
 		}
 		/* Else clear remaining cache entry */
 		else remove_pathman_relation_info(relid);
@@ -609,10 +649,10 @@ try_perform_parent_refresh(Oid parent)
 		parttype = DatumGetPartType(values[Anum_pathman_config_parttype - 1]);
 		attname = DatumGetTextP(values[Anum_pathman_config_attname - 1]);
 
-		/* If anything went wrong, return false (actually, it might throw ERROR) */
-		if (!PrelIsValid(refresh_pathman_relation_info(parent, parttype,
-													   text_to_cstring(attname))))
-			return false;
+		/* If anything went wrong, return false (actually, it might emit ERROR) */
+		refresh_pathman_relation_info(parent, parttype,
+									  text_to_cstring(attname),
+									  true); /* allow lazy */
 	}
 	/* Not a partitioned relation */
 	else return false;
diff --git a/src/relation_info.h b/src/relation_info.h
@@ -120,7 +120,8 @@ PrelLastChild(const PartRelationInfo *prel)
 
 const PartRelationInfo *refresh_pathman_relation_info(Oid relid,
 													  PartType partitioning_type,
-													  const char *part_column_name);
+													  const char *part_column_name,
+													  bool allow_incomplete);
 void invalidate_pathman_relation_info(Oid relid, bool *found);
 void remove_pathman_relation_info(Oid relid);
 const PartRelationInfo *get_pathman_relation_info(Oid relid);

Original file line number	Diff line number	Diff line change
`@@ -623,7 +623,9 @@ add_to_pathman_config(PG_FUNCTION_ARGS)`
`623`	`623`	`/* Some flags might change during refresh attempt */`
`624`	`624`	`save_pathman_init_state(&init_state);`
`625`	`625`
`626`		`- refresh_pathman_relation_info(relid, parttype, text_to_cstring(attname));`
	`626`	`+ refresh_pathman_relation_info(relid, parttype,`
	`627`	`+ text_to_cstring(attname),`
	`628`	`+ false); /* initialize immediately */`
`627`	`629`	`}`
`628`	`630`	`PG_CATCH();`
`629`	`631`	`{`