sql: add implicit SELECT FOR SHARE locking to FK parent checks

Add SELECT FOR SHARE locking to FK parent checks. Under serializable
isolation, this locking is only used when
`enable_implicit_fk_locking_for_serializable` is set. Under weaker
isolation levels (snapshot and read committed) this locking is always
used.

We only need to lock during the insertion-side FK checks, which verify
the existence of a parent row. Deletion-side FK checks verify the
non-existence of a child row, and these do not need to lock. Instead, to
prevent concurrent inserts or updates to the child that would violate
the FK constraint, we rely on the intent(s) created by the deletion
conflicting with the FK locking of those concurrent inserts or updates.

Fixes: #80683
Informs: #100156

Epic: CRDB-25322

Release note (sql change): Add a new session variable,
`enable_implicit_fk_locking_for_serializable`, which controls locking
during foreign key checks under serializable isolation. With this set to
true, foreign key checks of the referenced (parent) table, such as those
performed during an INSERT or UPDATE of the referencing (child) table,
will lock the referenced row using SELECT FOR SHARE locking. (This is
somewhat analogous to the existing `enable_implicit_select_for_update`
variable but applies to the foreign key checks of a mutation statement
instead of the initial row fetch.)

Under weaker isolation levels such as read committed, SELECT FOR SHARE
locking will always be used to ensure the database maintains the foreign
key constraint, regardless of the current setting of
`enable_implicit_fk_locking_for_serializable`.

pkg/sql/exec_util.go

@@ -3612,6 +3612,10 @@ func (m *sessionDataMutator) SetOptimizerUseImprovedJoinElimination(val bool) {
 	m.data.OptimizerUseImprovedJoinElimination = val
 }
 
+func (m *sessionDataMutator) SetImplicitFKLockingForSerializable(val bool) {
+	m.data.ImplicitFKLockingForSerializable = val
+}
+
 // Utility functions related to scrubbing sensitive information on SQL Stats.
 
 // quantizeCounts ensures that the Count field in the

pkg/sql/insert_fast_path.go

