sql: use the correct locking strength for FOR SHARE clause

Previously, FOR SHARE and FOR KEY SHARE would use non-locking KV scans.
Now that the lock table supports shared locks, we can use lock.Shared as
the locking strength for KV scans. This patch does that, and in doing
so, wires up SHARED locks end to end.

Informs #91545

Release note: None

docs/generated/settings/settings-for-tenants.txt

@@ -248,6 +248,7 @@ sql.insights.anomaly_detection.memory_limit	byte size	1.0 MiB	the maximum amount
 sql.insights.execution_insights_capacity	integer	1000	the size of the per-node store of execution insights	tenant-rw
 sql.insights.high_retry_count.threshold	integer	10	the number of retries a slow statement must have undergone for its high retry count to be highlighted as a potential problem	tenant-rw
 sql.insights.latency_threshold	duration	100ms	amount of time after which an executing statement is considered slow. Use 0 to disable.	tenant-rw
+sql.locking.enable_shared_locks	boolean	false	enable shared locking strength when using FOR SHARE or FOR KEY SHARE modifiers	tenant-ro
 sql.log.slow_query.experimental_full_table_scans.enabled	boolean	false	when set to true, statements that perform a full table/index scan will be logged to the slow query log even if they do not meet the latency threshold. Must have the slow query log enabled for this setting to have any effect.	tenant-rw
 sql.log.slow_query.internal_queries.enabled	boolean	false	when set to true, internal queries which exceed the slow query log threshold are logged to a separate log. Must have the slow query log enabled for this setting to have any effect.	tenant-rw
 sql.log.slow_query.latency_threshold	duration	0s	when set to non-zero, log statements whose service latency exceeds the threshold to a secondary logger on each node	tenant-rw

docs/generated/settings/settings.html

@@ -195,6 +195,7 @@
 <tr><td><div id="setting-sql-insights-execution-insights-capacity" class="anchored"><code>sql.insights.execution_insights_capacity</code></div></td><td>integer</td><td><code>1000</code></td><td>the size of the per-node store of execution insights</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-insights-high-retry-count-threshold" class="anchored"><code>sql.insights.high_retry_count.threshold</code></div></td><td>integer</td><td><code>10</code></td><td>the number of retries a slow statement must have undergone for its high retry count to be highlighted as a potential problem</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-insights-latency-threshold" class="anchored"><code>sql.insights.latency_threshold</code></div></td><td>duration</td><td><code>100ms</code></td><td>amount of time after which an executing statement is considered slow. Use 0 to disable.</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
+<tr><td><div id="setting-sql-locking-enable-shared-locks" class="anchored"><code>sql.locking.enable_shared_locks</code></div></td><td>boolean</td><td><code>false</code></td><td>enable shared locking strength when using FOR SHARE or FOR KEY SHARE modifiers</td><td>Serverless/Dedicated/Self-Hosted (read-only)</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-experimental-full-table-scans-enabled" class="anchored"><code>sql.log.slow_query.experimental_full_table_scans.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when set to true, statements that perform a full table/index scan will be logged to the slow query log even if they do not meet the latency threshold. Must have the slow query log enabled for this setting to have any effect.</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-internal-queries-enabled" class="anchored"><code>sql.log.slow_query.internal_queries.enabled</code></div></td><td>boolean</td><td><code>false</code></td><td>when set to true, internal queries which exceed the slow query log threshold are logged to a separate log. Must have the slow query log enabled for this setting to have any effect.</td><td>Serverless/Dedicated/Self-Hosted</td></tr>
 <tr><td><div id="setting-sql-log-slow-query-latency-threshold" class="anchored"><code>sql.log.slow_query.latency_threshold</code></div></td><td>duration</td><td><code>0s</code></td><td>when set to non-zero, log statements whose service latency exceeds the threshold to a secondary logger on each node</td><td>Serverless/Dedicated/Self-Hosted</td></tr>

pkg/kv/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//pkg/kv/kvserver/concurrency/isolation",
         "//pkg/roachpb",
         "//pkg/settings",
+        "//pkg/settings/cluster",
         "//pkg/sql/sessiondatapb",
         "//pkg/storage/enginepb",
         "//pkg/testutils",

pkg/kv/db.go

@@ -266,7 +266,7 @@ type DB struct {
 	// SQL code that all uses kv.DB.
 	//
 	// TODO(sumeer,irfansharif): Find a home for these in the SQL layer.
-	// Especially SettingsValue.
+	// Especially SettingsValues.
 	SQLKVResponseAdmissionQ *admission.WorkQueue
 	AdmissionPacerFactory   admission.PacerFactory
 	SettingsValues          *settings.Values

pkg/sql/backfill/backfill.go

@@ -172,6 +172,7 @@ func (cb *ColumnBackfiller) init(
 			Spec:                       &spec,
 			TraceKV:                    traceKV,
 			ForceProductionKVBatchSize: cb.evalCtx.TestingKnobs.ForceProductionValues,
+			Settings:                   cb.evalCtx.Settings,
 		},
 	)
 }
