/*------------------------------------------------------------------------- * * viewUpdate.c * routines for translating an SQL-92-compliant view definition into * INSERT/UPDATE/DELETE rules (i.e. updatable views). * * Portions Copyright (c) 1996-2006, PostgreSQL Global Development Group * Portions Copyright (c) 1994, Regents of the University of California * * ORIGINAL AUTHORS * Bernd Helmle, Jaime Casanova * * IDENTIFICATION * $PostgreSQL$ * *------------------------------------------------------------------------- */ #include "postgres.h" #include "access/heapam.h" #include "access/xact.h" #include "catalog/pg_operator.h" #include "catalog/pg_proc.h" #include "catalog/pg_rewrite.h" #include "catalog/pg_type.h" #include "lib/stringinfo.h" #include "nodes/makefuncs.h" #include "optimizer/clauses.h" #include "parser/parse_func.h" #include "parser/parse_oper.h" #include "parser/parsetree.h" #include "rewrite/rewriteDefine.h" #include "rewrite/rewriteHandler.h" #include "rewrite/rewriteManip.h" #include "rewrite/rewriteSupport.h" #include "rewrite/viewUpdate.h" #include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/memutils.h" #include "utils/syscache.h" typedef TargetEntry** ViewDefColumnList; typedef struct ViewBaseRelation { List *defs; /* List of all base relations (root starts * with only one relation because SQL92 * allows this only) */ Oid parentRelation; /* Oid of parent relation, 0 indicates root */ } ViewBaseRelation; typedef struct ViewBaseRelationItem { Relation rel; /* the Relation itself */ Query *rule; /* _RETURN rule of a view relation */ TargetEntry **tentries; /* saves order of column target list */ } ViewBaseRelationItem; typedef struct ViewExprContext { Index newRTE; Index oldRTE; Index baseRTE; Index subQueryLevel; ViewDefColumnList tentries; } ViewExprContext; /* * Defines fixed OID for view update check option validation * function (since we know it's a builtin function we don't want * to do a lookup via its name). */ #define PG_VIEW_UPDATE_CONTROL 2570 static Query *get_return_rule(Relation rel); static void read_rearranged_cols(ViewBaseRelation *tree); static bool checkTree(const Query *query, ViewBaseRelation *tree); static Oid get_reloid_from_select(const Query *select, int *rti, RangeTblEntry **rel_entry); static Query *transform_select_to_update(const Query *update, const Relation rel, const RangeVar *var, TargetEntry **tentries, bool checkOption, bool checkCascade); static Query *transform_select_to_insert(const Query *select, const Relation rel, const RangeVar *var, TargetEntry **tentries, bool checkOption, bool checkCascade); static Query *transform_select_to_delete(const Query *delete, const Relation rel, const RangeVar *var, TargetEntry **tentries, bool checkOption, bool checkCascade); static void get_base_relations(ViewBaseRelation *tree, List **baserelations); static void get_base_base_relations(const Query *view, Oid baserel, List **list); static void copyReversedTargetEntryPtr(List *targetList, ViewDefColumnList targets); static bool check_reltree(ViewBaseRelation *node); static RuleStmt *create_rule_stmt(Query *actions, const RangeVar *view, bool replace, char *rulename, CmdType event); static bool hasRule(Oid view, const char *name); static bool form_query(const Query *select, Query *query, ViewDefColumnList tentries, bool copyTargetList); static RangeTblEntry *get_relation_RTE(const Query *select, unsigned int *offset); static Index get_rtindex_for_rel(List *rte_list, const char *relname); static HeapTuple get_view_qualification_func(Oid procoid); static bool view_def_qual_walker(Node *node, ViewExprContext *context); static FuncExpr *create_qual_for_rule(RuleStmt *stmt, const Query *select, ViewDefColumnList tentries, Index newRTE, Index oldRTE); static bool replace_tlist_varno_walker(Node *node, ViewExprContext *ctxt); static bool replace_varnos_for_var(Node *node, Index rtIndex, Index oldIndex, Index subQueryLevel); static OpExpr *create_opexpr(Var *var_left, Var *var_right); static void form_where_for_updrule(const Query *select, FromExpr **from, const Relation rel, Oid baserel, Index baserti, Index oldrti); static void build_update_target_list(const Query *update, const Query *select, ViewDefColumnList tentries, Oid baserel, const Relation rel); static bool viewIsInsertable(const Query *select, const Relation rel); /* * System function to check CHECK OPTION expression */ PG_FUNCTION_INFO_V1(pg_view_update_error); Datum pg_view_update_error(PG_FUNCTION_ARGS); /*------------------------------------------------------------------------------ * Private functions * ----------------------------------------------------------------------------- */ /* * Returns the range table index for the specified relname. * * XXX This seems pretty grotty ... can't we do this in some other way? */ static Index get_rtindex_for_rel(List *rte_list, const char *relname) { ListCell *cell; int index = 0; Assert(relname != NULL); foreach(cell, rte_list) { RangeTblEntry *rte = (RangeTblEntry *) lfirst(cell); ++index; if (rte != NULL) { if (strncmp(rte->eref->aliasname, relname, NAMEDATALEN) == 0) break; } } Assert(index > 0); return (Index) index; } /* * Returns the RangeTblEntry starting at the specified offset. The function can * be used to iterate over the rtable list of the specified select query tree. * Returns NULL if nothing is found. * * NOTE: the function only returns those RangeTblEntry that do not match a * *NEW* or *OLD* RangeTblEntry. * * The offset is incremented as a side effect. */ static RangeTblEntry * get_relation_RTE(const Query *select, unsigned int *offset) { AssertArg(offset != NULL); AssertArg(select != NULL); if ((*offset + 1) > list_length(select->rtable)) return NULL; while (*offset <= list_length(select->rtable)) { RangeTblEntry *rte = rt_fetch(*offset, select->rtable); ++(*offset); /* skip non-table RTEs */ if (rte->rtekind != RTE_RELATION) continue; /* * Skip RTEs named *NEW* and *OLD*. * * XXX It would be nice to be able to use something else than just * the names here ... However, rtekind does not work as expected :-( */ if (((strncmp(rte->eref->aliasname, "*NEW*", 6) == 0) || (strncmp(rte->eref->aliasname, "*OLD*", 6) == 0))) continue; return rte; } return NULL; } /* * Replaces the varnos if node is of type Var. The varno is replaced only if * oldIndex matches. Set oldIndex to 0 if you want _all_ varnos to be replaced * with rtIndex. * * subQueryLevel gives the outer relation level replace_varnos_for_var is * allowed to change, and so only Var's with varlevelsup values that match * subQueryLevel are changed. * * Returns true in case of success. */ static bool replace_varnos_for_var(Node *node, Index rtIndex, Index oldIndex, Index subQueryLevel) { bool result = false; Assert(PointerIsValid(node)); Assert(rtIndex > 0); if (IsA(node, Var)) { Var *var = (Var *) node; if ((var->varno == oldIndex) && (oldIndex > 0) && (var->varlevelsup == subQueryLevel)) { elog(DEBUG1, "Replacing varno %d to %d in sublevel %d, varattno %d, varlevelsup %d", var->varno, rtIndex, subQueryLevel, var->varattno, var->varlevelsup); var->varno = rtIndex; var->varnoold = rtIndex; result = true; } else if ((oldIndex == 0) && (var->varlevelsup == subQueryLevel)) { elog(DEBUG1, "Replacing varno %d to %d in sublevel %d, varattno %d, varlevelsup %d", var->varno, rtIndex, subQueryLevel, var->varattno, var->varlevelsup); var->varno = rtIndex; var->varnoold = rtIndex; result = true; } } return result; } static void replace_varattno_for_var(Node *node, ViewDefColumnList tentries, Index subQueryLevel) { Assert((PointerIsValid(node)) && !(tentries == NULL)); if (IsA(node, Var)) { Var *var = (Var *) node; TargetEntry *entry; entry = tentries[var->varattno - 1]; if (entry == NULL) return; if (entry->resno != var->varattno && var->varlevelsup == subQueryLevel) { var->varattno = entry->resno; var->varoattno = entry->resno; } } } /* * Walks down the view qualifications node and moves all varnos/varnoold to * point to the *NEW* RTE. */ static bool view_def_qual_walker(Node *node, ViewExprContext *context) { if (!PointerIsValid(node)) return false; replace_varnos_for_var(node, context->newRTE, 0, context->subQueryLevel); replace_varattno_for_var(node, context->tentries, context->subQueryLevel); switch (node->type) { case T_CaseExpr: case T_NullTest: ChangeVarNodes((Node *)node, context->baseRTE, context->newRTE, 0); break; case T_RelabelType: ChangeVarNodes((Node *)((RelabelType *)node)->arg, context->baseRTE, context->newRTE, 0); break; case T_Query: { bool result; /* * Here comes the part where we dive into the subselect query * tree. we need this to detect any outer relation references * in this subquery tree, because they needs to be rewritten to * be pinned at the *NEW* pseudorelation. Note that this is * necessary only for outer references to *NEW* or *OLD*. * * According to the comments in src/include/nodes/primnodes.h, * we can rely on the fact that the subselect node is a Query * */ ++(context->subQueryLevel); result = query_tree_walker((Query *)node, view_def_qual_walker, (void *)context, 0); --(context->subQueryLevel); return result; } default: break; } return expression_tree_walker((Node *)node, view_def_qual_walker, (void *)context); } /* * Rewrite varno's and varattno for the specified Var node if * it is in a reversed order regarding to the * underlying relation. The lookup table tentries * holds all TargetEntries which are on a different * location in the view definition. If var isn't * on a different position in the current view than * on its original relation, nothing is done. * * Note: this function assumes that the caller * has already checked all parameters for NULL. */ static void adjustVarnoIfReversedCol(Var *var, Index newRTE, ViewDefColumnList tentries) { TargetEntry *entry = tentries[var->varattno - 1]; /* * tentries holds NULL if given var isn't * on a different location in the view * Only replace if column order is reversed */ if (entry != NULL) { if (entry->resno != var->varattno) { var->varattno = entry->resno; var->varoattno = entry->resno; } } /* Finally, make varno point to the *NEW* range table entry */ var->varno = newRTE; var->varnoold = newRTE; } /* * Creates an equal operator expression for the specified Vars. They are * assumed to be of the same type. */ static OpExpr * create_opexpr(Var *var_left, Var *var_right) { OpExpr *result; Operator tuple; Form_pg_operator operator; Assert((var_left != NULL) && (var_right != NULL)); if (var_left->vartype != var_right->vartype) elog(ERROR, "left and right var types do not match"); tuple = equality_oper(var_left->vartype, true); if (!HeapTupleIsValid(tuple)) elog(ERROR, "could not find equal-operator for UPDATE/DELETE WHERE condition"); operator = (Form_pg_operator) GETSTRUCT(tuple); result = makeNode(OpExpr); result->opno = HeapTupleGetOid(tuple); result->opfuncid = operator->oprcode; result->opresulttype = operator->oprresult; result->opretset = false; result->args = lappend(result->args, var_left); result->args = lappend(result->args, var_right); ReleaseSysCache(tuple); return result; } /* * Creates an expression tree for a WHERE clause. * * If from is not NULL, assigns the root node to the specified FromExpr of the * target query tree. * * Please note that the function appends the specified opExpr op to * the specified anchor (if anchor != NULL) and returns that immediately. * That way this function could be used to add operator nodes * to an existing BoolExpr tree or (if from is given), to create * a new Query qualification list. */ static Node * build_expression_tree(FromExpr *from, Node **anchor, BoolExpr *expr, OpExpr *op) { /* Already some nodes there? */ if (*anchor != NULL) { expr->args = lappend(expr->args, op); ((BoolExpr *)(*anchor))->args = lappend(((BoolExpr *)(*anchor))->args, expr); *anchor = (Node *)expr; } else { /* Currently no nodes... */ BoolExpr *boolexpr = makeNode(BoolExpr); expr->args = lappend(expr->args, op); boolexpr->args = lappend(boolexpr->args, expr); *anchor = (Node *) boolexpr; if (from != NULL) from->quals = *anchor; } return *anchor; } /* * Forms the WHERE clause for DELETE/UPDATE rules targeted to the specified * view. */ static void form_where_for_updrule(const Query *select, /* View retrieve rule */ FromExpr **from, /* FromExpr for stmt */ const Relation rel, /* base relation of view */ Oid baserel, /* Oid of base relation */ Index baserti, /* Index of base relation RTE */ Index oldrti) /* Index of *OLD* RTE */ { BoolExpr *expr = NULL; Node *anchor = NULL; Form_pg_attribute *attrs = rel->rd_att->attrs; ListCell *cell; AssertArg(baserti > 0); AssertArg(oldrti > 0); AssertArg(OidIsValid(baserel)); AssertArg(*from != NULL); AssertArg(rel != NULL); foreach(cell, select->targetList) { TargetEntry *te = (TargetEntry *) lfirst(cell); Var *var1; Var *var2; OpExpr *op; CaseExpr *newcase; CaseWhen *casewhen; NullTest *nulltest1; NullTest *nulltest2; /* If te->expr holds no Var pointer, continue ... */ if (!IsA(te->expr, Var)) continue; newcase = makeNode(CaseExpr); casewhen = makeNode(CaseWhen); nulltest1 = makeNode(NullTest); nulltest2 = makeNode(NullTest); /* * These are the new operands we had to check for equality. * * For DELETE/UPDATE rules, var1 points to the *OLD* RTE, var2 * references the base relation. */ var1 = copyObject((Var *) (te->expr)); /* * Look at varoattno to determine whether this attribute has a different * location in the underlying base table. If that case, retrieve the * attribute from the base table and assign it to var2; otherwise * simply copy it to var1. */ if (var1->varoattno > 0) { var2 = makeNode(Var); var2->varno = baserti; var2->varnoold = baserti; var2->varattno = attrs[var1->varoattno - 1]->attnum; var2->vartype = attrs[var1->varoattno - 1]->atttypid; var2->vartypmod = attrs[var1->varoattno - 1]->atttypmod; var2->varlevelsup = var1->varlevelsup; var2->varnoold = var2->varno; var2->varoattno = var2->varattno; } else { var2 = copyObject(var1); var2->varno = baserti; var2->varnoold = baserti; } var1->varno = oldrti; var1->varnoold = oldrti; /* * rewrite varattno of var2 to point to the right column in relation * *OLD* or *NEW* */ var2->varattno = te->resorigcol; var2->varoattno = te->resorigcol; /* * rewrite varattno of var1 to point to the right column in base * relation */ var1->varattno = te->resno; var1->varoattno = te->resno; op = create_opexpr(var1, var2); expr = makeNode(BoolExpr); expr->boolop = AND_EXPR; /* * Finally, create the OpExpr node, as part of a CaseExpr and include * the OpExpr as part of the Case for managing NULL's we will do this * everytime. That way we will have no problem with: * * ALTER TABLE ... ALTER COLUMN ... DROP NOT NULL; */ nulltest1->arg = (Expr *)var1; nulltest1->nulltesttype = IS_NOT_NULL; nulltest2->arg = (Expr *)var2; nulltest2->nulltesttype = IS_NULL; casewhen->expr = (Expr *)nulltest1; casewhen->result = (Expr *)op; newcase->args = list_make1(casewhen); newcase->defresult = (Expr *) nulltest2; op = copyObject(newcase); anchor = build_expression_tree(*from, (Node **) &anchor, expr, op); } } /* * Replaces the varnos for the specified targetlist to rtIndex */ static bool replace_tlist_varno_walker(Node *node, ViewExprContext *ctxt) { AssertArg(PointerIsValid(ctxt)); if (!PointerIsValid(node)) return false; switch(node->type) { case T_Var: { elog(DEBUG1, "adjust varno old %d to new %d", ((Var *)(node))->varno, ctxt->newRTE); ((Var *)(node))->varno = ctxt->newRTE; adjustVarnoIfReversedCol((Var *)node, ctxt->newRTE, ctxt->tentries); /* nothing more to do */ break; } case T_ArrayRef: { ArrayRef *array = (ArrayRef *) node; /* * Things are getting complicated here. We have found an array * subscripting operation. It's necessary to examine all varno's * found in this operation to make sure, we're getting right. This * covers cases where a view selects a single index or complete * array from a base table or view. */ /* * Look at expressions that evaluate upper array indexes. Make * sure all varno's are modified. This is done by walking the * expression tree recursively. */ expression_tree_walker((Node *) array->refupperindexpr, replace_tlist_varno_walker, (void *)ctxt); expression_tree_walker((Node *) array->reflowerindexpr, replace_tlist_varno_walker, (void *)ctxt); expression_tree_walker((Node *) array->refexpr, replace_tlist_varno_walker, (void *)ctxt); expression_tree_walker((Node *) array->refassgnexpr, replace_tlist_varno_walker, (void *)ctxt); } default: break; } return expression_tree_walker(node, replace_tlist_varno_walker, ctxt); } /* * Adds RTEs to form a query tree. * * select has to be a valid initialized view definition query tree (the * function assumes that this query has passed the checkTree() function). * * The function returns true if succesful, false otherwise. */ static bool form_query(const Query *select, Query *query, ViewDefColumnList tentries, bool copyTargetList) { RangeTblEntry *rte; ListCell *cell; Oid reloid; AssertArg(select != NULL); AssertArg(query != NULL); AssertArg(tentries != NULL); /* Copy the range table entries */ query->rtable = copyObject(select->rtable); /* Prepare other stuff */ query->canSetTag = true; query->jointree = makeNode(FromExpr); /* * Those entries in the range table which have the inFromCl flag set must be * changed. */ foreach(cell, query->rtable) { Node *node = (Node *) lfirst(cell); Assert(IsA(node, RangeTblEntry)); ((RangeTblEntry *)node)->inFromCl = false; ((RangeTblEntry *)node)->inh = false; } /* * Set result relation to the base relation. * * Since we currently only support SQL92 views, we simply extract the one * relation which isn't labeled as *OLD* or *NEW*. */ reloid = get_reloid_from_select(select, &(query->resultRelation), &rte); if (!OidIsValid(reloid)) elog(ERROR, "couldn't retrieve base relation OID"); Assert(query->resultRelation > 0); if (copyTargetList) { ViewExprContext ctxt; ListCell *cell; /* Copy all target entries. */ query->targetList = copyObject(select->targetList); /* * Replace all varnos to point to the *NEW* node in all targetentry * expressions. */ ctxt.newRTE = PRS2_NEW_VARNO; ctxt.tentries = tentries; foreach(cell, query->targetList) { Node *node = (Node *) lfirst(cell); expression_tree_walker(node, replace_tlist_varno_walker, (void *) &ctxt); } } return true; } /* * Rewrite a TargetEntry, based on the given arguments to match * the new Query tree of the new DELETE/UPDATE/INSERT rule and/or * its underlying base relation. * * form_te_for_update() needs to carefully reassign Varno's of * all Var expressions assigned to the given TargetEntry and to * adjust all type info values and attribute index locations so * that the rewritten TargetEntry corresponds to the correct * column in the underlying base relation. * * Someone should consider that columns could be in reversed * order in a view definition, so we need to take care to * "restore" the correct order of all columns in the target list * of the new view update rules. * * There's also some additional overhead if we have an array field * involved. In this case we have to loop recursively through the * array expressions to get all target entries right. */ static void form_te_for_update(int2 attnum, Form_pg_attribute attrs, Oid baserel, Expr *expr, TargetEntry *te_update) { /* * First, try if this is an array subscripting operation. If true, dive * recursively into the subscripting tree examining all varnos. */ if (IsA(expr, ArrayRef)) { ArrayRef *array = (ArrayRef *) expr; if (array->refassgnexpr != NULL) { form_te_for_update(attnum, attrs, baserel, array->refassgnexpr, te_update); } if (array->refupperindexpr != NIL) { ListCell *cell; foreach(cell, array->refupperindexpr) { Expr *subexpr = (Expr *) lfirst(cell); form_te_for_update(attnum, attrs, baserel, subexpr, te_update); } } if (array->reflowerindexpr != NIL) { ListCell *cell; foreach(cell, array->reflowerindexpr) { Expr *subexpr = (Expr *)lfirst(cell); form_te_for_update(attnum, attrs, baserel, subexpr, te_update); } } if (array->refexpr != NULL) { form_te_for_update(attnum, attrs, baserel, array->refexpr, te_update); } } else if (IsA(expr, Var)) { /* * Base case of recursion: actually build the TargetEntry. */ Var *upd_var = (Var *) (te_update->expr); upd_var->varattno = te_update->resno; upd_var->varoattno = te_update->resno; upd_var->vartype = attrs->atttypid; upd_var->vartypmod = attrs->atttypmod; upd_var->varnoold = upd_var->varno; te_update->resno = attnum; te_update->resname = pstrdup(get_attname(baserel, attnum)); te_update->ressortgroupref = 0; te_update->resorigcol = 0; te_update->resorigtbl = 0; te_update->resjunk = false; } } /* * Build the target list for a view UPDATE rule. * * Note: The function assumes a Query tree specified by update, * which was copied by form_query(). We need the original * Query tree to adjust the properties of each member * of the TargetList of the new query tree. */ static void build_update_target_list(const Query *update, const Query *select, ViewDefColumnList tentries, Oid baserel, Relation rel) { ListCell *cell1; ListCell *cell2; /* * This Assert() is needed, since we rely on a query tree * created by from_query() which copies the target list from * the original query tree specified by the argument select, * which holds the current view definition. * So both target lists have to be equal, indeed. */ Assert(list_length(update->targetList) == list_length(select->targetList)); forboth(cell1, select->targetList, cell2, update->targetList) { TargetEntry *entry = (TargetEntry *) lfirst(cell1); TargetEntry *upd_entry = (TargetEntry *) lfirst(cell2); int attindex; Form_pg_attribute attr; if (entry->resorigcol > 0) { /* * This column seems to have a different order than in the base * table. We get the attribute from the base relation referenced * by rel and create a new resdom. This new result domain is then * assigned instead of the old one. */ attindex = entry->resorigcol; } else attindex = entry->resno; attr = rel->rd_att->attrs[attindex - 1]; form_te_for_update(attindex, attr, baserel, upd_entry->expr, upd_entry); } } /* * Returns the pg_proc tuple for the view definition qualification check * function. Caller must heap_freetuple the result when done with it. * * An error is raised if the function is not found. */ static HeapTuple get_view_qualification_func(Oid procoid) { HeapTuple procTuple; Assert(procoid != InvalidOid); procTuple = SearchSysCacheCopy(PROCOID, ObjectIdGetDatum(procoid), 0, 0, 0); if (!HeapTupleIsValid(procTuple)) elog(ERROR, "cache lookup failure for function with OID %d", procoid); return procTuple; } /* * Create a rule qualification from the specified view definition query tree. * * This qualification is copied to the stmt's ev_action query tree and * evaluated to check a view definition's qualification. We get a WITH CHECK * OPTION behavior by creating a conditional rule here. The view definition * qualification is evaluated by the PG_VIEW_UPDATE_CONTROL function. * * The function returns the function node or null. */ static FuncExpr * create_qual_for_rule(RuleStmt *stmt, const Query *select, ViewDefColumnList tentries, Index newRTE, Index oldRTE) { FuncExpr *func = makeNode(FuncExpr); Index baseRTE = 0; HeapTuple proc; ViewExprContext context; Assert(stmt != NULL); Assert(list_length(stmt->actions) >= 1); Assert(select != NULL); Assert((newRTE > 0) && (oldRTE > 0)); if (select->jointree->quals == NULL) { /* nothing to do here */ return NULL; } /* Retrieve the function. */ proc = get_view_qualification_func(PG_VIEW_UPDATE_CONTROL); /* Create the FuncExpr node for evaluation */ func->funcid = HeapTupleGetOid(proc); func->funcresulttype = ((Form_pg_proc) GETSTRUCT(proc))->prorettype; func->funcretset = false; func->funcformat = 0; func->args = list_make1(copyObject(select->jointree->quals)); /* Move the view qualification over to the RuleStmt */ elog(DEBUG1, "previous WHERE clause was %s", nodeToString(stmt->whereClause)); stmt->whereClause = (Node *)func; /* * We have to dig into the WHERE clause and grep out all varnos that have * to point to the *NEW* or *OLD* RTE. */ baseRTE = 2; get_relation_RTE(select, &baseRTE); Assert(baseRTE > 2); context.newRTE = newRTE; context.baseRTE = baseRTE - 1; context.oldRTE = oldRTE; context.subQueryLevel = 0; context.tentries = tentries; expression_tree_walker((Node *) func, view_def_qual_walker, (void *) &context); /* free stuff */ heap_freetuple(proc); return func; } static RuleStmt * create_rule_stmt(Query *actions, const RangeVar *view, bool replace, char *rulename, CmdType event) { RuleStmt *rule = makeNode(RuleStmt); rule->relation = copyObject((RangeVar *) view); rule->rulename = pstrdup(rulename); rule->whereClause = NULL; rule->event = event; rule->instead = true; rule->actions = list_make1(actions); rule->replace = replace; return rule; } static bool hasRule(Oid view, const char *name) { Assert(OidIsValid(view)); return OidIsValid(GetSysCacheOid(RULERELNAME, ObjectIdGetDatum(view), PointerGetDatum(name), 0, 0)); } /* * Check if we can create the insert rule for this view, i.e. if we can create * DEFAULT values for all columns that need it. * * Up to now the only reason I can see to not create the rule is if the view * does not contain all not null without default fields of the relation. */ static bool viewIsInsertable(const Query *select, const Relation rel) { int2 natts = RelationGetNumberOfAttributes(rel); int num_sys_cols = 0; bool attr_ok[natts]; ListCell *cell; int i; /* Initialize attr_ok array with false values */ MemSet(attr_ok, 0, sizeof(attr_ok)); /* * Loop over the targetlist of the querytree and mark the table attributes * that are present in the view definition. */ foreach(cell, select->targetList) { TargetEntry *entry = (TargetEntry *) lfirst(cell); if (!IsA(entry->expr, Var) && !IsA(entry->expr, ArrayRef)) { /* * Count non-Var expressions. We need to check if there are * 'normal' table columns left. */ num_sys_cols++; } else if (entry->resorigcol == 0) { /* This is not a base rel attribute, it will be ignored. */ continue; } else if (entry->resorigcol <= 0) { /* * If resorigcol < 0 then this is a system attribute and will be * ignored. */ num_sys_cols++; continue; } /* * Mark the attribute as present in the view definition. */ attr_ok[entry->resorigcol - 1] = true; } /* * Loop over table attributes to check whether they were all marked in the * previous loop. */ for (i = 0; i < natts; i++) { /* Ignore dropped columns */ if (rel->rd_att->attrs[i]->attisdropped) continue; /* If the column is in the view definition, accept it */ if (attr_ok[i]) continue; /* * If there is a NOT NULL attribute which is not in the view definition * and doesn't have a DEFAULT value, then the view is not insertable. */ if (rel->rd_att->attrs[i]->attnotnull && !rel->rd_att->attrs[i]->atthasdef) return false; } /* * Finally, check if we have an empty * target list. */ if (list_length(select->targetList) - num_sys_cols <= 0) return false; return true; } /* * Examines the columns by the current view and initializes the lookup table * for all rearranged columns in base relations. The function requires a * relation tree initialized by get_base_base_relations(). */ static void read_rearranged_cols(ViewBaseRelation *tree) { Assert((tree != NULL)); if (tree->defs != NIL) { int num_items = list_length(tree->defs); int i; /* * Traverse the relation tree and look on all base relations for * reversed column order in their target lists. We have to perform a * look-ahead-read on the tree, because we need to know how much * columns the next base relation has, to allocate enough memory in * tentries. * * Note that if we only have one base relation (a "real" table, not a * view) exists, we have nothing to do, because this base relation * cannot have a reversed column order caused by a view definition * query. */ for (i = (num_items - 1); i > 0; --i) { ViewBaseRelationItem *item_current; ViewBaseRelationItem *item_next; ViewBaseRelation *current; ViewBaseRelation *next; current = (ViewBaseRelation *) list_nth(tree->defs, i); /* * We look ahead for the next base relation. We can do this here * safely, because the loop condition terminates before reaching * the list head. */ next = (ViewBaseRelation *) list_nth(tree->defs, i - 1); /* * Note that the code currently requires a SQL92 only relation * tree. This means we handle one base relation per loop, only. */ Assert(next != NULL); Assert(current != NULL); Assert(list_length(next->defs) == 1); Assert(list_length(current->defs) == 1); item_current = (ViewBaseRelationItem *) list_nth(current->defs, 0); item_next = (ViewBaseRelationItem *) list_nth(next->defs, 0); /* Allocate tentries buffer */ item_current->tentries = (ViewDefColumnList) palloc(sizeof(TargetEntry *) * RelationGetNumberOfAttributes(item_next->rel)); copyReversedTargetEntryPtr(item_current->rule->targetList, item_current->tentries); } } } /* * Returns the base table(s) for the specified relation OID. The result is a * list of all possible base table(s) the given view is based on. */ static void get_base_relations(ViewBaseRelation *tree, List **baserelations) { ListCell *acell; /* nothing to do? */ if (tree == NULL || tree->defs == NIL) return; foreach(acell, tree->defs) { ListCell *bcell; ViewBaseRelation *relations = (ViewBaseRelation *) lfirst(acell); /* current children holds a base table? */ foreach(bcell, relations->defs) { ViewBaseRelationItem *item = (ViewBaseRelationItem *) lfirst(bcell); if (item->rel->rd_rel->relkind != RELKIND_VIEW) { elog(DEBUG1, "Found base relation %s", RelationGetRelationName(item->rel)); *baserelations = lappend(*baserelations, item); } } } } /*------------------------------------------------------------------------------ * Retrieves all relations from the view that can be considered a "base * relation". The function returns a list that holds lists of all relation * OIDs found for the view. The list is filled top down, that means the head of * the list holds the relations for the "highest" view in the tree. * * Consider this view definition tree where each node is a relation the above * node is based on: * * 1 * / \ * 2 3 * / \ \ * 4 5 6 * / * 7 * * The function will then return a list with the following layout: * * Listindex Node(s) * -------------------------- * 1 7 * 2 4 5 6 * 3 2 3 * * As you can see in the table, all relations that are "childs" of the * given root relation (the view relation itself) are saved in the * tree, except the root node itself. *------------------------------------------------------------------------------ */ static void get_base_base_relations(const Query *view, Oid baserel, List **list) { RangeTblEntry *entry; unsigned int offset = 1; ViewBaseRelation *childRel; if (view == NULL) return; childRel = (ViewBaseRelation *) palloc(sizeof(ViewBaseRelation)); childRel->defs = NIL; childRel->parentRelation = baserel; /* Get all OIDs from the RTE list of view. */ while ((entry = get_relation_RTE(view, &offset)) != NULL) { Relation rel; ViewBaseRelationItem *item; /* * Is this really a view or relation? * * XXX -- maybe we don't need the lock here. It may be actively * dangerous, if somewhere else we acquire a stronger lock later ... */ rel = relation_open(entry->relid, AccessShareLock); if (rel->rd_rel->relkind != RELKIND_RELATION && rel->rd_rel->relkind != RELKIND_VIEW) { /* don't need this one */ relation_close(rel, AccessShareLock); continue; } item = (ViewBaseRelationItem *) palloc0(sizeof(ViewBaseRelationItem)); item->rel = rel; if (rel->rd_rel->relkind == RELKIND_VIEW) { /* * Get the rule _RETURN expression tree for the specified relation * OID. We need this to recurse into the view base relation tree. */ item->rule = get_return_rule(rel); /* recurse to child relations */ if (item->rule != NULL) get_base_base_relations(item->rule, RelationGetRelid(rel), list); } childRel->defs = lappend(childRel->defs, item); elog(DEBUG1, "extracted relation %s for relation tree", RelationGetRelationName(rel)); } if (childRel->defs != NIL) *list = lappend(*list, childRel); } static void copyReversedTargetEntryPtr(List *targetList, ViewDefColumnList targets) { ListCell *cell; AssertArg(targets != NULL); AssertArg(targetList != NIL); /* NOTE: we only reassign pointers. */ foreach(cell, targetList) { Node *node = (Node *) lfirst(cell); if (IsA(node, TargetEntry)) { /* * Look at the resdom's resorigcol to determine whether this is a * reversed column (meaning, it has a different column number than * the underlying base table). */ TargetEntry *entry = (TargetEntry *) node; if (!IsA(entry->expr, Var)) /* nothing to do here */ continue; if ((entry->resorigcol > 0) && (entry->resno != entry->resorigcol)) { /* * Save this TargetEntry to the appropiate place in the lookup * table. Do it only if not already occupied (this could * happen if the column is specified more than one time in the * view definition). */ if (targets[entry->resorigcol - 1] == NULL) targets[entry->resorigcol - 1] = entry; } } } } /* * Transforms the specified view definition into a DELETE rule. * * Note: The function assumes that the specified query tree has passed the * checkTree() function. */ static Query * transform_select_to_delete(const Query *select, const Relation rel, const RangeVar *var, ViewDefColumnList tentries, bool checkOption, bool checkCascade) { Query *gen_delete = NULL; /* only needed when checkOption = true */ Query *delete; RuleStmt *nothing; RuleStmt *rule; Oid baserel; bool replace_nothing; bool replace_delete; Index baserti; char option; RangeTblRef *oldref; RangeTblRef *baseref; RangeTblEntry *entry; RangeTblEntry *viewrte; AssertArg(tentries != NULL); AssertArg(rel != NULL); AssertArg(var != NULL); AssertArg(select != NULL); baserel = get_reloid_from_select(select, NULL, &entry); if (!OidIsValid(baserel)) elog(ERROR, "could not get the base relation from the view definition"); delete = makeNode(Query); delete->commandType = CMD_DELETE; /* We don't need a targetlist in DELETE */ if (!form_query(select, delete, tentries, false)) { elog(DEBUG1, "could not create DELETE rule"); return NULL; } /* * Pass this generic tree to create_rule_stmt(), it's sufficient to * create a INSTEAD DO NOTHING rule for deletion. * * This is required, because we can get conditional rules if the * checkOption is specified. We only save this generic tree here, the * work is done at the end of this function. * * Note: Do this only if the check option is specified. */ if (checkOption) { gen_delete = copyObject(delete); gen_delete->commandType = CMD_NOTHING; } /* * form_query() has prepared the jointree of the new DELETE rule. * However, a DELETE rule needs range table references to *OLD* and * base relation RTEs. */ baserti = get_rtindex_for_rel(delete->rtable, entry->eref->aliasname); Assert(baserti > 0); oldref = makeNode(RangeTblRef); oldref->rtindex = PRS2_OLD_VARNO; baseref = makeNode(RangeTblRef); baseref->rtindex = baserti; delete->jointree->fromlist = list_make2(baseref, oldref); /* Create the WHERE condition qualification for the rule action. */ form_where_for_updrule(select, &(delete->jointree), rel, baserel, baserti, PRS2_OLD_VARNO); /* * Get the relation name for the view * * XXX - I think it's safe here to rely on *NEW*, since * the rewriter itself makes heavily use of this * pseudorelation and it needs to be present in all * rule events. */ viewrte = rt_fetch(PRS2_NEW_VARNO, delete->rtable); /* Set ACL bit */ entry->requiredPerms |= ACL_DELETE; /* * Check out wether we need to replace any existing implicit * rules. This is required if we are going to replace an existing * view with implicit update rules. */ /* DELETE rule has to be replaced? */ replace_delete = hasRule(viewrte->relid, DELETERULENAME); /* NOTHING rule has to be replaced? */ replace_nothing = hasRule(viewrte->relid, NOTHING_DELETERULENAME); /* create system DELETE rule */ rule = create_rule_stmt(delete, var, replace_delete, DELETERULENAME, CMD_DELETE); if (checkOption) { /* Don't create INSTEAD DO NOTHING rule if not necessary */ nothing = create_rule_stmt(gen_delete, var, replace_nothing, NOTHING_DELETERULENAME, CMD_DELETE); DefineQueryRewrite(nothing, NO_OPTION_NOTHING_RULE); /* * Enter rule qualification creation. The DELETE rule has to reference * the *OLD* range table entry. */ create_qual_for_rule(rule, select, tentries, PRS2_OLD_VARNO, baserti); } option = makeViewCheckOption(checkOption, checkCascade); DefineQueryRewrite(rule, option); return delete; } /* * Transforms the specified SELECT query tree into an equivalent UPDATE * statement. * * Note: The function assumes that the specified query tree has passed the * checkTree() function before. */ static Query * transform_select_to_update(const Query *select, const Relation rel, const RangeVar *var, ViewDefColumnList tentries, bool checkOption, bool checkCascade) { Query *gen_update = NULL; /* only needed when checkOption = true */ Query *update; Oid baserel; RuleStmt *nothing; RuleStmt *rule; bool replace_update; bool replace_nothing; Index baserti; char option; RangeTblRef *oldref; RangeTblRef *baseref; RangeTblEntry *entry; RangeTblEntry *viewrte; AssertArg(tentries != NULL); AssertArg(rel != NULL); AssertArg(var != NULL); AssertArg(select != NULL); baserel = get_reloid_from_select(select, NULL, &entry); if (!OidIsValid(baserel)) elog(ERROR, "could not get the base relation from the view definition"); update = makeNode(Query); update->commandType = CMD_UPDATE; if (!form_query(select, update, tentries, true)) { elog(DEBUG1, "could not create UPDATE update rule"); return NULL; } /* * Our UPDATE rule needs range table references for the *NEW* and base * relation. * * Note: the jointree in the UPDATE tree is already prepared, we only * had to fill the fromlist list. */ baserti = get_rtindex_for_rel(update->rtable, entry->eref->aliasname); oldref = makeNode(RangeTblRef); baseref = makeNode(RangeTblRef); oldref->rtindex = PRS2_OLD_VARNO; baseref->rtindex = baserti; update->jointree->fromlist = list_make2(baseref, oldref); /* Create the WHERE condition qualification for the new rule */ form_where_for_updrule(select, &(update->jointree), rel, baserel, baserti, PRS2_OLD_VARNO); /* * We must reorder the columns in the targetlist to match the * underlying table. We do this after calling form_where_for_updrule() * because build_update_target_list() relies on the original resdoms in * the update tree. */ build_update_target_list(update, select, tentries, baserel, rel); /* * Pass this generic tree to create_rule_stmt(), it's sufficient to * create an INSTEAD DO NOTHING rule for deletion. * * This is required, because we can get conditional rules if the * checkOption is specified. We only save this generic tree here, the * work is done at the end of this function. * * Note: Do this only if the check option is specified. */ if (checkOption) { gen_update = copyObject(update); gen_update->commandType = CMD_NOTHING; } /* Get the relation name for the view */ viewrte = rt_fetch(PRS2_NEW_VARNO, update->rtable); /* Set ACL bit */ entry->requiredPerms |= ACL_UPDATE; /* * Check out wether we need to replace any existing implicit * rules. This is required if we are going to replace an existing * view with implicit update rules. */ /* UPDATE rule has to be replaced? */ replace_update = hasRule(viewrte->relid, UPDATERULENAME); /* NOTHING rule has to be replaced? */ replace_nothing = hasRule(viewrte->relid, NOTHING_UPDATERULENAME); rule = create_rule_stmt(update, var, replace_update, UPDATERULENAME, CMD_UPDATE); /* Ready to create the rule */ if (checkOption) { /* Don't create INSTEAD DO NOTHING rule if not necessary */ nothing = create_rule_stmt(gen_update, var, replace_nothing, NOTHING_UPDATERULENAME, CMD_UPDATE); DefineQueryRewrite(nothing, NO_OPTION_NOTHING_RULE); /* * Enter rule qualification creation. The UPDATE rule has to reference * the *OLD* range table entry. */ create_qual_for_rule(rule, select, tentries, PRS2_NEW_VARNO, baserti); } option = makeViewCheckOption(checkOption, checkCascade); DefineQueryRewrite(rule, option); return update; } /* * Transforms the specified SELECT query tree into an equivalent INSERT * statement. * * Note: The function assumes that the specified query tree has passed the * checkTree() function before. */ static Query * transform_select_to_insert(const Query *select, const Relation rel, const RangeVar *var, ViewDefColumnList tentries, bool checkOption, bool checkCascade) { Query *gen_insert = NULL; /* only needed when checkOption = true */ Query *insert; Oid baserel; Index baserti; char option; bool replace_insert; RuleStmt *rule; bool replace_nothing; RuleStmt *nothing; RangeTblEntry *viewrte; RangeTblEntry *entry; AssertArg(select != NULL); AssertArg(rel != NULL); AssertArg(var != NULL); AssertArg(tentries != NULL); baserel = get_reloid_from_select(select, NULL, &entry); if (!OidIsValid(baserel)) elog(ERROR, "could not get the base relation from the view definition"); option = makeViewCheckOption(checkOption, checkCascade); /* Check if we can create the insert rule for this view. */ if (!viewIsInsertable(select, rel)) { elog(DEBUG1, "Cannot create implicit insert rule for this view"); return NULL; } insert = makeNode(Query); insert->commandType = CMD_INSERT; if (!form_query(select, insert, tentries, true)) { elog(DEBUG1, "could not create INSERT update rule"); return NULL; } /* * Pass this generic tree to create_rule_stmt(), it's sufficient to * create a INSTEAD DO NOTHING rule. The INSTEAD DO NOTHING rule is * required, because we can get conditional rules if the checkOption is * specified. * * Note: Do this only if the check option is specified. */ if (checkOption) { gen_insert = copyObject(insert); gen_insert->commandType = CMD_NOTHING; } baserti = get_rtindex_for_rel(insert->rtable, entry->eref->aliasname); viewrte = rt_fetch(PRS2_NEW_VARNO, insert->rtable); /* * We must reorder the columns in the targetlist to match the underlying * table. */ build_update_target_list(insert, select, tentries, baserel, rel); /* Set the needed ACL bit */ entry->requiredPerms |= ACL_INSERT; /* * Check out wether we need to replace any existing implicit * rules. This is required if we are going to replace an existing * view with implicit update rules. */ /* INSERT rule has to be replaced? */ replace_insert = hasRule(viewrte->relid, INSERTRULENAME); /* NOTHING rule has to be replaced? */ replace_nothing = hasRule(viewrte->relid, NOTHING_INSERTRULENAME); rule = create_rule_stmt(insert, var, replace_insert, INSERTRULENAME, CMD_INSERT); if (checkOption) { /* Ready to create the rules */ nothing = create_rule_stmt(gen_insert, var, replace_nothing, NOTHING_INSERTRULENAME, CMD_INSERT); DefineQueryRewrite(nothing, NO_OPTION_NOTHING_RULE); /* * enter rule qualification creation */ create_qual_for_rule(rule, select, tentries, PRS2_NEW_VARNO, baserti); } DefineQueryRewrite(rule, option); return insert; } /* * Checks the specified Query for SQL92 compliance. * * Returns FALSE if the specified SELECT query tree does not pass the SQL92 * requirements, otherwise TRUE is returned. * */ static bool checkTree(const Query *query, ViewBaseRelation *tree) { ListCell *cell; AssertArg(query != NULL); AssertArg(tree != NULL); Assert(query->commandType == CMD_SELECT); if (!check_reltree(tree)) { elog(DEBUG1, "possible JOIN/UNION in relations found in view definition"); return false; } /* Check for unsupported conditions in the view definition */ if (query->hasAggs) { elog(DEBUG1, "aggregates are not supported in updatable views"); return false; } if (list_length(query->groupClause) >= 1) { elog(DEBUG1, "GROUP BY violates SQL92 view update rules"); return false; } if (list_length(query->distinctClause) >= 1) { elog(DEBUG1, "DISTINCT violates SQL92 view update rules"); return false; } if (query->havingQual != NULL) { elog(DEBUG1, "HAVING violates SQL92 view update rules"); return false; } /* * Test for number of involved relations. Since we assume to operate on a * view definition SELECT query tree, we must count 3 rtable entries. * Otherwise this seems not to be a view based on a single relation. */ if (list_length(query->rtable) > 3) { elog(DEBUG1, "view seems to have more than one base relation"); return false; } /* Any rtable entries involved? */ if (list_length(query->rtable) < 3) { elog(DEBUG1, "no base relation detected"); return false; } /* * Walk down the target list and look for nodes that aren't Vars. SQL92 * doesn't allow functions, hostvariables or constant expressions in the * target list. * * Also, check if any of the target list entries are indexed array * expressions, which aren't supported. */ foreach(cell, query->targetList) { Node *node = (Node *) lfirst(cell); if (IsA(node, TargetEntry)) { TargetEntry *te = (TargetEntry *) node; /* * TODO -- it would be nice to support Const nodes here as well * (but apparently it isn't in the standard) */ if (!IsA(te->expr, Var) && !IsA(te->expr, ArrayRef)) { elog(DEBUG1, "view target list has unsupported column entries"); return false; } /* This is an implementation shortcoming */ if (IsA(te->expr, ArrayRef)) { ArrayRef *ref = (ArrayRef *) te->expr; if (ref->refupperindexpr != NIL) { elog(DEBUG1, "indexed array fields aren't updatable"); return false; } } } } /* * Finally, check that all RTEs are valid. We have to look especially for * table functions, which cannot be ever updatable. */ foreach(cell, query->rtable) { RangeTblEntry *entry = (RangeTblEntry *) lfirst(cell); if (entry->rtekind != RTE_RELATION) { elog(DEBUG1, "range table entry %s is not a relation!", entry->eref->aliasname); return false; } } return true; } /* * Traverse the specified relation tree. The function stops at the base * relations at the leafs of the tree. If any of the relations has more than * one base relations, it is considered as a non-SQL92 updatable view and FALSE * is returned. * */ static bool check_reltree(ViewBaseRelation *node) { ListCell *cell; Assert(node != NULL); foreach(cell, node->defs) { /* Walk down the tree */ ViewBaseRelation *relations = (ViewBaseRelation *) lfirst(cell); if (list_length(relations->defs) > 1) return false; } return true; } /* * Given a SELECT query tree, return the OID of the first RTE_RELATION range * table entry found that is not *NEW* nor *OLD*. * * Also copies the RangeTblEntry into rel_entry, and the range table index * into rti, unless they are NULL. * * This function assumes that the specified query tree was checked by a * previous call to the checkTree() function. */ static Oid get_reloid_from_select(const Query *select, int *rti, RangeTblEntry **rel_entry) { ListCell *cell; Oid result = InvalidOid; int index; /* Check specified query tree. Return immediately on error. */ if (select == NULL || select->commandType != CMD_SELECT) return InvalidOid; /* * We loop through the RTEs to get information about all involved * relations. We return the first OID we find in the list that is not * *NEW* nor *OLD*. */ index = 0; foreach(cell, select->rtable) { RangeTblEntry *entry = (RangeTblEntry *) lfirst(cell); index++; if (entry == NULL) elog(ERROR, "null RTE pointer in get_reloid_from_select"); elog(DEBUG1, "extracted range table entry for %u", entry->relid); /* Return the first RELATION rte we find */ if (entry->rtekind == RTE_RELATION) { /* * XXX This is ugly. The parser prepends two RTEs with rtekind * RTE_RELATION named *NEW* and *OLD*. We have to exclude them by * name! It would be much better if it used RTE_SPECIAL * instead, but other parts of the system stop working if one * just changes it naively. */ if (strncmp(entry->eref->aliasname, "*NEW*", 6) == 0 || strncmp(entry->eref->aliasname, "*OLD*", 6) == 0) continue; result = entry->relid; if (rti != NULL) *rti = index; if (rel_entry != NULL) *rel_entry = copyObject(entry); break; } } return result; } /* * get_return_rule: returns the _RETURN rule of a view as a Query node. */ static Query * get_return_rule(Relation rel) { Query *query = NULL; int i; Assert(rel->rd_rel->relkind == RELKIND_VIEW); for (i = 0; i < rel->rd_rules->numLocks; i++) { RewriteRule *rule = rel->rd_rules->rules[i]; if (rule->event == CMD_SELECT) { /* A _RETURN rule should have only one action */ if (list_length(rule->actions) != 1) elog(ERROR, "invalid _RETURN rule action specification"); query = linitial(rule->actions); break; } } return query; } /*------------------------------------------------------------------------------ * Public functions *------------------------------------------------------------------------------ */ /* * CreateViewUpdateRules * * This is the main entry point to creating an updatable view's rules. Given a * rule definition, examine it, and create the rules if appropiate, or return * doing nothing if not. */ void CreateViewUpdateRules(const Query *viewDef, const RangeVar *var, bool checkOption, bool cascade) { Relation rel; Form_pg_attribute *attrs; ViewDefColumnList tentries; Oid baserel; MemoryContext cxt; MemoryContext oldcxt; ViewBaseRelation *tree; List *baserelations; ListCell *cell; /* * The routines in this file leak memory like crazy, so make sure we * allocate it all in an appropiate context. */ cxt = AllocSetContextCreate(TopTransactionContext, "UpdateRulesContext", ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE); oldcxt = MemoryContextSwitchTo(cxt); /* * Create the lookup table for the view definition target columns. We save * the RESDOMS in that manner to look quickly for reversed column orders. */ baserel = get_reloid_from_select(viewDef, NULL, NULL); /* Get relation tree */ tree = (ViewBaseRelation *) palloc(sizeof(ViewBaseRelation)); tree->parentRelation = InvalidOid; tree->defs = NIL; get_base_base_relations(viewDef, baserel, &(tree->defs)); baserelations = NIL; get_base_relations(tree, &baserelations); /* Check the query tree for SQL92 compliance */ if (!checkTree(viewDef, tree)) { elog(DEBUG1, "view is not updatable"); goto finish; } rel = heap_open(baserel, AccessExclusiveLock); attrs = rel->rd_att->attrs; /* * Copy TargetEntries to match the slot numbers in the target list with * their original column attribute number. Note that only pointers are * copied and they are valid only as long as the specified SELECT query * stays valid! */ tentries = (ViewDefColumnList) palloc0(rel->rd_rel->relnatts * sizeof(TargetEntry *)); copyReversedTargetEntryPtr(viewDef->targetList, tentries); /* * Now do the same for the base relation tree. read_rearranged_cols * traverses the relation tree and performs a copyReversedTargetEntry() * call to each base relation. */ read_rearranged_cols(tree); /* * Create update rules. Need a CCI here to see previously-created * rules. XXX Maybe this CCI should be elsewhere. */ CommandCounterIncrement(); elog(NOTICE, "CREATE VIEW will create implicit INSERT/UPDATE/DELETE rules"); transform_select_to_insert(viewDef, rel, var, tentries, checkOption, cascade); transform_select_to_delete(viewDef, rel, var, tentries, checkOption, cascade); transform_select_to_update(viewDef, rel, var, tentries, checkOption, cascade); /* free remaining stuff */ heap_close(rel, AccessExclusiveLock); finish: /* get_base_base_relations leaves some open relations */ foreach(cell, tree->defs) { ListCell *cell2; ViewBaseRelation *vbr = (ViewBaseRelation *) lfirst(cell); foreach(cell2, vbr->defs) { ViewBaseRelationItem *vbri = (ViewBaseRelationItem *) lfirst(cell2); /* XXX should we keep the locks here? */ relation_close(vbri->rel, AccessShareLock); } } MemoryContextSwitchTo(oldcxt); MemoryContextDelete(cxt); } char makeViewCheckOption(bool check, bool cascaded) { char result = NO_OPTION_IMPLICIT; if (check) { if (cascaded) result = CASCADED_OPTION_IMPLICIT; else result = LOCAL_OPTION_IMPLICIT; } return result; } /*------------------------------------------------------------------------------ * Required for update rule CHECK OPTION *------------------------------------------------------------------------------ */ Datum pg_view_update_error(PG_FUNCTION_ARGS) { bool qual = PG_GETARG_BOOL(0); if (PG_ARGISNULL(0)) { /* * A view's CHECK OPTION should never evaluate to NULL, * so we treat a NULL evaluated expression tree as false. */ qual = false; } if (!qual) ereport(ERROR, (errcode(ERRCODE_WITH_CHECK_OPTION_VIOLATION), errmsg("view update commands violates rule condition"))); PG_RETURN_BOOL(true); }