Re: WITH CHECK and Column-Level Privileges

All,

Apologies, forgot to mention- this is again 9.4.

Thanks,
Stephen

* Stephen Frost (sfrost@snowman.net) wrote:
> Dean, Robert,
>
> * Dean Rasheed (dean.a.rasheed@gmail.com) wrote:
> > On 29 October 2014 13:04, Stephen Frost <sfrost@snowman.net> wrote:
> > > * Robert Haas (robertmhaas@gmail.com) wrote:
> > >> On Wed, Oct 29, 2014 at 8:16 AM, Stephen Frost <sfrost@snowman.net> wrote:
> > >> > suggestions.  If the user does not have table-level SELECT rights,
> > >> > they'll see for the "Failing row contains" case, they'll get:
> > >> >
> > >> > Failing row contains (col1, col2, col3) = (1, 2, 3).
> > >> >
> > >> > Or, if they have no access to any columns:
> > >> >
> > >> > Failing row contains () = ()
> > >>
> > >> I haven't looked at the code, but that sounds nice, except that if
> > >> they have no access to any columns, I'd leave the message out
> > >> altogether instead of emitting it with no useful content.
> > >
> > > Alright, I can change things around to make that happen without too much
> > > trouble.
> >
> > Yes, skim-reading the patch, it looks like a good approach to me, and
> > also +1 for not emitting anything if no values are visible.
>
> Alright, here's an updated patch which doesn't return any detail if no
> values are visible or if only a partial key is visible.
>
> Please take a look.  I'm not thrilled with simply returning an empty
> string and then checking that for BuildIndexValueDescription and
> ExecBuildSlotValueDescription, but I figured we might have other users
> of BuildIndexValueDescription and I wasn't seeing a particularly better
> solution.  Suggestions welcome, of course.
>
>     Thanks!
>
>         Stephen

