opt: disallow SELECT FOR UPDATE under weak isolation levels

pkg/sql/catalog/descpb/locking.proto

@@ -115,7 +115,7 @@ enum ScanLockingStrength {
   // on each key scanned.
   FOR_UPDATE = 4;
 }
-  
+
 // LockingWaitPolicy controls the policy used for handling conflicting locks
 // held by other active transactions when attempting to lock rows due to FOR
 // UPDATE/SHARE clauses (i.e. it represents the NOWAIT and SKIP LOCKED options).

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/mutation.go

@@ -973,8 +973,12 @@ func (b *Builder) canAutoCommit(rel memo.RelExpr) bool {
 
 // forUpdateLocking is the row-level locking mode used by mutations during their
 // initial row scan, when such locking is deemed desirable. The locking mode is
-// equivalent that used by a SELECT ... FOR UPDATE statement.
-var forUpdateLocking = opt.Locking{Strength: tree.ForUpdate}
+// equivalent to that used by a SELECT FOR UPDATE statement, except not durable.
+var forUpdateLocking = opt.Locking{
+	Strength:   tree.ForUpdate,
+	WaitPolicy: tree.LockWaitBlock,
+	Durability: tree.LockDurabilityBestEffort,
+}
 
 // shouldApplyImplicitLockingToMutationInput determines whether or not the
 // builder should apply a FOR UPDATE row-level locking mode to the initial row