storage: plumb maxLockConflicts into MVCCClearTimeRange

This avoids returning an unbounded number of locks.

Release note: None

pkg/kv/kvserver/batcheval/cmd_revert_range.go

@@ -140,12 +140,13 @@ func RevertRange(
 
 	leftPeekBound, rightPeekBound := rangeTombstonePeekBounds(
 		args.Key, args.EndKey, desc.StartKey.AsRawKey(), desc.EndKey.AsRawKey())
+	maxLockConflicts := storage.MaxConflictsPerLockConflictError.Get(&cArgs.EvalCtx.ClusterSettings().SV)
 
 	log.VEventf(ctx, 2, "clearing keys with timestamp (%v, %v]", args.TargetTime, cArgs.Header.Timestamp)
 
 	resumeKey, err := storage.MVCCClearTimeRange(ctx, readWriter, cArgs.Stats, args.Key, args.EndKey,
 		args.TargetTime, cArgs.Header.Timestamp, leftPeekBound, rightPeekBound,
-		clearRangeThreshold, cArgs.Header.MaxSpanRequestKeys, maxRevertRangeBatchBytes)
+		clearRangeThreshold, cArgs.Header.MaxSpanRequestKeys, maxRevertRangeBatchBytes, maxLockConflicts)
 	if err != nil {
 		return result.Result{}, err
 	}

pkg/kv/kvserver/batcheval/cmd_revert_range_test.go

@@ -23,6 +23,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/batcheval/result"
 	"github.com/cockroachdb/cockroach/pkg/kv/kvserver/concurrency/isolation"
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
+	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/storage"
 	"github.com/cockroachdb/cockroach/pkg/storage/enginepb"
 	"github.com/cockroachdb/cockroach/pkg/testutils"
@@ -115,7 +116,12 @@ func TestCmdRevertRange(t *testing.T) {
 		EndKey:   roachpb.RKey(endKey),
 	}
 	cArgs := batcheval.CommandArgs{Header: kvpb.Header{RangeID: desc.RangeID, Timestamp: tsReq, MaxSpanRequestKeys: 2}}
-	evalCtx := &batcheval.MockEvalCtx{Desc: &desc, Clock: hlc.NewClockForTesting(nil), Stats: stats}
+	evalCtx := &batcheval.MockEvalCtx{
+		ClusterSettings: cluster.MakeTestingClusterSettings(),
+		Desc:            &desc,
+		Clock:           hlc.NewClockForTesting(nil),
+		Stats:           stats,
+	}
 	cArgs.EvalCtx = evalCtx.EvalContext()
 	afterStats, err := storage.ComputeStats(ctx, eng, keys.LocalMax, keys.MaxKey, 0)
 	require.NoError(t, err)
