Thanks for the review.
On 2017/02/27 2:39, Robert Haas wrote:
> On Tue, Feb 21, 2017 at 7:28 AM, Amit Langote
> <langote_amit...@lab.ntt.co.jp> wrote:
>> Simon pointed out in a nearby thread [0] that the detail part of
>> partition-not-found error should show just the partition keys. I posted a
>> patch on that thread [1], but to avoid confusion being caused by multitude
>> of patches over there I'm re-posting it here.
>
> Thanks. GetPartitionFailureData seems like a strange name for a
> datatype, particularly the "Get" part. How about
> PartitionRoutingFailureInfo? Or just two out parameters.
Went with two out parameters instead of a new struct.
> Spelling: BuildSlotPartitinKeyDescription (in comment).
Fixed.
> ExecBuildSlotPartitionKeyDescription could have a comment saying that
> it's LIKE BuildIndexValueDescription() instead of copy-and-pasting the
> comments. And maybe BuildIndexValueDescription() could also get a
> comment saying that if we change anything there, we should check
> whether ExecBuildSlotPartitionKeyDescription() needs a similar change.
OK, I modified the comments. Although, I kept comments that are a bit
different.
Updated patch is attached.
Thanks,
Amit
>From e8da812092c8cdd9d847467fbad668afbb0f3df0 Mon Sep 17 00:00:00 2001
From: amit <amitlangot...@gmail.com>
Date: Mon, 13 Feb 2017 19:52:19 +0900
Subject: [PATCH] Show only the partition key upon failing to find a partition
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 in format (key1, ...)=(val1, ...).
The key description shown in the error message is now built by the
new ExecBuildSlotPartitionKeyDescription(), which works along the
lines of BuildIndexValueDescription(), using similar rules about
what columns of the partition key to include depending on the user's
privileges to view the same.
A bunch of relevant tests are added.
---
src/backend/access/index/genam.c | 4 ++
src/backend/catalog/partition.c | 30 ++++----
src/backend/executor/execMain.c | 132 ++++++++++++++++++++++++++++++-----
src/backend/utils/adt/ruleutils.c | 37 +++++++---
src/include/catalog/partition.h | 8 ++-
src/include/utils/ruleutils.h | 2 +
src/test/regress/expected/insert.out | 38 ++++++++--
src/test/regress/sql/insert.sql | 30 ++++++++
8 files changed, 231 insertions(+), 50 deletions(-)
diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
index c4a393f34e..cb50b751c7 100644
--- a/src/backend/access/index/genam.c
+++ b/src/backend/access/index/genam.c
@@ -166,6 +166,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 we change anything there, we should check whether
+ * ExecBuildSlotPartitionKeyDescription() in execMain.c needs a similar
+ * change.
*/
char *
BuildIndexValueDescription(Relation indexRelation,
diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c
index 4bcef58763..e01ef864f0 100644
--- a/src/backend/catalog/partition.c
+++ b/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)
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
index 3f76a407d7..f5cd65d8a0 100644
--- a/src/backend/executor/execMain.c
+++ b/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;
+}
diff --git a/src/backend/utils/adt/ruleutils.c b/src/backend/utils/adt/ruleutils.c
index b27b77de21..cf79f4126e 100644
--- a/src/backend/utils/adt/ruleutils.c
+++ b/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);
diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h
index b195d1a5ab..421644ca77 100644
--- a/src/include/catalog/partition.h
+++ b/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 */
diff --git a/src/include/utils/ruleutils.h b/src/include/utils/ruleutils.h
index 3e8aad97e2..42fc872c4a 100644
--- a/src/include/utils/ruleutils.h
+++ b/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);
diff --git a/src/test/regress/expected/insert.out b/src/test/regress/expected/insert.out
index 81af3ef497..397238332b 100644
--- a/src/test/regress/expected/insert.out
+++ b/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;
diff --git a/src/test/regress/sql/insert.sql b/src/test/regress/sql/insert.sql
index 454e1ce2e7..4f19df5c25 100644
--- a/src/test/regress/sql/insert.sql
+++ b/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;
--
2.11.0
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
