improve error handling during pg_pathman's initialization

Author: funbringer

src/init.c

@@ -42,7 +42,6 @@
 #define PART_RELS_SIZE	10
 #define CHILD_FACTOR	500
 
-
 /* Storage for PartRelationInfos */
 HTAB		   *partitioned_rels = NULL;
 
@@ -52,6 +51,9 @@ HTAB		   *parent_cache = NULL;
 bool			initialization_needed = true;
 static bool		relcache_callback_needed = true;
 
+/* Help user in case of emergency */
+#define INIT_ERROR_HINT "pg_pathman will be disabled to allow you fix this"
+
 
 static bool init_pathman_relation_oids(void);
 static void fini_pathman_relation_oids(void);
@@ -284,9 +286,13 @@ fill_prel_with_partitions(const Oid *partitions,
 					if (validate_hash_constraint(con_expr, prel, &hash))
 						prel->children[hash] = partitions[i];
 					else
-						elog(ERROR,
-							 "Wrong constraint format for HASH partition \"%s\"",
-							 get_rel_name_or_relid(partitions[i]));
+					{
+						DisablePathman(); /* disable pg_pathman since config is broken */
+						ereport(ERROR,
+								(errmsg("Wrong constraint format for HASH partition \"%s\"",
+										get_rel_name_or_relid(partitions[i])),
+								 errhint(INIT_ERROR_HINT)));
+					}
 				}
 				break;
 
@@ -302,15 +308,24 @@ fill_prel_with_partitions(const Oid *partitions,
 						prel->ranges[i].max			= range_max;
 					}
 					else
-						elog(ERROR,
-							 "Wrong constraint format for RANGE partition \"%s\"",
-							 get_rel_name_or_relid(partitions[i]));
+					{
+						DisablePathman(); /* disable pg_pathman since config is broken */
+						ereport(ERROR,
+								(errmsg("Wrong constraint format for RANGE partition \"%s\"",
+										get_rel_name_or_relid(partitions[i])),
+								 errhint(INIT_ERROR_HINT)));
+					}
 				}
 				break;
 
 			default:
-				elog(ERROR, "Unknown partitioning type for relation \"%s\"",
-					 get_rel_name_or_relid(prel->key));
+			{
+				DisablePathman(); /* disable pg_pathman since config is broken */
+				ereport(ERROR,
+						(errmsg("Unknown partitioning type for relation \"%s\"",
+								get_rel_name_or_relid(prel->key)),
+						 errhint(INIT_ERROR_HINT)));
+			}
 		}
 	}
 
@@ -350,9 +365,12 @@ fill_prel_with_partitions(const Oid *partitions,
 		for (i = 0; i < PrelChildrenCount(prel); i++)
 		{
 			if (prel->children[i] == InvalidOid)
+			{
+				DisablePathman(); /* disable pg_pathman since config is broken */
 				elog(ERROR, "pg_pathman's cache for relation \"%s\" "
 							"has not been properly initialized",
 					 get_rel_name_or_relid(prel->key));
+			}
 		}
 #endif
 }
@@ -605,11 +623,10 @@ read_pathman_config(void)
 		if (get_rel_type_id(relid) == InvalidOid)
 		{
 			DisablePathman(); /* disable pg_pathman since config is broken */
-
 			ereport(ERROR,
 					(errmsg("Table \"%s\" contains nonexistent relation %u",
 							PATHMAN_CONFIG, relid),
-					 errdetail("pg_pathman will be disabled")));
+					 errhint(INIT_ERROR_HINT)));
 		}
 
 		/* Create or update PartRelationInfo for this partitioned table */
@@ -638,17 +655,30 @@ get_partition_constraint_expr(Oid partition, AttrNumber part_attno)
 	Expr	   *expr;			/* expression tree for constraint */
 
 	conname = build_check_constraint_name_internal(partition, part_attno);
-	conid = get_relation_constraint_oid(partition, conname, false);
+	conid = get_relation_constraint_oid(partition, conname, true);
+	if (conid == InvalidOid)
+	{
+		DisablePathman(); /* disable pg_pathman since config is broken */
+		ereport(ERROR,
+				(errmsg("constraint \"%s\" for partition \"%s\" does not exist",
+						conname, get_rel_name_or_relid(partition)),
+				 errhint(INIT_ERROR_HINT)));
+	}
 
 	con_tuple = SearchSysCache1(CONSTROID, ObjectIdGetDatum(conid));
 	conbin_datum = SysCacheGetAttr(CONSTROID, con_tuple,
 								   Anum_pg_constraint_conbin,
 								   &conbin_isnull);
 	if (conbin_isnull)
 	{
-		elog(DEBUG2, "conbin is null for constraint %s", conname);
+		DisablePathman(); /* disable pg_pathman since config is broken */
+		ereport(WARNING,
+				(errmsg("constraint \"%s\" for partition \"%s\" has NULL conbin",
+						conname, get_rel_name_or_relid(partition)),
+				 errhint(INIT_ERROR_HINT)));
 		pfree(conname);
-		return NULL;
+
+		return NULL; /* could not parse */
 	}
 	pfree(conname);
 
@@ -690,6 +720,9 @@ validate_range_constraint(const Expr *expr,
 	const BoolExpr		   *boolexpr = (const BoolExpr *) expr;
 	const OpExpr		   *opexpr;
 
+	if (!expr)
+		return false;
+
 	/* it should be an AND operator on top */
 	if (!and_clause((Node *) expr))
 		return false;
@@ -779,6 +812,9 @@ validate_hash_constraint(const Expr *expr,
 						   *type_hash_proc_expr;
 	const Var			   *var; /* partitioned column */
 
+	if (!expr)
+		return false;
+
 	if (!IsA(expr, OpExpr))
 		return false;
 	eq_expr = (const OpExpr *) expr;
@@ -825,7 +861,7 @@ validate_hash_constraint(const Expr *expr,
 			return false;
 
 		/* Check that PARTITIONS_COUNT is equal to total amount of partitions */
-		if (DatumGetUInt32(((Const*) second)->constvalue) != PrelChildrenCount(prel))
+		if (DatumGetUInt32(((Const *) second)->constvalue) != PrelChildrenCount(prel))
 			return false;
 
 		/* Check that CUR_PARTITION_HASH is Const */