@@ -857,6 +858,7 @@ func (ib *IndexBackfiller) BuildIndexEntriesChunk(
 			Spec:                       &spec,
 			TraceKV:                    traceKV,
 			ForceProductionKVBatchSize: ib.evalCtx.TestingKnobs.ForceProductionValues,
+			Settings:                   ib.evalCtx.Settings,
 		},
 	); err != nil {
 		return nil, nil, 0, err

pkg/sql/colexec/colbuilder/execplan.go

@@ -858,7 +858,7 @@ func NewColOperator(
 					// would be to simply include all key columns into the set
 					// of needed for the fetch and to project them away in the
 					// ColBatchDirectScan.
-					if row.GetKeyLockingStrength(core.TableReader.LockingStrength) != lock.None ||
+					if row.GetKeyLockingStrength(ctx, core.TableReader.LockingStrength, flowCtx.EvalCtx.Settings) != lock.None ||
 						core.TableReader.LockingWaitPolicy == descpb.ScanLockingWaitPolicy_SKIP_LOCKED {
 						return false
 					}

pkg/sql/colfetcher/colbatch_direct_scan.go

@@ -209,6 +209,7 @@ func NewColBatchDirectScan(
 	// should be able to modify the BatchRequest, but alas.
 	fetchSpec := spec.FetchSpec
 	fetcher := row.NewDirectKVBatchFetcher(
+		ctx,
 		flowCtx.Txn,
 		bsHeader,
 		&fetchSpec,
@@ -218,6 +219,7 @@ func NewColBatchDirectScan(
 		flowCtx.EvalCtx.SessionData().LockTimeout,
 		kvFetcherMemAcc,
 		flowCtx.EvalCtx.TestingKnobs.ForceProductionValues,
+		flowCtx.EvalCtx.Settings,
 	)
 	var hasDatumVec bool
 	for _, t := range tableArgs.typs {

pkg/sql/colfetcher/colbatch_scan.go

@@ -345,6 +345,7 @@ func NewColBatchScan(
 		return nil, nil, err
 	}
 	kvFetcher := row.NewKVFetcher(
+		ctx,
 		flowCtx.Txn,
 		bsHeader,
 		spec.Reverse,
@@ -353,6 +354,7 @@ func NewColBatchScan(
 		flowCtx.EvalCtx.SessionData().LockTimeout,
 		kvFetcherMemAcc,
 		flowCtx.EvalCtx.TestingKnobs.ForceProductionValues,
+		flowCtx.EvalCtx.Settings,
 	)
 	fetcher := cFetcherPool.Get().(*cFetcher)
 	fetcher.cFetcherArgs = cFetcherArgs{

pkg/sql/colfetcher/index_join.go

@@ -557,6 +557,7 @@ func NewColIndexJoin(
 		cFetcherMemoryLimit = int64(math.Ceil(float64(totalMemoryLimit) / 16.0))
 		streamerBudgetLimit := 14 * cFetcherMemoryLimit
 		kvFetcher = row.NewStreamingKVFetcher(
+			ctx,
 			flowCtx.Cfg.DistSender,
 			flowCtx.Stopper(),
 			txn,
@@ -575,6 +576,7 @@ func NewColIndexJoin(
 		)
 	} else {
 		kvFetcher = row.NewKVFetcher(
+			ctx,
 			txn,
 			nil,   /* bsHeader */
 			false, /* reverse */
@@ -583,6 +585,7 @@ func NewColIndexJoin(
 			flowCtx.EvalCtx.SessionData().LockTimeout,
 			kvFetcherMemAcc,
 			flowCtx.EvalCtx.TestingKnobs.ForceProductionValues,
+			flowCtx.EvalCtx.Settings,
 		)
 	}
 

pkg/sql/insert_fast_path.go

@@ -17,6 +17,7 @@ import (
 	"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/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/colinfo"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/descpb"
@@ -153,7 +154,7 @@ func (r *insertFastPathRun) inputRow(rowIdx int) tree.Datums {
 // addFKChecks adds Requests to fkBatch and entries in fkSpanInfo / fkSpanMap as
 // needed for checking foreign keys for the given row.
 func (r *insertFastPathRun) addFKChecks(
-	ctx context.Context, rowIdx int, inputRow tree.Datums,
+	ctx context.Context, rowIdx int, inputRow tree.Datums, settings *cluster.Settings,
 ) error {
 	for i := range r.fkChecks {
 		c := &r.fkChecks[i]
@@ -189,7 +190,7 @@ func (r *insertFastPathRun) addFKChecks(
 		if r.traceKV {
 			log.VEventf(ctx, 2, "FKScan %s", span)
 		}
-		lockStrength := row.GetKeyLockingStrength(descpb.ToScanLockingStrength(c.Locking.Strength))
+		lockStrength := row.GetKeyLockingStrength(ctx, descpb.ToScanLockingStrength(c.Locking.Strength), settings)
 		lockWaitPolicy := row.GetWaitPolicy(descpb.ToScanLockingWaitPolicy(c.Locking.WaitPolicy))
 		if r.fkBatch.Header.WaitPolicy != lockWaitPolicy {
 			return errors.AssertionFailedf(
@@ -310,7 +311,7 @@ func (n *insertFastPathNode) BatchedNext(params runParams) (bool, error) {
 
 		// Add FK existence checks.
 		if len(n.run.fkChecks) > 0 {
-			if err := n.run.addFKChecks(params.ctx, rowIdx, inputRow); err != nil {
+			if err := n.run.addFKChecks(params.ctx, rowIdx, inputRow, params.ExecCfg().Settings); err != nil {
 				return false, err
 			}
 		}

pkg/sql/logictest/testdata/logic_test/select_for_share

@@ -0,0 +1,75 @@
+# LogicTest: !local-mixed-22.2-23.1
+
+statement ok
+SET CLUSTER SETTING sql.locking.enable_shared_locks = true
+
+statement ok
+CREATE TABLE t(a INT PRIMARY KEY);
+INSERT INTO t VALUES(1);
+GRANT ALL ON t TO testuser;
+CREATE USER testuser2 WITH VIEWACTIVITY;
+GRANT ALL ON t TO testuser2;
+
+user testuser
+
+statement ok
+BEGIN
+
+query I
+SELECT * FROM t WHERE a = 1 FOR SHARE;
+----
+1
+
+# Start another transaction to show multiple transactions can acquire SHARED
+# locks at the same time.
+
+user root
+
+statement ok
+BEGIN
+
+query I
+SELECT * FROM t  WHERE a = 1 FOR SHARE;
+----
+1
+
+user testuser2
+
+statement async writeReq count 1
+UPDATE t SET a = 2 WHERE a = 1
+
+# TODO(arul): Until https://github.com/cockroachdb/cockroach/issues/107766 is
+# addressed, we'll incorrectly report shared locks as having "Exclusive" lock
+# strength; however, having this query in here is useful to make sure there are
+# two locks on our key and setting the cluster setting above actually did
+# something; otherwise, had we used non-locking reads, we'd have failed here.
+query TTTTTTTBB colnames,retry,rowsort
+SELECT database_name, schema_name, table_name, lock_key_pretty, lock_strength, durability, isolation_level, granted, contended FROM crdb_internal.cluster_locks
+----
+database_name  schema_name  table_name  lock_key_pretty   lock_strength  durability    isolation_level  granted  contended
+test           public       t           /Table/106/1/1/0  Exclusive      Unreplicated  SERIALIZABLE     true     true
+test           public       t           /Table/106/1/1/0  Exclusive      Unreplicated  SERIALIZABLE     false    true
+
+# Commit the first transaction and rollback the second.
+
+user testuser
+
+statement ok
+COMMIT
+
+user root
+
+statement ok
+ROLLBACK
+
+user testuser2
+
+# Now that both the transactions that issued shared lock reads have been
+# finalized, the write should be able to proceed.
+
+awaitstatement writeReq
+
+query I
+SELECT * FROM t;
+----
+2

pkg/sql/row/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
     importpath = "github.com/cockroachdb/cockroach/pkg/sql/row",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/clusterversion",
         "//pkg/col/coldata",
         "//pkg/jobs",
         "//pkg/jobs/jobspb",

pkg/sql/row/fetcher.go

@@ -22,6 +22,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvpb"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catenumpb"
 	"github.com/cockroachdb/cockroach/pkg/sql/catalog/catpb"
@@ -316,6 +317,7 @@ type FetcherInitArgs struct {
 	// row is being processed. In practice, this means that span IDs must be
 	// passed in when SpansCanOverlap is true.
 	SpansCanOverlap bool
+	Settings        *cluster.Settings
 }
 
 // Init sets up a Fetcher for a given table and index.
@@ -445,6 +447,7 @@ func (rf *Fetcher) Init(ctx context.Context, args FetcherInitArgs) error {
 			forceProductionKVBatchSize: args.ForceProductionKVBatchSize,
 			kvPairsRead:                &kvPairsRead,
 			batchRequestsIssued:        &batchRequestsIssued,
+			settings:                   args.Settings,
 		}
 		if args.Txn != nil {
 			fetcherArgs.sendFn = makeTxnKVFetcherDefaultSendFunc(args.Txn, &batchRequestsIssued)
@@ -453,7 +456,7 @@ func (rf *Fetcher) Init(ctx context.Context, args FetcherInitArgs) error {
 			fetcherArgs.admission.pacerFactory = args.Txn.DB().AdmissionPacerFactory
 			fetcherArgs.admission.settingsValues = args.Txn.DB().SettingsValues
 		}
-		rf.kvFetcher = newKVFetcher(newTxnKVFetcherInternal(fetcherArgs))
+		rf.kvFetcher = newKVFetcher(newTxnKVFetcherInternal(ctx, fetcherArgs))
 	}
 
 	return nil