postgresql/src/backend/executor/nodeLockRows.c

/*-------------------------------------------------------------------------
 *
 * nodeLockRows.c
 *	  Routines to handle FOR UPDATE/FOR SHARE row locking
 *
 * Portions Copyright (c) 1996-2010, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  $PostgreSQL: pgsql/src/backend/executor/nodeLockRows.c,v 1.3 2010/01/02 16:57:42 momjian Exp $
 *
 *-------------------------------------------------------------------------
 */
/*
 * INTERFACE ROUTINES
 *		ExecLockRows		- fetch locked rows
 *		ExecInitLockRows	- initialize node and subnodes..
 *		ExecEndLockRows		- shutdown node and subnodes
 */

#include "postgres.h"

#include "access/xact.h"
#include "executor/executor.h"
#include "executor/nodeLockRows.h"
#include "storage/bufmgr.h"
#include "utils/tqual.h"


/* ----------------------------------------------------------------
 *		ExecLockRows
 * ----------------------------------------------------------------
 */
TupleTableSlot *				/* return: a tuple or NULL */
ExecLockRows(LockRowsState *node)
{
	TupleTableSlot *slot;
	EState	   *estate;
	PlanState  *outerPlan;
	bool		epq_started;
	ListCell   *lc;

	/*
	 * get information from the node
	 */
	estate = node->ps.state;
	outerPlan = outerPlanState(node);

	/*
	 * Get next tuple from subplan, if any.
	 */
lnext:
	slot = ExecProcNode(outerPlan);

	if (TupIsNull(slot))
		return NULL;

	/*
	 * Attempt to lock the source tuple(s).  (Note we only have locking
	 * rowmarks in lr_rowMarks.)
	 */
	epq_started = false;
	foreach(lc, node->lr_rowMarks)
	{
		ExecRowMark *erm = (ExecRowMark *) lfirst(lc);
		Datum		datum;
		bool		isNull;
		HeapTupleData tuple;
		Buffer		buffer;
		ItemPointerData update_ctid;
		TransactionId update_xmax;
		LockTupleMode lockmode;
		HTSU_Result test;
		HeapTuple	copyTuple;

		/* clear any leftover test tuple for this rel */
		if (node->lr_epqstate.estate != NULL)
			EvalPlanQualSetTuple(&node->lr_epqstate, erm->rti, NULL);

		/* if child rel, must check whether it produced this row */
		if (erm->rti != erm->prti)
		{
			Oid			tableoid;

			datum = ExecGetJunkAttribute(slot,
										 erm->toidAttNo,
										 &isNull);
			/* shouldn't ever get a null result... */
			if (isNull)
				elog(ERROR, "tableoid is NULL");
			tableoid = DatumGetObjectId(datum);

			if (tableoid != RelationGetRelid(erm->relation))
			{
				/* this child is inactive right now */
				ItemPointerSetInvalid(&(erm->curCtid));
				continue;
			}
		}

		/* fetch the tuple's ctid */
		datum = ExecGetJunkAttribute(slot,
									 erm->ctidAttNo,
									 &isNull);
		/* shouldn't ever get a null result... */
		if (isNull)
			elog(ERROR, "ctid is NULL");
		tuple.t_self = *((ItemPointer) DatumGetPointer(datum));

		/* okay, try to lock the tuple */
		if (erm->markType == ROW_MARK_EXCLUSIVE)
			lockmode = LockTupleExclusive;
		else
			lockmode = LockTupleShared;

		test = heap_lock_tuple(erm->relation, &tuple, &buffer,
							   &update_ctid, &update_xmax,
							   estate->es_output_cid,
							   lockmode, erm->noWait);
		ReleaseBuffer(buffer);
		switch (test)
		{
			case HeapTupleSelfUpdated:
				/* treat it as deleted; do not process */
				goto lnext;

			case HeapTupleMayBeUpdated:
				/* got the lock successfully */
				break;

			case HeapTupleUpdated:
				if (IsXactIsoLevelSerializable)
					ereport(ERROR,
							(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
							 errmsg("could not serialize access due to concurrent update")));
				if (ItemPointerEquals(&update_ctid,
									  &tuple.t_self))
				{
					/* Tuple was deleted, so don't return it */
					goto lnext;
				}

				/* updated, so fetch and lock the updated version */
				copyTuple = EvalPlanQualFetch(estate, erm->relation, lockmode,
											  &update_ctid, update_xmax);

				if (copyTuple == NULL)
				{
					/* Tuple was deleted, so don't return it */
					goto lnext;
				}
				/* remember the actually locked tuple's TID */
				tuple.t_self = copyTuple->t_self;

				/*
				 * Need to run a recheck subquery.  Initialize EPQ state
				 * if we didn't do so already.
				 */
				if (!epq_started)
				{
					EvalPlanQualBegin(&node->lr_epqstate, estate);
					epq_started = true;
				}

				/* Store target tuple for relation's scan node */
				EvalPlanQualSetTuple(&node->lr_epqstate, erm->rti, copyTuple);

				/* Continue loop until we have all target tuples */
				break;

			default:
				elog(ERROR, "unrecognized heap_lock_tuple status: %u",
					 test);
		}

		/* Remember locked tuple's TID for WHERE CURRENT OF */
		erm->curCtid = tuple.t_self;
	}

	/*
	 * If we need to do EvalPlanQual testing, do so.
	 */
	if (epq_started)
	{
		/*
		 * First, fetch a copy of any rows that were successfully locked
		 * without any update having occurred.  (We do this in a separate
		 * pass so as to avoid overhead in the common case where there are
		 * no concurrent updates.)
		 */
		foreach(lc, node->lr_rowMarks)
		{
			ExecRowMark *erm = (ExecRowMark *) lfirst(lc);
			HeapTupleData tuple;
			Buffer		buffer;

			if (EvalPlanQualGetTuple(&node->lr_epqstate, erm->rti) != NULL)
				continue;		/* it was updated and fetched above */

			/* okay, fetch the tuple */
			tuple.t_self = erm->curCtid;
			if (!heap_fetch(erm->relation, SnapshotAny, &tuple, &buffer,
							false, NULL))
				elog(ERROR, "failed to fetch tuple for EvalPlanQual recheck");

			/* successful, copy and store tuple */
			EvalPlanQualSetTuple(&node->lr_epqstate, erm->rti,
								 heap_copytuple(&tuple));
			ReleaseBuffer(buffer);
		}
		/*
		 * Now fetch any non-locked source rows --- the EPQ logic knows
		 * how to do that.
		 */
		EvalPlanQualSetSlot(&node->lr_epqstate, slot);
		EvalPlanQualFetchRowMarks(&node->lr_epqstate);
		/*
		 * And finally we can re-evaluate the tuple.
		 */
		slot = EvalPlanQualNext(&node->lr_epqstate);
		if (TupIsNull(slot))
		{
			/* Updated tuple fails qual, so ignore it and go on */
			goto lnext;
		}
	}

	/* Got all locks, so return the current tuple */
	return slot;
}

