From 0363a94a38c84db2877ded679799670fb7c0a3e5 Mon Sep 17 00:00:00 2001
From: Michael Erickson <michae2@cockroachlabs.com>
Date: Mon, 25 Sep 2023 22:51:22 -0700
Subject: [PATCH 1/2] optbuilder: a few minor tweaks to building of FOR UPDATE

Make a few minor tweaks to semantic analysis of FOR UPDATE locking
clauses.

1. Disallow multiple FOR UPDATE clauses on parenthesized queries. We do
   not currently handle scopes of parenthesized queries correctly.
   Because of this, we disallow multiple ORDER BY and LIMIT clauses on
   parenthesized queries. The previous implementation of FOR UPDATE was
   simple enough that we could work around this, but the upcoming
   changes make it impossible to support.

2. Allow FOR UPDATE on statements with VALUES in the FROM list (but
   continue to disallow FOR UPDATE on VALUES directly). This matches
   Postgres.

Release note (sql change): Two minor changes to FOR UPDATE clauses:

1. Multiple FOR UPDATE clauses on fully-parenthesized queries are now
disallowed. For example, the following statements are now disallowed:

```
(SELECT 1 FOR UPDATE) FOR UPDATE;
SELECT * FROM ((SELECT 1 FOR UPDATE) FOR UPDATE) AS x;
```

whereas statements like the following are still allowed:

```
SELECT * FROM (SELECT 1 FOR UPDATE) AS x FOR UPDATE;
SELECT (SELECT 1 FOR UPDATE) FOR UPDATE;
```

This does not match PostgreSQL, which allows all of these, but does
match our behavior for ORDER BY and LIMIT.

2. FOR UPDATE is now allowed on statements with VALUES in the FROM list
or as a subquery. For example, the following statements are now allowed:

```
SELECT (VALUES (1)) FOR UPDATE;
SELECT * FROM (VALUES (1)) AS x FOR UPDATE;
```

Using FOR UPDATE directly on VALUES is still disallowed:

```
VALUES (1) FOR UPDATE;
(VALUES (1)) FOR UPDATE;
INSERT INTO t VALUES (1) FOR UPDATE;
```

This matches PostgreSQL.
---
 .../testdata/logic_test/select_for_update     | 67 +++++++++++++++++++
 pkg/sql/opt/optbuilder/select.go              | 27 ++++++--
 2 files changed, 87 insertions(+), 7 deletions(-)

diff --git a/pkg/sql/logictest/testdata/logic_test/select_for_update b/pkg/sql/logictest/testdata/logic_test/select_for_update
index 46d68fc7e467..cbdf2b0896f0 100644
--- a/pkg/sql/logictest/testdata/logic_test/select_for_update
+++ b/pkg/sql/logictest/testdata/logic_test/select_for_update
@@ -131,6 +131,29 @@ query I
 ----
 1
 
+# Unlike Postgres, we do not support multiple locking clauses on parenthesized
+# queries.
+
+statement error pgcode 42601 multiple FOR UPDATE clauses not allowed
+(SELECT 1 FOR UPDATE) FOR UPDATE
+
+statement error pgcode 42601 multiple FOR UPDATE clauses not allowed
+((SELECT 1 FOR UPDATE)) FOR UPDATE
+
+statement error pgcode 42601 multiple FOR UPDATE clauses not allowed
+((SELECT 1) FOR UPDATE) FOR UPDATE
+
+# But we do support locking clauses both inside and outside subqueries.
+
+statement ok
+SELECT (SELECT 1 FOR UPDATE) FOR UPDATE
+
+statement ok
+SELECT * FROM (SELECT 1 FOR UPDATE) AS x FOR UPDATE
+
+statement ok
+SELECT * FROM (SELECT 1 FOR UPDATE) AS x WHERE EXISTS (SELECT 1 FOR UPDATE) FOR UPDATE
+
 # FOR READ ONLY is ignored, like in Postgres.
 query I
 SELECT 1 FOR READ ONLY
@@ -140,6 +163,9 @@ SELECT 1 FOR READ ONLY
 # Various operations are not supported when locking clauses are provided.
 # FeatureNotSupported errors are thrown for each of them.
 
+statement ok
+CREATE TABLE i (i PRIMARY KEY) AS SELECT 1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with UNION/INTERSECT/EXCEPT
 SELECT 1 UNION SELECT 1 FOR UPDATE
 
@@ -150,7 +176,24 @@ statement error pgcode 0A000 FOR UPDATE is not allowed with VALUES
 VALUES (1) FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with VALUES
+(VALUES (1)) FOR UPDATE
+
+# VALUES within a subquery is allowed, like in Postgres.
+
+query I
+SELECT (VALUES (1)) FOR UPDATE
+----
+1
+
+query I
 SELECT * FROM (VALUES (1)) a FOR UPDATE
+----
+1
+
+query II
+SELECT * FROM (VALUES (1)) a, i FOR UPDATE
+----
+1  1
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
 SELECT DISTINCT 1 FOR UPDATE
@@ -158,36 +201,54 @@ SELECT DISTINCT 1 FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
 SELECT * FROM (SELECT DISTINCT 1) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
+SELECT * FROM (SELECT DISTINCT 1) a, i FOR UPDATE
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
 SELECT 1 GROUP BY 1 FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
 SELECT * FROM (SELECT 1 GROUP BY 1) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
+SELECT * FROM (SELECT 1 GROUP BY 1) a, i FOR UPDATE
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
 SELECT 1 HAVING TRUE FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
 SELECT * FROM (SELECT 1 HAVING TRUE) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
+SELECT * FROM (SELECT 1 HAVING TRUE) a, i FOR UPDATE
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
 SELECT count(1) FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
 SELECT * FROM (SELECT count(1)) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
+SELECT * FROM (SELECT count(1)) a, i FOR UPDATE
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
 SELECT count(1) OVER () FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
 SELECT * FROM (SELECT count(1) OVER ()) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
+SELECT * FROM (SELECT count(1) OVER ()) a, i FOR UPDATE
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
 SELECT generate_series(1, 2) FOR UPDATE
 
 statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
 SELECT * FROM (SELECT generate_series(1, 2)) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
+SELECT * FROM (SELECT generate_series(1, 2)) a, i FOR UPDATE
+
 # Set-returning functions in the from list are allowed.
 query I nosort
 SELECT * FROM generate_series(1, 2) FOR UPDATE
@@ -201,6 +262,12 @@ SELECT * FROM (SELECT * FROM generate_series(1, 2)) a FOR UPDATE
 1
 2
 
+query II nosort
+SELECT * FROM (SELECT * FROM generate_series(1, 2)) a, i FOR UPDATE
+----
+1  1
+2  1
+
 # Use of SELECT FOR UPDATE/SHARE requires SELECT and UPDATE privileges.
 
 statement ok
