rewritehandler.c

关系型数据库 Postgresql 6.5.2

				  int rt_index,
				  int relation_level,
				  Relation relation,
				  int *modified)
{
	Query	   *rule_action = NULL;
	Node	   *rule_qual;
	List	   *rtable,
			   *rt,
			   *l;
	int			nothing,
				rt_length;
	int			badsql = FALSE;

	rule_qual = rule->qual;
	if (rule->actions)
	{
		if (length(rule->actions) > 1)	/* ??? because we don't handle
										 * rules with more than one
										 * action? -ay */

			return;
		rule_action = copyObject(lfirst(rule->actions));
		nothing = FALSE;
	}
	else
		nothing = TRUE;

	rtable = copyObject(parsetree->rtable);
	foreach(rt, rtable)
	{
		RangeTblEntry *rte = lfirst(rt);

		/*
		 * this is to prevent add_missing_vars_to_base_rels() from adding
		 * a bogus entry to the new target list.
		 */
		rte->inFromCl = false;
	}
	rt_length = length(rtable);

	rtable = nconc(rtable, copyObject(rule_action->rtable));
	parsetree->rtable = rtable;

	/* FOR UPDATE of view... */
	foreach(l, parsetree->rowMark)
	{
		if (((RowMark *) lfirst(l))->rti == rt_index)
			break;
	}
	if (l != NULL)				/* oh, hell -:) */
	{
		RowMark    *newrm;
		Index		rti = 1;
		List	   *l2;

		CheckSelectForUpdate(rule_action);

		/*
		 * We believe that rt_index is VIEW - nothing should be marked for
		 * VIEW, but ACL check must be done. As for real tables of VIEW -
		 * their rows must be marked, but we have to skip ACL check for
		 * them.
		 */
		((RowMark *) lfirst(l))->info &= ~ROW_MARK_FOR_UPDATE;
		foreach(l2, rule_action->rtable)
		{

			/*
			 * RTable of VIEW has two entries of VIEW itself - we use
			 * relid to skip them.
			 */
			if (relation->rd_id != ((RangeTblEntry *) lfirst(l2))->relid)
			{
				newrm = makeNode(RowMark);
				newrm->rti = rti + rt_length;
				newrm->info = ROW_MARK_FOR_UPDATE;
				lnext(l) = lcons(newrm, lnext(l));
				l = lnext(l);
			}
			rti++;
		}
	}

	rule_action->rtable = rtable;
	OffsetVarNodes((Node *) rule_qual, rt_length, 0);
	OffsetVarNodes((Node *) rule_action, rt_length, 0);

	ChangeVarNodes((Node *) rule_qual,
				   PRS2_CURRENT_VARNO + rt_length, rt_index, 0);
	ChangeVarNodes((Node *) rule_action,
				   PRS2_CURRENT_VARNO + rt_length, rt_index, 0);

	if (relation_level)
	{
		apply_RIR_view((Node **) &parsetree, rt_index,
					   (RangeTblEntry *) nth(rt_index - 1, rtable),
					   rule_action->targetList, modified, 0);
		apply_RIR_view((Node **) &rule_action, rt_index,
					   (RangeTblEntry *) nth(rt_index - 1, rtable),
					   rule_action->targetList, modified, 0);
	}
	else
	{
		HandleRIRAttributeRule(parsetree, rtable, rule_action->targetList,
							   rt_index, rule->attrno, modified, &badsql);
	}
	if (*modified && !badsql)
	{
		AddQual(parsetree, rule_action->qual);
		AddGroupClause(parsetree, rule_action->groupClause,
					   rule_action->targetList);
		AddHavingQual(parsetree, rule_action->havingQual);
		parsetree->hasAggs = (rule_action->hasAggs || parsetree->hasAggs);
		parsetree->hasSubLinks = (rule_action->hasSubLinks || parsetree->hasSubLinks);
	}
}