/* ----------------------------------------------------------------
 *		ExecInitLockRows
 *
 *		This initializes the LockRows node state structures and
 *		the node's subplan.
 * ----------------------------------------------------------------
 */
LockRowsState *
ExecInitLockRows(LockRows *node, EState *estate, int eflags)
{
	LockRowsState *lrstate;
	Plan	   *outerPlan = outerPlan(node);
	ListCell   *lc;

	/* check for unsupported flags */
	Assert(!(eflags & EXEC_FLAG_MARK));

	/*
	 * create state structure
	 */
	lrstate = makeNode(LockRowsState);
	lrstate->ps.plan = (Plan *) node;
	lrstate->ps.state = estate;
	EvalPlanQualInit(&lrstate->lr_epqstate, estate, outerPlan, node->epqParam);

	/*
	 * Miscellaneous initialization
	 *
	 * LockRows nodes never call ExecQual or ExecProject.
	 */

	/*
	 * Tuple table initialization (XXX not actually used...)
	 */
	ExecInitResultTupleSlot(estate, &lrstate->ps);

	/*
	 * then initialize outer plan
	 */
	outerPlanState(lrstate) = ExecInitNode(outerPlan, estate, eflags);

	/*
	 * LockRows nodes do no projections, so initialize projection info for this
	 * node appropriately
	 */
	ExecAssignResultTypeFromTL(&lrstate->ps);
	lrstate->ps.ps_ProjInfo = NULL;

	/*
	 * Locate the ExecRowMark(s) that this node is responsible for.
	 * (InitPlan should already have built the global list of ExecRowMarks.)
	 */
	lrstate->lr_rowMarks = NIL;
	foreach(lc, node->rowMarks)
	{
		PlanRowMark *rc = (PlanRowMark *) lfirst(lc);
		ExecRowMark *erm = NULL;
		ListCell   *lce;

		Assert(IsA(rc, PlanRowMark));

		/* ignore "parent" rowmarks; they are irrelevant at runtime */
		if (rc->isParent)
			continue;

		foreach(lce, estate->es_rowMarks)
		{
			erm = (ExecRowMark *) lfirst(lce);
			if (erm->rti == rc->rti)
				break;
			erm = NULL;
		}
		if (erm == NULL)
			elog(ERROR, "failed to find ExecRowMark for PlanRowMark %u",
				 rc->rti);

		/*
		 * Only locking rowmarks go into our own list.  Non-locking marks
		 * are passed off to the EvalPlanQual machinery.  This is because
		 * we don't want to bother fetching non-locked rows unless we
		 * actually have to do an EPQ recheck.
		 */
		if (RowMarkRequiresRowShareLock(erm->markType))
			lrstate->lr_rowMarks = lappend(lrstate->lr_rowMarks, erm);
		else
			EvalPlanQualAddRowMark(&lrstate->lr_epqstate, erm);
	}

	return lrstate;
}

/* ----------------------------------------------------------------
 *		ExecEndLockRows
 *
 *		This shuts down the subplan and frees resources allocated
 *		to this node.
 * ----------------------------------------------------------------
 */
void
ExecEndLockRows(LockRowsState *node)
{
	EvalPlanQualEnd(&node->lr_epqstate);
	ExecEndNode(outerPlanState(node));
}


void
ExecReScanLockRows(LockRowsState *node, ExprContext *exprCtxt)
{
	/*
	 * if chgParam of subnode is not null then plan will be re-scanned by
	 * first ExecProcNode.
	 */
	if (((PlanState *) node)->lefttree->chgParam == NULL)
		ExecReScan(((PlanState *) node)->lefttree, exprCtxt);
}