diff --git a/pkg/sql/opt/optbuilder/select.go b/pkg/sql/opt/optbuilder/select.go
index c391cc2c93a0..195ba2347965 100644
--- a/pkg/sql/opt/optbuilder/select.go
+++ b/pkg/sql/opt/optbuilder/select.go
@@ -1027,7 +1027,7 @@ func (b *Builder) buildSelect(
 	with := stmt.With
 	orderBy := stmt.OrderBy
 	limit := stmt.Limit
-	locking.apply(stmt.Locking)
+	lockingClause := stmt.Locking
 
 	for s, ok := wrapped.(*tree.ParenSelect); ok; s, ok = wrapped.(*tree.ParenSelect) {
 		stmt = s.Select
@@ -1045,7 +1045,7 @@ func (b *Builder) buildSelect(
 		}
 		if stmt.OrderBy != nil {
 			if orderBy != nil {
-				panic(pgerror.Newf(
+				panic(pgerror.New(
 					pgcode.Syntax, "multiple ORDER BY clauses not allowed",
 				))
 			}
@@ -1053,20 +1053,25 @@ func (b *Builder) buildSelect(
 		}
 		if stmt.Limit != nil {
 			if limit != nil {
-				panic(pgerror.Newf(
+				panic(pgerror.New(
 					pgcode.Syntax, "multiple LIMIT clauses not allowed",
 				))
 			}
 			limit = stmt.Limit
 		}
 		if stmt.Locking != nil {
-			locking.apply(stmt.Locking)
+			if lockingClause != nil {
+				panic(pgerror.New(
+					pgcode.Syntax, "multiple FOR UPDATE clauses not allowed",
+				))
+			}
+			lockingClause = stmt.Locking
 		}
 	}
 
 	return b.processWiths(with, inScope, func(inScope *scope) *scope {
 		return b.buildSelectStmtWithoutParens(
-			wrapped, orderBy, limit, locking, desiredTypes, inScope,
+			wrapped, orderBy, limit, lockingClause, locking, desiredTypes, inScope,
 		)
 	})
 }
@@ -1081,14 +1086,22 @@ func (b *Builder) buildSelectStmtWithoutParens(
 	wrapped tree.SelectStatement,
 	orderBy tree.OrderBy,
 	limit *tree.Limit,
+	lockingClause tree.LockingClause,
 	locking lockingSpec,
 	desiredTypes []*types.T,
 	inScope *scope,
 ) (outScope *scope) {
+	locking.apply(lockingClause)
+
 	// NB: The case statements are sorted lexicographically.
 	switch t := wrapped.(type) {
 	case *tree.LiteralValuesClause:
-		b.rejectIfLocking(locking, "VALUES")
+		// To match Postgres, we only disallow VALUES with FOR UPDATE when the
+		// locking clause is directly on the VALUES statement. Unlike UNION,
+		// INTERSECT, etc, if the VALUES is within a subquery locked by FOR UPDATE
+		// we allow it. This means we only check the immediate lockingClause here,
+		// instead of checking all currently-applied locking.
+		b.rejectIfLocking(lockingSpec(lockingClause), "VALUES")
 		outScope = b.buildLiteralValuesClause(t, desiredTypes, inScope)
 
 	case *tree.ParenSelect:
@@ -1103,7 +1116,7 @@ func (b *Builder) buildSelectStmtWithoutParens(
 		outScope = b.buildUnionClause(t, desiredTypes, inScope)
 
 	case *tree.ValuesClause:
-		b.rejectIfLocking(locking, "VALUES")
+		b.rejectIfLocking(lockingSpec(lockingClause), "VALUES")
 		outScope = b.buildValuesClause(t, desiredTypes, inScope)
 
 	default:

From fb58047a7315330ed844340b6889ab42445eb03e Mon Sep 17 00:00:00 2001
From: Michael Erickson <michae2@cockroachlabs.com>
Date: Tue, 26 Sep 2023 01:42:33 -0700
Subject: [PATCH 2/2] optbuilder: refactor semantic analysis of FOR UPDATE

Locking clauses such as FOR UPDATE and FOR SHARE apply to some or all of
the data sources in a query's FROM list, depending on whether they have
targets (FOR UPDATE OF x). Without targets, they always apply within
subqueries in the FROM list. With targets, they apply within subqueries
if the subquery alias matches the target.

Because of this scope-like nature of FOR UPDATE and FOR SHARE, we
implement semantic analysis using a stack of locking items that grow as
we build inner subqueries deeper in the recursive optbuilder calls.

Prior to this change, we only used the stack of locking items during
buildScan, at the very bottom of the recursion. Because of this, calls
to `lockingSpec.filter` could afford to compress the stack into a single
locking item on our way deeper in the recursion.

As part of the upcoming fix for #75457, however, we will need to build a
new Lock operator when popping off locking items after returning from
the recursion. That Lock operator will need some information gathered
from buildScan at the bottom of the recursion.

To support this, we refactor the stack of locking items to be two
stacks: one that tracks all locking items in scope, and a second that
tracks which locking items currently apply. This will allow buildScan to
associate table information with the correct locking item(s), which can
then be used to build Lock operators when popping the locking items.

As a bonus, by using only the applied locking item stack in
`validateLockingInFrom` we can make validation of SELECT FOR UPDATE
queries a little more precise, which allows some queries we were
incorrectly disallowing.

Informs: #57031, #75457

Epic: CRDB-25322

Release note (sql change): Allow FOR UPDATE on some queries that were
previously disallowed. Queries that use the following operations are now
allowed to have FOR UPDATE OF as long as the prohibited operation is in
a subquery not locked by the FOR UPDATE OF:

- UNION
- INTERSECT
- EXCEPT
- DISTINCT
- GROUP BY
- HAVING
- aggregations
- window functions

For example, the following query is now allowed because the subquery
using the prohibited operations is not affected by the FOR UPDATE OF:

```
SELECT * FROM t,
  (SELECT DISTINCT 0, 0 UNION SELECT a, count(*) FROM t GROUP BY a HAVING a > 0) AS u
  FOR UPDATE OF t;
```

This matches PostgreSQL.
---
 .../testdata/logic_test/select_for_update     |  67 +++++
 pkg/sql/opt/optbuilder/builder.go             |   4 +-
 pkg/sql/opt/optbuilder/join.go                |   6 +-
 pkg/sql/opt/optbuilder/locking.go             | 281 ++++++++++++------
 pkg/sql/opt/optbuilder/mutation_builder.go    |   4 +-
 .../optbuilder/mutation_builder_arbiter.go    |  55 ++--
 pkg/sql/opt/optbuilder/mutation_builder_fk.go |  12 +-
 .../opt/optbuilder/mutation_builder_unique.go |  20 +-
 pkg/sql/opt/optbuilder/select.go              | 186 ++++--------
 pkg/sql/opt/optbuilder/update.go              |   2 +-
 10 files changed, 378 insertions(+), 259 deletions(-)

diff --git a/pkg/sql/logictest/testdata/logic_test/select_for_update b/pkg/sql/logictest/testdata/logic_test/select_for_update
index cbdf2b0896f0..b7b8b705e5d8 100644
--- a/pkg/sql/logictest/testdata/logic_test/select_for_update
+++ b/pkg/sql/logictest/testdata/logic_test/select_for_update
@@ -172,6 +172,14 @@ SELECT 1 UNION SELECT 1 FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with UNION/INTERSECT/EXCEPT
 SELECT * FROM (SELECT 1 UNION SELECT 1) a FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with UNION/INTERSECT/EXCEPT
+SELECT * FROM (SELECT 1 UNION SELECT 1) a, i FOR UPDATE
+
+query II
+SELECT * FROM (SELECT 1 UNION SELECT 1) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with VALUES
 VALUES (1) FOR UPDATE
 
@@ -195,6 +203,16 @@ SELECT * FROM (VALUES (1)) a, i FOR UPDATE
 ----
 1  1
 
+query II
+SELECT * FROM (VALUES (1)) a, i FOR UPDATE OF a
+----
+1  1
+
+query II
+SELECT * FROM (VALUES (1)) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
 SELECT DISTINCT 1 FOR UPDATE
 
@@ -204,6 +222,14 @@ SELECT * FROM (SELECT DISTINCT 1) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
 SELECT * FROM (SELECT DISTINCT 1) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with DISTINCT clause
+SELECT * FROM (SELECT DISTINCT 1) a, i FOR UPDATE OF a
+
+query II
+SELECT * FROM (SELECT DISTINCT 1) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
 SELECT 1 GROUP BY 1 FOR UPDATE
 
@@ -213,6 +239,14 @@ SELECT * FROM (SELECT 1 GROUP BY 1) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
 SELECT * FROM (SELECT 1 GROUP BY 1) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with GROUP BY clause
+SELECT * FROM (SELECT 1 GROUP BY 1) a, i FOR UPDATE OF a
+
+query II
+SELECT * FROM (SELECT 1 GROUP BY 1) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
 SELECT 1 HAVING TRUE FOR UPDATE
 
@@ -222,6 +256,14 @@ SELECT * FROM (SELECT 1 HAVING TRUE) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
 SELECT * FROM (SELECT 1 HAVING TRUE) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with HAVING clause
+SELECT * FROM (SELECT 1 HAVING TRUE) a, i FOR UPDATE OF a
+
+query II
+SELECT * FROM (SELECT 1 HAVING TRUE) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
 SELECT count(1) FOR UPDATE
 
@@ -231,6 +273,14 @@ SELECT * FROM (SELECT count(1)) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
 SELECT * FROM (SELECT count(1)) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with aggregate functions
+SELECT * FROM (SELECT count(1)) a, i FOR UPDATE OF a
+
+query II
+SELECT * FROM (SELECT count(1)) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
 SELECT count(1) OVER () FOR UPDATE
 
@@ -240,6 +290,14 @@ SELECT * FROM (SELECT count(1) OVER ()) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
 SELECT * FROM (SELECT count(1) OVER ()) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with window functions
+SELECT * FROM (SELECT count(1) OVER ()) a, i FOR UPDATE OF a
+
+query II
+SELECT * FROM (SELECT count(1) OVER ()) a, i FOR UPDATE OF i
+----
+1  1
+
 statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
 SELECT generate_series(1, 2) FOR UPDATE
 
@@ -249,6 +307,15 @@ SELECT * FROM (SELECT generate_series(1, 2)) a FOR UPDATE
 statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
 SELECT * FROM (SELECT generate_series(1, 2)) a, i FOR UPDATE
 
+statement error pgcode 0A000 FOR UPDATE is not allowed with set-returning functions in the target list
+SELECT * FROM (SELECT generate_series(1, 2)) a, i FOR UPDATE OF a
+
+query II nosort
+SELECT * FROM (SELECT generate_series(1, 2)) a, i FOR UPDATE OF i
+----
+1  1
+2  1
+
 # Set-returning functions in the from list are allowed.
 query I nosort
 SELECT * FROM generate_series(1, 2) FOR UPDATE
diff --git a/pkg/sql/opt/optbuilder/builder.go b/pkg/sql/opt/optbuilder/builder.go
index 44151f09a0b9..5732d669ba60 100644
--- a/pkg/sql/opt/optbuilder/builder.go
+++ b/pkg/sql/opt/optbuilder/builder.go
@@ -332,10 +332,10 @@ func (b *Builder) buildStmt(
 
 	switch stmt := stmt.(type) {
 	case *tree.Select:
-		return b.buildSelect(stmt, noRowLocking, desiredTypes, inScope)
+		return b.buildSelect(stmt, noLocking, desiredTypes, inScope)
 
 	case *tree.ParenSelect:
-		return b.buildSelect(stmt.Select, noRowLocking, desiredTypes, inScope)
+		return b.buildSelect(stmt.Select, noLocking, desiredTypes, inScope)
 
 	case *tree.Delete:
 		return b.processWiths(stmt.With, inScope, func(inScope *scope) *scope {
diff --git a/pkg/sql/opt/optbuilder/join.go b/pkg/sql/opt/optbuilder/join.go
index be883ffd63a6..b3492f674672 100644
--- a/pkg/sql/opt/optbuilder/join.go
+++ b/pkg/sql/opt/optbuilder/join.go
@@ -31,9 +31,9 @@ import (
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildJoin(
-	join *tree.JoinTableExpr, locking lockingSpec, inScope *scope,
+	join *tree.JoinTableExpr, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
-	leftScope := b.buildDataSource(join.Left, nil /* indexFlags */, locking, inScope)
+	leftScope := b.buildDataSource(join.Left, nil /* indexFlags */, lockCtx, inScope)
 
 	inScopeRight := inScope
 	isLateral := b.exprIsLateral(join.Right)
@@ -45,7 +45,7 @@ func (b *Builder) buildJoin(
 		inScopeRight.context = exprKindLateralJoin
 	}
 
-	rightScope := b.buildDataSource(join.Right, nil /* indexFlags */, locking, inScopeRight)
+	rightScope := b.buildDataSource(join.Right, nil /* indexFlags */, lockCtx, inScopeRight)
 
 	// Check that the same table name is not used on both sides.
 	b.validateJoinTableNames(leftScope, rightScope)
diff --git a/pkg/sql/opt/optbuilder/locking.go b/pkg/sql/opt/optbuilder/locking.go
index f4234b99be0a..48391a56e784 100644
--- a/pkg/sql/opt/optbuilder/locking.go
+++ b/pkg/sql/opt/optbuilder/locking.go
@@ -12,17 +12,18 @@ package optbuilder
 
 import (
 	"github.com/cockroachdb/cockroach/pkg/sql/opt"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgcode"
+	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
+	"github.com/cockroachdb/cockroach/pkg/util/intsets"
+	"github.com/cockroachdb/errors"
 )
 
-// lockingSpec maintains a collection of FOR [KEY] UPDATE/SHARE items that apply
-// to a given scope. Locking clauses can be applied to the lockingSpec as they
-// come into scope in the AST. The lockingSpec can then be consolidated down to
-// a single row-level locking specification for different tables to determine
-// how scans over those tables should perform row-level locking, if at all.
+// lockingItem represents a single FOR UPDATE / FOR SHARE item in a locking
+// clause (perhaps with multiple targets). It wraps a tree.LockingItem with
+// extra information needed for semantic analysis and plan building.
 //
-// A SELECT statement may contain zero, one, or more than one row-level locking
-// clause. Each of these clauses consist of two different properties.
+// A locking item specifies several locking properties.
 //
 // The first property is locking strength (see tree.LockingStrength). Locking
 // strength represents the degree of protection that a row-level lock provides.
@@ -48,21 +49,49 @@ import (
 //	SKIP LOCKED
 //	NOWAIT
 //
-// In addition to these two properties, locking clauses can contain an optional
-// list of target relations. When provided, the locking clause applies only to
-// those relations in the target list. When not provided, the locking clause
-// applies to all relations in the current scope.
+// In addition to these properties, locking items can contain an optional list
+// of target relations. When provided, the locking item applies only to those
+// relations in the target list. When not provided, the locking item applies to
+// all relations in the current scope.
+//
+// Locking clauses consist of multiple locking items.
 //
-// Put together, a complex locking spec might look like:
+// For example, a complex locking clause might look like:
 //
 //	SELECT ... FROM ... FOR SHARE NOWAIT FOR UPDATE OF t1, t2
 //
-// which would be represented as:
+// which would be represented as two locking items:
 //
 //	[ {ForShare, LockWaitError, []}, {ForUpdate, LockWaitBlock, [t1, t2]} ]
-type lockingSpec []*tree.LockingItem
+type lockingItem struct {
+	item *tree.LockingItem
+
+	// targetsFound is used to validate that we matched all of the lock targets.
+	targetsFound intsets.Fast
+}
+
+// lockingSpec maintains a collection of FOR [KEY] UPDATE/SHARE items that apply
+// to the current scope. Locking items can apply as they come into scope in the
+// AST, or as data sources match locking targets within FROM lists.
+//
+// For example, for a statement like:
+//
+//	SELECT * FROM a, (SELECT * FROM b, c FOR SHARE NOWAIT FOR UPDATE OF c) FOR SHARE
+//
+// while building each scan, the lockingSpec would look different:
+//
+//   - while building a it would be:
+//     [{ForShare, LockWaitBlock, []}]
+//
+//   - while building b it would be:
+//     [{ForShare, LockWaitBlock, []}, {ForShare, LockWaitError, []}]
+//
+//   - while building c it would be:
+//     [{ForShare, LockWaitBlock, []}, {ForShare, LockWaitError, []}, {ForUpdate, LockWaitBlock, [c]}]
+type lockingSpec []*lockingItem
 
-// noRowLocking indicates that no row-level locking has been specified.
+// noRowLocking indicates that no row-level locking applies to the current
+// scope.
 var noRowLocking lockingSpec
 
 // isSet returns whether the spec contains any row-level locking modes.
@@ -70,97 +99,94 @@ func (lm lockingSpec) isSet() bool {
 	return len(lm) != 0
 }
 
-// get returns the first row-level locking mode in the spec. If the spec was the
-// outcome of filter operation, this will be the only locking mode in the spec.
+// get returns the combined row-level locking mode from all currently-applied
+// locking items.
 func (lm lockingSpec) get() opt.Locking {
-	if lm.isSet() {
-		spec := lm[0]
-		return opt.Locking{
+	var l opt.Locking
+	for _, li := range lm {
+		spec := li.item
+		l = l.Max(opt.Locking{
 			Strength:   spec.Strength,
 			WaitPolicy: spec.WaitPolicy,
 			Form:       spec.Form,
-		}
+		})
 	}
-	return opt.Locking{}
+	return l
+}
+
+// lockingContext holds the locking information for the current scope.
+type lockingContext struct {
+	// lockScope is the stack of locking items that are currently in scope. This
+	// might include locking items that do not currently apply because they have
+	// an unmatched target.
+	lockScope []*lockingItem
+
+	// locking is the stack of locking items that apply to the current scope,
+	// either because they did not have a target or because one of their targets
+	// matched an ancestor of this scope.
+	locking lockingSpec
 }
 
-// apply merges the locking clause into the current locking spec. The effect of
-// applying new locking clauses to an existing spec is always to strengthen the
-// locking approaches it represents, either through increasing locking strength
-// or using more aggressive wait policies.
-func (lm *lockingSpec) apply(locking tree.LockingClause) {
-	// TODO(nvanbenschoten): If we wanted to eagerly prune superfluous locking
-	// items so that they don't need to get merged away in each call to filter,
-	// this would be the place to do it. We don't expect to see multiple FOR
-	// UPDATE clauses very often, so it's probably not worth it.
-	if len(*lm) == 0 {
-		// NB: avoid allocation, but also prevent future mutation of AST.
-		l := len(locking)
-		*lm = lockingSpec(locking[:l:l])
-		return
+// noLocking indicates that no row-level locking has been specified.
+var noLocking lockingContext
+
+// push pushes a locking item onto the scope stack, and also applies it if it
+// has no targets.
+func (lockCtx *lockingContext) push(li *tree.LockingItem) {
+	item := &lockingItem{
+		item: li,
+	}
+	lockCtx.lockScope = append(lockCtx.lockScope, item)
+	if len(li.Targets) == 0 {
+		lockCtx.locking = append(lockCtx.locking, item)
 	}
-	*lm = append(*lm, locking...)
 }
 
-// filter returns the desired row-level locking mode for the specified table as
-// a new consolidated lockingSpec. If no matching locking mode is found then the
-// resulting spec will remain un-set. If a matching locking mode for the table
-// is found then the resulting spec will contain exclusively that locking mode
-// and will no longer be restricted to specific target relations.
-func (lm lockingSpec) filter(alias tree.Name) lockingSpec {
-	var ret lockingSpec
-	var copied bool
-	updateRet := func(li *tree.LockingItem, len1 []*tree.LockingItem) {
-		if ret == nil && len(li.Targets) == 0 {
-			// Fast-path. We don't want the resulting spec to include targets,
-			// so we only allow this if the item we want to copy has none.
-			ret = len1
-			return
-		}
-		if !copied {
-			retCpy := make(lockingSpec, 1)
-			retCpy[0] = new(tree.LockingItem)
-			if len(ret) == 1 {
-				*retCpy[0] = *ret[0]
-			}
-			ret = retCpy
-			copied = true
-		}
-		// From https://www.postgresql.org/docs/12/sql-select.html#SQL-FOR-UPDATE-SHARE
-		// > If the same table is mentioned (or implicitly affected) by more
-		// > than one locking clause, then it is processed as if it was only
-		// > specified by the strongest one.
-		ret[0].Strength = ret[0].Strength.Max(li.Strength)
-		// > Similarly, a table is processed as NOWAIT if that is specified in
-		// > any of the clauses affecting it. Otherwise, it is processed as SKIP
-		// > LOCKED if that is specified in any of the clauses affecting it.
-		ret[0].WaitPolicy = ret[0].WaitPolicy.Max(li.WaitPolicy)
-		// We assume the same behavior for locking form.
-		ret[0].Form = ret[0].Form.Max(li.Form)
+// pop removes and returns the topmost locking item from the scope stack.
+func (lockCtx *lockingContext) pop() *lockingItem {
+	n := len(lockCtx.lockScope)
+	if n == 0 {
+		panic(errors.AssertionFailedf("tried to pop non-existent locking item"))
 	}
+	item := lockCtx.lockScope[n-1]
+	lockCtx.lockScope = lockCtx.lockScope[:n-1]
+	// For now we do not bother explicitly popping the lockingSpec stack. Instead
+	// we rely on passing lockingContext by value in optbuilder, meaning
+	// lockingSpec is implicitly popped when returning.
+	return item
+}
+
+// blankLockingScope is a sentinel locking item that, when pushed, prevents
+// lockCtx.filter from matching targets outside it.
+var blankLockingScope lockingItem = lockingItem{item: &tree.LockingItem{}}
 
-	for i, li := range lm {
-		len1 := lm[i : i+1 : i+1]
-		if len(li.Targets) == 0 {
-			// If no targets are specified, the clause affects all tables.
-			updateRet(li, len1)
-		} else {
-			// If targets are specified, the clause affects only those tables.
-			for _, target := range li.Targets {
-				if target.ObjectName == alias {
-					updateRet(li, len1)
-					break
-				}
+// filter applies any locking items that match the specified data source alias.
+func (lockCtx *lockingContext) filter(alias tree.Name) {
+	// Search backward through locking scopes to find all of the matching items
+	// inside the innermost blankLockingScope. Unlike for variable scopes, for
+	// locking scopes *all* of the matching items apply, not just the first. This
+	// means in some cases we might apply the same locking item multiple times, if
+	// it has multiple targets and they match more than once. This is fine.
+	for i := len(lockCtx.lockScope) - 1; i >= 0; i-- {
+		item := lockCtx.lockScope[i]
+		if item == &blankLockingScope {
+			break
+		}
+		// Only consider locking items with targets. (Locking items without targets
+		// were already applied in push.)
+		for i, target := range item.item.Targets {
+			if target.ObjectName == alias {
+				lockCtx.locking = append(lockCtx.locking, item)
+				item.targetsFound.Add(i)
 			}
 		}
 	}
-	return ret
 }
 
-// withoutTargets returns a new lockingSpec with all locking clauses that apply
-// only to a subset of tables removed.
-func (lm lockingSpec) withoutTargets() lockingSpec {
-	return lm.filter("")
+// withoutTargets hides all unapplied locking items in scope, so that they
+// cannot be applied. Already applied locking items remain applied.
+func (lockCtx *lockingContext) withoutTargets() {
+	lockCtx.lockScope = append(lockCtx.lockScope, &blankLockingScope)
 }
 
 // ignoreLockingForCTE is a placeholder for the following comment:
@@ -175,3 +201,80 @@ func (lm lockingSpec) withoutTargets() lockingSpec {
 // > If you want row locking to occur within a WITH query, specify a locking
 // > clause within the WITH query.
 func (lm lockingSpec) ignoreLockingForCTE() {}
+
+// validate checks that the locking item is well-formed, and that all of its
+// targets matched a data source in the FROM clause.
+func (item *lockingItem) validate() {
+	li := item.item
+
+	// Validate locking strength.
+	switch li.Strength {
+	case tree.ForNone:
+		// AST nodes should not be created with this locking strength.
+		panic(errors.AssertionFailedf("locking item without strength"))
+	case tree.ForUpdate:
+		// Exclusive locking on the entire row.
+	case tree.ForNoKeyUpdate:
+		// Exclusive locking on only non-key(s) of the row. Currently unimplemented
+		// and treated identically to ForUpdate.
+	case tree.ForShare:
+		// Shared locking on the entire row.
+	case tree.ForKeyShare:
+		// Shared locking on only key(s) of the row. Currently unimplemented and
+		// treated identically to ForShare.
+	default:
+		panic(errors.AssertionFailedf("unknown locking strength: %d", li.Strength))
+	}
+
+	// Validating locking wait policy.
+	switch li.WaitPolicy {
+	case tree.LockWaitBlock:
+		// Default. Block on conflicting locks.
+	case tree.LockWaitSkipLocked:
+		// Skip rows that can't be locked.
+	case tree.LockWaitError:
+		// Raise an error on conflicting locks.
+	default:
+		panic(errors.AssertionFailedf("unknown locking wait policy: %d", li.WaitPolicy))
+	}
+
+	// Validate locking form.
+	switch li.Form {
+	case tree.LockRecord:
+		// Default. Only lock existing rows.
+	case tree.LockPredicate:
+		// Lock both existing rows and gaps between rows.
+	default:
+		panic(errors.AssertionFailedf("unknown locking form: %d", li.Form))
+	}
+
+	// Validate locking targets by checking that all targets are well-formed and
+	// all were found somewhere in the FROM clause.
+	for i, target := range li.Targets {
+		// Insist on unqualified alias names here. We could probably do
+		// something smarter, but it's better to just mirror Postgres
+		// exactly. See transformLockingClause in Postgres' source.
+		if target.CatalogName != "" || target.SchemaName != "" {
+			panic(pgerror.Newf(pgcode.Syntax,
+				"%s must specify unqualified relation names", li.Strength))
+		}
+		// Validate that at some point we found this target.
+		if !item.targetsFound.Contains(i) {
+			panic(pgerror.Newf(
+				pgcode.UndefinedTable,
+				"relation %q in %s clause not found in FROM clause",
+				target.ObjectName, li.Strength,
+			))
+		}
+	}
+}
+
+// lockingSpecForClause converts a lockingClause to a lockingSpec.
+func lockingSpecForClause(lockingClause tree.LockingClause) (lm lockingSpec) {
+	for _, li := range lockingClause {
+		lm = append(lm, &lockingItem{
+			item: li,
+		})
+	}
+	return lm
+}
diff --git a/pkg/sql/opt/optbuilder/mutation_builder.go b/pkg/sql/opt/optbuilder/mutation_builder.go
index 3b5583d4d536..8a31ff49a9b6 100644
--- a/pkg/sql/opt/optbuilder/mutation_builder.go
+++ b/pkg/sql/opt/optbuilder/mutation_builder.go
@@ -314,7 +314,7 @@ func (mb *mutationBuilder) buildInputForUpdate(
 	// together with the table being updated.
 	fromClausePresent := len(from) > 0
 	if fromClausePresent {
-		fromScope := mb.b.buildFromTables(from, noRowLocking, inScope)
+		fromScope := mb.b.buildFromTables(from, noLocking, inScope)
 
 		// Check that the same table name is not used multiple times.
 		mb.b.validateJoinTableNames(mb.fetchScope, fromScope)
@@ -424,7 +424,7 @@ func (mb *mutationBuilder) buildInputForDelete(
 	// USING
 	usingClausePresent := len(using) > 0
 	if usingClausePresent {
-		usingScope := mb.b.buildFromTables(using, noRowLocking, inScope)
+		usingScope := mb.b.buildFromTables(using, noLocking, inScope)
 
 		// Check that the same table name is not used multiple times.
 		mb.b.validateJoinTableNames(mb.fetchScope, usingScope)
diff --git a/pkg/sql/opt/optbuilder/mutation_builder_arbiter.go b/pkg/sql/opt/optbuilder/mutation_builder_arbiter.go
index 664348b058ec..bf21d5432566 100644
--- a/pkg/sql/opt/optbuilder/mutation_builder_arbiter.go
+++ b/pkg/sql/opt/optbuilder/mutation_builder_arbiter.go
@@ -292,20 +292,23 @@ func (mb *mutationBuilder) buildAntiJoinForDoNothingArbiter(
 		h := &mb.uniqueCheckHelper
 		if h.init(mb, uniqueOrd) {
 			locking = lockingSpec{
-				&tree.LockingItem{
-					// TODO(michae2): Change this to ForKeyShare when it is supported.
-					// Actually, for INSERT ON CONFLICT DO NOTHING, I think this could be
-					// ForNone if we supported predicate locking at that strength. I'm
-					// pretty sure we don't need to lock existing rows at *any* locking
-					// strength, only need to prevent insertion of new non-existing rows.
-					Strength:   tree.ForShare,
-					Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(mb.tab.Name())},
-					WaitPolicy: tree.LockWaitBlock,
-					// Unique arbiters must ensure the non-existence of certain rows, so
-					// we use predicate locks instead of record locks to prevent insertion
-					// of new rows into the locked span(s) by other concurrent
-					// transactions.
-					Form: tree.LockPredicate,
+				&lockingItem{
+					item: &tree.LockingItem{
+						// TODO(michae2): Change this to ForKeyShare when it is supported.
+						// Actually, for INSERT ON CONFLICT DO NOTHING, I think this could
+						// be ForNone if we supported predicate locking at that
+						// strength. I'm pretty sure we don't need to lock existing rows at
+						// *any* locking strength, only need to prevent insertion of new
+						// non-existing rows.
+						Strength:   tree.ForShare,
+						Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(mb.tab.Name())},
+						WaitPolicy: tree.LockWaitBlock,
+						// Unique arbiters must ensure the non-existence of certain rows, so
+						// we use predicate locks instead of record locks to prevent
+						// insertion of new rows into the locked span(s) by other concurrent
+						// transactions.
+						Form: tree.LockPredicate,
+					},
 				},
 			}
 		}
@@ -416,17 +419,19 @@ func (mb *mutationBuilder) buildLeftJoinForUpsertArbiter(
 		h := &mb.uniqueCheckHelper
 		if h.init(mb, uniqueOrd) {
 			locking = lockingSpec{
-				&tree.LockingItem{
-					// If the row exists, we're about to update it, so take an exclusive
-					// lock to prevent a lock promotion.
-					Strength:   tree.ForUpdate,
-					Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(mb.tab.Name())},
-					WaitPolicy: tree.LockWaitBlock,
-					// Unique arbiters must ensure the non-existence of certain rows, so
-					// we use predicate locks instead of record locks to prevent insertion
-					// of new rows into the locked span(s) by other concurrent
-					// transactions.
-					Form: tree.LockPredicate,
+				&lockingItem{
+					item: &tree.LockingItem{
+						// If the row exists, we're about to update it, so take an exclusive
+						// lock to prevent a lock promotion.
+						Strength:   tree.ForUpdate,
+						Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(mb.tab.Name())},
+						WaitPolicy: tree.LockWaitBlock,
+						// Unique arbiters must ensure the non-existence of certain rows, so
+						// we use predicate locks instead of record locks to prevent
+						// insertion of new rows into the locked span(s) by other concurrent
+						// transactions.
+						Form: tree.LockPredicate,
+					},
 				},
 			}
 		}
diff --git a/pkg/sql/opt/optbuilder/mutation_builder_fk.go b/pkg/sql/opt/optbuilder/mutation_builder_fk.go
index c16f1425bdd9..7206f01a0df6 100644
--- a/pkg/sql/opt/optbuilder/mutation_builder_fk.go
+++ b/pkg/sql/opt/optbuilder/mutation_builder_fk.go
@@ -648,11 +648,13 @@ func (h *fkCheckHelper) buildOtherTableScan(parent bool) (outScope *scope, tabMe
 	if parent && (h.mb.b.evalCtx.TxnIsoLevel != isolation.Serializable ||
 		h.mb.b.evalCtx.SessionData().ImplicitFKLockingForSerializable) {
 		locking = lockingSpec{
-			&tree.LockingItem{
-				// TODO(michae2): Change this to ForKeyShare when it is supported.
-				Strength:   tree.ForShare,
-				Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(h.otherTab.Name())},
-				WaitPolicy: tree.LockWaitBlock,
+			&lockingItem{
+				item: &tree.LockingItem{
+					// TODO(michae2): Change this to ForKeyShare when it is supported.
+					Strength:   tree.ForShare,
+					Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(h.otherTab.Name())},
+					WaitPolicy: tree.LockWaitBlock,
+				},
 			},
 		}
 	}
diff --git a/pkg/sql/opt/optbuilder/mutation_builder_unique.go b/pkg/sql/opt/optbuilder/mutation_builder_unique.go
index f31317599818..e2873cebf528 100644
--- a/pkg/sql/opt/optbuilder/mutation_builder_unique.go
+++ b/pkg/sql/opt/optbuilder/mutation_builder_unique.go
@@ -582,15 +582,17 @@ func (h *uniqueCheckHelper) buildTableScan() (outScope *scope, ordinals []int) {
 	// unique constraint.
 	if h.mb.b.evalCtx.TxnIsoLevel != isolation.Serializable {
 		locking = lockingSpec{
-			&tree.LockingItem{
-				// TODO(michae2): Change this to ForKeyShare when it is supported.
-				Strength:   tree.ForShare,
-				Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(h.mb.tab.Name())},
-				WaitPolicy: tree.LockWaitBlock,
-				// Unique checks must ensure the non-existence of certain rows, so we
-				// use predicate locks instead of record locks to prevent insertion of
-				// new rows into the locked span(s) by other concurrent transactions.
-				Form: tree.LockPredicate,
+			&lockingItem{
+				item: &tree.LockingItem{
+					// TODO(michae2): Change this to ForKeyShare when it is supported.
+					Strength:   tree.ForShare,
+					Targets:    []tree.TableName{tree.MakeUnqualifiedTableName(h.mb.tab.Name())},
+					WaitPolicy: tree.LockWaitBlock,
+					// Unique checks must ensure the non-existence of certain rows, so we
+					// use predicate locks instead of record locks to prevent insertion of
+					// new rows into the locked span(s) by other concurrent transactions.
+					Form: tree.LockPredicate,
+				},
 			},
 		}
 	}
diff --git a/pkg/sql/opt/optbuilder/select.go b/pkg/sql/opt/optbuilder/select.go
index 195ba2347965..f572462a02b0 100644
--- a/pkg/sql/opt/optbuilder/select.go
+++ b/pkg/sql/opt/optbuilder/select.go
@@ -50,7 +50,7 @@ import (
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildDataSource(
-	texpr tree.TableExpr, indexFlags *tree.IndexFlags, locking lockingSpec, inScope *scope,
+	texpr tree.TableExpr, indexFlags *tree.IndexFlags, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
 	defer func(prevAtRoot bool, prevInsideDataSource bool) {
 		inScope.atRoot = prevAtRoot
@@ -87,10 +87,11 @@ func (b *Builder) buildDataSource(
 
 		if source.As.Alias != "" {
 			inScope.alias = &source.As
-			locking = locking.filter(source.As.Alias)
+			lockCtx.filter(source.As.Alias)
+			lockCtx.withoutTargets()
 		}
 
-		outScope = b.buildDataSource(source.Expr, indexFlags, locking, inScope)
+		outScope = b.buildDataSource(source.Expr, indexFlags, lockCtx, inScope)
 
 		if source.Ordinality {
 			outScope = b.buildWithOrdinality(outScope)
@@ -102,14 +103,14 @@ func (b *Builder) buildDataSource(
 		return outScope
 
 	case *tree.JoinTableExpr:
-		return b.buildJoin(source, locking, inScope)
+		return b.buildJoin(source, lockCtx, inScope)
 
 	case *tree.TableName:
 		tn := source
 
 		// CTEs take precedence over other data sources.
 		if cte := inScope.resolveCTE(tn); cte != nil {
-			locking.ignoreLockingForCTE()
+			lockCtx.locking.ignoreLockingForCTE()
 			outScope = inScope.push()
 			inCols := make(opt.ColList, len(cte.cols), len(cte.cols)+len(inScope.ordering))
 			outCols := make(opt.ColList, len(cte.cols), len(cte.cols)+len(inScope.ordering))
@@ -136,8 +137,8 @@ func (b *Builder) buildDataSource(
 		}
 
 		ds, depName, resName := b.resolveDataSource(tn, privilege.SELECT)
-		locking = locking.filter(tn.ObjectName)
-		if locking.isSet() {
+		lockCtx.filter(tn.ObjectName)
+		if lockCtx.locking.isSet() {
 			// SELECT ... FOR [KEY] UPDATE/SHARE also requires UPDATE privileges.
 			b.checkPrivilege(depName, ds, privilege.UPDATE)
 		}
@@ -152,7 +153,7 @@ func (b *Builder) buildDataSource(
 					includeSystem:    true,
 					includeInverted:  false,
 				}),
-				indexFlags, locking, inScope,
+				indexFlags, lockCtx.locking, inScope,
 				false, /* disableNotVisibleIndex */
 			)
 
@@ -160,14 +161,14 @@ func (b *Builder) buildDataSource(
 			return b.buildSequenceSelect(t, &resName, inScope)
 
 		case cat.View:
-			return b.buildView(t, &resName, locking, inScope)
+			return b.buildView(t, &resName, lockCtx, inScope)
 
 		default:
 			panic(errors.AssertionFailedf("unknown DataSource type %T", ds))
 		}
 
 	case *tree.ParenTableExpr:
-		return b.buildDataSource(source.Expr, indexFlags, locking, inScope)
+		return b.buildDataSource(source.Expr, indexFlags, lockCtx, inScope)
 
 	case *tree.RowsFromExpr:
 		return b.buildZip(source.Items, inScope)
@@ -175,11 +176,11 @@ func (b *Builder) buildDataSource(
 	case *tree.Subquery:
 		// Remove any target relations from the current scope's locking spec, as
 		// those only apply to relations in this statement. Interestingly, this
-		// would not be necessary if we required all subqueries to have aliases
-		// like Postgres does.
-		locking = locking.withoutTargets()
+		// would not be necessary if we required all subqueries to have aliases like
+		// Postgres does, because then the AliasedTableExpr case would handle this.
+		lockCtx.withoutTargets()
 
-		outScope = b.buildSelectStmt(source.Select, locking, nil /* desiredTypes */, inScope)
+		outScope = b.buildSelectStmt(source.Select, lockCtx, nil /* desiredTypes */, inScope)
 
 		// Treat the subquery result as an anonymous data source (i.e. column names
 		// are not qualified). Remove hidden columns, as they are not accessible
@@ -225,7 +226,7 @@ func (b *Builder) buildDataSource(
 			outCols[i] = b.factory.Metadata().AddColumn(col.Alias, c.Type)
 		}
 
-		locking.ignoreLockingForCTE()
+		lockCtx.locking.ignoreLockingForCTE()
 		outScope = inScope.push()
 		// Similar to appendColumnsFromScope, but with re-numbering the column IDs.
 		for i, col := range innerScope.cols {
@@ -248,15 +249,15 @@ func (b *Builder) buildDataSource(
 	case *tree.TableRef:
 		ds, depName := b.resolveDataSourceRef(source, privilege.SELECT)
 
-		locking = locking.filter(source.As.Alias)
-		if locking.isSet() {
+		lockCtx.filter(source.As.Alias)
+		if lockCtx.locking.isSet() {
 			// SELECT ... FOR [KEY] UPDATE/SHARE also requires UPDATE privileges.
 			b.checkPrivilege(depName, ds, privilege.UPDATE)
 		}
 
 		switch t := ds.(type) {
 		case cat.Table:
-			outScope = b.buildScanFromTableRef(t, source, indexFlags, locking, inScope)
+			outScope = b.buildScanFromTableRef(t, source, indexFlags, lockCtx.locking, inScope)
 		case cat.View:
 			if source.Columns != nil {
 				panic(pgerror.Newf(pgcode.FeatureNotSupported,
@@ -264,7 +265,7 @@ func (b *Builder) buildDataSource(
 			}
 			tn := tree.MakeUnqualifiedTableName(t.Name())
 
-			outScope = b.buildView(t, &tn, locking, inScope)
+			outScope = b.buildView(t, &tn, lockCtx, inScope)
 		case cat.Sequence:
 			tn := tree.MakeUnqualifiedTableName(t.Name())
 			// Any explicitly listed columns are ignored.
@@ -282,7 +283,7 @@ func (b *Builder) buildDataSource(
 
 // buildView parses the view query text and builds it as a Select expression.
 func (b *Builder) buildView(
-	view cat.View, viewName *tree.TableName, locking lockingSpec, inScope *scope,
+	view cat.View, viewName *tree.TableName, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
 	if b.sourceViews == nil {
 		b.sourceViews = make(map[string]struct{})
@@ -345,7 +346,8 @@ func (b *Builder) buildView(
 	// to the existing scope chain because we want the rest of the query to be
 	// able to refer to the higher scopes (see #46180).
 	emptyScope := b.allocScope()
-	outScope = b.buildSelect(sel, locking, nil /* desiredTypes */, emptyScope)
+	lockCtx.withoutTargets()
+	outScope = b.buildSelect(sel, lockCtx, nil /* desiredTypes */, emptyScope)
 	emptyScope.parent = inScope
 
 	// Update data source name to be the name of the view. And if view columns
@@ -458,7 +460,9 @@ func (b *Builder) buildScanFromTableRef(
 	tn := tree.MakeUnqualifiedTableName(tab.Name())
 	tabMeta := b.addTable(tab, &tn)
 
-	return b.buildScan(tabMeta, ordinals, indexFlags, locking, inScope, false /* disableNotVisibleIndex */)
+	return b.buildScan(
+		tabMeta, ordinals, indexFlags, locking, inScope, false, /* disableNotVisibleIndex */
+	)
 }
 
 // addTable adds a table to the metadata and returns the TableMeta. The table
@@ -985,7 +989,7 @@ func (b *Builder) buildWithOrdinality(inScope *scope) (outScope *scope) {
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildSelectStmt(
-	stmt tree.SelectStatement, locking lockingSpec, desiredTypes []*types.T, inScope *scope,
+	stmt tree.SelectStatement, lockCtx lockingContext, desiredTypes []*types.T, inScope *scope,
 ) (outScope *scope) {
 	// The top level in a select statement is not considered a data source.
 	oldInsideDataSource := b.insideDataSource
@@ -999,10 +1003,10 @@ func (b *Builder) buildSelectStmt(
 		return b.buildLiteralValuesClause(stmt, desiredTypes, inScope)
 
 	case *tree.ParenSelect:
-		return b.buildSelect(stmt.Select, locking, desiredTypes, inScope)
+		return b.buildSelect(stmt.Select, lockCtx, desiredTypes, inScope)
 
 	case *tree.SelectClause:
-		return b.buildSelectClause(stmt, nil /* orderBy */, locking, desiredTypes, inScope)
+		return b.buildSelectClause(stmt, nil /* orderBy */, lockCtx, desiredTypes, inScope)
 
 	case *tree.UnionClause:
 		return b.buildUnionClause(stmt, desiredTypes, inScope)
@@ -1021,7 +1025,7 @@ func (b *Builder) buildSelectStmt(
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildSelect(
-	stmt *tree.Select, locking lockingSpec, desiredTypes []*types.T, inScope *scope,
+	stmt *tree.Select, lockCtx lockingContext, desiredTypes []*types.T, inScope *scope,
 ) (outScope *scope) {
 	wrapped := stmt.Select
 	with := stmt.With
@@ -1071,7 +1075,7 @@ func (b *Builder) buildSelect(
 
 	return b.processWiths(with, inScope, func(inScope *scope) *scope {
 		return b.buildSelectStmtWithoutParens(
-			wrapped, orderBy, limit, lockingClause, locking, desiredTypes, inScope,
+			wrapped, orderBy, limit, lockingClause, lockCtx, desiredTypes, inScope,
 		)
 	})
 }
@@ -1087,11 +1091,14 @@ func (b *Builder) buildSelectStmtWithoutParens(
 	orderBy tree.OrderBy,
 	limit *tree.Limit,
 	lockingClause tree.LockingClause,
-	locking lockingSpec,
+	lockCtx lockingContext,
 	desiredTypes []*types.T,
 	inScope *scope,
 ) (outScope *scope) {
-	locking.apply(lockingClause)
+	// Push locking items (FOR UPDATE) in reverse order to match Postgres.
+	for i := len(lockingClause) - 1; i >= 0; i-- {
+		lockCtx.push(lockingClause[i])
+	}
 
 	// NB: The case statements are sorted lexicographically.
 	switch t := wrapped.(type) {
@@ -1101,7 +1108,7 @@ func (b *Builder) buildSelectStmtWithoutParens(
 		// INTERSECT, etc, if the VALUES is within a subquery locked by FOR UPDATE
 		// we allow it. This means we only check the immediate lockingClause here,
 		// instead of checking all currently-applied locking.
-		b.rejectIfLocking(lockingSpec(lockingClause), "VALUES")
+		b.rejectIfLocking(lockingSpecForClause(lockingClause), "VALUES")
 		outScope = b.buildLiteralValuesClause(t, desiredTypes, inScope)
 
 	case *tree.ParenSelect:
@@ -1109,14 +1116,14 @@ func (b *Builder) buildSelectStmtWithoutParens(
 			"%T in buildSelectStmtWithoutParens", wrapped))
 
 	case *tree.SelectClause:
-		outScope = b.buildSelectClause(t, orderBy, locking, desiredTypes, inScope)
+		outScope = b.buildSelectClause(t, orderBy, lockCtx, desiredTypes, inScope)
 
 	case *tree.UnionClause:
-		b.rejectIfLocking(locking, "UNION/INTERSECT/EXCEPT")
+		b.rejectIfLocking(lockCtx.locking, "UNION/INTERSECT/EXCEPT")
 		outScope = b.buildUnionClause(t, desiredTypes, inScope)
 
 	case *tree.ValuesClause:
-		b.rejectIfLocking(lockingSpec(lockingClause), "VALUES")
+		b.rejectIfLocking(lockingSpecForClause(lockingClause), "VALUES")
 		outScope = b.buildValuesClause(t, desiredTypes, inScope)
 
 	default:
@@ -1143,6 +1150,13 @@ func (b *Builder) buildSelectStmtWithoutParens(
 		b.buildLimit(limit, inScope, outScope)
 	}
 
+	// Remove locking items from scope, and validate that they were found within
+	// the FROM clause.
+	for range lockingClause {
+		item := lockCtx.pop()
+		item.validate()
+	}
+
 	// TODO(rytaft): Support FILTER expression.
 	return outScope
 }
@@ -1158,11 +1172,11 @@ func (b *Builder) buildSelectStmtWithoutParens(
 func (b *Builder) buildSelectClause(
 	sel *tree.SelectClause,
 	orderBy tree.OrderBy,
-	locking lockingSpec,
+	lockCtx lockingContext,
 	desiredTypes []*types.T,
 	inScope *scope,
 ) (outScope *scope) {
-	fromScope := b.buildFrom(sel.From, locking, inScope)
+	fromScope := b.buildFrom(sel.From, lockCtx, inScope)
 
 	b.processWindowDefs(sel, fromScope)
 	b.buildWhere(sel.Where, fromScope)
@@ -1207,7 +1221,7 @@ func (b *Builder) buildSelectClause(
 	}
 
 	b.buildWindow(outScope, fromScope)
-	b.validateLockingInFrom(sel, locking, fromScope)
+	b.validateLockingInFrom(sel, lockCtx.locking, fromScope)
 
 	// Construct the projection.
 	b.constructProjectForScope(outScope, projectionsScope)
@@ -1232,7 +1246,9 @@ func (b *Builder) buildSelectClause(
 //
 // See Builder.buildStmt for a description of the remaining input and return
 // values.
-func (b *Builder) buildFrom(from tree.From, locking lockingSpec, inScope *scope) (outScope *scope) {
+func (b *Builder) buildFrom(
+	from tree.From, lockCtx lockingContext, inScope *scope,
+) (outScope *scope) {
 	// The root AS OF clause is recognized and handled by the executor. The only
 	// thing that must be done at this point is to ensure that if any timestamps
 	// are specified, the root SELECT was an AS OF SYSTEM TIME and that the time
@@ -1242,7 +1258,7 @@ func (b *Builder) buildFrom(from tree.From, locking lockingSpec, inScope *scope)
 	}
 
 	if len(from.Tables) > 0 {
-		outScope = b.buildFromTables(from.Tables, locking, inScope)
+		outScope = b.buildFromTables(from.Tables, lockCtx, inScope)
 	} else {
 		outScope = inScope.push()
 		outScope.expr = b.factory.ConstructValues(memo.ScalarListWithEmptyTuple, &memo.ValuesPrivate{
@@ -1305,17 +1321,17 @@ func (b *Builder) buildWhere(where *tree.Where, inScope *scope) {
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildFromTables(
-	tables tree.TableExprs, locking lockingSpec, inScope *scope,
+	tables tree.TableExprs, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
 	// If there are any lateral data sources, we need to build the join tree
 	// left-deep instead of right-deep.
 	for i := range tables {
 		if b.exprIsLateral(tables[i]) {
 			telemetry.Inc(sqltelemetry.LateralJoinUseCounter)
-			return b.buildFromWithLateral(tables, locking, inScope)
+			return b.buildFromWithLateral(tables, lockCtx, inScope)
 		}
 	}
-	return b.buildFromTablesRightDeep(tables, locking, inScope)
+	return b.buildFromTablesRightDeep(tables, lockCtx, inScope)
 }
 
 // buildFromTablesRightDeep recursively builds a series of InnerJoin
@@ -1336,16 +1352,16 @@ func (b *Builder) buildFromTables(
 // See Builder.buildStmt for a description of the remaining input and
 // return values.
 func (b *Builder) buildFromTablesRightDeep(
-	tables tree.TableExprs, locking lockingSpec, inScope *scope,
+	tables tree.TableExprs, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
-	outScope = b.buildDataSource(tables[0], nil /* indexFlags */, locking, inScope)
+	outScope = b.buildDataSource(tables[0], nil /* indexFlags */, lockCtx, inScope)
 
 	// Recursively build table join.
 	tables = tables[1:]
 	if len(tables) == 0 {
 		return outScope
 	}
-	tableScope := b.buildFromTablesRightDeep(tables, locking, inScope)
+	tableScope := b.buildFromTablesRightDeep(tables, lockCtx, inScope)
 
 	// Check that the same table name is not used multiple times.
 	b.validateJoinTableNames(outScope, tableScope)
@@ -1386,9 +1402,9 @@ func (b *Builder) exprIsLateral(t tree.TableExpr) bool {
 //	buildFromTablesRightDeep: a JOIN (b JOIN c)
 //	buildFromWithLateral:     (a JOIN b) JOIN c
 func (b *Builder) buildFromWithLateral(
-	tables tree.TableExprs, locking lockingSpec, inScope *scope,
+	tables tree.TableExprs, lockCtx lockingContext, inScope *scope,
 ) (outScope *scope) {
-	outScope = b.buildDataSource(tables[0], nil /* indexFlags */, locking, inScope)
+	outScope = b.buildDataSource(tables[0], nil /* indexFlags */, lockCtx, inScope)
 	for i := 1; i < len(tables); i++ {
 		scope := inScope
 		// Lateral expressions need to be able to refer to the expressions that
@@ -1397,7 +1413,7 @@ func (b *Builder) buildFromWithLateral(
 			scope = outScope
 			scope.context = exprKindLateralJoin
 		}
-		tableScope := b.buildDataSource(tables[i], nil /* indexFlags */, locking, scope)
+		tableScope := b.buildDataSource(tables[i], nil /* indexFlags */, lockCtx, scope)
 
 		// Check that the same table name is not used multiple times.
 		b.validateJoinTableNames(outScope, tableScope)
@@ -1470,82 +1486,6 @@ func (b *Builder) validateLockingInFrom(
 	case len(fromScope.srfs) != 0:
 		b.raiseLockingContextError(locking, "set-returning functions in the target list")
 	}
-
-	for _, li := range locking {
-		// Validate locking strength.
-		switch li.Strength {
-		case tree.ForNone:
-			// AST nodes should not be created with this locking strength.
-			panic(errors.AssertionFailedf("locking item without strength"))
-		case tree.ForUpdate:
-			// Exclusive locking on the entire row.
-		case tree.ForNoKeyUpdate:
-			// Exclusive locking on only non-key(s) of the row. Currently
-			// unimplemented and treated identically to ForUpdate.
-		case tree.ForShare:
-			// Shared locking on the entire row.
-		case tree.ForKeyShare:
-			// Shared locking on only key(s) of the row. Currently unimplemented and
-			// treated identically to ForShare.
-		default:
-			panic(errors.AssertionFailedf("unknown locking strength: %d", li.Strength))
-		}
-
-		// Validating locking wait policy.
-		switch li.WaitPolicy {
-		case tree.LockWaitBlock:
-			// Default. Block on conflicting locks.
-		case tree.LockWaitSkipLocked:
-			// Skip rows that can't be locked.
-		case tree.LockWaitError:
-			// Raise an error on conflicting locks.
-		default:
-			panic(errors.AssertionFailedf("unknown locking wait policy: %d", li.WaitPolicy))
-		}
-
-		// Validate locking form.
-		switch li.Form {
-		case tree.LockRecord:
-			// Default. Only lock existing rows.
-		case tree.LockPredicate:
-			// Lock both existing rows and gaps between rows.
-		default:
-			panic(errors.AssertionFailedf("unknown locking form: %d", li.Form))
-		}
-
-		// Validate locking targets by checking that all targets are well-formed
-		// and all point to real relations present in the FROM clause.
-		for _, target := range li.Targets {
-			// Insist on unqualified alias names here. We could probably do
-			// something smarter, but it's better to just mirror Postgres
-			// exactly. See transformLockingClause in Postgres' source.
-			if target.CatalogName != "" || target.SchemaName != "" {
-				panic(pgerror.Newf(pgcode.Syntax,
-					"%s must specify unqualified relation names", li.Strength))
-			}
-
-			// Search for the target in fromScope. If a target is missing from
-			// the scope then raise an error. This will end up looping over all
-			// columns in scope for each of the locking targets. We could use a
-			// more efficient data structure (e.g. a hash map of relation names)
-			// to improve the time complexity here, but we expect the number of
-			// columns to be small enough that doing so is likely not worth it.
-			found := false
-			for _, col := range fromScope.cols {
-				if target.ObjectName == col.table.ObjectName {
-					found = true
-					break
-				}
-			}
-			if !found {
-				panic(pgerror.Newf(
-					pgcode.UndefinedTable,
-					"relation %q in %s clause not found in FROM clause",
-					target.ObjectName, li.Strength,
-				))
-			}
-		}
-	}
 }
 
 // rejectIfLocking raises a locking error if a locking clause was specified.
diff --git a/pkg/sql/opt/optbuilder/update.go b/pkg/sql/opt/optbuilder/update.go
index 7e2057a26d16..c70400421c04 100644
--- a/pkg/sql/opt/optbuilder/update.go
+++ b/pkg/sql/opt/optbuilder/update.go
@@ -142,7 +142,7 @@ func (mb *mutationBuilder) addTargetColsForUpdate(exprs tree.UpdateExprs) {
 				for i := range desiredTypes {
 					desiredTypes[i] = mb.md.ColumnMeta(mb.targetColList[targetIdx+i]).Type
 				}
-				outScope := mb.b.buildSelectStmt(t.Select, noRowLocking, desiredTypes, mb.outScope)
+				outScope := mb.b.buildSelectStmt(t.Select, noLocking, desiredTypes, mb.outScope)
 				mb.subqueries = append(mb.subqueries, outScope)
 				n = len(outScope.cols)