diff options
author | Robert Haas <rhaas@postgresql.org> | 2017-03-03 09:07:41 +0530 |
---|---|---|
committer | Robert Haas <rhaas@postgresql.org> | 2017-03-03 09:09:52 +0530 |
commit | 5a73e17317e91912b2755f7960d5bf31d374cf31 (patch) | |
tree | 9df368817ebdf24e6eca7a2a47dbccdbc67808b0 /src/backend/executor/execMain.c | |
parent | be6ed6451c693d9121d357996cbc21b06058b9c1 (diff) | |
download | postgresql-5a73e17317e91912b2755f7960d5bf31d374cf31.tar.gz postgresql-5a73e17317e91912b2755f7960d5bf31d374cf31.zip |
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
Diffstat (limited to 'src/backend/executor/execMain.c')
-rw-r--r-- | src/backend/executor/execMain.c | 132 |
1 files changed, 113 insertions, 19 deletions
diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 3f76a407d7a..f5cd65d8a0d 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; +} |