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/logictest/testdata/logic_test/select_for_update

@@ -485,3 +485,114 @@ SELECT * FROM t94290 WHERE b = 2 FOR UPDATE;
 
 statement ok
 RESET statement_timeout;
+
+# SELECT FOR UPDATE is prohibited under weaker isolation levels until we improve
+# locking. See #57031, #75457, #100144.
+
+# Try to catch the children.
+statement ok
+CREATE TABLE supermarket (person STRING PRIMARY KEY, aisle INT NOT NULL)
+
+statement ok
+INSERT INTO supermarket VALUES ('abbie', 1), ('gus', 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 = 'gus' 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. TODO(michae2): uncomment after #103488 is merged.
+# 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 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
+
+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"
@@ -622,14 +623,22 @@ func (b *Builder) scanParams(
 	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())
+	if 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 b.evalCtx.TxnReadOnly {
+			return exec.ScanParams{}, opt.ColMap{}, pgerror.Newf(pgcode.ReadOnlySQLTransaction,
+				"cannot execute %s in a read-only transaction", locking.Strength.String())
+		}
+		if locking.Durability == tree.LockDurabilityGuaranteed &&
+			b.evalCtx.Txn.IsoLevel() != isolation.Serializable {
+			return exec.ScanParams{}, opt.ColMap{}, unimplemented.NewWithIssuef(
+				100144, "cannot execute SELECT FOR UPDATE statements under %s isolation",
+				b.evalCtx.Txn.IsoLevel(),
+			)
+		}
+		b.ContainsNonDefaultKeyLocking = true
 	}
 
 	needed, outputMap := b.getColumns(scan.Cols, scan.Table)