introduce function shout_if_prel_is_invalid(), fixes

Author: funbringer

src/init.c

@@ -155,13 +155,10 @@ init_pathman_relation_oids(void)
 
 	/* Cache PATHMAN_CONFIG relation's Oid */
 	pathman_config_relid = get_relname_relid(PATHMAN_CONFIG, schema);
-	/* NOTE: add more relations to be cached right here ^^^ */
-
-	/* Return false if *any* relation doesn't exist yet */
 	if (pathman_config_relid == InvalidOid)
-	{
 		return false;
-	}
+
+	/* NOTE: add more relations to be cached right here ^^^ */
 
 	/* Everything is fine, proceed */
 	return true;

src/pg_pathman.c

@@ -813,9 +813,11 @@ create_partitions_internal(Oid relid, Datum value, Oid value_type)
 		Datum					values[Natts_pathman_config];
 		bool					isnull[Natts_pathman_config];
 
+		prel = get_pathman_relation_info(relid);
+		shout_if_prel_is_invalid(relid, prel, PT_RANGE);
+
 		/* Get both PartRelationInfo & PATHMAN_CONFIG contents for this relation */
-		if ((prel = get_pathman_relation_info(relid)) != NULL &&
-			pathman_config_contains_relation(relid, values, isnull, NULL))
+		if (pathman_config_contains_relation(relid, values, isnull, NULL))
 		{
 			Datum		min_rvalue,
 						max_rvalue;
@@ -827,10 +829,6 @@ create_partitions_internal(Oid relid, Datum value, Oid value_type)
 
 			FmgrInfo	interval_type_cmp;
 
-			if (prel->parttype != PT_RANGE)
-				elog(ERROR, "Relation \"%s\" is not partitioned by RANGE",
-					 get_rel_name_or_relid(relid));
-
 			/* Fill the FmgrInfo struct with a cmp(value, part_attribute) function */
 			fill_type_cmp_fmgr_info(&interval_type_cmp, value_type, prel->atttype);
 
@@ -892,7 +890,7 @@ create_partitions_internal(Oid relid, Datum value, Oid value_type)
 			SPI_finish(); /* close SPI connection */
 		}
 		else
-			elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
+			elog(ERROR, "pg_pathman's config does not contain relation \"%s\"",
 				 get_rel_name_or_relid(relid));
 	}
 	PG_CATCH();
@@ -931,7 +929,8 @@ create_partitions(Oid relid, Datum value, Oid value_type)
 	if (pathman_config_contains_relation(relid, NULL, NULL, &rel_xmin))
 	{
 		/* If table was partitioned in some previous xact, run BGWorker */
-		if (TransactionIdPrecedes(rel_xmin, GetCurrentTransactionId()))
+		if (TransactionIdPrecedes(rel_xmin, GetCurrentTransactionId()) ||
+			TransactionIdEquals(rel_xmin, FrozenTransactionId))
 		{
 			elog(DEBUG2, "create_partitions(): chose BGW [%u]", MyProcPid);
 			last_partition = create_partitions_bg_worker(relid, value, value_type);
@@ -1178,6 +1177,9 @@ handle_binary_opexpr(WalkerContext *context, WrapperNode *result,
 										result);
 				return;
 			}
+
+		default:
+			elog(ERROR, "Unknown partitioning type %u", prel->parttype);
 	}
 
 	result->rangeset = list_make1_irange(make_irange(0,

src/pl_funcs.c

@@ -180,14 +180,12 @@ find_or_create_range_partition(PG_FUNCTION_ARGS)
 	search_rangerel_result	search_state;
 
 	prel = get_pathman_relation_info(parent_oid);
-
-	if (!prel)
-		PG_RETURN_NULL();
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	fill_type_cmp_fmgr_info(&cmp_func, value_type, prel->atttype);
 
-	/* FIXME: does this function even work? */
-	search_state = search_range_partition_eq(value, &cmp_func,prel,
+	/* Use available PartRelationInfo to find partition */
+	search_state = search_range_partition_eq(value, &cmp_func, prel,
 											 &found_rentry);
 
 	/*
@@ -246,9 +244,7 @@ get_range_by_part_oid(PG_FUNCTION_ARGS)
 	const PartRelationInfo *prel;
 
 	prel = get_pathman_relation_info(parent_oid);
-	if (!prel)
-		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
-			 get_rel_name_or_relid(parent_oid));
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	ranges = PrelGetRangesArray(prel);
 
@@ -266,6 +262,7 @@ get_range_by_part_oid(PG_FUNCTION_ARGS)
 			PG_RETURN_ARRAYTYPE_P(arr);
 		}
 
+	/* No partition found, report error */
 	elog(ERROR, "Relation \"%s\" has no partition \"%s\"",
 		 get_rel_name_or_relid(parent_oid),
 		 get_rel_name_or_relid(child_oid));
@@ -290,9 +287,7 @@ get_range_by_idx(PG_FUNCTION_ARGS)
 	const PartRelationInfo *prel;
 
 	prel = get_pathman_relation_info(parent_oid);
-	if (!prel)
-		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
-			 get_rel_name_or_relid(parent_oid));
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	if (((uint32) abs(idx)) >= PrelChildrenCount(prel))
 		elog(ERROR, "Partition #%d does not exist (total amount is %u)",
@@ -326,14 +321,7 @@ get_min_range_value(PG_FUNCTION_ARGS)
 	const PartRelationInfo *prel;
 
 	prel = get_pathman_relation_info(parent_oid);
-	if (!prel)
-		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
-			 get_rel_name_or_relid(parent_oid));
-
-	if (prel->parttype != PT_RANGE)
-		if (!prel)
-			elog(ERROR, "Relation \"%s\" is not partitioned by RANGE",
-				 get_rel_name_or_relid(parent_oid));
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	ranges = PrelGetRangesArray(prel);
 
@@ -351,14 +339,7 @@ get_max_range_value(PG_FUNCTION_ARGS)
 	const PartRelationInfo *prel;
 
 	prel = get_pathman_relation_info(parent_oid);
-	if (!prel)
-		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
-			 get_rel_name_or_relid(parent_oid));
-
-	if (prel->parttype != PT_RANGE)
-		if (!prel)
-			elog(ERROR, "Relation \"%s\" is not partitioned by RANGE",
-				 get_rel_name_or_relid(parent_oid));
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	ranges = PrelGetRangesArray(prel);
 
@@ -388,14 +369,7 @@ check_overlap(PG_FUNCTION_ARGS)
 	const PartRelationInfo *prel;
 
 	prel = get_pathman_relation_info(parent_oid);
-	if (!prel)
-		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
-			 get_rel_name_or_relid(parent_oid));
-
-	if (prel->parttype != PT_RANGE)
-		if (!prel)
-			elog(ERROR, "Relation \"%s\" is not partitioned by RANGE",
-				 get_rel_name_or_relid(parent_oid));
+	shout_if_prel_is_invalid(parent_oid, prel, PT_RANGE);
 
 	/* comparison functions */
 	fill_type_cmp_fmgr_info(&cmp_func_1, p1_type, prel->atttype);