@@ -200,7 +206,12 @@ func TestCmdRevertRange(t *testing.T) {
 	sumD := hashRange(t, eng, startKey, endKey)
 
 	// Re-set EvalCtx to pick up revised stats.
-	cArgs.EvalCtx = (&batcheval.MockEvalCtx{Desc: &desc, Clock: hlc.NewClockForTesting(nil), Stats: stats}).EvalContext( /* maxOffset */ )
+	cArgs.EvalCtx = (&batcheval.MockEvalCtx{
+		ClusterSettings: cluster.MakeTestingClusterSettings(),
+		Desc:            &desc,
+		Clock:           hlc.NewClockForTesting(nil),
+		Stats:           stats,
+	}).EvalContext( /* maxOffset */ )
 	for _, tc := range []struct {
 		name        string
 		ts          hlc.Timestamp
@@ -296,9 +307,10 @@ func TestCmdRevertRangeMVCCRangeTombstones(t *testing.T) {
 		var ms enginepb.MVCCStats
 		cArgs := batcheval.CommandArgs{
 			EvalCtx: (&batcheval.MockEvalCtx{
-				Desc:  &desc,
-				Clock: hlc.NewClockForTesting(nil),
-				Stats: ms,
+				ClusterSettings: cluster.MakeTestingClusterSettings(),
+				Desc:            &desc,
+				Clock:           hlc.NewClockForTesting(nil),
+				Stats:           ms,
 			}).EvalContext(),
 			Header: kvpb.Header{
 				RangeID:   desc.RangeID,

pkg/storage/metamorphic/operations.go

@@ -403,7 +403,7 @@ func (m mvccClearTimeRangeOp) run(ctx context.Context) string {
 		return "no-op due to no non-conflicting key range"
 	}
 	span, err := storage.MVCCClearTimeRange(ctx, m.m.engine, &enginepb.MVCCStats{}, m.key, m.endKey,
-		m.startTime, m.endTime, nil, nil, math.MaxInt64, math.MaxInt64, math.MaxInt64)
+		m.startTime, m.endTime, nil, nil, math.MaxInt64, math.MaxInt64, math.MaxInt64, 0)
 	if err != nil {
 		return fmt.Sprintf("error: %s", err)
 	}

pkg/storage/mvcc.go

@@ -3144,6 +3144,7 @@ func MVCCClearTimeRange(
 	leftPeekBound, rightPeekBound roachpb.Key,
 	clearRangeThreshold int,
 	maxBatchSize, maxBatchByteSize int64,
+	maxLockConflicts int64,
 ) (roachpb.Key, error) {
 	var batchSize, batchByteSize int64
 	var resumeKey roachpb.Key
@@ -3181,7 +3182,6 @@ func MVCCClearTimeRange(
 	// be resolved. We don't _expect_ to hit any since the RevertRange is only
 	// intended for non-live key spans, but there could be an intent or lock
 	// leftover from before the keyspace become non-live.
-	maxLockConflicts := int64(0) // TODO(nvanbenschoten): plumb this in.
 	if locks, err := ScanLocks(
 		ctx, rw, key, endKey, maxLockConflicts, 0, BatchEvalReadCategory); err != nil {
 		return nil, err

pkg/storage/mvcc_history_test.go

@@ -1231,7 +1231,7 @@ func cmdClearTimeRange(e *evalCtx) error {
 
 	rw, leftPeekBound, rightPeekBound := e.metamorphicPeekBounds(batch, key, endKey)
 	resume, err := storage.MVCCClearTimeRange(e.ctx, rw, e.ms, key, endKey, targetTs, ts,
-		leftPeekBound, rightPeekBound, clearRangeThreshold, int64(maxBatchSize), int64(maxBatchByteSize))
+		leftPeekBound, rightPeekBound, clearRangeThreshold, int64(maxBatchSize), int64(maxBatchByteSize), 0)
 	if err != nil {
 		return err
 	}

pkg/storage/mvcc_stats_test.go

@@ -2055,7 +2055,7 @@ func TestMVCCStatsRandomized(t *testing.T) {
 
 		desc := fmt.Sprintf("mvccClearTimeRange=%s, startTime=%s, endTime=%s", keySpan, startTime, endTime)
 		_, err := MVCCClearTimeRange(ctx, s.batch, s.MSDelta, keySpan.Key, keySpan.EndKey,
-			startTime, endTime, nil /* leftPeekBound */, nil /* rightPeekBound */, clearRangeThreshold, 0, 0)
+			startTime, endTime, nil /* leftPeekBound */, nil /* rightPeekBound */, clearRangeThreshold, 0, 0, 0)
 		if err != nil {
 			desc += " " + err.Error()
 			return false, desc + ": " + err.Error()

pkg/storage/mvcc_test.go

@@ -1978,11 +1978,11 @@ func TestMVCCClearTimeRange(t *testing.T) {
 		sz int64,
 		byteLimit int64,
 	) int {
-		resume, err := MVCCClearTimeRange(ctx, rw, ms, key, endKey, ts, endTs, nil, nil, 64, sz, byteLimit)
+		resume, err := MVCCClearTimeRange(ctx, rw, ms, key, endKey, ts, endTs, nil, nil, 64, sz, byteLimit, 0)
 		require.NoError(t, err)
 		attempts := 1
 		for resume != nil {
-			resume, err = MVCCClearTimeRange(ctx, rw, ms, resume, endKey, ts, endTs, nil, nil, 64, sz, byteLimit)
+			resume, err = MVCCClearTimeRange(ctx, rw, ms, resume, endKey, ts, endTs, nil, nil, 64, sz, byteLimit, 0)
 			require.NoError(t, err)
 			attempts++
 		}
@@ -1991,7 +1991,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear > ts0", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts0, ts0Content)
 		assertKVs(t, b, ts1, ts0Content)
@@ -2047,7 +2047,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear > ts4 (nothing) ", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts4, ts5, nil, nil, 64, 10, kb)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts4, ts5, nil, nil, 64, 10, kb, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts4, ts4Content)
 		assertKVs(t, b, ts5, ts4Content)
@@ -2056,7 +2056,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear > ts5 (nothing)", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts5, ts5.Next(), nil, nil, 64, 10, kb)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts5, ts5.Next(), nil, nil, 64, 10, kb, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts4, ts4Content)
 		assertKVs(t, b, ts5, ts4Content)
@@ -2073,7 +2073,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear > ts0 in empty span (nothing)", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey5, ts0, ts5, nil, nil, 64, 10, kb)
+		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey5, ts0, ts5, nil, nil, 64, 10, kb, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts2, ts2Content)
 		assertKVs(t, b, ts5, ts4Content)
@@ -2082,7 +2082,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear > ts0 in empty span [k3,k5) (nothing)", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey5, ts0, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey5, ts0, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts2, ts2Content)
 		assertKVs(t, b, ts5, ts4Content)
@@ -2091,7 +2091,7 @@ func TestMVCCClearTimeRange(t *testing.T) {
 	t.Run("clear k3 and up in ts0 > x >= ts1 (nothing)", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
-		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, keyMax, ts0, ts1, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, keyMax, ts0, ts1, nil, nil, 64, 10, 1<<10, 0)
 		require.NoError(t, err)
 		assertKVs(t, b, ts2, ts2Content)
 		assertKVs(t, b, ts5, ts4Content)
@@ -2107,31 +2107,31 @@ func TestMVCCClearTimeRange(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addIntent(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db3\"")
 	})
 
 	t.Run("clear exactly hitting intent fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addIntent(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey4, ts2, ts3, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, testKey3, testKey4, ts2, ts3, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db3\"")
 	})
 
 	t.Run("clear everything above intent fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addIntent(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts3, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts3, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db3\"")
 	})
 
 	t.Run("clear below intent fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addIntent(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts1, ts2, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts1, ts2, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db3\"")
 	})
 
@@ -2144,31 +2144,31 @@ func TestMVCCClearTimeRange(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addLock(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts0, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db1\"")
 	})
 
 	t.Run("clear exactly hitting lock fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addLock(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts2, ts3, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts2, ts3, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db1\"")
 	})
 
 	t.Run("clear everything above lock fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addLock(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts3, ts5, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts3, ts5, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db1\"")
 	})
 
 	t.Run("clear below lock fails", func(t *testing.T) {
 		b := eng.NewBatch()
 		defer b.Close()
 		addLock(t, b)
-		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts1, ts2, nil, nil, 64, 10, 1<<10)
+		_, err := MVCCClearTimeRange(ctx, b, nil, localMax, keyMax, ts1, ts2, nil, nil, 64, 10, 1<<10, 0)
 		require.EqualError(t, err, "conflicting locks on \"/db1\"")
 	})
 }
@@ -2275,7 +2275,7 @@ func TestMVCCClearTimeRangeOnRandomData(t *testing.T) {
 				attempts++
 				batch := e.NewBatch()
 				startKey, err = MVCCClearTimeRange(ctx, batch, &ms, startKey, keyMax, revertTo, now,
-					nil, nil, clearRangeThreshold, keyLimit, byteLimit)
+					nil, nil, clearRangeThreshold, keyLimit, byteLimit, 0)
 				require.NoError(t, err)
 				require.NoError(t, batch.Commit(false))
 				batch.Close()