static void
fireRIRonSubselect(Node *node)
{
	if (node == NULL)
		return;

	switch (nodeTag(node))
	{
		case T_TargetEntry:
			{
				TargetEntry *tle = (TargetEntry *) node;

				fireRIRonSubselect(
								   (Node *) (tle->expr));
			}
			break;

		case T_Aggref:
			{
				Aggref	   *aggref = (Aggref *) node;

				fireRIRonSubselect(
								   (Node *) (aggref->target));
			}
			break;

		case T_GroupClause:
			break;

		case T_Expr:
			{
				Expr	   *exp = (Expr *) node;

				fireRIRonSubselect(
								   (Node *) (exp->args));
			}
			break;

		case T_Iter:
			{
				Iter	   *iter = (Iter *) node;

				fireRIRonSubselect(
								   (Node *) (iter->iterexpr));
			}
			break;

		case T_ArrayRef:
			{
				ArrayRef   *ref = (ArrayRef *) node;

				fireRIRonSubselect(
								   (Node *) (ref->refupperindexpr));
				fireRIRonSubselect(
								   (Node *) (ref->reflowerindexpr));
				fireRIRonSubselect(
								   (Node *) (ref->refexpr));
				fireRIRonSubselect(
								   (Node *) (ref->refassgnexpr));
			}
			break;

		case T_Var:
			break;

		case T_Param:
			break;

		case T_Const:
			break;

		case T_List:
			{
				List	   *l;

				foreach(l, (List *) node)
					fireRIRonSubselect(
									   (Node *) (lfirst(l)));
			}
			break;

		case T_SubLink:
			{
				SubLink    *sub = (SubLink *) node;
				Query	   *qry;

				fireRIRonSubselect(
								   (Node *) (sub->lefthand));

				qry = fireRIRrules((Query *) (sub->subselect));

				fireRIRonSubselect(
								   (Node *) qry);

				sub->subselect = (Node *) qry;
			}
			break;

		case T_CaseExpr:
			{
				CaseExpr   *exp = (CaseExpr *) node;

				fireRIRonSubselect(
								   (Node *) (exp->args));

				fireRIRonSubselect(
								   (Node *) (exp->defresult));
			}
			break;

		case T_CaseWhen:
			{
				CaseWhen   *exp = (CaseWhen *) node;

				fireRIRonSubselect(
								   (Node *) (exp->expr));

				fireRIRonSubselect(
								   (Node *) (exp->result));
			}
			break;

		case T_Query:
			{
				Query	   *qry = (Query *) node;

				fireRIRonSubselect(
								   (Node *) (qry->targetList));

				fireRIRonSubselect(
								   (Node *) (qry->qual));

				fireRIRonSubselect(
								   (Node *) (qry->havingQual));
			}
			break;

		default:
			elog(NOTICE, "unknown node tag %d in fireRIRonSubselect()", nodeTag(node));
			elog(NOTICE, "Node is: %s", nodeToString(node));
			break;


	}
}


/*
 * fireRIRrules -
 *	Apply all RIR rules on each rangetable entry in a query
 */
