Improve error reporting for tuple-routing failures.

Currently, the whole row is shown without column names.  Instead,
adopt a style similar to _bt_check_unique() in ExecFindPartition()
and show the failing key: (key1, ...) = (val1, ...).

Amit Langote, per a complaint from Simon Riggs.  Reviewed by me;
I also adjusted the grammar in one of the comments.

Discussion: http://postgr.es/m/9f9dc7ae-14f0-4a25-5485-964d9bfc19bd@lab.ntt.co.jp

src/backend/access/index/genam.c

@@ -168,6 +168,10 @@ IndexScanEnd(IndexScanDesc scan)
  * The passed-in values/nulls arrays are the "raw" input to the index AM,
  * e.g. results of FormIndexDatum --- this is not necessarily what is stored
  * in the index, but it's what the user perceives to be stored.
+ *
+ * Note: if you change anything here, check whether
+ * ExecBuildSlotPartitionKeyDescription() in execMain.c needs a similar
+ * change.
  */
 char *
 BuildIndexValueDescription(Relation indexRelation,

src/backend/catalog/partition.c

@@ -140,13 +140,6 @@ static int partition_bound_bsearch(PartitionKey key,
 						PartitionBoundInfo boundinfo,
 						void *probe, bool probe_is_bound, bool *is_equal);
 
-/* Support get_partition_for_tuple() */
-static void FormPartitionKeyDatum(PartitionDispatch pd,
-					  TupleTableSlot *slot,
-					  EState *estate,
-					  Datum *values,
-					  bool *isnull);
-
 /*
  * RelationBuildPartitionDesc
  *		Form rel's partition descriptor
@@ -1608,7 +1601,7 @@ generate_partition_qual(Relation rel)
  * the heap tuple passed in.
  * ----------------
  */
-static void
+void
 FormPartitionKeyDatum(PartitionDispatch pd,
 					  TupleTableSlot *slot,
 					  EState *estate,
@@ -1672,7 +1665,8 @@ int
 get_partition_for_tuple(PartitionDispatch *pd,
 						TupleTableSlot *slot,
 						EState *estate,
-						Oid *failed_at)
+						PartitionDispatchData **failed_at,
+						TupleTableSlot **failed_slot)
 {
 	PartitionDispatch parent;
 	Datum		values[PARTITION_MAX_KEYS];
@@ -1693,13 +1687,6 @@ get_partition_for_tuple(PartitionDispatch *pd,
 		TupleTableSlot *myslot = parent->tupslot;
 		TupleConversionMap *map = parent->tupmap;
 
-		/* Quick exit */
-		if (partdesc->nparts == 0)
-		{
-			*failed_at = RelationGetRelid(parent->reldesc);
-			return -1;
-		}
-
 		if (myslot != NULL && map != NULL)
 		{
 			HeapTuple	tuple = ExecFetchSlotTuple(slot);
@@ -1710,6 +1697,14 @@ get_partition_for_tuple(PartitionDispatch *pd,
 			slot = myslot;
 		}
 
+		/* Quick exit */
+		if (partdesc->nparts == 0)
+		{
+			*failed_at = parent;
+			*failed_slot = slot;
+			return -1;
+		}
+
 		/*
 		 * Extract partition key from tuple. Expression evaluation machinery
 		 * that FormPartitionKeyDatum() invokes expects ecxt_scantuple to
@@ -1774,7 +1769,8 @@ get_partition_for_tuple(PartitionDispatch *pd,
 		if (cur_index < 0)
 		{
 			result = -1;
-			*failed_at = RelationGetRelid(parent->reldesc);
+			*failed_at = parent;
+			*failed_slot = slot;
 			break;
 		}
 		else if (parent->indexes[cur_index] >= 0)

src/backend/executor/execMain.c

@@ -60,6 +60,7 @@
 #include "utils/lsyscache.h"
 #include "utils/memutils.h"
 #include "utils/rls.h"
+#include "utils/ruleutils.h"
 #include "utils/snapmgr.h"
 #include "utils/tqual.h"
 
@@ -95,6 +96,10 @@ static char *ExecBuildSlotValueDescription(Oid reloid,
 							  TupleDesc tupdesc,
 							  Bitmapset *modifiedCols,
 							  int maxfieldlen);
+static char *ExecBuildSlotPartitionKeyDescription(Relation rel,
+									 Datum *values,
+									 bool *isnull,
+									 int maxfieldlen);
 static void EvalPlanQualStart(EPQState *epqstate, EState *parentestate,
 				  Plan *planTree);
 
@@ -3189,33 +3194,122 @@ ExecFindPartition(ResultRelInfo *resultRelInfo, PartitionDispatch *pd,
 				  TupleTableSlot *slot, EState *estate)
 {
 	int			result;
-	Oid			failed_at;
+	PartitionDispatchData *failed_at;
+	TupleTableSlot *failed_slot;
 
-	result = get_partition_for_tuple(pd, slot, estate, &failed_at);
+	result = get_partition_for_tuple(pd, slot, estate,
+									 &failed_at, &failed_slot);
 	if (result < 0)
 	{
-		Relation	rel = resultRelInfo->ri_RelationDesc;
+		Relation	failed_rel;
+		Datum		key_values[PARTITION_MAX_KEYS];
+		bool		key_isnull[PARTITION_MAX_KEYS];
 		char	   *val_desc;
-		Bitmapset  *insertedCols,
-				   *updatedCols,
-				   *modifiedCols;
-		TupleDesc	tupDesc = RelationGetDescr(rel);
-
-		insertedCols = GetInsertedColumns(resultRelInfo, estate);
-		updatedCols = GetUpdatedColumns(resultRelInfo, estate);
-		modifiedCols = bms_union(insertedCols, updatedCols);
-		val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
-												 slot,
-												 tupDesc,
-												 modifiedCols,
-												 64);
-		Assert(OidIsValid(failed_at));
+		ExprContext *ecxt = GetPerTupleExprContext(estate);
+
+		failed_rel = failed_at->reldesc;
+		ecxt->ecxt_scantuple = failed_slot;
+		FormPartitionKeyDatum(failed_at, failed_slot, estate,
+							  key_values, key_isnull);
+		val_desc = ExecBuildSlotPartitionKeyDescription(failed_rel,
+														key_values,
+														key_isnull,
+														64);
+		Assert(OidIsValid(RelationGetRelid(failed_rel)));
 		ereport(ERROR,
 				(errcode(ERRCODE_CHECK_VIOLATION),
 				 errmsg("no partition of relation \"%s\" found for row",
-						get_rel_name(failed_at)),
-			val_desc ? errdetail("Failing row contains %s.", val_desc) : 0));
+						RelationGetRelationName(failed_rel)),
+			val_desc ? errdetail("Partition key of the failing row contains %s.", val_desc) : 0));
 	}
 
 	return result;
 }
+
+/*
+ * BuildSlotPartitionKeyDescription
+ *
+ * This works very much like BuildIndexValueDescription() and is currently
+ * used for building error messages when ExecFindPartition() fails to find
+ * partition for a row.
+ */
+static char *
+ExecBuildSlotPartitionKeyDescription(Relation rel,
+									 Datum *values,
+									 bool *isnull,
+									 int maxfieldlen)
+{
+	StringInfoData	buf;
+	PartitionKey	key = RelationGetPartitionKey(rel);
+	int			partnatts = get_partition_natts(key);
+	int			i;
+	Oid			relid = RelationGetRelid(rel);
+	AclResult	aclresult;
+
+	if (check_enable_rls(relid, InvalidOid, true) == RLS_ENABLED)
+		return NULL;
+
+	/* If the user has table-level access, just go build the description. */
+	aclresult = pg_class_aclcheck(relid, GetUserId(), ACL_SELECT);
+	if (aclresult != ACLCHECK_OK)
+	{
+		/*
+		 * Step through the columns of the partition key and make sure the
+		 * user has SELECT rights on all of them.
+		 */
+		for (i = 0; i < partnatts; i++)
+		{
+			AttrNumber	attnum = get_partition_col_attnum(key, i);
+
+			/*
+			 * If this partition key column is an expression, we return no
+			 * detail rather than try to figure out what column(s) the
+			 * expression includes and if the user has SELECT rights on them.
+			 */
+			if (attnum == InvalidAttrNumber ||
+				pg_attribute_aclcheck(relid, attnum, GetUserId(),
+									  ACL_SELECT) != ACLCHECK_OK)
+				return NULL;
+		}
+	}
+
+	initStringInfo(&buf);
+	appendStringInfo(&buf, "(%s) = (",
+					 pg_get_partkeydef_columns(relid, true));
+
+	for (i = 0; i < partnatts; i++)
+	{
+		char	   *val;
+		int			vallen;
+
+		if (isnull[i])
+			val = "null";
+		else
+		{
+			Oid			foutoid;
+			bool		typisvarlena;
+
+			getTypeOutputInfo(get_partition_col_typid(key, i),
+							  &foutoid, &typisvarlena);
+			val = OidOutputFunctionCall(foutoid, values[i]);
+		}
+
+		if (i > 0)
+			appendStringInfoString(&buf, ", ");
+
+		/* truncate if needed */
+		vallen = strlen(val);
+		if (vallen <= maxfieldlen)
+			appendStringInfoString(&buf, val);
+		else
+		{
+			vallen = pg_mbcliplen(val, vallen, maxfieldlen);
+			appendBinaryStringInfo(&buf, val, vallen);
+			appendStringInfoString(&buf, "...");
+		}
+	}
+
+	appendStringInfoChar(&buf, ')');
+
+	return buf.data;
+}

src/backend/utils/adt/ruleutils.c

@@ -317,7 +317,8 @@ static char *pg_get_indexdef_worker(Oid indexrelid, int colno,
 					   const Oid *excludeOps,
 					   bool attrsOnly, bool showTblSpc,
 					   int prettyFlags, bool missing_ok);
-static char *pg_get_partkeydef_worker(Oid relid, int prettyFlags);
+static char *pg_get_partkeydef_worker(Oid relid, int prettyFlags,
+						 bool attrsOnly);
 static char *pg_get_constraintdef_worker(Oid constraintId, bool fullCommand,
 							int prettyFlags, bool missing_ok);
 static text *pg_get_expr_worker(text *expr, Oid relid, const char *relname,
@@ -1431,14 +1432,26 @@ pg_get_partkeydef(PG_FUNCTION_ARGS)
 	Oid			relid = PG_GETARG_OID(0);
 
 	PG_RETURN_TEXT_P(string_to_text(pg_get_partkeydef_worker(relid,
-														PRETTYFLAG_INDENT)));
+														PRETTYFLAG_INDENT,
+															 false)));
+}
+
+/* Internal version that just reports the column definitions */
+char *
+pg_get_partkeydef_columns(Oid relid, bool pretty)
+{
+	int			prettyFlags;
+
+	prettyFlags = pretty ? PRETTYFLAG_PAREN | PRETTYFLAG_INDENT : PRETTYFLAG_INDENT;
+	return pg_get_partkeydef_worker(relid, prettyFlags, true);
 }
 
 /*
  * Internal workhorse to decompile a partition key definition.
  */
 static char *
-pg_get_partkeydef_worker(Oid relid, int prettyFlags)
+pg_get_partkeydef_worker(Oid relid, int prettyFlags,
+						 bool attrsOnly)
 {
 	Form_pg_partitioned_table form;
 	HeapTuple	tuple;
@@ -1508,17 +1521,20 @@ pg_get_partkeydef_worker(Oid relid, int prettyFlags)
 	switch (form->partstrat)
 	{
 		case PARTITION_STRATEGY_LIST:
-			appendStringInfo(&buf, "LIST");
+			if (!attrsOnly)
+				appendStringInfo(&buf, "LIST");
 			break;
 		case PARTITION_STRATEGY_RANGE:
-			appendStringInfo(&buf, "RANGE");
+			if (!attrsOnly)
+				appendStringInfo(&buf, "RANGE");
 			break;
 		default:
 			elog(ERROR, "unexpected partition strategy: %d",
 				 (int) form->partstrat);
 	}
 
-	appendStringInfo(&buf, " (");
+	if (!attrsOnly)
+		appendStringInfo(&buf, " (");
 	sep = "";
 	for (keyno = 0; keyno < form->partnatts; keyno++)
 	{
@@ -1561,14 +1577,17 @@ pg_get_partkeydef_worker(Oid relid, int prettyFlags)
 
 		/* Add collation, if not default for column */
 		partcoll = partcollation->values[keyno];
-		if (OidIsValid(partcoll) && partcoll != keycolcollation)
+		if (!attrsOnly && OidIsValid(partcoll) && partcoll != keycolcollation)
 			appendStringInfo(&buf, " COLLATE %s",
 							 generate_collation_name((partcoll)));
 
 		/* Add the operator class name, if not default */
-		get_opclass_name(partclass->values[keyno], keycoltype, &buf);
+		if (!attrsOnly)
+			get_opclass_name(partclass->values[keyno], keycoltype, &buf);
 	}
-	appendStringInfoChar(&buf, ')');
+
+	if (!attrsOnly)
+		appendStringInfoChar(&buf, ')');
 
 	/* Clean up */
 	ReleaseSysCache(tuple);

src/include/catalog/partition.h

@@ -85,8 +85,14 @@ extern List *RelationGetPartitionQual(Relation rel);
 extern PartitionDispatch *RelationGetPartitionDispatchInfo(Relation rel,
 								 int lockmode, int *num_parted,
 								 List **leaf_part_oids);
+extern void FormPartitionKeyDatum(PartitionDispatch pd,
+					  TupleTableSlot *slot,
+					  EState *estate,
+					  Datum *values,
+					  bool *isnull);
 extern int get_partition_for_tuple(PartitionDispatch *pd,
 						TupleTableSlot *slot,
 						EState *estate,
-						Oid *failed_at);
+						PartitionDispatchData **failed_at,
+						TupleTableSlot **failed_slot);
 #endif   /* PARTITION_H */

src/include/utils/ruleutils.h

@@ -21,6 +21,8 @@
 extern char *pg_get_indexdef_string(Oid indexrelid);
 extern char *pg_get_indexdef_columns(Oid indexrelid, bool pretty);
 
+extern char *pg_get_partkeydef_columns(Oid relid, bool pretty);
+
 extern char *pg_get_constraintdef_command(Oid constraintId);
 extern char *deparse_expression(Node *expr, List *dpcontext,
 				   bool forceprefix, bool showimplicit);

src/test/regress/expected/insert.out

@@ -234,14 +234,14 @@ insert into part_ee_ff2 values ('ff', 11);
 -- fail
 insert into range_parted values ('a', 0);
 ERROR:  no partition of relation "range_parted" found for row
-DETAIL:  Failing row contains (a, 0).
+DETAIL:  Partition key of the failing row contains (a, (b + 0)) = (a, 0).
 -- ok
 insert into range_parted values ('a', 1);
 insert into range_parted values ('a', 10);
 -- fail
 insert into range_parted values ('a', 20);
 ERROR:  no partition of relation "range_parted" found for row
-DETAIL:  Failing row contains (a, 20).
+DETAIL:  Partition key of the failing row contains (a, (b + 0)) = (a, 20).
 -- ok
 insert into range_parted values ('b', 1);
 insert into range_parted values ('b', 10);
@@ -265,10 +265,10 @@ insert into list_parted (a) values ('aA');
 -- fail (partition of part_ee_ff not found in both cases)
 insert into list_parted values ('EE', 0);
 ERROR:  no partition of relation "part_ee_ff" found for row
-DETAIL:  Failing row contains (EE, 0).
+DETAIL:  Partition key of the failing row contains (b) = (0).
 insert into part_ee_ff values ('EE', 0);
 ERROR:  no partition of relation "part_ee_ff" found for row
-DETAIL:  Failing row contains (EE, 0).
+DETAIL:  Partition key of the failing row contains (b) = (0).
 -- ok
 insert into list_parted values ('EE', 1);
 insert into part_ee_ff values ('EE', 10);
@@ -351,6 +351,10 @@ select tableoid::regclass, * from p;
  p11      | 1 | 2
 (1 row)
 
+-- check that proper message is shown after failure to route through p1
+insert into p (a, b) values (1, 5);
+ERROR:  no partition of relation "p1" found for row
+DETAIL:  Partition key of the failing row contains ((b + 0)) = (5).
 truncate p;
 alter table p add constraint check_b check (b = 3);
 -- check that correct input row is shown when constraint check_b fails on p11
@@ -386,5 +390,31 @@ with ins (a, b, c) as
  p4  | 1 |  30 |  39
 (5 rows)
 
+-- check that message shown after failure to find a partition shows the
+-- appropriate key description (or none) in various situations
+create table key_desc (a int, b int) partition by list ((a+0));
+create table key_desc_1 partition of key_desc for values in (1) partition by range (b);
+create user someone_else;
+grant select (a) on key_desc_1 to someone_else;
+grant insert on key_desc to someone_else;
+set role someone_else;
+-- no key description is shown
+insert into key_desc values (1, 1);
+ERROR:  no partition of relation "key_desc_1" found for row
+reset role;
+grant select (b) on key_desc_1 to someone_else;
+set role someone_else;
+-- key description (b)=(1) is now shown
+insert into key_desc values (1, 1);
+ERROR:  no partition of relation "key_desc_1" found for row
+DETAIL:  Partition key of the failing row contains (b) = (1).
+-- key description is not shown if key contains expression
+insert into key_desc values (2, 1);
+ERROR:  no partition of relation "key_desc" found for row
+reset role;
+revoke all on key_desc from someone_else;
+revoke all on key_desc_1 from someone_else;
+drop role someone_else;
+drop table key_desc, key_desc_1;
 -- cleanup
 drop table p, p1, p11, p12, p2, p3, p4;

src/test/regress/sql/insert.sql

@@ -215,6 +215,9 @@ alter table p attach partition p1 for values from (1, 2) to (1, 10);
 insert into p values (1, 2);
 select tableoid::regclass, * from p;
 
+-- check that proper message is shown after failure to route through p1
+insert into p (a, b) values (1, 5);
+
 truncate p;
 alter table p add constraint check_b check (b = 3);
 -- check that correct input row is shown when constraint check_b fails on p11
@@ -240,5 +243,32 @@ with ins (a, b, c) as
   (insert into p (b, a) select s.a, 1 from generate_series(2, 39) s(a) returning tableoid::regclass, *)
   select a, b, min(c), max(c) from ins group by a, b order by 1;
 
+-- check that message shown after failure to find a partition shows the
+-- appropriate key description (or none) in various situations
+create table key_desc (a int, b int) partition by list ((a+0));
+create table key_desc_1 partition of key_desc for values in (1) partition by range (b);
+
+create user someone_else;
+grant select (a) on key_desc_1 to someone_else;
+grant insert on key_desc to someone_else;
+
+set role someone_else;
+-- no key description is shown
+insert into key_desc values (1, 1);
+
+reset role;
+grant select (b) on key_desc_1 to someone_else;
+set role someone_else;
+-- key description (b)=(1) is now shown
+insert into key_desc values (1, 1);
+
+-- key description is not shown if key contains expression
+insert into key_desc values (2, 1);
+reset role;
+revoke all on key_desc from someone_else;
+revoke all on key_desc_1 from someone_else;
+drop role someone_else;
+drop table key_desc, key_desc_1;
+
 -- cleanup
 drop table p, p1, p11, p12, p2, p3, p4;