src/relation_info.h

@@ -19,11 +19,12 @@
 
 
 /*
- * Partitioning type
+ * Partitioning type.
  */
 typedef enum
 {
-	PT_HASH = 1,
+	PT_INDIFFERENT = 0, /* for part type traits (virtual type) */
+	PT_HASH,
 	PT_RANGE
 } PartType;
 

src/utils.c

@@ -711,6 +711,53 @@ is_string_type_internal(Oid typid)
 		   typid == CSTRINGOID;
 }
 
+/*
+ * Common PartRelationInfo checks. Emit ERROR if anything is wrong.
+ */
+void
+shout_if_prel_is_invalid(Oid parent_oid,
+						 const PartRelationInfo *prel,
+						 PartType expected_part_type)
+{
+	if (!prel)
+		elog(ERROR, "Relation \"%s\" is not partitioned by pg_pathman",
+			 get_rel_name_or_relid(parent_oid));
+
+	if (!PrelIsValid(prel))
+		elog(ERROR, "pg_pathman's cache contains invalid entry "
+					"for relation \"%s\" [%u]",
+			 get_rel_name_or_relid(parent_oid),
+			 MyProcPid);
+
+	/* Check partitioning type unless it's "indifferent" */
+	if (expected_part_type != PT_INDIFFERENT &&
+		expected_part_type != prel->parttype)
+	{
+		char *expected_str;
+
+		switch (expected_part_type)
+		{
+			case PT_HASH:
+				expected_str = "HASH";
+				break;
+
+			case PT_RANGE:
+				expected_str = "RANGE";
+				break;
+
+			default:
+				elog(ERROR,
+					 "expected_str selection not implemented for type %d",
+					 expected_part_type);
+		}
+
+		elog(ERROR, "Relation \"%s\" is not partitioned by %s",
+			 get_rel_name_or_relid(parent_oid),
+			 expected_str);
+	}
+}
+
+
 /*
  * Try to find binary operator.
  *

src/utils.h

@@ -55,6 +55,9 @@ bool is_string_type_internal(Oid typid);
 bool check_rinfo_for_partitioned_attr(List *rinfo,
 									  Index varno,
 									  AttrNumber varattno);
+void shout_if_prel_is_invalid(Oid parent_oid,
+							  const PartRelationInfo *prel,
+							  PartType expected_part_type);
 
 /*
  * Misc.