> From e00cc2f5be4524989e8d670296f82bb99f3774a1 Mon Sep 17 00:00:00 2001
> From: Stephen Frost <sfrost@snowman.net>
> Date: Mon, 12 Jan 2015 17:04:11 -0500
> Subject: [PATCH] Fix column-privilege leak in error-message paths
>
> While building error messages to return to the user,
> BuildIndexValueDescription and ExecBuildSlotValueDescription would
> happily include the entire key or entire row in the result returned to
> the user, even if the user didn't have access to view all of the columns
> being included.
>
> Instead, include only those columns which the user is providing or which
> the user has select rights on.  If the user does not have any rights
> to view the table or any of the columns involved then no detail is
> provided.  Note that, for duplicate key cases, the user must have access
> to all of the columns for the key to be shown; a partial key will not be
> returned.
>
> Back-patch all the way, as column-level privileges are now in all
> supported versions.
> ---
>  src/backend/access/index/genam.c         |  59 ++++++++-
>  src/backend/access/nbtree/nbtinsert.c    |  30 +++--
>  src/backend/commands/copy.c              |   6 +-
>  src/backend/executor/execMain.c          | 215 +++++++++++++++++++++++--------
>  src/backend/executor/execUtils.c         |  12 +-
>  src/backend/utils/sort/tuplesort.c       |  28 ++--
>  src/test/regress/expected/privileges.out |  31 +++++
>  src/test/regress/sql/privileges.sql      |  24 ++++
>  8 files changed, 328 insertions(+), 77 deletions(-)
>
> diff --git a/src/backend/access/index/genam.c b/src/backend/access/index/genam.c
> index 850008b..1090568 100644
> --- a/src/backend/access/index/genam.c
> +++ b/src/backend/access/index/genam.c
> @@ -25,10 +25,12 @@
>  #include "lib/stringinfo.h"
>  #include "miscadmin.h"
>  #include "storage/bufmgr.h"
> +#include "utils/acl.h"
>  #include "utils/builtins.h"
>  #include "utils/lsyscache.h"
>  #include "utils/rel.h"
>  #include "utils/snapmgr.h"
> +#include "utils/syscache.h"
>  #include "utils/tqual.h"
>
>
> @@ -154,6 +156,9 @@ IndexScanEnd(IndexScanDesc scan)
>   * form "(key_name, ...)=(key_value, ...)".  This is currently used
>   * for building unique-constraint and exclusion-constraint error messages.
>   *
> + * Note that if the user does not have permissions to view the columns
> + * involved, an empty string "" will be returned instead.
> + *
>   * 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.
> @@ -163,13 +168,63 @@ BuildIndexValueDescription(Relation indexRelation,
>                             Datum *values, bool *isnull)
>  {
>      StringInfoData buf;
> +    Form_pg_index idxrec;
> +    HeapTuple    ht_idx;
>      int            natts = indexRelation->rd_rel->relnatts;
>      int            i;
> +    int            keyno;
> +    Oid            indexrelid = RelationGetRelid(indexRelation);
> +    Oid            indrelid;
> +    AclResult    aclresult;
> +
> +    /*
> +     * Check permissions- if the users does not have access to view the
> +     * data in the key columns, we return "" instead, which callers can
> +     * test for and use or not accordingly.
> +     *
> +     * First we need to check table-level SELECT access and then, if
> +     * there is no access there, check column-level permissions.
> +     */
> +
> +    /*
> +     * Fetch the pg_index tuple by the Oid of the index
> +     */
> +    ht_idx = SearchSysCache1(INDEXRELID, ObjectIdGetDatum(indexrelid));
> +    if (!HeapTupleIsValid(ht_idx))
> +        elog(ERROR, "cache lookup failed for index %u", indexrelid);
> +    idxrec = (Form_pg_index) GETSTRUCT(ht_idx);
> +
> +    indrelid = idxrec->indrelid;
> +    Assert(indexrelid == idxrec->indexrelid);
> +
> +    /* Table-level SELECT is enough, if the user has it */
> +    aclresult = pg_class_aclcheck(indrelid, GetUserId(), ACL_SELECT);
> +    if (aclresult != ACLCHECK_OK)
> +    {
> +        /*
> +         * No table-level access, so step through the columns in the
> +         * index and make sure the user has SELECT rights on all of them.
> +         */
> +        for (keyno = 0; keyno < idxrec->indnatts; keyno++)
> +        {
> +            AttrNumber    attnum = idxrec->indkey.values[keyno];
> +
> +            aclresult = pg_attribute_aclcheck(indrelid, attnum, GetUserId(),
> +                                              ACL_SELECT);
> +
> +            if (aclresult != ACLCHECK_OK)
> +            {
> +                /* No access, so clean up and just return "" */
> +                ReleaseSysCache(ht_idx);
> +                return "";
> +            }
> +        }
> +    }
> +    ReleaseSysCache(ht_idx);
>
>      initStringInfo(&buf);
>      appendStringInfo(&buf, "(%s)=(",
> -                     pg_get_indexdef_columns(RelationGetRelid(indexRelation),
> -                                             true));
> +                     pg_get_indexdef_columns(indexrelid, true));
>
>      for (i = 0; i < natts; i++)
>      {
> diff --git a/src/backend/access/nbtree/nbtinsert.c b/src/backend/access/nbtree/nbtinsert.c
> index 59d7006..4279416 100644
> --- a/src/backend/access/nbtree/nbtinsert.c
> +++ b/src/backend/access/nbtree/nbtinsert.c
> @@ -388,18 +388,30 @@ _bt_check_unique(Relation rel, IndexTuple itup, Relation heapRel,
>                      {
>                          Datum        values[INDEX_MAX_KEYS];
>                          bool        isnull[INDEX_MAX_KEYS];
> +                        char       *key_desc;
>
>                          index_deform_tuple(itup, RelationGetDescr(rel),
>                                             values, isnull);
> -                        ereport(ERROR,
> -                                (errcode(ERRCODE_UNIQUE_VIOLATION),
> -                                 errmsg("duplicate key value violates unique constraint \"%s\"",
> -                                        RelationGetRelationName(rel)),
> -                                 errdetail("Key %s already exists.",
> -                                           BuildIndexValueDescription(rel,
> -                                                            values, isnull)),
> -                                 errtableconstraint(heapRel,
> -                                             RelationGetRelationName(rel))));
> +
> +                        key_desc = BuildIndexValueDescription(rel, values,
> +                                                              isnull);
> +
> +                        if (!strlen(key_desc))
> +                            ereport(ERROR,
> +                                    (errcode(ERRCODE_UNIQUE_VIOLATION),
> +                                     errmsg("duplicate key value violates unique constraint \"%s\"",
> +                                            RelationGetRelationName(rel)),
> +                                     errtableconstraint(heapRel,
> +                                                RelationGetRelationName(rel))));
> +                        else
> +                            ereport(ERROR,
> +                                    (errcode(ERRCODE_UNIQUE_VIOLATION),
> +                                     errmsg("duplicate key value violates unique constraint \"%s\"",
> +                                            RelationGetRelationName(rel)),
> +                                     errdetail("Key %s already exists.",
> +                                                key_desc),
> +                                     errtableconstraint(heapRel,
> +                                                RelationGetRelationName(rel))));
>                      }
>                  }
>                  else if (all_dead)
> diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c
> index fbd7492..9ab1e19 100644
> --- a/src/backend/commands/copy.c
> +++ b/src/backend/commands/copy.c
> @@ -160,6 +160,7 @@ typedef struct CopyStateData
>      int           *defmap;            /* array of default att numbers */
>      ExprState **defexprs;        /* array of default att expressions */
>      bool        volatile_defexprs;        /* is any of defexprs volatile? */
> +    List       *range_table;
>
>      /*
>       * These variables are used to reduce overhead in textual COPY FROM.
> @@ -784,6 +785,7 @@ DoCopy(const CopyStmt *stmt, const char *queryString, uint64 *processed)
>      bool        pipe = (stmt->filename == NULL);
>      Relation    rel;
>      Oid            relid;
> +    RangeTblEntry *rte;
>
>      /* Disallow COPY to/from file or program except to superusers. */
>      if (!pipe && !superuser())
> @@ -806,7 +808,6 @@ DoCopy(const CopyStmt *stmt, const char *queryString, uint64 *processed)
>      {
>          TupleDesc    tupDesc;
>          AclMode        required_access = (is_from ? ACL_INSERT : ACL_SELECT);
> -        RangeTblEntry *rte;
>          List       *attnums;
>          ListCell   *cur;
>
> @@ -856,6 +857,7 @@ DoCopy(const CopyStmt *stmt, const char *queryString, uint64 *processed)
>
>          cstate = BeginCopyFrom(rel, stmt->filename, stmt->is_program,
>                                 stmt->attlist, stmt->options);
> +        cstate->range_table = list_make1(rte);
>          *processed = CopyFrom(cstate);    /* copy from file to database */
>          EndCopyFrom(cstate);
>      }
> @@ -864,6 +866,7 @@ DoCopy(const CopyStmt *stmt, const char *queryString, uint64 *processed)
>          cstate = BeginCopyTo(rel, stmt->query, queryString,
>                               stmt->filename, stmt->is_program,
>                               stmt->attlist, stmt->options);
> +        cstate->range_table = list_make1(rte);
>          *processed = DoCopyTo(cstate);    /* copy from database to file */
>          EndCopyTo(cstate);
>      }
> @@ -2184,6 +2187,7 @@ CopyFrom(CopyState cstate)
>      estate->es_result_relations = resultRelInfo;
>      estate->es_num_result_relations = 1;
>      estate->es_result_relation_info = resultRelInfo;
> +    estate->es_range_table = cstate->range_table;
>
>      /* Set up a tuple slot too */
>      myslot = ExecInitExtraTupleSlot(estate);
> diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c
> index 01eda70..49e4c81 100644
> --- a/src/backend/executor/execMain.c
> +++ b/src/backend/executor/execMain.c
> @@ -82,12 +82,17 @@ static void ExecutePlan(EState *estate, PlanState *planstate,
>              DestReceiver *dest);
>  static bool ExecCheckRTEPerms(RangeTblEntry *rte);
>  static void ExecCheckXactReadOnly(PlannedStmt *plannedstmt);
> -static char *ExecBuildSlotValueDescription(TupleTableSlot *slot,
> +static char *ExecBuildSlotValueDescription(Oid reloid,
> +                              TupleTableSlot *slot,
>                                TupleDesc tupdesc,
> +                              Bitmapset *modifiedCols,
>                                int maxfieldlen);
>  static void EvalPlanQualStart(EPQState *epqstate, EState *parentestate,
>                    Plan *planTree);
>
> +#define GetModifiedColumns(relinfo, estate) \
> +    (rt_fetch((relinfo)->ri_RangeTableIndex, (estate)->es_range_table)->modifiedCols)
> +
>  /* end of local decls */
>
>
> @@ -1590,9 +1595,12 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
>      Relation    rel = resultRelInfo->ri_RelationDesc;
>      TupleDesc    tupdesc = RelationGetDescr(rel);
>      TupleConstr *constr = tupdesc->constr;
> +    Bitmapset  *modifiedCols;
>
>      Assert(constr);
>
> +    modifiedCols = GetModifiedColumns(resultRelInfo, estate);
> +
>      if (constr->has_not_null)
>      {
>          int            natts = tupdesc->natts;
> @@ -1602,15 +1610,29 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
>          {
>              if (tupdesc->attrs[attrChk - 1]->attnotnull &&
>                  slot_attisnull(slot, attrChk))
> -                ereport(ERROR,
> -                        (errcode(ERRCODE_NOT_NULL_VIOLATION),
> -                         errmsg("null value in column \"%s\" violates not-null constraint",
> -                              NameStr(tupdesc->attrs[attrChk - 1]->attname)),
> -                         errdetail("Failing row contains %s.",
> -                                   ExecBuildSlotValueDescription(slot,
> -                                                                 tupdesc,
> -                                                                 64)),
> -                         errtablecol(rel, attrChk)));
> +            {
> +                char       *val_desc;
> +
> +                val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
> +                                                         slot,
> +                                                         tupdesc,
> +                                                         modifiedCols,
> +                                                         64);
> +
> +                if (!strlen(val_desc))
> +                    ereport(ERROR,
> +                            (errcode(ERRCODE_NOT_NULL_VIOLATION),
> +                             errmsg("null value in column \"%s\" violates not-null constraint",
> +                                  NameStr(tupdesc->attrs[attrChk - 1]->attname)),
> +                             errtablecol(rel, attrChk)));
> +                else
> +                    ereport(ERROR,
> +                            (errcode(ERRCODE_NOT_NULL_VIOLATION),
> +                             errmsg("null value in column \"%s\" violates not-null constraint",
> +                                  NameStr(tupdesc->attrs[attrChk - 1]->attname)),
> +                             errdetail("Failing row contains %s.", val_desc),
> +                             errtablecol(rel, attrChk)));
> +            }
>          }
>      }
>
> @@ -1619,15 +1641,28 @@ ExecConstraints(ResultRelInfo *resultRelInfo,
>          const char *failed;
>
>          if ((failed = ExecRelCheck(resultRelInfo, slot, estate)) != NULL)
> -            ereport(ERROR,
> -                    (errcode(ERRCODE_CHECK_VIOLATION),
> -                     errmsg("new row for relation \"%s\" violates check constraint \"%s\"",
> -                            RelationGetRelationName(rel), failed),
> -                     errdetail("Failing row contains %s.",
> -                               ExecBuildSlotValueDescription(slot,
> -                                                             tupdesc,
> -                                                             64)),
> -                     errtableconstraint(rel, failed)));
> +        {
> +            char       *val_desc;
> +
> +            val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
> +                                                     slot,
> +                                                     tupdesc,
> +                                                     modifiedCols,
> +                                                     64);
> +            if (!strlen(val_desc))
> +                ereport(ERROR,
> +                        (errcode(ERRCODE_CHECK_VIOLATION),
> +                         errmsg("new row for relation \"%s\" violates check constraint \"%s\"",
> +                                RelationGetRelationName(rel), failed),
> +                         errtableconstraint(rel, failed)));
> +            else
> +                ereport(ERROR,
> +                        (errcode(ERRCODE_CHECK_VIOLATION),
> +                         errmsg("new row for relation \"%s\" violates check constraint \"%s\"",
> +                                RelationGetRelationName(rel), failed),
> +                         errdetail("Failing row contains %s.", val_desc),
> +                         errtableconstraint(rel, failed)));
> +        }
>      }
>  }
>
> @@ -1638,9 +1673,13 @@ void
>  ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
>                       TupleTableSlot *slot, EState *estate)
>  {
> +    Relation    rel = resultRelInfo->ri_RelationDesc;
>      ExprContext *econtext;
>      ListCell   *l1,
>                 *l2;
> +    Bitmapset  *modifiedCols;
> +
> +    modifiedCols = GetModifiedColumns(resultRelInfo, estate);
>
>      /*
>       * We will use the EState's per-tuple context for evaluating constraint
> @@ -1666,14 +1705,27 @@ ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
>           * above for CHECK constraints).
>           */
>          if (!ExecQual((List *) wcoExpr, econtext, false))
> -            ereport(ERROR,
> -                    (errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION),
> -                 errmsg("new row violates WITH CHECK OPTION for view \"%s\"",
> -                        wco->viewname),
> -                     errdetail("Failing row contains %s.",
> -                               ExecBuildSlotValueDescription(slot,
> -                            RelationGetDescr(resultRelInfo->ri_RelationDesc),
> -                                                             64))));
> +        {
> +            char       *val_desc;
> +
> +            val_desc = ExecBuildSlotValueDescription(RelationGetRelid(rel),
> +                                                     slot,
> +                             RelationGetDescr(resultRelInfo->ri_RelationDesc),
> +                                                      modifiedCols,
> +                                                     64);
> +
> +            if (!strlen(val_desc))
> +                ereport(ERROR,
> +                        (errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION),
> +                     errmsg("new row violates WITH CHECK OPTION for view \"%s\"",
> +                            wco->viewname)));
> +            else
> +                ereport(ERROR,
> +                        (errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION),
> +                     errmsg("new row violates WITH CHECK OPTION for view \"%s\"",
> +                            wco->viewname),
> +                         errdetail("Failing row contains %s.", val_desc)));
> +        }
>      }
>  }
>
> @@ -1689,25 +1741,51 @@ ExecWithCheckOptions(ResultRelInfo *resultRelInfo,
>   * dropped columns.  We used to use the slot's tuple descriptor to decode the
>   * data, but the slot's descriptor doesn't identify dropped columns, so we
>   * now need to be passed the relation's descriptor.
> + *
> + * Note that, like BuildIndexValueDescription, if the user does not have
> + * permission to view any of the columns involved, an empty string is returned.
>   */
>  static char *
> -ExecBuildSlotValueDescription(TupleTableSlot *slot,
> +ExecBuildSlotValueDescription(Oid reloid,
> +                              TupleTableSlot *slot,
>                                TupleDesc tupdesc,
> +                              Bitmapset *modifiedCols,
>                                int maxfieldlen)
>  {
>      StringInfoData buf;
> +    StringInfoData collist;
>      bool        write_comma = false;
> +    bool        write_comma_collist = false;
>      int            i;
> -
> -    /* Make sure the tuple is fully deconstructed */
> -    slot_getallattrs(slot);
> +    AclResult    aclresult;
> +    bool        table_perm = false;
> +    bool        any_perm = false;
>
>      initStringInfo(&buf);
>
>      appendStringInfoChar(&buf, '(');
>
> +    /*
> +     * Check that the user has permissions to see the row.  If the user
> +     * has table-level SELECT then that is good enough.  If not, we have
> +     * to check all the columns.
> +     */
> +    aclresult = pg_class_aclcheck(reloid, GetUserId(), ACL_SELECT);
> +    if (aclresult != ACLCHECK_OK)
> +    {
> +        /* Set up the buffer for the column list */
> +        initStringInfo(&collist);
> +        appendStringInfoChar(&collist, '(');
> +    }
> +    else
> +        table_perm = any_perm = true;
> +
> +    /* Make sure the tuple is fully deconstructed */
> +    slot_getallattrs(slot);
> +
>      for (i = 0; i < tupdesc->natts; i++)
>      {
> +        bool        column_perm = false;
>          char       *val;
>          int            vallen;
>
> @@ -1715,37 +1793,70 @@ ExecBuildSlotValueDescription(TupleTableSlot *slot,
>          if (tupdesc->attrs[i]->attisdropped)
>              continue;
>
> -        if (slot->tts_isnull[i])
> -            val = "null";
> -        else
> +        if (!table_perm)
>          {
> -            Oid            foutoid;
> -            bool        typisvarlena;
> +            aclresult = pg_attribute_aclcheck(reloid, tupdesc->attrs[i]->attnum,
> +                                              GetUserId(), ACL_SELECT);
> +            if (bms_is_member(tupdesc->attrs[i]->attnum - FirstLowInvalidHeapAttributeNumber,
> +                              modifiedCols) || aclresult == ACLCHECK_OK)
> +            {
> +                column_perm = any_perm = true;
>
> -            getTypeOutputInfo(tupdesc->attrs[i]->atttypid,
> -                              &foutoid, &typisvarlena);
> -            val = OidOutputFunctionCall(foutoid, slot->tts_values[i]);
> -        }
> +                if (write_comma_collist)
> +                    appendStringInfoString(&collist, ", ");
> +                else
> +                    write_comma_collist = true;
>
> -        if (write_comma)
> -            appendStringInfoString(&buf, ", ");
> -        else
> -            write_comma = true;
> +                appendStringInfoString(&collist, NameStr(tupdesc->attrs[i]->attname));
> +            }
> +        }
>
> -        /* truncate if needed */
> -        vallen = strlen(val);
> -        if (vallen <= maxfieldlen)
> -            appendStringInfoString(&buf, val);
> -        else
> +        if (table_perm || column_perm)
>          {
> -            vallen = pg_mbcliplen(val, vallen, maxfieldlen);
> -            appendBinaryStringInfo(&buf, val, vallen);
> -            appendStringInfoString(&buf, "...");
> +            if (slot->tts_isnull[i])
> +                val = "null";
> +            else
> +            {
> +                Oid            foutoid;
> +                bool        typisvarlena;
> +
> +                getTypeOutputInfo(tupdesc->attrs[i]->atttypid,
> +                                  &foutoid, &typisvarlena);
> +                val = OidOutputFunctionCall(foutoid, slot->tts_values[i]);
> +            }
> +
> +            if (write_comma)
> +                appendStringInfoString(&buf, ", ");
> +            else
> +                write_comma = true;
> +
> +            /* 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, "...");
> +            }
>          }
>      }
>
> +    /* If there were no permissions found, return an empty string. */
> +    if (!any_perm)
> +        return "";
> +
>      appendStringInfoChar(&buf, ')');
>
> +    if (!table_perm)
> +    {
> +        appendStringInfoString(&collist, ") = ");
> +        appendStringInfoString(&collist, buf.data);
> +
> +        return collist.data;
> +    }
> +
>      return buf.data;
>  }
>
> diff --git a/src/backend/executor/execUtils.c b/src/backend/executor/execUtils.c
> index d5e1273..4860356 100644
> --- a/src/backend/executor/execUtils.c
> +++ b/src/backend/executor/execUtils.c
> @@ -1323,8 +1323,10 @@ retry:
>                      (errcode(ERRCODE_EXCLUSION_VIOLATION),
>                       errmsg("could not create exclusion constraint \"%s\"",
>                              RelationGetRelationName(index)),
> -                     errdetail("Key %s conflicts with key %s.",
> -                               error_new, error_existing),
> +                     strlen(error_new) == 0 || strlen(error_existing) == 0 ?
> +                         errdetail("Key conflicts exist.") :
> +                         errdetail("Key %s conflicts with key %s.",
> +                                  error_new, error_existing),
>                       errtableconstraint(heap,
>                                          RelationGetRelationName(index))));
>          else
> @@ -1332,8 +1334,10 @@ retry:
>                      (errcode(ERRCODE_EXCLUSION_VIOLATION),
>                       errmsg("conflicting key value violates exclusion constraint \"%s\"",
>                              RelationGetRelationName(index)),
> -                     errdetail("Key %s conflicts with existing key %s.",
> -                               error_new, error_existing),
> +                     strlen(error_new) == 0 || strlen(error_existing) == 0 ?
> +                         errdetail("Key conflicts with existing key.") :
> +                         errdetail("Key %s conflicts with existing key %s.",
> +                                  error_new, error_existing),
>                       errtableconstraint(heap,
>                                          RelationGetRelationName(index))));
>      }
> diff --git a/src/backend/utils/sort/tuplesort.c b/src/backend/utils/sort/tuplesort.c
> index aa0f6d8..6aba499 100644
> --- a/src/backend/utils/sort/tuplesort.c
> +++ b/src/backend/utils/sort/tuplesort.c
> @@ -3240,6 +3240,7 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
>      {
>          Datum        values[INDEX_MAX_KEYS];
>          bool        isnull[INDEX_MAX_KEYS];
> +        char       *key_desc;
>
>          /*
>           * Some rather brain-dead implementations of qsort (such as the one in
> @@ -3250,15 +3251,24 @@ comparetup_index_btree(const SortTuple *a, const SortTuple *b,
>          Assert(tuple1 != tuple2);
>
>          index_deform_tuple(tuple1, tupDes, values, isnull);
> -        ereport(ERROR,
> -                (errcode(ERRCODE_UNIQUE_VIOLATION),
> -                 errmsg("could not create unique index \"%s\"",
> -                        RelationGetRelationName(state->indexRel)),
> -                 errdetail("Key %s is duplicated.",
> -                           BuildIndexValueDescription(state->indexRel,
> -                                                      values, isnull)),
> -                 errtableconstraint(state->heapRel,
> -                                 RelationGetRelationName(state->indexRel))));
> +
> +        key_desc = BuildIndexValueDescription(state->indexRel, values, isnull);
> +
> +        if (!strlen(key_desc))
> +            ereport(ERROR,
> +                    (errcode(ERRCODE_UNIQUE_VIOLATION),
> +                     errmsg("could not create unique index \"%s\"",
> +                            RelationGetRelationName(state->indexRel)),
> +                     errtableconstraint(state->heapRel,
> +                                     RelationGetRelationName(state->indexRel))));
> +        else
> +            ereport(ERROR,
> +                    (errcode(ERRCODE_UNIQUE_VIOLATION),
> +                     errmsg("could not create unique index \"%s\"",
> +                            RelationGetRelationName(state->indexRel)),
> +                     errdetail("Key %s is duplicated.", key_desc),
> +                     errtableconstraint(state->heapRel,
> +                                     RelationGetRelationName(state->indexRel))));
>      }
>
>      /*
> diff --git a/src/test/regress/expected/privileges.out b/src/test/regress/expected/privileges.out
> index 1675075..b1ecd39 100644
> --- a/src/test/regress/expected/privileges.out
> +++ b/src/test/regress/expected/privileges.out
> @@ -381,6 +381,37 @@ SELECT atest6 FROM atest6; -- ok
>  (0 rows)
>
>  COPY atest6 TO stdout; -- ok
> +-- check error reporting with column privs
> +SET SESSION AUTHORIZATION regressuser1;
> +CREATE TABLE t1 (c1 int, c2 int, c3 int check (c3 < 5), primary key (c1, c2));
> +GRANT SELECT (c1) ON t1 TO regressuser2;
> +GRANT INSERT (c1, c2, c3) ON t1 TO regressuser2;
> +GRANT UPDATE (c1, c2, c3) ON t1 TO regressuser2;
> +-- seed data
> +INSERT INTO t1 VALUES (1, 1, 1);
> +INSERT INTO t1 VALUES (1, 2, 1);
> +INSERT INTO t1 VALUES (2, 1, 2);
> +INSERT INTO t1 VALUES (2, 2, 2);
> +INSERT INTO t1 VALUES (3, 1, 3);
> +SET SESSION AUTHORIZATION regressuser2;
> +INSERT INTO t1 (c1, c2) VALUES (1, 1); -- fail, but row not shown
> +ERROR:  duplicate key value violates unique constraint "t1_pkey"
> +UPDATE t1 SET c2 = 1; -- fail, but row not shown
> +ERROR:  duplicate key value violates unique constraint "t1_pkey"
> +INSERT INTO t1 (c1, c2) VALUES (null, null); -- fail, but see columns being inserted
> +ERROR:  null value in column "c1" violates not-null constraint
> +DETAIL:  Failing row contains (c1, c2) = (null, null).
> +INSERT INTO t1 (c3) VALUES (null); -- fail, but see columns being inserted or have SELECT
> +ERROR:  null value in column "c1" violates not-null constraint
> +DETAIL:  Failing row contains (c1, c3) = (null, null).
> +INSERT INTO t1 (c1) VALUES (5); -- fail, but see columns being inserted or have SELECT
> +ERROR:  null value in column "c2" violates not-null constraint
> +DETAIL:  Failing row contains (c1) = (5).
> +UPDATE t1 SET c3 = 10; -- fail, but see columns with SELECT rights, or being modified
> +ERROR:  new row for relation "t1" violates check constraint "t1_c3_check"
> +DETAIL:  Failing row contains (c1, c3) = (1, 10).
> +DROP TABLE t1;
> +ERROR:  must be owner of relation t1
>  -- test column-level privileges when involved with DELETE
>  SET SESSION AUTHORIZATION regressuser1;
>  ALTER TABLE atest6 ADD COLUMN three integer;
> diff --git a/src/test/regress/sql/privileges.sql b/src/test/regress/sql/privileges.sql
> index a0ff953..c850a2e 100644
> --- a/src/test/regress/sql/privileges.sql
> +++ b/src/test/regress/sql/privileges.sql
> @@ -256,6 +256,30 @@ UPDATE atest5 SET one = 1; -- fail
>  SELECT atest6 FROM atest6; -- ok
>  COPY atest6 TO stdout; -- ok
>
> +-- check error reporting with column privs
> +SET SESSION AUTHORIZATION regressuser1;
> +CREATE TABLE t1 (c1 int, c2 int, c3 int check (c3 < 5), primary key (c1, c2));
> +GRANT SELECT (c1) ON t1 TO regressuser2;
> +GRANT INSERT (c1, c2, c3) ON t1 TO regressuser2;
> +GRANT UPDATE (c1, c2, c3) ON t1 TO regressuser2;
> +
> +-- seed data
> +INSERT INTO t1 VALUES (1, 1, 1);
> +INSERT INTO t1 VALUES (1, 2, 1);
> +INSERT INTO t1 VALUES (2, 1, 2);
> +INSERT INTO t1 VALUES (2, 2, 2);
> +INSERT INTO t1 VALUES (3, 1, 3);
> +
> +SET SESSION AUTHORIZATION regressuser2;
> +INSERT INTO t1 (c1, c2) VALUES (1, 1); -- fail, but row not shown
> +UPDATE t1 SET c2 = 1; -- fail, but row not shown
> +INSERT INTO t1 (c1, c2) VALUES (null, null); -- fail, but see columns being inserted
> +INSERT INTO t1 (c3) VALUES (null); -- fail, but see columns being inserted or have SELECT
> +INSERT INTO t1 (c1) VALUES (5); -- fail, but see columns being inserted or have SELECT
> +UPDATE t1 SET c3 = 10; -- fail, but see columns with SELECT rights, or being modified
> +
> +DROP TABLE t1;
> +
>  -- test column-level privileges when involved with DELETE
>  SET SESSION AUTHORIZATION regressuser1;
>  ALTER TABLE atest6 ADD COLUMN three integer;
> --
> 1.9.1
>