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-2006, PostgreSQL Global Development Group
   7  * Portions Copyright (c) 1994, Regents of the University of California
   8  *
   9  *
  10  * IDENTIFICATION
  11  *        $PostgreSQL: pgsql/src/backend/rewrite/rewriteDefine.c,v 1.109 2006/03/05 15:58:36 momjian Exp $
  12  *
  13  *-------------------------------------------------------------------------
  14  */
  15 #include "postgres.h"
  16
  17 #include "access/heapam.h"
  18 #include "catalog/dependency.h"
  19 #include "catalog/indexing.h"
  20 #include "catalog/pg_rewrite.h"
  21 #include "commands/view.h"
  22 #include "miscadmin.h"
  23 #include "optimizer/clauses.h"
  24 #include "parser/parse_expr.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/lsyscache.h"
  33 #include "utils/syscache.h"
  34
  35
  36 static void setRuleCheckAsUser_Query(Query *qry, Oid userid);
  37 static void setRuleCheckAsUser_Expr(Node *node, Oid userid);
  38 static bool setRuleCheckAsUser_walker(Node *node, Oid *context);
  39
  40
  41 /*
  42  * InsertRule -
  43  *        takes the arguments and inserts them as a row into the system
  44  *        relation "pg_rewrite"
  45  */
  46 static Oid
  47 InsertRule(char *rulname,
  48                    int evtype,
  49                    Oid eventrel_oid,
  50                    AttrNumber evslot_index,
  51                    bool evinstead,
  52                    Node *event_qual,
  53                    List *action,
  54                    bool replace)
  55 {
  56         char       *evqual = nodeToString(event_qual);
  57         char       *actiontree = nodeToString((Node *) action);
  58         int                     i;
  59         Datum           values[Natts_pg_rewrite];
  60         char            nulls[Natts_pg_rewrite];
  61         char            replaces[Natts_pg_rewrite];
  62         NameData        rname;
  63         Relation        pg_rewrite_desc;
  64         HeapTuple       tup,
  65                                 oldtup;
  66         Oid                     rewriteObjectId;
  67         ObjectAddress myself,
  68                                 referenced;
  69         bool            is_update = false;
  70
  71         /*
  72          * Set up *nulls and *values arrays
  73          */
  74         MemSet(nulls, ' ', sizeof(nulls));
  75
  76         i = 0;
  77         namestrcpy(&rname, rulname);
  78         values[i++] = NameGetDatum(&rname); /* rulename */
  79         values[i++] = ObjectIdGetDatum(eventrel_oid);           /* ev_class */
  80         values[i++] = Int16GetDatum(evslot_index);      /* ev_attr */
  81         values[i++] = CharGetDatum(evtype + '0');       /* ev_type */
  82         values[i++] = BoolGetDatum(evinstead);          /* is_instead */
  83         values[i++] = DirectFunctionCall1(textin, CStringGetDatum(evqual)); /* ev_qual */
  84         values[i++] = DirectFunctionCall1(textin, CStringGetDatum(actiontree));         /* ev_action */
  85
  86         /*
  87          * Ready to store new pg_rewrite tuple
  88          */
  89         pg_rewrite_desc = heap_open(RewriteRelationId, RowExclusiveLock);
  90
  91         /*
  92          * Check to see if we are replacing an existing tuple
  93          */
  94         oldtup = SearchSysCache(RULERELNAME,
  95                                                         ObjectIdGetDatum(eventrel_oid),
  96                                                         PointerGetDatum(rulname),
  97                                                         0, 0);
  98
  99         if (HeapTupleIsValid(oldtup))
 100         {
 101                 if (!replace)
 102                         ereport(ERROR,
 103                                         (errcode(ERRCODE_DUPLICATE_OBJECT),
 104                                          errmsg("rule \"%s\" for relation \"%s\" already exists",
 105                                                         rulname, get_rel_name(eventrel_oid))));
 106
 107                 /*
 108                  * When replacing, we don't need to replace every attribute
 109                  */
 110                 MemSet(replaces, ' ', sizeof(replaces));
 111                 replaces[Anum_pg_rewrite_ev_attr - 1] = 'r';
 112                 replaces[Anum_pg_rewrite_ev_type - 1] = 'r';
 113                 replaces[Anum_pg_rewrite_is_instead - 1] = 'r';
 114                 replaces[Anum_pg_rewrite_ev_qual - 1] = 'r';
 115                 replaces[Anum_pg_rewrite_ev_action - 1] = 'r';
 116
 117                 tup = heap_modifytuple(oldtup, RelationGetDescr(pg_rewrite_desc),
 118                                                            values, nulls, replaces);
 119
 120                 simple_heap_update(pg_rewrite_desc, &tup->t_self, tup);
 121
 122                 ReleaseSysCache(oldtup);
 123
 124                 rewriteObjectId = HeapTupleGetOid(tup);
 125                 is_update = true;
 126         }
 127         else
 128         {
 129                 tup = heap_formtuple(pg_rewrite_desc->rd_att, values, nulls);
 130
 131                 rewriteObjectId = simple_heap_insert(pg_rewrite_desc, tup);
 132         }
 133
 134         /* Need to update indexes in either case */
 135         CatalogUpdateIndexes(pg_rewrite_desc, tup);
 136
 137         heap_freetuple(tup);
 138
 139         /* If replacing, get rid of old dependencies and make new ones */
 140         if (is_update)
 141                 deleteDependencyRecordsFor(RewriteRelationId, rewriteObjectId);
 142
 143         /*
 144          * Install dependency on rule's relation to ensure it will go away on
 145          * relation deletion.  If the rule is ON SELECT, make the dependency
 146          * implicit --- this prevents deleting a view's SELECT rule.  Other kinds
 147          * of rules can be AUTO.
 148          */
 149         myself.classId = RewriteRelationId;
 150         myself.objectId = rewriteObjectId;
 151         myself.objectSubId = 0;
 152
 153         referenced.classId = RelationRelationId;
 154         referenced.objectId = eventrel_oid;
 155         referenced.objectSubId = 0;
 156
 157         recordDependencyOn(&myself, &referenced,
 158                          (evtype == CMD_SELECT) ? DEPENDENCY_INTERNAL : DEPENDENCY_AUTO);
 159
 160         /*
 161          * Also install dependencies on objects referenced in action and qual.
 162          */
 163         recordDependencyOnExpr(&myself, (Node *) action, NIL,
 164                                                    DEPENDENCY_NORMAL);
 165
 166         if (event_qual != NULL)
 167         {
 168                 /* Find query containing OLD/NEW rtable entries */
 169                 Query      *qry = (Query *) linitial(action);
 170
 171                 qry = getInsertSelectQuery(qry, NULL);
 172                 recordDependencyOnExpr(&myself, event_qual, qry->rtable,
 173                                                            DEPENDENCY_NORMAL);
 174         }
 175
 176         heap_close(pg_rewrite_desc, RowExclusiveLock);
 177
 178         return rewriteObjectId;
 179 }
 180
 181 void
 182 DefineQueryRewrite(RuleStmt *stmt)
 183 {
 184         RangeVar   *event_obj = stmt->relation;
 185         Node       *event_qual = stmt->whereClause;
 186         CmdType         event_type = stmt->event;
 187         bool            is_instead = stmt->instead;
 188         bool            replace = stmt->replace;
 189         List       *action = stmt->actions;
 190         Relation        event_relation;
 191         Oid                     ev_relid;
 192         Oid                     ruleId;
 193         int                     event_attno;
 194         ListCell   *l;
 195         Query      *query;
 196         AclResult       aclresult;
 197         bool            RelisBecomingView = false;
 198
 199         /*
 200          * If we are installing an ON SELECT rule, we had better grab
 201          * AccessExclusiveLock to ensure no SELECTs are currently running on the
 202          * event relation.      For other types of rules, it might be sufficient to
 203          * grab ShareLock to lock out insert/update/delete actions.  But for now,
 204          * let's just grab AccessExclusiveLock all the time.
 205          */
 206         event_relation = heap_openrv(event_obj, AccessExclusiveLock);
 207         ev_relid = RelationGetRelid(event_relation);
 208
 209         /*
 210          * Check user has permission to apply rules to this relation.
 211          */
 212         aclresult = pg_class_aclcheck(ev_relid, GetUserId(), ACL_RULE);
 213         if (aclresult != ACLCHECK_OK)
 214                 aclcheck_error(aclresult, ACL_KIND_CLASS,
 215                                            RelationGetRelationName(event_relation));
 216
 217         /*
 218          * No rule actions that modify OLD or NEW
 219          */
 220         foreach(l, action)
 221         {
 222                 query = (Query *) lfirst(l);
 223                 if (query->resultRelation == 0)
 224                         continue;
 225                 /* Don't be fooled by INSERT/SELECT */
 226                 if (query != getInsertSelectQuery(query, NULL))
 227                         continue;
 228                 if (query->resultRelation == PRS2_OLD_VARNO)
 229                         ereport(ERROR,
 230                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 231                                          errmsg("rule actions on OLD are not implemented"),
 232                                          errhint("Use views or triggers instead.")));
 233                 if (query->resultRelation == PRS2_NEW_VARNO)
 234                         ereport(ERROR,
 235                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 236                                          errmsg("rule actions on NEW are not implemented"),
 237                                          errhint("Use triggers instead.")));
 238         }
 239
 240         /*
 241          * Rules ON SELECT are restricted to view definitions
 242          */
 243         if (event_type == CMD_SELECT)
 244         {
 245                 ListCell   *tllist;
 246                 int                     i;
 247
 248                 /*
 249                  * So there cannot be INSTEAD NOTHING, ...
 250                  */
 251                 if (list_length(action) == 0)
 252                         ereport(ERROR,
 253                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 254                            errmsg("INSTEAD NOTHING rules on SELECT are not implemented"),
 255                                          errhint("Use views instead.")));
 256
 257                 /*
 258                  * ... there cannot be multiple actions, ...
 259                  */
 260                 if (list_length(action) > 1)
 261                         ereport(ERROR,
 262                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 263                                          errmsg("multiple actions for rules on SELECT are not implemented")));
 264
 265                 /*
 266                  * ... the one action must be a SELECT, ...
 267                  */
 268                 query = (Query *) linitial(action);
 269                 if (!is_instead || query->commandType != CMD_SELECT)
 270                         ereport(ERROR,
 271                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 272                                  errmsg("rules on SELECT must have action INSTEAD SELECT")));
 273
 274                 /*
 275                  * ... there can be no rule qual, ...
 276                  */
 277                 if (event_qual != NULL)
 278                         ereport(ERROR,
 279                                         (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 280                                          errmsg("event qualifications are not implemented for rules on SELECT")));
 281
 282                 /*
 283                  * ... the targetlist of the SELECT action must exactly match the
 284                  * event relation, ...
 285                  */
 286                 i = 0;
 287                 foreach(tllist, query->targetList)
 288                 {
 289                         TargetEntry *tle = (TargetEntry *) lfirst(tllist);
 290                         int32           tletypmod;
 291                         Form_pg_attribute attr;
 292                         char       *attname;
 293
 294                         if (tle->resjunk)
 295                                 continue;
 296                         i++;
 297                         if (i > event_relation->rd_att->natts)
 298                                 ereport(ERROR,
 299                                                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 300                                   errmsg("SELECT rule's target list has too many entries")));
 301
 302                         attr = event_relation->rd_att->attrs[i - 1];
 303                         attname = NameStr(attr->attname);
 304
 305                         /*
 306                          * Disallow dropped columns in the relation.  This won't happen in
 307                          * the cases we actually care about (namely creating a view via
 308                          * CREATE TABLE then CREATE RULE).      Trying to cope with it is much
 309                          * more trouble than it's worth, because we'd have to modify the
 310                          * rule to insert dummy NULLs at the right positions.
 311                          */
 312                         if (attr->attisdropped)
 313                                 ereport(ERROR,
 314                                                 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 315                                                  errmsg("cannot convert relation containing dropped columns to view")));
 316
 317                         if (strcmp(tle->resname, attname) != 0)
 318                                 ereport(ERROR,
 319                                                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 320                                                  errmsg("SELECT rule's target entry %d has different column name from \"%s\"", i, attname)));
 321
 322                         if (attr->atttypid != exprType((Node *) tle->expr))
 323                                 ereport(ERROR,
 324                                                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 325                                                  errmsg("SELECT rule's target entry %d has different type from column \"%s\"", i, attname)));
 326
 327                         /*
 328                          * Allow typmods to be different only if one of them is -1, ie,
 329                          * "unspecified".  This is necessary for cases like "numeric",
 330                          * where the table will have a filled-in default length but the
 331                          * select rule's expression will probably have typmod = -1.
 332                          */
 333                         tletypmod = exprTypmod((Node *) tle->expr);
 334                         if (attr->atttypmod != tletypmod &&
 335                                 attr->atttypmod != -1 && tletypmod != -1)
 336                                 ereport(ERROR,
 337                                                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 338                                                  errmsg("SELECT rule's target entry %d has different size from column \"%s\"", i, attname)));
 339                 }
 340
 341                 if (i != event_relation->rd_att->natts)
 342                         ereport(ERROR,
 343                                         (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 344                                    errmsg("SELECT rule's target list has too few entries")));
 345
 346                 /*
 347                  * ... there must not be another ON SELECT rule already ...
 348                  */
 349                 if (!replace && event_relation->rd_rules != NULL)
 350                 {
 351                         for (i = 0; i < event_relation->rd_rules->numLocks; i++)
 352                         {
 353                                 RewriteRule *rule;
 354
 355                                 rule = event_relation->rd_rules->rules[i];
 356                                 if (rule->event == CMD_SELECT)
 357                                         ereport(ERROR,
 358                                                   (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 359                                                    errmsg("\"%s\" is already a view",
 360                                                                   RelationGetRelationName(event_relation))));
 361                         }
 362                 }
 363
 364                 /*
 365                  * ... and finally the rule must be named _RETURN.
 366                  */
 367                 if (strcmp(stmt->rulename, ViewSelectRuleName) != 0)
 368                 {
 369                         /*
 370                          * In versions before 7.3, the expected name was _RETviewname. For
 371                          * backwards compatibility with old pg_dump output, accept that
 372                          * and silently change it to _RETURN.  Since this is just a quick
 373                          * backwards-compatibility hack, limit the number of characters
 374                          * checked to a few less than NAMEDATALEN; this saves having to
 375                          * worry about where a multibyte character might have gotten
 376                          * truncated.
 377                          */
 378                         if (strncmp(stmt->rulename, "_RET", 4) != 0 ||
 379                                 strncmp(stmt->rulename + 4, event_obj->relname,
 380                                                 NAMEDATALEN - 4 - 4) != 0)
 381                                 ereport(ERROR,
 382                                                 (errcode(ERRCODE_INVALID_OBJECT_DEFINITION),
 383                                                  errmsg("view rule for \"%s\" must be named \"%s\"",
 384                                                                 event_obj->relname, ViewSelectRuleName)));
 385                         stmt->rulename = pstrdup(ViewSelectRuleName);
 386                 }
 387
 388                 /*
 389                  * Are we converting a relation to a view?
 390                  *
 391                  * If so, check that the relation is empty because the storage for the
 392                  * relation is going to be deleted.  Also insist that the rel not have
 393                  * any triggers, indexes, or child tables.
 394                  */
 395                 if (event_relation->rd_rel->relkind != RELKIND_VIEW)
 396                 {
 397                         HeapScanDesc scanDesc;
 398
 399                         scanDesc = heap_beginscan(event_relation, SnapshotNow, 0, NULL);
 400                         if (heap_getnext(scanDesc, ForwardScanDirection) != NULL)
 401                                 ereport(ERROR,
 402                                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 403                                                  errmsg("could not convert table \"%s\" to a view because it is not empty",
 404                                                                 event_obj->relname)));
 405                         heap_endscan(scanDesc);
 406
 407                         if (event_relation->rd_rel->reltriggers != 0)
 408                                 ereport(ERROR,
 409                                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 410                                                  errmsg("could not convert table \"%s\" to a view because it has triggers",
 411                                                                 event_obj->relname),
 412                                                  errhint("In particular, the table may not be involved in any foreign key relationships.")));
 413
 414                         if (event_relation->rd_rel->relhasindex)
 415                                 ereport(ERROR,
 416                                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 417                                                  errmsg("could not convert table \"%s\" to a view because it has indexes",
 418                                                                 event_obj->relname)));
 419
 420                         if (event_relation->rd_rel->relhassubclass)
 421                                 ereport(ERROR,
 422                                                 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 423                                                  errmsg("could not convert table \"%s\" to a view because it has child tables",
 424                                                                 event_obj->relname)));
 425
 426                         RelisBecomingView = true;
 427                 }
 428         }
 429
 430         /*
 431          * This rule is allowed - prepare to install it.
 432          */
 433         event_attno = -1;
 434
 435         /*
 436          * We want the rule's table references to be checked as though by the rule
 437          * owner, not the user referencing the rule.  Therefore, scan through the
 438          * rule's rtables and set the checkAsUser field on all rtable entries.  We
 439          * have to look at event_qual as well, in case it contains sublinks.
 440          */
 441         foreach(l, action)
 442         {
 443                 query = (Query *) lfirst(l);
 444                 setRuleCheckAsUser_Query(query, GetUserId());
 445         }
 446         setRuleCheckAsUser_Expr(event_qual, GetUserId());
 447
 448         /* discard rule if it's null action and not INSTEAD; it's a no-op */
 449         if (action != NIL || is_instead)
 450         {
 451                 ruleId = InsertRule(stmt->rulename,
 452                                                         event_type,
 453                                                         ev_relid,
 454                                                         event_attno,
 455                                                         is_instead,
 456                                                         event_qual,
 457                                                         action,
 458                                                         replace);
 459
 460                 /*
 461                  * Set pg_class 'relhasrules' field TRUE for event relation. If
 462                  * appropriate, also modify the 'relkind' field to show that the
 463                  * relation is now a view.
 464                  *
 465                  * Important side effect: an SI notice is broadcast to force all
 466                  * backends (including me!) to update relcache entries with the new
 467                  * rule.
 468                  */
 469                 SetRelationRuleStatus(ev_relid, true, RelisBecomingView);
 470         }
 471
 472         /*
 473          * IF the relation is becoming a view, delete the storage files associated
 474          * with it.  NB: we had better have AccessExclusiveLock to do this ...
 475          *
 476          * XXX what about getting rid of its TOAST table?  For now, we don't.
 477          */
 478         if (RelisBecomingView)
 479         {
 480                 RelationOpenSmgr(event_relation);
 481                 smgrscheduleunlink(event_relation->rd_smgr, event_relation->rd_istemp);
 482         }
 483
 484         /* Close rel, but keep lock till commit... */
 485         heap_close(event_relation, NoLock);
 486 }
 487
 488 /*
 489  * setRuleCheckAsUser_Query
 490  *              Recursively scan a query and set the checkAsUser field to the
 491  *              given userid in all rtable entries.
 492  *
 493  * Note: for a view (ON SELECT rule), the checkAsUser field of the *OLD*
 494  * RTE entry will be overridden when the view rule is expanded, and the
 495  * checkAsUser field of the *NEW* entry is irrelevant because that entry's
 496  * requiredPerms bits will always be zero.      However, for other types of rules
 497  * it's important to set these fields to match the rule owner.  So we just set
 498  * them always.
 499  */
 500 static void
 501 setRuleCheckAsUser_Query(Query *qry, Oid userid)
 502 {
 503         ListCell   *l;
 504
 505         /* Set all the RTEs in this query node */
 506         foreach(l, qry->rtable)
 507         {
 508                 RangeTblEntry *rte = (RangeTblEntry *) lfirst(l);
 509
 510                 if (rte->rtekind == RTE_SUBQUERY)
 511                 {
 512                         /* Recurse into subquery in FROM */
 513                         setRuleCheckAsUser_Query(rte->subquery, userid);
 514                 }
 515                 else
 516                         rte->checkAsUser = userid;
 517         }
 518
 519         /* If there are sublinks, search for them and process their RTEs */
 520         /* ignore subqueries in rtable because we already processed them */
 521         if (qry->hasSubLinks)
 522                 query_tree_walker(qry, setRuleCheckAsUser_walker, (void *) &userid,
 523                                                   QTW_IGNORE_RT_SUBQUERIES);
 524 }
 525
 526 /*
 527  * Expression-tree walker to find sublink queries
 528  */
 529 static void
 530 setRuleCheckAsUser_Expr(Node *node, Oid userid)
 531 {
 532         (void) setRuleCheckAsUser_walker(node, &userid);
 533 }
 534
 535 static bool
 536 setRuleCheckAsUser_walker(Node *node, Oid *context)
 537 {
 538         if (node == NULL)
 539                 return false;
 540         if (IsA(node, Query))
 541         {
 542                 Query      *qry = (Query *) node;
 543
 544                 setRuleCheckAsUser_Query(qry, *context);
 545                 return false;
 546         }
 547         return expression_tree_walker(node, setRuleCheckAsUser_walker,
 548                                                                   (void *) context);
 549 }
 550
 551
 552 /*
 553  * Rename an existing rewrite rule.
 554  *
 555  * This is unused code at the moment.
 556  */
 557 void
 558 RenameRewriteRule(Oid owningRel, const char *oldName,
 559                                   const char *newName)
 560 {
 561         Relation        pg_rewrite_desc;
 562         HeapTuple       ruletup;
 563
 564         pg_rewrite_desc = heap_open(RewriteRelationId, RowExclusiveLock);
 565
 566         ruletup = SearchSysCacheCopy(RULERELNAME,
 567                                                                  ObjectIdGetDatum(owningRel),
 568                                                                  PointerGetDatum(oldName),
 569                                                                  0, 0);
 570         if (!HeapTupleIsValid(ruletup))
 571                 ereport(ERROR,
 572                                 (errcode(ERRCODE_UNDEFINED_OBJECT),
 573                                  errmsg("rule \"%s\" for relation \"%s\" does not exist",
 574                                                 oldName, get_rel_name(owningRel))));
 575
 576         /* should not already exist */
 577         if (IsDefinedRewriteRule(owningRel, newName))
 578                 ereport(ERROR,
 579                                 (errcode(ERRCODE_DUPLICATE_OBJECT),
 580                                  errmsg("rule \"%s\" for relation \"%s\" already exists",
 581                                                 newName, get_rel_name(owningRel))));
 582
 583         namestrcpy(&(((Form_pg_rewrite) GETSTRUCT(ruletup))->rulename), newName);
 584
 585         simple_heap_update(pg_rewrite_desc, &ruletup->t_self, ruletup);
 586
 587         /* keep system catalog indexes current */
 588         CatalogUpdateIndexes(pg_rewrite_desc, ruletup);
 589
 590         heap_freetuple(ruletup);
 591         heap_close(pg_rewrite_desc, RowExclusiveLock);
 592 }