@@ -15,6 +15,7 @@ import (
 	"sync"
 
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
+	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/lock"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
@@ -188,10 +189,20 @@ func (r *insertFastPathRun) addFKChecks(
 		if r.traceKV {
 			log.VEventf(ctx, 2, "FKScan %s", span)
 		}
+		lockStrength := row.GetKeyLockingStrength(descpb.ToScanLockingStrength(c.Locking.Strength))
+		lockWaitPolicy := row.GetWaitPolicy(descpb.ToScanLockingWaitPolicy(c.Locking.WaitPolicy))
+		if r.fkBatch.Header.WaitPolicy != lockWaitPolicy {
+			return errors.AssertionFailedf(
+				"FK check lock wait policy %s did not match %s",
+				lockWaitPolicy, r.fkBatch.Header.WaitPolicy,
+			)
+		}
 		reqIdx := len(r.fkBatch.Requests)
 		r.fkBatch.Requests = append(r.fkBatch.Requests, kvpb.RequestUnion{})
 		r.fkBatch.Requests[reqIdx].MustSetInner(&kvpb.ScanRequest{
 			RequestHeader: kvpb.RequestHeaderFromSpan(span),
+			KeyLocking:    lockStrength,
+			// TODO(michae2): Once #100193 is finished, also include c.Locking.Durability.
 		})
 		r.fkSpanInfo = append(r.fkSpanInfo, insertFastPathFKSpanInfo{
 			check:  c,
@@ -248,6 +259,8 @@ func (n *insertFastPathNode) startExec(params runParams) error {
 			}
 		}
 		maxSpans := len(n.run.fkChecks) * len(n.input)
+		// Any FK checks using locking should have lock wait policy BLOCK.
+		n.run.fkBatch.Header.WaitPolicy = lock.WaitPolicy_Block
 		n.run.fkBatch.Requests = make([]kvpb.RequestUnion, 0, maxSpans)
 		n.run.fkSpanInfo = make([]insertFastPathFKSpanInfo, 0, maxSpans)
 		if len(n.input) > 1 {

pkg/sql/logictest/testdata/logic_test/fk_read_committed

@@ -0,0 +1,46 @@
+# LogicTest: !local-mixed-22.2-23.1
+
+# Some foreign key checks are prohibited under weaker isolation levels until we
+# improve locking. See #80683, #100156, #100193.
+
+statement ok
+CREATE TABLE jars (j INT PRIMARY KEY)
+
+statement ok
+CREATE TABLE cookies (c INT PRIMARY KEY, j INT REFERENCES jars (j))
+
+statement ok
+SET SESSION CHARACTERISTICS AS TRANSACTION ISOLATION LEVEL READ COMMITTED
+
+statement ok
+INSERT INTO jars VALUES (1), (2)
+
+# Foreign key checks of the parent require durable shared locking under weaker
+# isolation levels, and are not yet supported.
+query error pgcode 0A000 guaranteed-durable locking not yet implemented
+INSERT INTO cookies VALUES (1, 1)
+
+statement ok
+BEGIN TRANSACTION ISOLATION LEVEL SERIALIZABLE
+
+statement ok
+INSERT INTO cookies VALUES (1, 1)
+
+statement ok
+COMMIT
+
+query error pgcode 0A000 guaranteed-durable locking not yet implemented
+UPDATE cookies SET j = 2 WHERE c = 1
+
+# Foreign key checks of the child do not require locking.
+query error violates foreign key constraint
+UPDATE jars SET j = j + 4
+
+query error violates foreign key constraint
+DELETE FROM jars WHERE j = 1
+
+statement ok
+DELETE FROM cookies WHERE c = 1
+
+statement ok
+DELETE FROM jars WHERE j = 1

pkg/sql/logictest/testdata/logic_test/information_schema

@@ -5289,6 +5289,7 @@ enable_auto_rehoming                                       off
 enable_create_stats_using_extremes                         off
 enable_drop_enum_value                                     on
 enable_experimental_alter_column_type_general              off
+enable_implicit_fk_locking_for_serializable                off
 enable_implicit_select_for_update                          on
 enable_implicit_transaction_for_batch_statements           on
 enable_insert_fast_path                                    on

pkg/sql/logictest/testdata/logic_test/pg_catalog

@@ -2727,6 +2727,7 @@ distsql                                                    off                 N
 enable_auto_rehoming                                       off                 NULL      NULL        NULL        string
 enable_create_stats_using_extremes                         off                 NULL      NULL        NULL        string
 enable_experimental_alter_column_type_general              off                 NULL      NULL        NULL        string
+enable_implicit_fk_locking_for_serializable                off                 NULL      NULL        NULL        string
 enable_implicit_select_for_update                          on                  NULL      NULL        NULL        string
 enable_implicit_transaction_for_batch_statements           on                  NULL      NULL        NULL        string
 enable_insert_fast_path                                    on                  NULL      NULL        NULL        string
@@ -2887,6 +2888,7 @@ distsql                                                    off                 N
 enable_auto_rehoming                                       off                 NULL  user     NULL      off                 off
 enable_create_stats_using_extremes                         off                 NULL  user     NULL      off                 off
 enable_experimental_alter_column_type_general              off                 NULL  user     NULL      off                 off
+enable_implicit_fk_locking_for_serializable                off                 NULL  user     NULL      off                 off
 enable_implicit_select_for_update                          on                  NULL  user     NULL      on                  on
 enable_implicit_transaction_for_batch_statements           on                  NULL  user     NULL      on                  on
 enable_insert_fast_path                                    on                  NULL  user     NULL      on                  on
@@ -3044,6 +3046,7 @@ distsql_workmem                                            NULL    NULL     NULL
 enable_auto_rehoming                                       NULL    NULL     NULL     NULL        NULL
 enable_create_stats_using_extremes                         NULL    NULL     NULL     NULL        NULL
 enable_experimental_alter_column_type_general              NULL    NULL     NULL     NULL        NULL
+enable_implicit_fk_locking_for_serializable                NULL    NULL     NULL     NULL        NULL
 enable_implicit_select_for_update                          NULL    NULL     NULL     NULL        NULL
 enable_implicit_transaction_for_batch_statements           NULL    NULL     NULL     NULL        NULL
 enable_insert_fast_path                                    NULL    NULL     NULL     NULL        NULL

pkg/sql/logictest/testdata/logic_test/read_committed

@@ -3,7 +3,7 @@
 subtest select_for_update
 
 # SELECT FOR UPDATE is prohibited under weaker isolation levels until we improve
-# locking. See #57031, #75457, #100144.
+# locking. See #57031, #75457, #100144, #100193.
 
 statement ok
 CREATE TABLE supermarket (

pkg/sql/logictest/testdata/logic_test/show_source

@@ -62,6 +62,7 @@ distsql                                                    off
 enable_auto_rehoming                                       off
 enable_create_stats_using_extremes                         off
 enable_experimental_alter_column_type_general              off
+enable_implicit_fk_locking_for_serializable                off
 enable_implicit_select_for_update                          on
 enable_implicit_transaction_for_batch_statements           on
 enable_insert_fast_path                                    on

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

@@ -200,6 +200,11 @@ func (b *Builder) tryBuildFastPathInsert(ins *memo.InsertExpr) (_ execPlan, ok b
 			return execPlan{}, false, nil
 		}
 
+		locking, err := b.buildLocking(lookupJoin.Locking)
+		if err != nil {
+			return execPlan{}, false, err
+		}
+
 		out := &fkChecks[i]
 		out.InsertCols = make([]exec.TableColumnOrdinal, len(lookupJoin.KeyCols))
 		for i, keyCol := range lookupJoin.KeyCols {
@@ -220,6 +225,7 @@ func (b *Builder) tryBuildFastPathInsert(ins *memo.InsertExpr) (_ execPlan, ok b
 		out.ReferencedTable = md.Table(lookupJoin.Table)
 		out.ReferencedIndex = out.ReferencedTable.Index(lookupJoin.Index)
 		out.MatchMethod = fk.MatchMethod()
+		out.Locking = locking
 		out.MkErr = func(values tree.Datums) error {
 			if len(values) != len(out.InsertCols) {
 				return errors.AssertionFailedf("invalid FK violation values")