static Query *
fireRIRrules(Query *parsetree)
{
	int			rt_index;
	RangeTblEntry *rte;
	Relation	rel;
	List	   *locks;
	RuleLock   *rules;
	RewriteRule *rule;
	RewriteRule RIRonly;
	int			modified;
	int			i;
	List	   *l;

	rt_index = 0;
	while (rt_index < length(parsetree->rtable))
	{
		++rt_index;

		rte = nth(rt_index - 1, parsetree->rtable);

		if (!rangeTableEntry_used((Node *) parsetree, rt_index, 0))
		{

			/*
			 * Unused range table entries must not be marked as coming
			 * from a clause. Otherwise the planner will generate joins
			 * over relations that in fact shouldn't be scanned at all and
			 * the result will contain duplicates
			 *
			 * Jan
			 *
			 */
			rte->inFromCl = FALSE;
			continue;
		}

		rel = heap_openr(rte->relname);
		if (rel->rd_rules == NULL)
		{
			heap_close(rel);
			continue;
		}

		rules = rel->rd_rules;
		locks = NIL;

		/*
		 * Collect the RIR rules that we must apply
		 */
		for (i = 0; i < rules->numLocks; i++)
		{
			rule = rules->rules[i];
			if (rule->event != CMD_SELECT)
				continue;

			if (rule->attrno > 0 &&
				!attribute_used((Node *) parsetree,
								rt_index,
								rule->attrno, 0))
				continue;

			locks = lappend(locks, rule);
		}

		/*
		 * Check permissions
		 */
		checkLockPerms(locks, parsetree, rt_index);

		/*
		 * Now apply them
		 */
		foreach(l, locks)
		{
			rule = lfirst(l);

			RIRonly.event = rule->event;
			RIRonly.attrno = rule->attrno;
			RIRonly.qual = rule->qual;
			RIRonly.actions = rule->actions;

			ApplyRetrieveRule(parsetree,
							  &RIRonly,
							  rt_index,
							  RIRonly.attrno == -1,
							  rel,
							  &modified);
		}

		heap_close(rel);
	}

	fireRIRonSubselect((Node *) parsetree);
	modifyAggrefQual((Node **) &(parsetree->qual), parsetree);

	return parsetree;
}


/*
 * idea is to fire regular rules first, then qualified instead
 * rules and unqualified instead rules last. Any lemming is counted for.
 */
static List *
orderRules(List *locks)
{
	List	   *regular = NIL;
	List	   *instead_rules = NIL;
	List	   *instead_qualified = NIL;
	List	   *i;

	foreach(i, locks)
	{
		RewriteRule *rule_lock = (RewriteRule *) lfirst(i);

		if (rule_lock->isInstead)
		{
			if (rule_lock->qual == NULL)
				instead_rules = lappend(instead_rules, rule_lock);
			else
				instead_qualified = lappend(instead_qualified, rule_lock);
		}
		else
			regular = lappend(regular, rule_lock);
	}
	regular = nconc(regular, instead_qualified);
	return nconc(regular, instead_rules);
}



static Query *
CopyAndAddQual(Query *parsetree,
			   List *actions,
			   Node *rule_qual,
			   int rt_index,
			   CmdType event)
{
	Query	   *new_tree = (Query *) copyObject(parsetree);
	Node	   *new_qual = NULL;
	Query	   *rule_action = NULL;

	if (actions)
		rule_action = lfirst(actions);
	if (rule_qual != NULL)
		new_qual = (Node *) copyObject(rule_qual);
	if (rule_action != NULL)
	{
		List	   *rtable;
		int			rt_length;

		rtable = new_tree->rtable;
		rt_length = length(rtable);
		rtable = nconc(rtable, copyObject(rule_action->rtable));
		new_tree->rtable = rtable;
		OffsetVarNodes(new_qual, rt_length, 0);
		ChangeVarNodes(new_qual, PRS2_CURRENT_VARNO + rt_length, rt_index, 0);
	}
	/* XXX -- where current doesn't work for instead nothing.... yet */
	AddNotQual(new_tree, new_qual);

	return new_tree;
}



/*
 *	fireRules -
 *	   Iterate through rule locks applying rules.
 *	   All rules create their own parsetrees. Instead rules
 *	   with rule qualification save the original parsetree
 *	   and add their negated qualification to it. Real instead
 *	   rules finally throw away the original parsetree.
 *
 *	   remember: reality is for dead birds -- glass
 *
 */
