granicus.if.org Git - postgresql/blob - src/backend/rewrite/rewriteDefine.c

   1 /*-------------------------------------------------------------------------
   2  *
   3  * rewriteDefine.c
   4  *        routines for defining a rewrite rule
   5  *
   6  * Portions Copyright (c) 1996-2002, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $Header: /cvsroot/pgsql/src/backend/rewrite/rewriteDefine.c,v 1.74 2002/07/12 18:43:17 tgl Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/heapam.h"
  18 #include "catalog/catname.h"
  19 #include "catalog/dependency.h"
  20 #include "catalog/indexing.h"
  21 #include "catalog/pg_rewrite.h"
  22 #include "commands/view.h"
  23 #include "miscadmin.h"
  24 #include "optimizer/clauses.h"
  25 #include "parser/parse_relation.h"
  26 #include "rewrite/rewriteDefine.h"
  27 #include "rewrite/rewriteManip.h"
  28 #include "rewrite/rewriteSupport.h"
  29 #include "storage/smgr.h"
  30 #include "utils/acl.h"
  31 #include "utils/builtins.h"
  32 #include "utils/syscache.h"
  33
  34
  35 static void setRuleCheckAsUser(Query *qry, Oid userid);
  36 static bool setRuleCheckAsUser_walker(Node *node, Oid *context);
  37
  38
  39 /*
  40  * InsertRule -
  41  *        takes the arguments and inserts them as a row into the system
  42  *        relation "pg_rewrite"
  43  */
  44 static Oid
  45 InsertRule(char *rulname,
  46                    int evtype,
  47                    Oid eventrel_oid,
  48                    AttrNumber evslot_index,
  49                    bool evinstead,
  50                    char *evqual,
  51                    char *actiontree)
  52 {
  53         int                     i;
  54         Datum           values[Natts_pg_rewrite];
  55         char            nulls[Natts_pg_rewrite];
  56         NameData        rname;
  57         Relation        pg_rewrite_desc;
  58         TupleDesc       tupDesc;
  59         HeapTuple       tup;
  60         Oid                     rewriteObjectId;
  61         ObjectAddress   myself,
  62                                         referenced;
  63
  64         if (IsDefinedRewriteRule(eventrel_oid, rulname))
  65                 elog(ERROR, "Attempt to insert rule \"%s\" failed: already exists",
  66                          rulname);
  67
  68         /*
  69          * Set up *nulls and *values arrays
  70          */
  71         MemSet(nulls, ' ', sizeof(nulls));
  72
  73         i = 0;
  74         namestrcpy(&rname, rulname);
  75         values[i++] = NameGetDatum(&rname);                             /* rulename */
  76         values[i++] = ObjectIdGetDatum(eventrel_oid);   /* ev_class */
  77         values[i++] = Int16GetDatum(evslot_index);              /* ev_attr */
  78         values[i++] = CharGetDatum(evtype + '0');               /* ev_type */
  79         values[i++] = BoolGetDatum(evinstead);                  /* is_instead */
  80         values[i++] = DirectFunctionCall1(textin, CStringGetDatum(evqual));     /* ev_qual */
  81         values[i++] = DirectFunctionCall1(textin, CStringGetDatum(actiontree)); /* ev_action */
  82
  83         /*
  84          * create a new pg_rewrite tuple
  85          */
  86         pg_rewrite_desc = heap_openr(RewriteRelationName, RowExclusiveLock);
  87
  88         tupDesc = pg_rewrite_desc->rd_att;
  89
  90         tup = heap_formtuple(tupDesc,
  91                                                  values,
  92                                                  nulls);
  93
  94         rewriteObjectId = simple_heap_insert(pg_rewrite_desc, tup);
  95
  96         if (RelationGetForm(pg_rewrite_desc)->relhasindex)
  97         {
  98                 Relation        idescs[Num_pg_rewrite_indices];
  99
 100                 CatalogOpenIndices(Num_pg_rewrite_indices, Name_pg_rewrite_indices,
 101                                                    idescs);
 102                 CatalogIndexInsert(idescs, Num_pg_rewrite_indices, pg_rewrite_desc,
 103                                                    tup);
 104                 CatalogCloseIndices(Num_pg_rewrite_indices, idescs);
 105         }
 106
 107         heap_freetuple(tup);
 108
 109         /*
 110          * Install dependency on rule's relation to ensure it will go away
 111          * on relation deletion.  If the rule is ON SELECT, make the dependency
 112          * implicit --- this prevents deleting a view's SELECT rule.  Other
 113          * kinds of rules can be AUTO.
 114          */
 115         myself.classId = RelationGetRelid(pg_rewrite_desc);
 116         myself.objectId = rewriteObjectId;
 117         myself.objectSubId = 0;
 118
 119         referenced.classId = RelOid_pg_class;
 120         referenced.objectId = eventrel_oid;
 121         referenced.objectSubId = 0;
 122
 123         recordDependencyOn(&myself, &referenced,
 124                 (evtype == CMD_SELECT) ? DEPENDENCY_INTERNAL : DEPENDENCY_AUTO);
 125
 126         heap_close(pg_rewrite_desc, RowExclusiveLock);
 127
 128         return rewriteObjectId;
 129 }
 130
 131 void
 132 DefineQueryRewrite(RuleStmt *stmt)
 133 {
 134         RangeVar   *event_obj = stmt->relation;
 135         Node       *event_qual = stmt->whereClause;
 136         CmdType         event_type = stmt->event;
 137         bool            is_instead = stmt->instead;
 138         List       *action = stmt->actions;
 139         Relation        event_relation;
 140         Oid                     ev_relid;
 141         Oid                     ruleId;
 142         int                     event_attno;
 143         Oid                     event_attype;
 144         char       *actionP,
 145                            *event_qualP;
 146         List       *l;
 147         Query      *query;
 148         AclResult       aclresult;
 149         bool            RelisBecomingView = false;
 150
 151         /*
 152          * If we are installing an ON SELECT rule, we had better grab
 153          * AccessExclusiveLock to ensure no SELECTs are currently running on
 154          * the event relation.  For other types of rules, it might be
 155          * sufficient to grab ShareLock to lock out insert/update/delete
 156          * actions.  But for now, let's just grab AccessExclusiveLock all the
 157          * time.
 158          */
 159         event_relation = heap_openrv(event_obj, AccessExclusiveLock);
 160         ev_relid = RelationGetRelid(event_relation);
 161
 162         /*
 163          * Check user has permission to apply rules to this relation.
 164          */
 165         aclresult = pg_class_aclcheck(ev_relid, GetUserId(), ACL_RULE);
 166         if (aclresult != ACLCHECK_OK)
 167                 aclcheck_error(aclresult, RelationGetRelationName(event_relation));
 168
 169         /*
 170          * No rule actions that modify OLD or NEW
 171          */
 172         foreach(l, action)
 173         {
 174                 query = (Query *) lfirst(l);
 175                 if (query->resultRelation == 0)
 176                         continue;
 177                 /* Don't be fooled by INSERT/SELECT */
 178                 if (query != getInsertSelectQuery(query, NULL))
 179                         continue;
 180                 if (query->resultRelation == PRS2_OLD_VARNO)
 181                         elog(ERROR, "rule actions on OLD currently not supported"
 182                                  "\n\tuse views or triggers instead");
 183                 if (query->resultRelation == PRS2_NEW_VARNO)
 184                         elog(ERROR, "rule actions on NEW currently not supported"
 185                                  "\n\tuse triggers instead");
 186         }
 187
 188         /*
 189          * Rules ON SELECT are restricted to view definitions
 190          */
 191         if (event_type == CMD_SELECT)
 192         {
 193                 List       *tllist;
 194                 int                     i;
 195
 196                 /*
 197                  * So there cannot be INSTEAD NOTHING, ...
 198                  */
 199                 if (length(action) == 0)
 200                 {
 201                         elog(ERROR, "instead nothing rules on select currently not supported"
 202                                  "\n\tuse views instead");
 203                 }
 204
 205                 /*
 206                  * ... there cannot be multiple actions, ...
 207                  */
 208                 if (length(action) > 1)
 209                         elog(ERROR, "multiple action rules on select currently not supported");
 210
 211                 /*
 212                  * ... the one action must be a SELECT, ...
 213                  */
 214                 query = (Query *) lfirst(action);
 215                 if (!is_instead || query->commandType != CMD_SELECT)
 216                         elog(ERROR, "only instead-select rules currently supported on select");
 217
 218                 /*
 219                  * ... there can be no rule qual, ...
 220                  */
 221                 if (event_qual != NULL)
 222                         elog(ERROR, "event qualifications not supported for rules on select");
 223
 224                 /*
 225                  * ... the targetlist of the SELECT action must exactly match the
 226                  * event relation, ...
 227                  */
 228                 i = 0;
 229                 foreach(tllist, query->targetList)
 230                 {
 231                         TargetEntry *tle = (TargetEntry *) lfirst(tllist);
 232                         Resdom     *resdom = tle->resdom;
 233                         Form_pg_attribute attr;
 234                         char       *attname;
 235
 236                         if (resdom->resjunk)
 237                                 continue;
 238                         i++;
 239                         if (i > event_relation->rd_att->natts)
 240                                 elog(ERROR, "select rule's target list has too many entries");
 241
 242                         attr = event_relation->rd_att->attrs[i - 1];
 243                         attname = NameStr(attr->attname);
 244
 245                         if (strcmp(resdom->resname, attname) != 0)
 246                                 elog(ERROR, "select rule's target entry %d has different column name from %s", i, attname);
 247
 248                         if (attr->atttypid != resdom->restype)
 249                                 elog(ERROR, "select rule's target entry %d has different type from attribute %s", i, attname);
 250
 251                         /*
 252                          * Allow typmods to be different only if one of them is -1,
 253                          * ie, "unspecified".  This is necessary for cases like
 254                          * "numeric", where the table will have a filled-in default
 255                          * length but the select rule's expression will probably have
 256                          * typmod = -1.
 257                          */
 258                         if (attr->atttypmod != resdom->restypmod &&
 259                                 attr->atttypmod != -1 && resdom->restypmod != -1)
 260                                 elog(ERROR, "select rule's target entry %d has different size from attribute %s", i, attname);
 261                 }
 262
 263                 if (i != event_relation->rd_att->natts)
 264                         elog(ERROR, "select rule's target list has too few entries");
 265
 266                 /*
 267                  * ... there must not be another ON SELECT rule already ...
 268                  */
 269                 if (event_relation->rd_rules != NULL)
 270                 {
 271                         for (i = 0; i < event_relation->rd_rules->numLocks; i++)
 272                         {
 273                                 RewriteRule *rule;
 274
 275                                 rule = event_relation->rd_rules->rules[i];
 276                                 if (rule->event == CMD_SELECT)
 277                                         elog(ERROR, "\"%s\" is already a view",
 278                                                  RelationGetRelationName(event_relation));
 279                         }
 280                 }
 281
 282                 /*
 283                  * ... and finally the rule must be named _RETURN.
 284                  */
 285                 if (strcmp(stmt->rulename, ViewSelectRuleName) != 0)
 286                 {
 287                         /*
 288                          * In versions before 7.3, the expected name was _RETviewname.
 289                          * For backwards compatibility with old pg_dump output, accept
 290                          * that and silently change it to _RETURN.  Since this is just
 291                          * a quick backwards-compatibility hack, limit the number of
 292                          * characters checked to a few less than NAMEDATALEN; this
 293                          * saves having to worry about where a multibyte character might
 294                          * have gotten truncated.
 295                          */
 296                         if (strncmp(stmt->rulename, "_RET", 4) != 0 ||
 297                                 strncmp(stmt->rulename + 4, event_obj->relname,
 298                                                 NAMEDATALEN - 4 - 4) != 0)
 299                                 elog(ERROR, "view rule for \"%s\" must be named \"%s\"",
 300                                          event_obj->relname, ViewSelectRuleName);
 301                         stmt->rulename = pstrdup(ViewSelectRuleName);
 302                 }
 303
 304                 /*
 305                  * Are we converting a relation to a view?
 306                  *
 307                  * If so, check that the relation is empty because the storage for
 308                  * the relation is going to be deleted.
 309                  */
 310                 if (event_relation->rd_rel->relkind != RELKIND_VIEW)
 311                 {
 312                         HeapScanDesc scanDesc;
 313
 314                         scanDesc = heap_beginscan(event_relation, SnapshotNow, 0, NULL);
 315                         if (heap_getnext(scanDesc, ForwardScanDirection) != NULL)
 316                                 elog(ERROR, "Relation \"%s\" is not empty. Cannot convert it to view",
 317                                          event_obj->relname);
 318                         heap_endscan(scanDesc);
 319
 320                         RelisBecomingView = true;
 321                 }
 322         }
 323
 324         /*
 325          * This rule is allowed - prepare to install it.
 326          */
 327         event_attno = -1;
 328         event_attype = InvalidOid;
 329
 330         /*
 331          * We want the rule's table references to be checked as though by the
 332          * rule owner, not the user referencing the rule.  Therefore, scan
 333          * through the rule's rtables and set the checkAsUser field on all
 334          * rtable entries.
 335          */
 336         foreach(l, action)
 337         {
 338                 query = (Query *) lfirst(l);
 339                 setRuleCheckAsUser(query, GetUserId());
 340         }
 341
 342         /* discard rule if it's null action and not INSTEAD; it's a no-op */
 343         if (action != NIL || is_instead)
 344         {
 345                 event_qualP = nodeToString(event_qual);
 346                 actionP = nodeToString(action);
 347
 348                 ruleId = InsertRule(stmt->rulename,
 349                                                         event_type,
 350                                                         ev_relid,
 351                                                         event_attno,
 352                                                         is_instead,
 353                                                         event_qualP,
 354                                                         actionP);
 355
 356                 /*
 357                  * Set pg_class 'relhasrules' field TRUE for event relation. If
 358                  * appropriate, also modify the 'relkind' field to show that the
 359                  * relation is now a view.
 360                  *
 361                  * Important side effect: an SI notice is broadcast to force all
 362                  * backends (including me!) to update relcache entries with the
 363                  * new rule.
 364                  */
 365                 SetRelationRuleStatus(ev_relid, true, RelisBecomingView);
 366         }
 367
 368         /*
 369          * IF the relation is becoming a view, delete the storage files
 370          * associated with it.  NB: we had better have AccessExclusiveLock to
 371          * do this ...
 372          */
 373         if (RelisBecomingView)
 374                 smgrunlink(DEFAULT_SMGR, event_relation);
 375
 376         /* Close rel, but keep lock till commit... */
 377         heap_close(event_relation, NoLock);
 378 }
 379
 380 /*
 381  * setRuleCheckAsUser
 382  *              Recursively scan a query and set the checkAsUser field to the
 383  *              given userid in all rtable entries.
 384  *
 385  * Note: for a view (ON SELECT rule), the checkAsUser field of the *OLD*
 386  * RTE entry will be overridden when the view rule is expanded, and the
 387  * checkAsUser field of the *NEW* entry is irrelevant because that entry's
 388  * checkFor bits will never be set.  However, for other types of rules it's
 389  * important to set these fields to match the rule owner.  So we just set
 390  * them always.
 391  */
 392 static void
 393 setRuleCheckAsUser(Query *qry, Oid userid)
 394 {
 395         List       *l;
 396
 397         /* Set all the RTEs in this query node */
 398         foreach(l, qry->rtable)
 399         {
 400                 RangeTblEntry *rte = (RangeTblEntry *) lfirst(l);
 401
 402                 if (rte->rtekind == RTE_SUBQUERY)
 403                 {
 404                         /* Recurse into subquery in FROM */
 405                         setRuleCheckAsUser(rte->subquery, userid);
 406                 }
 407                 else
 408                         rte->checkAsUser = userid;
 409         }
 410
 411         /* If there are sublinks, search for them and process their RTEs */
 412         if (qry->hasSubLinks)
 413                 query_tree_walker(qry, setRuleCheckAsUser_walker, (void *) &userid,
 414                                                   false /* already did the ones in rtable */ );
 415 }
 416
 417 /*
 418  * Expression-tree walker to find sublink queries
 419  */
 420 static bool
 421 setRuleCheckAsUser_walker(Node *node, Oid *context)
 422 {
 423         if (node == NULL)
 424                 return false;
 425         if (IsA(node, Query))
 426         {
 427                 Query      *qry = (Query *) node;
 428
 429                 setRuleCheckAsUser(qry, *context);
 430                 return false;
 431         }
 432         return expression_tree_walker(node, setRuleCheckAsUser_walker,
 433                                                                   (void *) context);
 434 }
 435
 436
 437 /*
 438  * Rename an existing rewrite rule.
 439  *
 440  * This is unused code at the moment.
 441  */
 442 void
 443 RenameRewriteRule(Oid owningRel, const char *oldName,
 444                                   const char *newName)
 445 {
 446         Relation        pg_rewrite_desc;
 447         HeapTuple       ruletup;
 448
 449         pg_rewrite_desc = heap_openr(RewriteRelationName, RowExclusiveLock);
 450
 451         ruletup = SearchSysCacheCopy(RULERELNAME,
 452                                                                  ObjectIdGetDatum(owningRel),
 453                                                                  PointerGetDatum(oldName),
 454                                                                  0, 0);
 455         if (!HeapTupleIsValid(ruletup))
 456                 elog(ERROR, "RenameRewriteRule: rule \"%s\" does not exist", oldName);
 457
 458         /* should not already exist */
 459         if (IsDefinedRewriteRule(owningRel, newName))
 460                 elog(ERROR, "Attempt to rename rule \"%s\" failed: \"%s\" already exists",
 461                          oldName, newName);
 462
 463         namestrcpy(&(((Form_pg_rewrite) GETSTRUCT(ruletup))->rulename), newName);
 464
 465         simple_heap_update(pg_rewrite_desc, &ruletup->t_self, ruletup);
 466
 467         /* keep system catalog indices current */
 468         if (RelationGetForm(pg_rewrite_desc)->relhasindex)
 469         {
 470                 Relation        idescs[Num_pg_rewrite_indices];
 471
 472                 CatalogOpenIndices(Num_pg_rewrite_indices, Name_pg_rewrite_indices,
 473                                                    idescs);
 474                 CatalogIndexInsert(idescs, Num_pg_rewrite_indices, pg_rewrite_desc,
 475                                                    ruletup);
 476                 CatalogCloseIndices(Num_pg_rewrite_indices, idescs);
 477         }
 478
 479         heap_freetuple(ruletup);
 480         heap_close(pg_rewrite_desc, RowExclusiveLock);
 481 }