opt: disallow SELECT FOR UPDATE under weak isolation levels

Temporarily disallow `SELECT FOR UPDATE` statements under all isolation
levels that are not `SERIALIZABLE` (i.e. `SNAPSHOT` and `READ
COMMITTED`). We will allow them again when the following issues are
fixed:

- cockroachdb#57031
- cockroachdb#75457
- cockroachdb#100193
- cockroachdb#100194

Fixes: cockroachdb#100144

Release note: None

pkg/sql/conn_executor.go

@@ -3512,6 +3512,7 @@ func (ex *connExecutor) resetEvalCtx(evalCtx *extendedEvalContext, txn *kv.Txn,
 	evalCtx.TxnReadOnly = ex.state.readOnly
 	evalCtx.TxnImplicit = ex.implicitTxn()
 	evalCtx.TxnIsSingleStmt = false
+	evalCtx.TxnIsoLevel = ex.state.isolationLevel
 	if newTxn || !ex.implicitTxn() {
 		// Only update the stmt timestamp if in a new txn or an explicit txn. This is because this gets
 		// called multiple times during an extended protocol implicit txn, but we

pkg/sql/logictest/testdata/logic_test/select_for_update_read_committed

@@ -0,0 +1,142 @@
+# LogicTest: !local-mixed-22.2-23.1
+
+# SELECT FOR UPDATE is prohibited under weaker isolation levels until we improve
+# locking. See #57031, #75457, #100144.
+
+statement ok
+CREATE TABLE supermarket (
+  person STRING PRIMARY KEY,
+  aisle INT NOT NULL,
+  starts_with STRING GENERATED ALWAYS AS (left(person, 1)) STORED,
+  ends_with STRING GENERATED ALWAYS AS (right(person, 3)) STORED,
+  INDEX (starts_with),
+  INDEX (ends_with)
+)
+
+statement ok
+INSERT INTO supermarket (person, aisle)
+  VALUES ('abbie', 1), ('gideon', 2), ('matilda', 3), ('michael', 4)
+
+# SELECT FOR UPDATE should still work under serializable isolation.
+statement ok
+BEGIN
+
+query I
+SELECT aisle FROM supermarket WHERE person = 'gideon' FOR UPDATE
+----
+2
+
+statement ok
+UPDATE supermarket SET aisle = 2 WHERE person = 'abbie'
+
+statement ok
+COMMIT
+
+# It should fail under read committed isolation.
+statement ok
+BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED
+
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+SELECT aisle FROM supermarket WHERE person = 'matilda' FOR UPDATE
+
+statement ok
+ROLLBACK
+
+# It should also fail under snapshot isolation.
+statement ok
+SET CLUSTER SETTING sql.txn.snapshot_isolation.enabled = true
+
+statement ok
+BEGIN TRANSACTION ISOLATION LEVEL SNAPSHOT
+
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under Snapshot isolation
+SELECT aisle FROM supermarket WHERE person = 'matilda' FOR UPDATE
+
+statement ok
+ROLLBACK
+
+statement ok
+RESET CLUSTER SETTING sql.txn.snapshot_isolation.enabled
+
+# SELECT FOR UPDATE in a subquery should also fail under read committed.
+statement ok
+BEGIN TRANSACTION
+
+statement ok
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+UPDATE supermarket
+  SET aisle = (SELECT aisle FROM supermarket WHERE person = 'matilda' FOR UPDATE)
+  WHERE person = 'michael'
+
+statement ok
+ROLLBACK
+
+# It should also fail in a CTE.
+statement ok
+BEGIN TRANSACTION
+
+statement ok
+SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
+
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+WITH s AS
+  (SELECT aisle FROM supermarket WHERE person = 'matilda' FOR UPDATE)
+SELECT aisle + 1 FROM s
+
+statement ok
+ROLLBACK
+
+statement ok
+SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL READ COMMITTED
+
+# Creating a UDF using SELECT FOR UPDATE should succeed under read committed.
+statement ok
+CREATE FUNCTION wrangle (name STRING) RETURNS INT LANGUAGE SQL AS $$
+  SELECT aisle FROM supermarket WHERE person = name FOR UPDATE
+$$
+
+# But calling that function should fail.
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+INSERT INTO supermarket (person, aisle) VALUES ('grandma', wrangle('matilda'))
+
+statement ok
+DROP FUNCTION wrangle
+
+# Preparing a SELECT FOR UPDATE should succeed under read committed.
+statement ok
+PREPARE psa AS SELECT aisle FROM supermarket WHERE person = $1::STRING FOR UPDATE
+
+# But execution should fail.
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+EXECUTE psa('matilda')
+
+statement ok
+DEALLOCATE psa
+
+# SELECT FOR UPDATE using a lookup join should also fail.
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+WITH names AS MATERIALIZED
+  (SELECT 'matilda' AS person)
+SELECT aisle
+  FROM names
+  NATURAL INNER LOOKUP JOIN supermarket
+  FOR UPDATE
+
+# SELECT FOR UPDATE using an index join should also fail.
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+SELECT aisle
+  FROM supermarket@supermarket_starts_with_idx
+  WHERE starts_with = 'm'
+  FOR UPDATE
+
+# SELECT FOR UPDATE using a zigzag join should also fail.
+query error pgcode 0A000 cannot execute SELECT FOR UPDATE statements under ReadCommitted isolation
+SELECT aisle
+  FROM supermarket@{FORCE_ZIGZAG}
+  WHERE starts_with = 'm' AND ends_with = 'lda'
+  FOR UPDATE
+
+statement ok
+SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL SERIALIZABLE

pkg/sql/opt/exec/execbuilder/BUILD.bazel

@@ -15,6 +15,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/opt/exec/execbuilder",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/server/telemetry",
         "//pkg/sql/catalog/colinfo",
         "//pkg/sql/catalog/descpb",

pkg/sql/opt/exec/execbuilder/relational.go

@@ -17,6 +17,7 @@ import (
 	"math"
 	"strings"
 
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/isolation"
 	"github.com/cockroachdb/cockroach/pkg/server/telemetry"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -618,18 +619,9 @@ func (b *Builder) scanParams(
 		return exec.ScanParams{}, opt.ColMap{}, err
 	}
 
-	locking := scan.Locking
-	if b.forceForUpdateLocking {
-		locking = locking.Max(forUpdateLocking)
-	}
-	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
-
-	// Raise error if row-level locking is part of a read-only transaction.
-	// TODO(nvanbenschoten): this check should be shared across all expressions
-	// that can perform row-level locking.
-	if locking.IsLocking() && b.evalCtx.TxnReadOnly {
-		return exec.ScanParams{}, opt.ColMap{}, pgerror.Newf(pgcode.ReadOnlySQLTransaction,
-			"cannot execute %s in a read-only transaction", locking.Strength.String())
+	locking, err := b.buildLocking(scan.Locking)
+	if err != nil {
+		return exec.ScanParams{}, opt.ColMap{}, err
 	}
 
 	needed, outputMap := b.getColumns(scan.Cols, scan.Table)
@@ -2170,11 +2162,10 @@ func (b *Builder) buildIndexJoin(join *memo.IndexJoinExpr) (execPlan, error) {
 	cols := join.Cols
 	needed, output := b.getColumns(cols, join.Table)
 
-	locking := join.Locking
-	if b.forceForUpdateLocking {
-		locking = locking.Max(forUpdateLocking)
+	locking, err := b.buildLocking(join.Locking)
+	if err != nil {
+		return execPlan{}, err
 	}
-	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
 	res := execPlan{outputCols: output}
 	b.recordJoinAlgorithm(exec.IndexJoin)
@@ -2491,11 +2482,10 @@ func (b *Builder) buildLookupJoin(join *memo.LookupJoinExpr) (execPlan, error) {
 	idx := tab.Index(join.Index)
 	b.IndexesUsed = util.CombineUnique(b.IndexesUsed, []string{fmt.Sprintf("%d@%d", tab.ID(), idx.ID())})
 
-	locking := join.Locking
-	if b.forceForUpdateLocking {
-		locking = locking.Max(forUpdateLocking)
+	locking, err := b.buildLocking(join.Locking)
+	if err != nil {
+		return execPlan{}, err
 	}
-	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
 	joinType, err := joinOpToJoinType(join.JoinType)
 	if err != nil {
@@ -2733,11 +2723,10 @@ func (b *Builder) buildInvertedJoin(join *memo.InvertedJoinExpr) (execPlan, erro
 		return execPlan{}, err
 	}
 
-	locking := join.Locking
-	if b.forceForUpdateLocking {
-		locking = locking.Max(forUpdateLocking)
+	locking, err := b.buildLocking(join.Locking)
+	if err != nil {
+		return execPlan{}, err
 	}
-	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || locking.IsLocking()
 
 	joinType, err := joinOpToJoinType(join.JoinType)
 	if err != nil {
@@ -2813,13 +2802,14 @@ func (b *Builder) buildZigzagJoin(join *memo.ZigzagJoinExpr) (execPlan, error) {
 	leftOrdinals, leftColMap := b.getColumns(leftCols, join.LeftTable)
 	rightOrdinals, rightColMap := b.getColumns(rightCols, join.RightTable)
 
-	leftLocking := join.LeftLocking
-	rightLocking := join.RightLocking
-	if b.forceForUpdateLocking {
-		leftLocking = leftLocking.Max(forUpdateLocking)
-		rightLocking = rightLocking.Max(forUpdateLocking)
+	leftLocking, err := b.buildLocking(join.LeftLocking)
+	if err != nil {
+		return execPlan{}, err
+	}
+	rightLocking, err := b.buildLocking(join.RightLocking)
+	if err != nil {
+		return execPlan{}, err
 	}
-	b.ContainsNonDefaultKeyLocking = b.ContainsNonDefaultKeyLocking || leftLocking.IsLocking() || rightLocking.IsLocking()
 
 	allCols := joinOutputMap(leftColMap, rightColMap)
 
@@ -2884,6 +2874,29 @@ func (b *Builder) buildZigzagJoin(join *memo.ZigzagJoinExpr) (execPlan, error) {
 	return b.applySimpleProject(res, join, join.Cols, join.ProvidedPhysical().Ordering)
 }
 
+func (b *Builder) buildLocking(locking opt.Locking) (opt.Locking, error) {
+	if b.forceForUpdateLocking {
+		locking = locking.Max(forUpdateLocking)
+	}
+	if locking.IsLocking() {
+		// Raise error if row-level locking is part of a read-only transaction.
+		if b.evalCtx.TxnReadOnly {
+			return opt.Locking{}, pgerror.Newf(pgcode.ReadOnlySQLTransaction,
+				"cannot execute %s in a read-only transaction", locking.Strength.String(),
+			)
+		}
+		if locking.Durability == tree.LockDurabilityGuaranteed &&
+			b.evalCtx.TxnIsoLevel != isolation.Serializable {
+			return opt.Locking{}, unimplemented.NewWithIssuef(
+				100144, "cannot execute SELECT FOR UPDATE statements under %s isolation",
+				b.evalCtx.TxnIsoLevel,
+			)
+		}
+		b.ContainsNonDefaultKeyLocking = true
+	}
+	return locking, nil
+}
+
 func (b *Builder) buildMax1Row(max1Row *memo.Max1RowExpr) (execPlan, error) {
 	input, err := b.buildRelational(max1Row.Input)
 	if err != nil {

pkg/sql/sem/eval/BUILD.bazel

@@ -44,6 +44,7 @@ go_library(
         "//pkg/keys",
         "//pkg/kv",
         "//pkg/kv/kvpb",
+        "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/kv/kvserver/kvserverbase",
         "//pkg/repstream/streampb",
         "//pkg/roachpb",