static List *
fireRules(Query *parsetree,
		  int rt_index,
		  CmdType event,
		  bool *instead_flag,
		  List *locks,
		  List **qual_products)
{
	RewriteInfo *info;
	List	   *results = NIL;
	List	   *i;

	/* choose rule to fire from list of rules */
	if (locks == NIL)
		return NIL;

	locks = orderRules(locks);	/* real instead rules last */
	foreach(i, locks)
	{
		RewriteRule *rule_lock = (RewriteRule *) lfirst(i);
		Node	   *qual,
				   *event_qual;
		List	   *actions;
		List	   *r;

		/*
		 * Instead rules change the resultRelation of the query. So the
		 * permission checks on the initial resultRelation would never be
		 * done (this is normally done in the executor deep down). So we
		 * must do it here. The result relations resulting from earlier
		 * rewrites are already checked against the rules eventrelation
		 * owner (during matchLocks) and have the skipAcl flag set.
		 */
		if (rule_lock->isInstead &&
			parsetree->commandType != CMD_SELECT)
		{
			RangeTblEntry *rte;
			int32		acl_rc;
			int32		reqperm;

			switch (parsetree->commandType)
			{
				case CMD_INSERT:
					reqperm = ACL_AP;
					break;
				default:
					reqperm = ACL_WR;
					break;
			}

			rte = (RangeTblEntry *) nth(parsetree->resultRelation - 1,
										parsetree->rtable);
			if (!rte->skipAcl)
			{
				acl_rc = pg_aclcheck(rte->relname,
									 GetPgUserName(), reqperm);
				if (acl_rc != ACLCHECK_OK)
				{
					elog(ERROR, "%s: %s",
						 rte->relname,
						 aclcheck_error_strings[acl_rc]);
				}
			}
		}

		/* multiple rule action time */
		*instead_flag = rule_lock->isInstead;
		event_qual = rule_lock->qual;
		actions = rule_lock->actions;
		if (event_qual != NULL && *instead_flag)
		{
			Query	   *qual_product;
			RewriteInfo qual_info;

			/* ----------
			 * If there are instead rules with qualifications,
			 * the original query is still performed. But all
			 * the negated rule qualifications of the instead
			 * rules are added so it does it's actions only
			 * in cases where the rule quals of all instead
			 * rules are false. Think of it as the default
			 * action in a case. We save this in *qual_products
			 * so deepRewriteQuery() can add it to the query
			 * list after we mangled it up enough.
			 * ----------
			 */
			if (*qual_products == NIL)
				qual_product = parsetree;
			else
				qual_product = (Query *) nth(0, *qual_products);

			qual_info.event = qual_product->commandType;
			qual_info.new_varno = length(qual_product->rtable) + 2;
			qual_product = CopyAndAddQual(qual_product,
										  actions,
										  event_qual,
										  rt_index,
										  event);

			qual_info.rule_action = qual_product;

			if (event == CMD_INSERT || event == CMD_UPDATE)
				FixNew(&qual_info, qual_product);

			*qual_products = lappend(NIL, qual_product);
		}

		foreach(r, actions)
		{
			Query	   *rule_action = lfirst(r);
			Node	   *rule_qual = copyObject(event_qual);

			if (rule_action->commandType == CMD_NOTHING)
				continue;

			/*--------------------------------------------------
			 * We copy the qualifications of the parsetree
			 * to the action and vice versa. So force
			 * hasSubLinks if one of them has it.
			 *
			 * As of 6.4 only parsetree qualifications can
			 * have sublinks. If this changes, we must make
			 * this a node lookup at the end of rewriting.
			 *
			 * Jan
			 *--------------------------------------------------
			 */
			if (parsetree->hasSubLinks && !rule_action->hasSubLinks)
			{
				rule_action = copyObject(rule_action);
				rule_action->hasSubLinks = TRUE;
			}
			if (!parsetree->hasSubLinks && rule_action->hasSubLinks)
				parsetree->hasSubLinks = TRUE;

			/*--------------------------------------------------
			 * Step 1:
			 *	  Rewrite current.attribute or current to tuple variable
			 *	  this appears to be done in parser?
			 *--------------------------------------------------
			 */
			info = gatherRewriteMeta(parsetree, rule_action, rule_qual,
									 rt_index, event, instead_flag);

			/* handle escapable cases, or those handled by other code */
			if (info->nothing)
			{
				if (*instead_flag)
					return NIL;