diff --git a/doc/src/sgml/ref/create_view.sgml b/doc/src/sgml/ref/create_view.sgml new file mode 100644 index e0fbe1e..888410f *** a/doc/src/sgml/ref/create_view.sgml --- b/doc/src/sgml/ref/create_view.sgml *************** CREATE VIEW vista AS SELECT text 'Hello *** 323,334 **** or set-returning functions. - - - - The view must not have the security_barrier property. - - --- 323,328 ---- diff --git a/src/backend/commands/tablecmds.c b/src/backend/commands/tablecmds.c new file mode 100644 index 466d757..5b7e31a *** a/src/backend/commands/tablecmds.c --- b/src/backend/commands/tablecmds.c *************** ATExecSetRelOptions(Relation rel, List * *** 8806,8812 **** List *view_options = untransformRelOptions(newOptions); ListCell *cell; bool check_option = false; - bool security_barrier = false; foreach(cell, view_options) { --- 8806,8811 ---- *************** ATExecSetRelOptions(Relation rel, List * *** 8814,8821 **** if (pg_strcasecmp(defel->defname, "check_option") == 0) check_option = true; - if (pg_strcasecmp(defel->defname, "security_barrier") == 0) - security_barrier = defGetBoolean(defel); } /* --- 8813,8818 ---- *************** ATExecSetRelOptions(Relation rel, List * *** 8825,8832 **** if (check_option) { const char *view_updatable_error = ! view_query_is_auto_updatable(view_query, ! security_barrier, true); if (view_updatable_error) ereport(ERROR, --- 8822,8828 ---- if (check_option) { const char *view_updatable_error = ! view_query_is_auto_updatable(view_query, true); if (view_updatable_error) ereport(ERROR, diff --git a/src/backend/commands/view.c b/src/backend/commands/view.c new file mode 100644 index 1735762..bc08566 *** a/src/backend/commands/view.c --- b/src/backend/commands/view.c *************** DefineView(ViewStmt *stmt, const char *q *** 396,402 **** RangeVar *view; ListCell *cell; bool check_option; - bool security_barrier; /* * Run parse analysis to convert the raw parse tree to a Query. Note this --- 396,401 ---- *************** DefineView(ViewStmt *stmt, const char *q *** 451,457 **** * specified. */ check_option = false; - security_barrier = false; foreach(cell, stmt->options) { --- 450,455 ---- *************** DefineView(ViewStmt *stmt, const char *q *** 459,466 **** if (pg_strcasecmp(defel->defname, "check_option") == 0) check_option = true; - if (pg_strcasecmp(defel->defname, "security_barrier") == 0) - security_barrier = defGetBoolean(defel); } /* --- 457,462 ---- *************** DefineView(ViewStmt *stmt, const char *q *** 470,476 **** if (check_option) { const char *view_updatable_error = ! view_query_is_auto_updatable(viewParse, security_barrier, true); if (view_updatable_error) ereport(ERROR, --- 466,472 ---- if (check_option) { const char *view_updatable_error = ! view_query_is_auto_updatable(viewParse, true); if (view_updatable_error) ereport(ERROR, diff --git a/src/backend/nodes/copyfuncs.c b/src/backend/nodes/copyfuncs.c new file mode 100644 index fb4ce2c..e8cdbfb *** a/src/backend/nodes/copyfuncs.c --- b/src/backend/nodes/copyfuncs.c *************** _copyRangeTblEntry(const RangeTblEntry * *** 1998,2003 **** --- 1998,2004 ---- COPY_SCALAR_FIELD(checkAsUser); COPY_BITMAPSET_FIELD(selectedCols); COPY_BITMAPSET_FIELD(modifiedCols); + COPY_NODE_FIELD(securityQuals); return newnode; } diff --git a/src/backend/nodes/equalfuncs.c b/src/backend/nodes/equalfuncs.c new file mode 100644 index ccf7267..0411519 *** a/src/backend/nodes/equalfuncs.c --- b/src/backend/nodes/equalfuncs.c *************** _equalRangeTblEntry(const RangeTblEntry *** 2278,2283 **** --- 2278,2284 ---- COMPARE_SCALAR_FIELD(checkAsUser); COMPARE_BITMAPSET_FIELD(selectedCols); COMPARE_BITMAPSET_FIELD(modifiedCols); + COMPARE_NODE_FIELD(securityQuals); return true; } diff --git a/src/backend/nodes/nodeFuncs.c b/src/backend/nodes/nodeFuncs.c new file mode 100644 index 123f2a6..1e48a7f *** a/src/backend/nodes/nodeFuncs.c --- b/src/backend/nodes/nodeFuncs.c *************** range_table_walker(List *rtable, *** 2020,2025 **** --- 2020,2028 ---- return true; break; } + + if (walker(rte->securityQuals, context)) + return true; } return false; } *************** range_table_mutator(List *rtable, *** 2755,2760 **** --- 2758,2764 ---- MUTATE(newrte->values_lists, rte->values_lists, List *); break; } + MUTATE(newrte->securityQuals, rte->securityQuals, List *); newrt = lappend(newrt, newrte); } return newrt; diff --git a/src/backend/nodes/outfuncs.c b/src/backend/nodes/outfuncs.c new file mode 100644 index 568c3b8..a0e3286 *** a/src/backend/nodes/outfuncs.c --- b/src/backend/nodes/outfuncs.c *************** _outRangeTblEntry(StringInfo str, const *** 2409,2414 **** --- 2409,2415 ---- WRITE_OID_FIELD(checkAsUser); WRITE_BITMAPSET_FIELD(selectedCols); WRITE_BITMAPSET_FIELD(modifiedCols); + WRITE_NODE_FIELD(securityQuals); } static void diff --git a/src/backend/nodes/readfuncs.c b/src/backend/nodes/readfuncs.c new file mode 100644 index 8a7a662..c81c5cb *** a/src/backend/nodes/readfuncs.c --- b/src/backend/nodes/readfuncs.c *************** _readRangeTblEntry(void) *** 1250,1255 **** --- 1250,1256 ---- READ_OID_FIELD(checkAsUser); READ_BITMAPSET_FIELD(selectedCols); READ_BITMAPSET_FIELD(modifiedCols); + READ_NODE_FIELD(securityQuals); READ_DONE(); } diff --git a/src/backend/optimizer/plan/planner.c b/src/backend/optimizer/plan/planner.c new file mode 100644 index 35bda67..1beb6c8 *** a/src/backend/optimizer/plan/planner.c --- b/src/backend/optimizer/plan/planner.c *************** inheritance_planner(PlannerInfo *root) *** 916,921 **** --- 916,927 ---- subplan = grouping_planner(&subroot, 0.0 /* retrieve all tuples */ ); /* + * Planning may have modified the query result relation (if there + * were security barrier quals on the result RTE). + */ + appinfo->child_relid = subroot.parse->resultRelation; + + /* * If this child rel was excluded by constraint exclusion, exclude it * from the result plan. */ *************** inheritance_planner(PlannerInfo *root) *** 932,940 **** if (final_rtable == NIL) final_rtable = subroot.parse->rtable; else ! final_rtable = list_concat(final_rtable, list_copy_tail(subroot.parse->rtable, list_length(final_rtable))); /* * We need to collect all the RelOptInfos from all child plans into --- 938,969 ---- if (final_rtable == NIL) final_rtable = subroot.parse->rtable; else ! { ! List *tmp_rtable = NIL; ! ListCell *cell1, *cell2; ! ! /* ! * Planning this new child may have turned some of the original ! * RTEs into subqueries (if they had security barrier quals). If ! * so, we want to use these in the final rtable. ! */ ! forboth(cell1, final_rtable, cell2, subroot.parse->rtable) ! { ! RangeTblEntry *rte1 = (RangeTblEntry *) lfirst(cell1); ! RangeTblEntry *rte2 = (RangeTblEntry *) lfirst(cell2); ! ! if (rte1->rtekind == RTE_RELATION && ! rte1->securityQuals != NIL && ! rte2->rtekind == RTE_SUBQUERY) ! tmp_rtable = lappend(tmp_rtable, rte2); ! else ! tmp_rtable = lappend(tmp_rtable, rte1); ! } ! ! final_rtable = list_concat(tmp_rtable, list_copy_tail(subroot.parse->rtable, list_length(final_rtable))); + } /* * We need to collect all the RelOptInfos from all child plans into *************** grouping_planner(PlannerInfo *root, doub *** 1163,1168 **** --- 1192,1203 ---- tlist = preprocess_targetlist(root, tlist); /* + * Expand any rangetable entries that have security barrier quals. + * This may add new security barrier subquery RTEs to the rangetable. + */ + expand_security_quals(root, tlist); + + /* * Locate any window functions in the tlist. (We don't need to look * anywhere else, since expressions used in ORDER BY will be in there * too.) Note that they could all have been eliminated by constant diff --git a/src/backend/optimizer/prep/Makefile b/src/backend/optimizer/prep/Makefile new file mode 100644 index 86301bf..5195d9b *** a/src/backend/optimizer/prep/Makefile --- b/src/backend/optimizer/prep/Makefile *************** subdir = src/backend/optimizer/prep *** 12,17 **** top_builddir = ../../../.. include $(top_builddir)/src/Makefile.global ! OBJS = prepjointree.o prepqual.o preptlist.o prepunion.o include $(top_srcdir)/src/backend/common.mk --- 12,17 ---- top_builddir = ../../../.. include $(top_builddir)/src/Makefile.global ! OBJS = prepjointree.o prepqual.o prepsecurity.o preptlist.o prepunion.o include $(top_srcdir)/src/backend/common.mk diff --git a/src/backend/optimizer/prep/prepsecurity.c b/src/backend/optimizer/prep/prepsecurity.c new file mode 100644 index ...e7a3f56 *** a/src/backend/optimizer/prep/prepsecurity.c --- b/src/backend/optimizer/prep/prepsecurity.c *************** *** 0 **** --- 1,423 ---- + /*------------------------------------------------------------------------- + * + * prepsecurity.c + * Routines for preprocessing security barrier quals. + * + * Portions Copyright (c) 1996-2013, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, Regents of the University of California + * + * + * IDENTIFICATION + * src/backend/optimizer/prep/prepsecurity.c + * + *------------------------------------------------------------------------- + */ + #include "postgres.h" + + #include "access/heapam.h" + #include "access/sysattr.h" + #include "catalog/heap.h" + #include "nodes/makefuncs.h" + #include "nodes/nodeFuncs.h" + #include "optimizer/prep.h" + #include "parser/parsetree.h" + #include "rewrite/rewriteManip.h" + #include "utils/rel.h" + + + typedef struct + { + int rt_index; /* Index of security barrier RTE */ + int sublevels_up; /* Current nesting depth */ + Relation rel; /* RTE relation at rt_index */ + List *targetlist; /* Targetlist for new subquery RTE */ + List *colnames; /* Column names in subquery RTE */ + List *vars_processed; /* List of Vars already processed */ + } security_barrier_replace_vars_context; + + static void expand_security_qual(Query *parse, List *tlist, int rt_index, Node *qual); + + static void security_barrier_replace_vars(Node *node, + security_barrier_replace_vars_context *context); + + static bool security_barrier_replace_vars_walker(Node *node, + security_barrier_replace_vars_context *context); + + + /* + * expand_security_quals - + * expands any security barrier quals on RTEs in the query rtable, turning + * them into security barrier subqueries. + * + * Any given RTE may have multiple security barrier quals in a list, from which + * we create a set of nested subqueries to isolate each security barrier from + * the others, providing protection against malicious user-defined security + * barriers. The first security barrier qual in the list will be used in the + * innermost subquery. + */ + void + expand_security_quals(PlannerInfo *root, List *tlist) + { + Query *parse = root->parse; + int rt_index; + + /* + * Process each RTE in the rtable list. + * + * Note that this is deliberately not a foreach loop, since the rtable may + * be modified each time through the loop. + */ + rt_index = 0; + while (rt_index < list_length(parse->rtable)) + { + RangeTblEntry *rte; + + rt_index++; + rte = rt_fetch(rt_index, parse->rtable); + + if (rte->securityQuals == NIL) + continue; + + /* + * Ignore any RTEs that aren't used in the query (such RTEs may be + * present for permissions checks). + */ + if (rt_index != parse->resultRelation && + !rangeTableEntry_used((Node *) parse, rt_index, 0)) + continue; + + /* + * If this RTE is the target then we need to make a copy of it before + * expanding it. The unexpanded copy will become the new target, and + * the original RTE will be expanded to become the source of rows to + * update/delete. + */ + if (rt_index == parse->resultRelation) + { + RangeTblEntry *newrte = copyObject(rte); + parse->rtable = lappend(parse->rtable, newrte); + parse->resultRelation = list_length(parse->rtable); + + /* + * Wipe out any copied security barrier quals on the new target to + * prevent infinite recursion. + */ + newrte->securityQuals = NIL; + + /* + * There's no need to do permissions checks twice, so wipe out the + * permissions info for the original RTE (we prefer to keep the + * bits set on the result RTE). + */ + rte->requiredPerms = 0; + rte->checkAsUser = InvalidOid; + rte->selectedCols = NULL; + rte->modifiedCols = NULL; + + /* + * For the most part, Vars referencing the original relation should + * remain as they are, meaning that they pull OLD values from the + * expanded RTE. But in the RETURNING list and in any WITH CHECK + * OPTION quals, we want such Vars to represent NEW values, so + * change them to reference the new RTE. + */ + ChangeVarNodes((Node *) parse->returningList, rt_index, + parse->resultRelation, 0); + + ChangeVarNodes((Node *) parse->withCheckOptions, rt_index, + parse->resultRelation, 0); + } + + /* + * Process each security barrier qual in turn, starting with the + * innermost one (the first in the list) and working outwards. + * + * We remove each qual from the list before processing it, so that its + * variables aren't modified by expand_security_qual. Also we don't + * necessarily want the attributes referred to by the qual to be + * exposed by the newly built subquery. + */ + while (rte->securityQuals != NIL) + { + Node *qual = (Node *) linitial(rte->securityQuals); + rte->securityQuals = list_delete_first(rte->securityQuals); + + ChangeVarNodes(qual, rt_index, 1, 0); + expand_security_qual(parse, tlist, rt_index, qual); + } + } + } + + + /* + * expand_security_qual - + * expand the specified security barrier qual on a query RTE, turning the + * RTE into a security barrier subquery. + */ + static void + expand_security_qual(Query *parse, List *tlist, int rt_index, Node *qual) + { + RangeTblEntry *rte; + Oid relid; + Query *subquery; + RangeTblEntry *subrte; + RangeTblRef *subrtr; + security_barrier_replace_vars_context context; + ListCell *cell; + + rte = rt_fetch(rt_index, parse->rtable); + relid = rte->relid; + + /* + * There should only be 2 possible cases: + * + * 1. A relation RTE, which we turn into a subquery RTE containing all + * referenced columns. + * + * 2. A subquery RTE (either from a prior call to this function or from an + * expanded view). In this case we build a new subquery on top of it to + * isolate this security barrier qual from any other quals. + */ + switch (rte->rtekind) + { + case RTE_RELATION: + /* + * Turn the relation RTE into a security barrier subquery RTE, + * moving all permissions checks down into the subquery. + */ + subquery = makeNode(Query); + subquery->commandType = CMD_SELECT; + subquery->querySource = QSRC_INSTEAD_RULE; + + subrte = copyObject(rte); + subrte->inFromCl = true; + subrte->securityQuals = NIL; + subquery->rtable = list_make1(subrte); + + subrtr = makeNode(RangeTblRef); + subrtr->rtindex = 1; + subquery->jointree = makeFromExpr(list_make1(subrtr), qual); + subquery->hasSubLinks = checkExprHasSubLink(qual); + + rte->rtekind = RTE_SUBQUERY; + rte->relid = InvalidOid; + rte->subquery = subquery; + rte->security_barrier = true; + rte->inh = false; /* must not be set for a subquery */ + + /* the permissions checks have now been moved down */ + rte->requiredPerms = 0; + rte->checkAsUser = InvalidOid; + rte->selectedCols = NULL; + rte->modifiedCols = NULL; + + /* + * Replace any variables in the outer query that refer to the + * original relation RTE with references to columns that we will + * expose in the new subquery, building the subquery's targetlist + * as we go. + */ + context.rt_index = rt_index; + context.sublevels_up = 0; + context.rel = heap_open(relid, NoLock); + context.targetlist = NIL; + context.colnames = NIL; + context.vars_processed = NIL; + + security_barrier_replace_vars((Node *) parse, &context); + security_barrier_replace_vars((Node *) tlist, &context); + + heap_close(context.rel, NoLock); + + /* Now we know what columns the subquery needs to expose */ + rte->subquery->targetList = context.targetlist; + rte->eref = makeAlias(rte->eref->aliasname, context.colnames); + + break; + + case RTE_SUBQUERY: + /* + * Build a new subquery that includes all the same columns as the + * original subquery. + */ + subquery = makeNode(Query); + subquery->commandType = CMD_SELECT; + subquery->querySource = QSRC_INSTEAD_RULE; + subquery->targetList = NIL; + + foreach(cell, rte->subquery->targetList) + { + TargetEntry *tle; + Var *var; + + tle = (TargetEntry *) lfirst(cell); + var = makeVarFromTargetEntry(1, tle); + + tle = makeTargetEntry((Expr *) var, + list_length(subquery->targetList) + 1, + pstrdup(tle->resname), + tle->resjunk); + subquery->targetList = lappend(subquery->targetList, tle); + } + + subrte = makeNode(RangeTblEntry); + subrte->rtekind = RTE_SUBQUERY; + subrte->subquery = rte->subquery; + subrte->security_barrier = rte->security_barrier; + subrte->eref = copyObject(rte->eref); + subrte->inFromCl = true; + subquery->rtable = list_make1(subrte); + + subrtr = makeNode(RangeTblRef); + subrtr->rtindex = 1; + subquery->jointree = makeFromExpr(list_make1(subrtr), qual); + subquery->hasSubLinks = checkExprHasSubLink(qual); + + rte->subquery = subquery; + rte->security_barrier = true; + + break; + + default: + elog(ERROR, "invalid range table entry for security barrier qual"); + } + } + + + /* + * security_barrier_replace_vars - + * Apply security barrier variable replacement to an expression tree. + * + * This also builds/updates a targetlist with entries for each replacement + * variable that needs to be exposed by the security barrier subquery RTE. + * + * NOTE: although this has the form of a walker, we cheat and modify the + * nodes in-place. The given expression tree should have been copied + * earlier to ensure that no unwanted side-effects occur! + */ + static void + security_barrier_replace_vars(Node *node, + security_barrier_replace_vars_context *context) + { + /* + * Must be prepared to start with a Query or a bare expression tree; if + * it's a Query, go straight to query_tree_walker to make sure that + * sublevels_up doesn't get incremented prematurely. + */ + if (node && IsA(node, Query)) + query_tree_walker((Query *) node, + security_barrier_replace_vars_walker, + (void *) context, 0); + else + security_barrier_replace_vars_walker(node, context); + } + + static bool + security_barrier_replace_vars_walker(Node *node, + security_barrier_replace_vars_context *context) + { + if (node == NULL) + return false; + if (IsA(node, Var)) + { + Var *var = (Var *) node; + + /* + * Note that the same Var may be present in different lists, so we + * need to take care not to process it multiple times. + */ + if (var->varno == context->rt_index && + var->varlevelsup == context->sublevels_up && + !list_member_ptr(context->vars_processed, var)) + { + /* + * Found a matching variable. Make sure that it is in the subquery + * targetlist and map its attno accordingly. + */ + AttrNumber attno; + ListCell *l; + TargetEntry *tle; + char *attname; + Var *newvar; + + /* Search for the base attribute in the subquery targetlist */ + attno = InvalidAttrNumber; + foreach(l, context->targetlist) + { + tle = (TargetEntry *) lfirst(l); + attno++; + + Assert(IsA(tle->expr, Var)); + if (((Var *) tle->expr)->varattno == var->varattno && + ((Var *) tle->expr)->varcollid == var->varcollid) + { + /* Map the variable onto this subquery targetlist entry */ + var->varattno = attno; + return false; + } + } + + /* Not in the subquery targetlist, so add it. Get its name. */ + if (var->varattno < 0) + { + Form_pg_attribute att_tup; + + att_tup = SystemAttributeDefinition(var->varattno, + context->rel->rd_rel->relhasoids); + attname = NameStr(att_tup->attname); + } + else if (var->varattno == InvalidAttrNumber) + { + attname = "wholerow"; + } + else if (var->varattno <= context->rel->rd_att->natts) + { + Form_pg_attribute att_tup; + + att_tup = context->rel->rd_att->attrs[var->varattno - 1]; + attname = NameStr(att_tup->attname); + } + else + { + elog(ERROR, "invalid attribute number %d in security_barrier_replace_vars", var->varattno); + } + + /* New variable for subquery targetlist */ + newvar = copyObject(var); + newvar->varno = 1; + + attno = list_length(context->targetlist) + 1; + tle = makeTargetEntry((Expr *) newvar, + attno, + pstrdup(attname), + false); + + context->targetlist = lappend(context->targetlist, tle); + + context->colnames = lappend(context->colnames, + makeString(pstrdup(attname))); + + /* Update the outer query's variable */ + var->varattno = attno; + + /* Remember this Var so that we don't process it again */ + context->vars_processed = lappend(context->vars_processed, var); + } + return false; + } + + if (IsA(node, Query)) + { + /* Recurse into subselects */ + bool result; + + context->sublevels_up++; + result = query_tree_walker((Query *) node, + security_barrier_replace_vars_walker, + (void *) context, 0); + context->sublevels_up--; + return result; + } + return expression_tree_walker(node, security_barrier_replace_vars_walker, + (void *) context); + } diff --git a/src/backend/optimizer/prep/prepunion.c b/src/backend/optimizer/prep/prepunion.c new file mode 100644 index 52dcc72..c7e0199 *** a/src/backend/optimizer/prep/prepunion.c --- b/src/backend/optimizer/prep/prepunion.c *************** typedef struct *** 55,60 **** --- 55,61 ---- { PlannerInfo *root; AppendRelInfo *appinfo; + int sublevels_up; } adjust_appendrel_attrs_context; static Plan *recurse_set_operations(Node *setOp, PlannerInfo *root, *************** translate_col_privs(const Bitmapset *par *** 1580,1587 **** * child rel instead. We also update rtindexes appearing outside Vars, * such as resultRelation and jointree relids. * ! * Note: this is only applied after conversion of sublinks to subplans, ! * so we don't need to cope with recursion into sub-queries. * * Note: this is not hugely different from what pullup_replace_vars() does; * maybe we should try to fold the two routines together. --- 1581,1589 ---- * child rel instead. We also update rtindexes appearing outside Vars, * such as resultRelation and jointree relids. * ! * Note: this is applied after conversion of sublinks to subplans in the ! * query jointree, but there may still be sublinks in the security barrier ! * quals of RTEs, so we do need to cope with recursion into sub-queries. * * Note: this is not hugely different from what pullup_replace_vars() does; * maybe we should try to fold the two routines together. *************** adjust_appendrel_attrs(PlannerInfo *root *** 1594,1602 **** context.root = root; context.appinfo = appinfo; /* ! * Must be prepared to start with a Query or a bare expression tree. */ if (node && IsA(node, Query)) { --- 1596,1607 ---- context.root = root; context.appinfo = appinfo; + context.sublevels_up = 0; /* ! * Must be prepared to start with a Query or a bare expression tree; if ! * it's a Query, go straight to query_tree_walker to make sure that ! * sublevels_up doesn't get incremented prematurely. */ if (node && IsA(node, Query)) { *************** adjust_appendrel_attrs_mutator(Node *nod *** 1635,1641 **** { Var *var = (Var *) copyObject(node); ! if (var->varlevelsup == 0 && var->varno == appinfo->parent_relid) { var->varno = appinfo->child_relid; --- 1640,1646 ---- { Var *var = (Var *) copyObject(node); ! if (var->varlevelsup == context->sublevels_up && var->varno == appinfo->parent_relid) { var->varno = appinfo->child_relid; *************** adjust_appendrel_attrs_mutator(Node *nod *** 1652,1657 **** --- 1657,1663 ---- if (newnode == NULL) elog(ERROR, "attribute %d of relation \"%s\" does not exist", var->varattno, get_rel_name(appinfo->parent_reloid)); + ((Var *) newnode)->varlevelsup += context->sublevels_up; return newnode; } else if (var->varattno == 0) *************** adjust_appendrel_attrs_mutator(Node *nod *** 1694,1703 **** --- 1700,1715 ---- RowExpr *rowexpr; List *fields; RangeTblEntry *rte; + ListCell *lc; rte = rt_fetch(appinfo->parent_relid, context->root->parse->rtable); fields = (List *) copyObject(appinfo->translated_vars); + foreach(lc, fields) + { + Var *field = (Var *) lfirst(lc); + field->varlevelsup += context->sublevels_up; + } rowexpr = makeNode(RowExpr); rowexpr->args = fields; rowexpr->row_typeid = var->vartype; *************** adjust_appendrel_attrs_mutator(Node *nod *** 1716,1722 **** { CurrentOfExpr *cexpr = (CurrentOfExpr *) copyObject(node); ! if (cexpr->cvarno == appinfo->parent_relid) cexpr->cvarno = appinfo->child_relid; return (Node *) cexpr; } --- 1728,1735 ---- { CurrentOfExpr *cexpr = (CurrentOfExpr *) copyObject(node); ! if (context->sublevels_up == 0 && ! cexpr->cvarno == appinfo->parent_relid) cexpr->cvarno = appinfo->child_relid; return (Node *) cexpr; } *************** adjust_appendrel_attrs_mutator(Node *nod *** 1724,1730 **** { RangeTblRef *rtr = (RangeTblRef *) copyObject(node); ! if (rtr->rtindex == appinfo->parent_relid) rtr->rtindex = appinfo->child_relid; return (Node *) rtr; } --- 1737,1744 ---- { RangeTblRef *rtr = (RangeTblRef *) copyObject(node); ! if (context->sublevels_up == 0 && ! rtr->rtindex == appinfo->parent_relid) rtr->rtindex = appinfo->child_relid; return (Node *) rtr; } *************** adjust_appendrel_attrs_mutator(Node *nod *** 1737,1743 **** adjust_appendrel_attrs_mutator, (void *) context); /* now fix JoinExpr's rtindex (probably never happens) */ ! if (j->rtindex == appinfo->parent_relid) j->rtindex = appinfo->child_relid; return (Node *) j; } --- 1751,1758 ---- adjust_appendrel_attrs_mutator, (void *) context); /* now fix JoinExpr's rtindex (probably never happens) */ ! if (context->sublevels_up == 0 && ! j->rtindex == appinfo->parent_relid) j->rtindex = appinfo->child_relid; return (Node *) j; } *************** adjust_appendrel_attrs_mutator(Node *nod *** 1750,1756 **** adjust_appendrel_attrs_mutator, (void *) context); /* now fix PlaceHolderVar's relid sets */ ! if (phv->phlevelsup == 0) phv->phrels = adjust_relid_set(phv->phrels, appinfo->parent_relid, appinfo->child_relid); --- 1765,1771 ---- adjust_appendrel_attrs_mutator, (void *) context); /* now fix PlaceHolderVar's relid sets */ ! if (phv->phlevelsup == context->sublevels_up) phv->phrels = adjust_relid_set(phv->phrels, appinfo->parent_relid, appinfo->child_relid); *************** adjust_appendrel_attrs_mutator(Node *nod *** 1822,1833 **** return (Node *) newinfo; } ! /* ! * NOTE: we do not need to recurse into sublinks, because they should ! * already have been converted to subplans before we see them. ! */ ! Assert(!IsA(node, SubLink)); ! Assert(!IsA(node, Query)); return expression_tree_mutator(node, adjust_appendrel_attrs_mutator, (void *) context); --- 1837,1862 ---- return (Node *) newinfo; } ! if (IsA(node, Query)) ! { ! /* ! * Recurse into sublink subqueries. This should only be possible in ! * security barrier quals of top-level RTEs. All other sublinks should ! * have already been converted to subplans during expression ! * preprocessing, but this doesn't happen for security barrier quals, ! * since they are destined to become quals of a subquery RTE, which ! * will be recursively planned, and so should not be preprocessed at ! * this stage. ! */ ! Query *newnode; ! ! context->sublevels_up++; ! newnode = query_tree_mutator((Query *) node, ! adjust_appendrel_attrs_mutator, ! (void *) context, 0); ! context->sublevels_up--; ! return (Node *) newnode; ! } return expression_tree_mutator(node, adjust_appendrel_attrs_mutator, (void *) context); diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c new file mode 100644 index 0b13645..370b5aa *** a/src/backend/rewrite/rewriteHandler.c --- b/src/backend/rewrite/rewriteHandler.c *************** view_col_is_auto_updatable(RangeTblRef * *** 1973,1980 **** * updatable. */ const char * ! view_query_is_auto_updatable(Query *viewquery, bool security_barrier, ! bool check_cols) { RangeTblRef *rtr; RangeTblEntry *base_rte; --- 1973,1979 ---- * updatable. */ const char * ! view_query_is_auto_updatable(Query *viewquery, bool check_cols) { RangeTblRef *rtr; RangeTblEntry *base_rte; *************** view_query_is_auto_updatable(Query *view *** 2048,2061 **** return gettext_noop("Views that return set-returning functions are not automatically updatable."); /* - * For now, we also don't support security-barrier views, because of the - * difficulty of keeping upper-level qual expressions away from - * lower-level data. This might get relaxed in the future. - */ - if (security_barrier) - return gettext_noop("Security-barrier views are not automatically updatable."); - - /* * The view query should select from a single base relation, which must be * a table or another view. */ --- 2047,2052 ---- *************** relation_is_updatable(Oid reloid, *** 2303,2311 **** { Query *viewquery = get_view_query(rel); ! if (view_query_is_auto_updatable(viewquery, ! RelationIsSecurityView(rel), ! false) == NULL) { Bitmapset *updatable_cols; int auto_events; --- 2294,2300 ---- { Query *viewquery = get_view_query(rel); ! if (view_query_is_auto_updatable(viewquery, false) == NULL) { Bitmapset *updatable_cols; int auto_events; *************** rewriteTargetView(Query *parsetree, Rela *** 2460,2466 **** auto_update_detail = view_query_is_auto_updatable(viewquery, - RelationIsSecurityView(view), parsetree->commandType != CMD_DELETE); if (auto_update_detail) --- 2449,2454 ---- *************** rewriteTargetView(Query *parsetree, Rela *** 2664,2669 **** --- 2652,2665 ---- view_targetlist); /* + * Move any security barrier quals from the view RTE onto the new target + * RTE. Any such quals should now apply to the new target RTE and will not + * reference the original view RTE in the rewritten query. + */ + new_rte->securityQuals = view_rte->securityQuals; + view_rte->securityQuals = NIL; + + /* * For UPDATE/DELETE, rewriteTargetListUD will have added a wholerow junk * TLE for the view to the end of the targetlist, which we no longer need. * Remove it to avoid unnecessary work when we process the targetlist. *************** rewriteTargetView(Query *parsetree, Rela *** 2743,2748 **** --- 2739,2748 ---- * only adjust their varnos to reference the new target (just the same as * we did with the view targetlist). * + * Note that there is special-case handling for the quals of a security + * barrier view, since they need to be kept separate from any user-supplied + * quals, so these quals are kept on the new target RTE. + * * For INSERT, the view's quals can be ignored in the main query. */ if (parsetree->commandType != CMD_INSERT && *************** rewriteTargetView(Query *parsetree, Rela *** 2751,2757 **** Node *viewqual = (Node *) copyObject(viewquery->jointree->quals); ChangeVarNodes(viewqual, base_rt_index, new_rt_index, 0); ! AddQual(parsetree, (Node *) viewqual); } /* --- 2751,2775 ---- Node *viewqual = (Node *) copyObject(viewquery->jointree->quals); ChangeVarNodes(viewqual, base_rt_index, new_rt_index, 0); ! ! if (RelationIsSecurityView(view)) ! { ! /* ! * Note: the parsetree has been mutated, so the new_rte pointer is ! * stale and needs to be re-computed. ! */ ! new_rte = rt_fetch(new_rt_index, parsetree->rtable); ! new_rte->securityQuals = lcons(viewqual, new_rte->securityQuals); ! ! /* ! * Make sure that the query is marked correctly if the added qual ! * has sublinks. ! */ ! if (!parsetree->hasSubLinks) ! parsetree->hasSubLinks = checkExprHasSubLink(viewqual); ! } ! else ! AddQual(parsetree, (Node *) viewqual); } /* *************** rewriteTargetView(Query *parsetree, Rela *** 2813,2821 **** * Make sure that the query is marked correctly if the added * qual has sublinks. We can skip this check if the query is * already marked, or if the command is an UPDATE, in which ! * case the same qual will have already been added to the ! * query's WHERE clause, and AddQual will have already done ! * this check. */ if (!parsetree->hasSubLinks && parsetree->commandType != CMD_UPDATE) --- 2831,2838 ---- * Make sure that the query is marked correctly if the added * qual has sublinks. We can skip this check if the query is * already marked, or if the command is an UPDATE, in which ! * case the same qual will have already been added, and this ! * check will already have been done. */ if (!parsetree->hasSubLinks && parsetree->commandType != CMD_UPDATE) diff --git a/src/include/nodes/parsenodes.h b/src/include/nodes/parsenodes.h new file mode 100644 index 9a3a5d7..9cdef61 *** a/src/include/nodes/parsenodes.h --- b/src/include/nodes/parsenodes.h *************** typedef struct RangeTblEntry *** 801,806 **** --- 801,807 ---- Oid checkAsUser; /* if valid, check access as this role */ Bitmapset *selectedCols; /* columns needing SELECT permission */ Bitmapset *modifiedCols; /* columns needing INSERT/UPDATE permission */ + List *securityQuals; /* any security barrier quals to apply */ } RangeTblEntry; /* diff --git a/src/include/optimizer/prep.h b/src/include/optimizer/prep.h new file mode 100644 index 0f5a7d3..f5fc7e8 *** a/src/include/optimizer/prep.h --- b/src/include/optimizer/prep.h *************** extern Node *negate_clause(Node *node); *** 36,41 **** --- 36,46 ---- extern Expr *canonicalize_qual(Expr *qual); /* + * prototypes for prepsecurity.c + */ + extern void expand_security_quals(PlannerInfo *root, List *tlist); + + /* * prototypes for preptlist.c */ extern List *preprocess_targetlist(PlannerInfo *root, List *tlist); diff --git a/src/include/rewrite/rewriteHandler.h b/src/include/rewrite/rewriteHandler.h new file mode 100644 index e4027bd..8da0af5 *** a/src/include/rewrite/rewriteHandler.h --- b/src/include/rewrite/rewriteHandler.h *************** extern void AcquireRewriteLocks(Query *p *** 23,29 **** extern Node *build_column_default(Relation rel, int attrno); extern Query *get_view_query(Relation view); extern const char *view_query_is_auto_updatable(Query *viewquery, - bool security_barrier, bool check_cols); extern int relation_is_updatable(Oid reloid, bool include_triggers, --- 23,28 ---- diff --git a/src/test/regress/expected/create_view.out b/src/test/regress/expected/create_view.out new file mode 100644 index 91d1639..f6db582 *** a/src/test/regress/expected/create_view.out --- b/src/test/regress/expected/create_view.out *************** CREATE VIEW mysecview4 WITH (security_ba *** 252,258 **** AS SELECT * FROM tbl1 WHERE a <> 0; CREATE VIEW mysecview5 WITH (security_barrier=100) -- Error AS SELECT * FROM tbl1 WHERE a > 100; ! ERROR: security_barrier requires a Boolean value CREATE VIEW mysecview6 WITH (invalid_option) -- Error AS SELECT * FROM tbl1 WHERE a < 100; ERROR: unrecognized parameter "invalid_option" --- 252,258 ---- AS SELECT * FROM tbl1 WHERE a <> 0; CREATE VIEW mysecview5 WITH (security_barrier=100) -- Error AS SELECT * FROM tbl1 WHERE a > 100; ! ERROR: invalid value for boolean option "security_barrier": 100 CREATE VIEW mysecview6 WITH (invalid_option) -- Error AS SELECT * FROM tbl1 WHERE a < 100; ERROR: unrecognized parameter "invalid_option" diff --git a/src/test/regress/expected/updatable_views.out b/src/test/regress/expected/updatable_views.out new file mode 100644 index 99c9165..d12d384 *** a/src/test/regress/expected/updatable_views.out --- b/src/test/regress/expected/updatable_views.out *************** CREATE VIEW rw_view14 AS SELECT ctid, a, *** 22,33 **** CREATE VIEW rw_view15 AS SELECT a, upper(b) FROM base_tbl; -- Expression/function may be part of an updatable view CREATE VIEW rw_view16 AS SELECT a, b, a AS aa FROM base_tbl; -- Repeated column may be part of an updatable view CREATE VIEW ro_view17 AS SELECT * FROM ro_view1; -- Base relation not updatable ! CREATE VIEW ro_view18 WITH (security_barrier = true) ! AS SELECT * FROM base_tbl; -- Security barrier views not updatable ! CREATE VIEW ro_view19 AS SELECT * FROM (VALUES(1)) AS tmp(a); -- VALUES in rangetable CREATE SEQUENCE seq; ! CREATE VIEW ro_view20 AS SELECT * FROM seq; -- View based on a sequence ! CREATE VIEW ro_view21 AS SELECT a, b, generate_series(1, a) g FROM base_tbl; -- SRF in targetlist not supported SELECT table_name, is_insertable_into FROM information_schema.tables WHERE table_name LIKE E'r_\\_view%' --- 22,31 ---- CREATE VIEW rw_view15 AS SELECT a, upper(b) FROM base_tbl; -- Expression/function may be part of an updatable view CREATE VIEW rw_view16 AS SELECT a, b, a AS aa FROM base_tbl; -- Repeated column may be part of an updatable view CREATE VIEW ro_view17 AS SELECT * FROM ro_view1; -- Base relation not updatable ! CREATE VIEW ro_view18 AS SELECT * FROM (VALUES(1)) AS tmp(a); -- VALUES in rangetable CREATE SEQUENCE seq; ! CREATE VIEW ro_view19 AS SELECT * FROM seq; -- View based on a sequence ! CREATE VIEW ro_view20 AS SELECT a, b, generate_series(1, a) g FROM base_tbl; -- SRF in targetlist not supported SELECT table_name, is_insertable_into FROM information_schema.tables WHERE table_name LIKE E'r_\\_view%' *************** SELECT table_name, is_insertable_into *** 44,50 **** ro_view19 | NO ro_view2 | NO ro_view20 | NO - ro_view21 | NO ro_view3 | NO ro_view4 | NO ro_view5 | NO --- 42,47 ---- *************** SELECT table_name, is_insertable_into *** 55,61 **** rw_view14 | YES rw_view15 | YES rw_view16 | YES ! (21 rows) SELECT table_name, is_updatable, is_insertable_into FROM information_schema.views --- 52,58 ---- rw_view14 | YES rw_view15 | YES rw_view16 | YES ! (20 rows) SELECT table_name, is_updatable, is_insertable_into FROM information_schema.views *************** SELECT table_name, is_updatable, is_inse *** 73,79 **** ro_view19 | NO | NO ro_view2 | NO | NO ro_view20 | NO | NO - ro_view21 | NO | NO ro_view3 | NO | NO ro_view4 | NO | NO ro_view5 | NO | NO --- 70,75 ---- *************** SELECT table_name, is_updatable, is_inse *** 84,90 **** rw_view14 | YES | YES rw_view15 | YES | YES rw_view16 | YES | YES ! (21 rows) SELECT table_name, column_name, is_updatable FROM information_schema.columns --- 80,86 ---- rw_view14 | YES | YES rw_view15 | YES | YES rw_view16 | YES | YES ! (20 rows) SELECT table_name, column_name, is_updatable FROM information_schema.columns *************** SELECT table_name, column_name, is_updat *** 103,125 **** ro_view17 | a | NO ro_view17 | b | NO ro_view18 | a | NO ! ro_view18 | b | NO ! ro_view19 | a | NO ro_view2 | a | NO ro_view2 | b | NO ! ro_view20 | sequence_name | NO ! ro_view20 | last_value | NO ! ro_view20 | start_value | NO ! ro_view20 | increment_by | NO ! ro_view20 | max_value | NO ! ro_view20 | min_value | NO ! ro_view20 | cache_value | NO ! ro_view20 | log_cnt | NO ! ro_view20 | is_cycled | NO ! ro_view20 | is_called | NO ! ro_view21 | a | NO ! ro_view21 | b | NO ! ro_view21 | g | NO ro_view3 | ?column? | NO ro_view4 | count | NO ro_view5 | a | NO --- 99,119 ---- ro_view17 | a | NO ro_view17 | b | NO ro_view18 | a | NO ! ro_view19 | sequence_name | NO ! ro_view19 | last_value | NO ! ro_view19 | start_value | NO ! ro_view19 | increment_by | NO ! ro_view19 | max_value | NO ! ro_view19 | min_value | NO ! ro_view19 | cache_value | NO ! ro_view19 | log_cnt | NO ! ro_view19 | is_cycled | NO ! ro_view19 | is_called | NO ro_view2 | a | NO ro_view2 | b | NO ! ro_view20 | a | NO ! ro_view20 | b | NO ! ro_view20 | g | NO ro_view3 | ?column? | NO ro_view4 | count | NO ro_view5 | a | NO *************** SELECT table_name, column_name, is_updat *** 140,146 **** rw_view16 | a | YES rw_view16 | b | YES rw_view16 | aa | YES ! (48 rows) -- Read-only views DELETE FROM ro_view1; --- 134,140 ---- rw_view16 | a | YES rw_view16 | b | YES rw_view16 | aa | YES ! (46 rows) -- Read-only views DELETE FROM ro_view1; *************** INSERT INTO ro_view17 VALUES (3, 'ROW 3' *** 268,291 **** ERROR: cannot insert into view "ro_view1" DETAIL: Views containing DISTINCT are not automatically updatable. HINT: To enable inserting into the view, provide an INSTEAD OF INSERT trigger or an unconditional ON INSERT DO INSTEAD rule. ! INSERT INTO ro_view18 VALUES (3, 'ROW 3'); ! ERROR: cannot insert into view "ro_view18" ! DETAIL: Security-barrier views are not automatically updatable. ! HINT: To enable inserting into the view, provide an INSTEAD OF INSERT trigger or an unconditional ON INSERT DO INSTEAD rule. ! DELETE FROM ro_view19; ! ERROR: cannot delete from view "ro_view19" DETAIL: Views that do not select from a single table or view are not automatically updatable. HINT: To enable deleting from the view, provide an INSTEAD OF DELETE trigger or an unconditional ON DELETE DO INSTEAD rule. ! UPDATE ro_view20 SET max_value=1000; ! ERROR: cannot update view "ro_view20" DETAIL: Views that do not select from a single table or view are not automatically updatable. HINT: To enable updating the view, provide an INSTEAD OF UPDATE trigger or an unconditional ON UPDATE DO INSTEAD rule. ! UPDATE ro_view21 SET b=upper(b); ! ERROR: cannot update view "ro_view21" DETAIL: Views that return set-returning functions are not automatically updatable. HINT: To enable updating the view, provide an INSTEAD OF UPDATE trigger or an unconditional ON UPDATE DO INSTEAD rule. DROP TABLE base_tbl CASCADE; ! NOTICE: drop cascades to 17 other objects DETAIL: drop cascades to view ro_view1 drop cascades to view ro_view17 drop cascades to view ro_view2 --- 262,281 ---- ERROR: cannot insert into view "ro_view1" DETAIL: Views containing DISTINCT are not automatically updatable. HINT: To enable inserting into the view, provide an INSTEAD OF INSERT trigger or an unconditional ON INSERT DO INSTEAD rule. ! DELETE FROM ro_view18; ! ERROR: cannot delete from view "ro_view18" DETAIL: Views that do not select from a single table or view are not automatically updatable. HINT: To enable deleting from the view, provide an INSTEAD OF DELETE trigger or an unconditional ON DELETE DO INSTEAD rule. ! UPDATE ro_view19 SET max_value=1000; ! ERROR: cannot update view "ro_view19" DETAIL: Views that do not select from a single table or view are not automatically updatable. HINT: To enable updating the view, provide an INSTEAD OF UPDATE trigger or an unconditional ON UPDATE DO INSTEAD rule. ! UPDATE ro_view20 SET b=upper(b); ! ERROR: cannot update view "ro_view20" DETAIL: Views that return set-returning functions are not automatically updatable. HINT: To enable updating the view, provide an INSTEAD OF UPDATE trigger or an unconditional ON UPDATE DO INSTEAD rule. DROP TABLE base_tbl CASCADE; ! NOTICE: drop cascades to 16 other objects DETAIL: drop cascades to view ro_view1 drop cascades to view ro_view17 drop cascades to view ro_view2 *************** drop cascades to view ro_view11 *** 299,311 **** drop cascades to view ro_view13 drop cascades to view rw_view15 drop cascades to view rw_view16 ! drop cascades to view ro_view18 ! drop cascades to view ro_view21 drop cascades to view ro_view4 drop cascades to view rw_view14 ! DROP VIEW ro_view10, ro_view12, ro_view19; DROP SEQUENCE seq CASCADE; ! NOTICE: drop cascades to view ro_view20 -- simple updatable view CREATE TABLE base_tbl (a int PRIMARY KEY, b text DEFAULT 'Unspecified'); INSERT INTO base_tbl SELECT i, 'Row ' || i FROM generate_series(-2, 2) g(i); --- 289,300 ---- drop cascades to view ro_view13 drop cascades to view rw_view15 drop cascades to view rw_view16 ! drop cascades to view ro_view20 drop cascades to view ro_view4 drop cascades to view rw_view14 ! DROP VIEW ro_view10, ro_view12, ro_view18; DROP SEQUENCE seq CASCADE; ! NOTICE: drop cascades to view ro_view19 -- simple updatable view CREATE TABLE base_tbl (a int PRIMARY KEY, b text DEFAULT 'Unspecified'); INSERT INTO base_tbl SELECT i, 'Row ' || i FROM generate_series(-2, 2) g(i); *************** INSERT INTO rw_view2 VALUES (2,3); -- ok *** 1739,1742 **** --- 1728,1923 ---- DROP TABLE base_tbl CASCADE; NOTICE: drop cascades to 2 other objects DETAIL: drop cascades to view rw_view1 + drop cascades to view rw_view2 + -- security barrier view + CREATE TABLE base_tbl (person text, visibility text); + INSERT INTO base_tbl VALUES ('Tom', 'public'), + ('Dick', 'private'), + ('Harry', 'public'); + CREATE VIEW rw_view1 AS + SELECT person FROM base_tbl WHERE visibility = 'public'; + CREATE FUNCTION snoop(val text) + RETURNS boolean AS + $$ + BEGIN + RAISE NOTICE 'snooped value: %', val; + RETURN true; + END; + $$ + LANGUAGE plpgsql COST 0.000001; + SELECT * FROM rw_view1 WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Dick + NOTICE: snooped value: Harry + person + -------- + Tom + Harry + (2 rows) + + UPDATE rw_view1 SET person=person WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Dick + NOTICE: snooped value: Harry + DELETE FROM rw_view1 WHERE NOT snoop(person); + NOTICE: snooped value: Dick + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + ALTER VIEW rw_view1 SET (security_barrier = true); + SELECT table_name, is_insertable_into + FROM information_schema.tables + WHERE table_name = 'rw_view1'; + table_name | is_insertable_into + ------------+-------------------- + rw_view1 | YES + (1 row) + + SELECT table_name, is_updatable, is_insertable_into + FROM information_schema.views + WHERE table_name = 'rw_view1'; + table_name | is_updatable | is_insertable_into + ------------+--------------+-------------------- + rw_view1 | YES | YES + (1 row) + + SELECT table_name, column_name, is_updatable + FROM information_schema.columns + WHERE table_name = 'rw_view1' + ORDER BY ordinal_position; + table_name | column_name | is_updatable + ------------+-------------+-------------- + rw_view1 | person | YES + (1 row) + + SELECT * FROM rw_view1 WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + person + -------- + Tom + Harry + (2 rows) + + UPDATE rw_view1 SET person=person WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + DELETE FROM rw_view1 WHERE NOT snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + EXPLAIN (costs off) SELECT * FROM rw_view1 WHERE snoop(person); + QUERY PLAN + ----------------------------------------------- + Subquery Scan on rw_view1 + Filter: snoop(rw_view1.person) + -> Seq Scan on base_tbl + Filter: (visibility = 'public'::text) + (4 rows) + + EXPLAIN (costs off) UPDATE rw_view1 SET person=person WHERE snoop(person); + QUERY PLAN + ----------------------------------------------------- + Update on base_tbl base_tbl_1 + -> Subquery Scan on base_tbl + Filter: snoop(base_tbl.person) + -> Seq Scan on base_tbl base_tbl_2 + Filter: (visibility = 'public'::text) + (5 rows) + + EXPLAIN (costs off) DELETE FROM rw_view1 WHERE NOT snoop(person); + QUERY PLAN + ----------------------------------------------------- + Delete on base_tbl base_tbl_1 + -> Subquery Scan on base_tbl + Filter: (NOT snoop(base_tbl.person)) + -> Seq Scan on base_tbl base_tbl_2 + Filter: (visibility = 'public'::text) + (5 rows) + + -- security barrier view on top of security barrier view + CREATE VIEW rw_view2 WITH (security_barrier = true) AS + SELECT * FROM rw_view1 WHERE snoop(person); + SELECT table_name, is_insertable_into + FROM information_schema.tables + WHERE table_name = 'rw_view2'; + table_name | is_insertable_into + ------------+-------------------- + rw_view2 | YES + (1 row) + + SELECT table_name, is_updatable, is_insertable_into + FROM information_schema.views + WHERE table_name = 'rw_view2'; + table_name | is_updatable | is_insertable_into + ------------+--------------+-------------------- + rw_view2 | YES | YES + (1 row) + + SELECT table_name, column_name, is_updatable + FROM information_schema.columns + WHERE table_name = 'rw_view2' + ORDER BY ordinal_position; + table_name | column_name | is_updatable + ------------+-------------+-------------- + rw_view2 | person | YES + (1 row) + + SELECT * FROM rw_view2 WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + NOTICE: snooped value: Harry + person + -------- + Tom + Harry + (2 rows) + + UPDATE rw_view2 SET person=person WHERE snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + NOTICE: snooped value: Harry + DELETE FROM rw_view2 WHERE NOT snoop(person); + NOTICE: snooped value: Tom + NOTICE: snooped value: Tom + NOTICE: snooped value: Harry + NOTICE: snooped value: Harry + EXPLAIN (costs off) SELECT * FROM rw_view2 WHERE snoop(person); + QUERY PLAN + ----------------------------------------------------- + Subquery Scan on rw_view2 + Filter: snoop(rw_view2.person) + -> Subquery Scan on rw_view1 + Filter: snoop(rw_view1.person) + -> Seq Scan on base_tbl + Filter: (visibility = 'public'::text) + (6 rows) + + EXPLAIN (costs off) UPDATE rw_view2 SET person=person WHERE snoop(person); + QUERY PLAN + ----------------------------------------------------------- + Update on base_tbl base_tbl_1 + -> Subquery Scan on base_tbl + Filter: snoop(base_tbl.person) + -> Subquery Scan on base_tbl_2 + Filter: snoop(base_tbl_2.person) + -> Seq Scan on base_tbl base_tbl_3 + Filter: (visibility = 'public'::text) + (7 rows) + + EXPLAIN (costs off) DELETE FROM rw_view2 WHERE NOT snoop(person); + QUERY PLAN + ----------------------------------------------------------- + Delete on base_tbl base_tbl_1 + -> Subquery Scan on base_tbl + Filter: (NOT snoop(base_tbl.person)) + -> Subquery Scan on base_tbl_2 + Filter: snoop(base_tbl_2.person) + -> Seq Scan on base_tbl base_tbl_3 + Filter: (visibility = 'public'::text) + (7 rows) + + DROP TABLE base_tbl CASCADE; + NOTICE: drop cascades to 2 other objects + DETAIL: drop cascades to view rw_view1 drop cascades to view rw_view2 diff --git a/src/test/regress/sql/updatable_views.sql b/src/test/regress/sql/updatable_views.sql new file mode 100644 index a77cf19..6b352c1 *** a/src/test/regress/sql/updatable_views.sql --- b/src/test/regress/sql/updatable_views.sql *************** CREATE VIEW rw_view14 AS SELECT ctid, a, *** 25,36 **** CREATE VIEW rw_view15 AS SELECT a, upper(b) FROM base_tbl; -- Expression/function may be part of an updatable view CREATE VIEW rw_view16 AS SELECT a, b, a AS aa FROM base_tbl; -- Repeated column may be part of an updatable view CREATE VIEW ro_view17 AS SELECT * FROM ro_view1; -- Base relation not updatable ! CREATE VIEW ro_view18 WITH (security_barrier = true) ! AS SELECT * FROM base_tbl; -- Security barrier views not updatable ! CREATE VIEW ro_view19 AS SELECT * FROM (VALUES(1)) AS tmp(a); -- VALUES in rangetable CREATE SEQUENCE seq; ! CREATE VIEW ro_view20 AS SELECT * FROM seq; -- View based on a sequence ! CREATE VIEW ro_view21 AS SELECT a, b, generate_series(1, a) g FROM base_tbl; -- SRF in targetlist not supported SELECT table_name, is_insertable_into FROM information_schema.tables --- 25,34 ---- CREATE VIEW rw_view15 AS SELECT a, upper(b) FROM base_tbl; -- Expression/function may be part of an updatable view CREATE VIEW rw_view16 AS SELECT a, b, a AS aa FROM base_tbl; -- Repeated column may be part of an updatable view CREATE VIEW ro_view17 AS SELECT * FROM ro_view1; -- Base relation not updatable ! CREATE VIEW ro_view18 AS SELECT * FROM (VALUES(1)) AS tmp(a); -- VALUES in rangetable CREATE SEQUENCE seq; ! CREATE VIEW ro_view19 AS SELECT * FROM seq; -- View based on a sequence ! CREATE VIEW ro_view20 AS SELECT a, b, generate_series(1, a) g FROM base_tbl; -- SRF in targetlist not supported SELECT table_name, is_insertable_into FROM information_schema.tables *************** SELECT * FROM base_tbl; *** 87,99 **** DELETE FROM rw_view16 WHERE a=-3; -- should be OK -- Read-only views INSERT INTO ro_view17 VALUES (3, 'ROW 3'); ! INSERT INTO ro_view18 VALUES (3, 'ROW 3'); ! DELETE FROM ro_view19; ! UPDATE ro_view20 SET max_value=1000; ! UPDATE ro_view21 SET b=upper(b); DROP TABLE base_tbl CASCADE; ! DROP VIEW ro_view10, ro_view12, ro_view19; DROP SEQUENCE seq CASCADE; -- simple updatable view --- 85,96 ---- DELETE FROM rw_view16 WHERE a=-3; -- should be OK -- Read-only views INSERT INTO ro_view17 VALUES (3, 'ROW 3'); ! DELETE FROM ro_view18; ! UPDATE ro_view19 SET max_value=1000; ! UPDATE ro_view20 SET b=upper(b); DROP TABLE base_tbl CASCADE; ! DROP VIEW ro_view10, ro_view12, ro_view18; DROP SEQUENCE seq CASCADE; -- simple updatable view *************** CREATE VIEW rw_view2 AS *** 828,830 **** --- 825,902 ---- SELECT * FROM rw_view1 WHERE a > b WITH LOCAL CHECK OPTION; INSERT INTO rw_view2 VALUES (2,3); -- ok, but not in view (doesn't fail rw_view2's check) DROP TABLE base_tbl CASCADE; + + -- security barrier view + + CREATE TABLE base_tbl (person text, visibility text); + INSERT INTO base_tbl VALUES ('Tom', 'public'), + ('Dick', 'private'), + ('Harry', 'public'); + + CREATE VIEW rw_view1 AS + SELECT person FROM base_tbl WHERE visibility = 'public'; + + CREATE FUNCTION snoop(val text) + RETURNS boolean AS + $$ + BEGIN + RAISE NOTICE 'snooped value: %', val; + RETURN true; + END; + $$ + LANGUAGE plpgsql COST 0.000001; + + SELECT * FROM rw_view1 WHERE snoop(person); + UPDATE rw_view1 SET person=person WHERE snoop(person); + DELETE FROM rw_view1 WHERE NOT snoop(person); + + ALTER VIEW rw_view1 SET (security_barrier = true); + + SELECT table_name, is_insertable_into + FROM information_schema.tables + WHERE table_name = 'rw_view1'; + + SELECT table_name, is_updatable, is_insertable_into + FROM information_schema.views + WHERE table_name = 'rw_view1'; + + SELECT table_name, column_name, is_updatable + FROM information_schema.columns + WHERE table_name = 'rw_view1' + ORDER BY ordinal_position; + + SELECT * FROM rw_view1 WHERE snoop(person); + UPDATE rw_view1 SET person=person WHERE snoop(person); + DELETE FROM rw_view1 WHERE NOT snoop(person); + + EXPLAIN (costs off) SELECT * FROM rw_view1 WHERE snoop(person); + EXPLAIN (costs off) UPDATE rw_view1 SET person=person WHERE snoop(person); + EXPLAIN (costs off) DELETE FROM rw_view1 WHERE NOT snoop(person); + + -- security barrier view on top of security barrier view + + CREATE VIEW rw_view2 WITH (security_barrier = true) AS + SELECT * FROM rw_view1 WHERE snoop(person); + + SELECT table_name, is_insertable_into + FROM information_schema.tables + WHERE table_name = 'rw_view2'; + + SELECT table_name, is_updatable, is_insertable_into + FROM information_schema.views + WHERE table_name = 'rw_view2'; + + SELECT table_name, column_name, is_updatable + FROM information_schema.columns + WHERE table_name = 'rw_view2' + ORDER BY ordinal_position; + + SELECT * FROM rw_view2 WHERE snoop(person); + UPDATE rw_view2 SET person=person WHERE snoop(person); + DELETE FROM rw_view2 WHERE NOT snoop(person); + + EXPLAIN (costs off) SELECT * FROM rw_view2 WHERE snoop(person); + EXPLAIN (costs off) UPDATE rw_view2 SET person=person WHERE snoop(person); + EXPLAIN (costs off) DELETE FROM rw_view2 WHERE NOT snoop(person); + + DROP TABLE base_tbl CASCADE;