Merge #150406

150406: kvcoord: make GetRequest transforms configurable  r=miraradeva a=stevendanna

Making this metamorphic might help us discover other edge cases. Also,
it gives us the option to flip this off by default on 25.3 if we choose
to.

Release note: None

Epic: none

Co-authored-by: Steven Danna <[email protected]>

Author: craig[bot]

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -51,6 +51,13 @@ var bufferedWritesScanTransformEnabled = settings.RegisterBoolSetting(
 	metamorphic.ConstantWithTestBool("kv.transaction.write_buffering.transformations.scans.enabled", true /* defaultValue */),
 )
 
+var bufferedWritesGetTransformEnabled = settings.RegisterBoolSetting(
+	settings.ApplicationLevel,
+	"kv.transaction.write_buffering.transformations.get.enabled",
+	"if enabled, locking get requests with replicated durability are transformed to unreplicated durability",
+	metamorphic.ConstantWithTestBool("kv.transaction.write_buffering.transformations.get.enabled", true /* defaultValue */),
+)
+
 const defaultBufferSize = 1 << 22 // 4MB
 var bufferedWritesMaxBufferSize = settings.RegisterByteSizeSetting(
 	settings.ApplicationLevel,
@@ -222,6 +229,11 @@ type txnWriteBuffer struct {
 	testingOverrideCPutEvalFn func(expBytes []byte, actVal *roachpb.Value, actValPresent bool, allowNoExisting bool) *kvpb.ConditionFailedError
 }
 
+type transformConfig struct {
+	transformScans bool
+	transformGets  bool
+}
+
 type pipelineEnabler interface {
 	enableImplicitPipelining()
 }
@@ -274,17 +286,20 @@ func (twb *txnWriteBuffer) SendLocked(
 
 	// We check if scan transforms are enabled once and use that answer until the
 	// end of SendLocked.
-	transformScans := bufferedWritesScanTransformEnabled.Get(&twb.st.SV)
+	cfg := transformConfig{
+		transformScans: bufferedWritesScanTransformEnabled.Get(&twb.st.SV),
+		transformGets:  bufferedWritesGetTransformEnabled.Get(&twb.st.SV),
+	}
 
-	if twb.batchRequiresFlush(ctx, ba, transformScans) {
+	if twb.batchRequiresFlush(ctx, ba, cfg) {
 		return twb.flushBufferAndSendBatch(ctx, ba)
 	}
 
 	// Check if buffering writes from the supplied batch will run us over
 	// budget. If it will, we shouldn't buffer writes from the current batch,
 	// and flush the buffer.
 	maxSize := bufferedWritesMaxBufferSize.Get(&twb.st.SV)
-	bufSize := twb.estimateSize(ba, transformScans) + twb.bufferSize
+	bufSize := twb.estimateSize(ba, cfg) + twb.bufferSize
 
 	// NB: if bufferedWritesMaxBufferSize is set to 0 then we effectively disable
 	// any buffer limiting.
@@ -302,7 +317,7 @@ func (twb *txnWriteBuffer) SendLocked(
 		return nil, kvpb.NewError(err)
 	}
 
-	transformedBa, rr, pErr := twb.applyTransformations(ctx, ba, transformScans)
+	transformedBa, rr, pErr := twb.applyTransformations(ctx, ba, cfg)
 	if pErr != nil {
 		return nil, pErr
 	}
@@ -338,7 +353,7 @@ func (twb *txnWriteBuffer) SendLocked(
 }
 
 func (twb *txnWriteBuffer) batchRequiresFlush(
-	ctx context.Context, ba *kvpb.BatchRequest, transformScans bool,
+	ctx context.Context, ba *kvpb.BatchRequest, _ transformConfig,
 ) bool {
 	for _, ru := range ba.Requests {
 		req := ru.GetInner()
@@ -465,7 +480,7 @@ func unsupportedOptionError(m kvpb.Method, option string) error {
 
 // estimateSize returns a conservative estimate by which the buffer will grow in
 // size if the writes from the supplied batch request are buffered.
-func (twb *txnWriteBuffer) estimateSize(ba *kvpb.BatchRequest, transformScans bool) int64 {
+func (twb *txnWriteBuffer) estimateSize(ba *kvpb.BatchRequest, cfg transformConfig) int64 {
 	var scratch bufferedWrite
 	scratch.vals = scratch.valsScratch[:1]
 	estimate := int64(0)
@@ -485,7 +500,7 @@ func (twb *txnWriteBuffer) estimateSize(ba *kvpb.BatchRequest, transformScans bo
 			estimate += scratch.size()
 			estimate += lockKeyInfoSize
 		case *kvpb.GetRequest:
-			if IsReplicatedLockingRequest(t) {
+			if IsReplicatedLockingRequest(t) && cfg.transformGets {
 				scratch.key = t.Key
 				estimate += scratch.size()
 				estimate += lockKeyInfoSize
@@ -522,7 +537,7 @@ func (twb *txnWriteBuffer) estimateSize(ba *kvpb.BatchRequest, transformScans bo
 			// the buffer. Here, we assume at least 1 key will be returned that is
 			// about the size of the scan start boundary. We try to protect from large
 			// buffer overflows by transforming the batch's MaxSpanRequestKeys later.
-			if IsReplicatedLockingRequest(t) && transformScans {
+			if IsReplicatedLockingRequest(t) && cfg.transformScans {
 				scratch.key = t.Key
 				scratch.vals[0] = bufferedValue{
 					seq: t.Sequence,
@@ -531,7 +546,7 @@ func (twb *txnWriteBuffer) estimateSize(ba *kvpb.BatchRequest, transformScans bo
 			}
 		case *kvpb.ReverseScanRequest:
 			// See the comment on the ScanRequest case for more details.
-			if IsReplicatedLockingRequest(t) && transformScans {
+			if IsReplicatedLockingRequest(t) && cfg.transformScans {
 				scratch.key = t.Key
 				scratch.vals[0] = bufferedValue{
 					seq: t.Sequence,
@@ -830,7 +845,7 @@ func (twb *txnWriteBuffer) closeLocked() {}
 // locking Get request if it can be served from the buffer (i.e if a lock of
 // sufficient strength has been acquired and a value has been buffered).
 func (twb *txnWriteBuffer) applyTransformations(
-	ctx context.Context, ba *kvpb.BatchRequest, transformScans bool,
+	ctx context.Context, ba *kvpb.BatchRequest, cfg transformConfig,
 ) (*kvpb.BatchRequest, requestRecords, *kvpb.Error) {
 	baRemote := ba.ShallowCopy()
 	// TODO(arul): We could improve performance here by pre-allocating
@@ -939,7 +954,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 			requiresLockTransform := IsReplicatedLockingRequest(t)
 			requestRequired := requiresAdditionalLocking || !served
 
-			if requestRequired && requiresLockTransform {
+			if requestRequired && requiresLockTransform && cfg.transformGets {
 				var getReqU kvpb.RequestUnion
 				getReq := t.ShallowCopy().(*kvpb.GetRequest)
 				getReq.KeyLockingDurability = lock.Unreplicated
@@ -957,7 +972,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 			// Regardless of whether the scan overlaps with any writes in the buffer
 			// or not, we must send the request to the KV layer. We can't know for
 			// sure that there's nothing else to read.
-			if IsReplicatedLockingRequest(t) && transformScans {
+			if IsReplicatedLockingRequest(t) && cfg.transformScans {
 				var scanReqU kvpb.RequestUnion
 				scanReq := t.ShallowCopy().(*kvpb.ScanRequest)
 				scanReq.KeyLockingDurability = lock.Unreplicated
@@ -975,7 +990,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 			// Regardless of whether the reverse scan overlaps with any writes in the
 			// buffer or not, we must send the request to the KV layer. We can't know
 			// for sure that there's nothing else to read.
-			if IsReplicatedLockingRequest(t) && transformScans {
+			if IsReplicatedLockingRequest(t) && cfg.transformScans {
 				var rScanReqU kvpb.RequestUnion
 				rScanReq := t.ShallowCopy().(*kvpb.ReverseScanRequest)
 				rScanReq.KeyLockingDurability = lock.Unreplicated

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -31,6 +31,7 @@ func makeMockTxnWriteBuffer(
 ) (txnWriteBuffer, *mockLockedSender, *cluster.Settings) {
 	st := cluster.MakeClusterSettings()
 	bufferedWritesScanTransformEnabled.Override(ctx, &st.SV, true)
+	bufferedWritesGetTransformEnabled.Override(ctx, &st.SV, true)
 	bufferedWritesMaxBufferSize.Override(ctx, &st.SV, defaultBufferSize)
 
 	var metrics TxnMetrics
@@ -1406,6 +1407,12 @@ func TestTxnWriteBufferEstimateSize(t *testing.T) {
 	txn.Sequence = 10
 	keyA := roachpb.Key("a")
 	keyB := roachpb.Key("b")
+	cfg := transformConfig{
+		transformScans: true,
+		transformGets:  true,
+	}
+	noScanCfg := cfg
+	noScanCfg.transformScans = false
 
 	valAStr := "valA"
 	valA := roachpb.MakeValueFromString(valAStr)
@@ -1419,23 +1426,23 @@ func TestTxnWriteBufferEstimateSize(t *testing.T) {
 	ba.Add(putA)
 
 	expectedUnlockedPutSize := int64(len(keyA)+len(valA.RawBytes)) + bufferedWriteStructOverhead + bufferedValueStructOverhead
-	require.Equal(t, expectedUnlockedPutSize, twb.estimateSize(ba, true))
+	require.Equal(t, expectedUnlockedPutSize, twb.estimateSize(ba, cfg))
 
 	ba = &kvpb.BatchRequest{}
 	ba.Header = kvpb.Header{Txn: &txn}
 	putA = putArgs(keyA, valAStr, txn.Sequence)
 	putA.MustAcquireExclusiveLock = true
 	ba.Add(putA)
 
-	require.Equal(t, expectedUnlockedPutSize+lockKeyInfoSize, twb.estimateSize(ba, true))
+	require.Equal(t, expectedUnlockedPutSize+lockKeyInfoSize, twb.estimateSize(ba, cfg))
 
 	ba = &kvpb.BatchRequest{}
 	cputLarge := cputArgs(keyLarge, valLargeStr, "", txn.Sequence)
 	ba.Add(cputLarge)
 
 	require.Equal(t,
 		int64(len(keyLarge)+len(valLarge.RawBytes))+bufferedWriteStructOverhead+bufferedValueStructOverhead+lockKeyInfoSize,
-		twb.estimateSize(ba, true),
+		twb.estimateSize(ba, cfg),
 	)
 
 	ba = &kvpb.BatchRequest{}
@@ -1445,14 +1452,14 @@ func TestTxnWriteBufferEstimateSize(t *testing.T) {
 	// NB: note that we're overcounting here, as we're deleting a key that's
 	// already present in the buffer. But that's what estimating is about.
 	expectedUnlockedDelSize := int64(len(keyA)) + bufferedWriteStructOverhead + bufferedValueStructOverhead
-	require.Equal(t, expectedUnlockedDelSize, twb.estimateSize(ba, true))
+	require.Equal(t, expectedUnlockedDelSize, twb.estimateSize(ba, cfg))
 
 	ba = &kvpb.BatchRequest{}
 	delA = delArgs(keyA, txn.Sequence)
 	delA.MustAcquireExclusiveLock = true
 	ba.Add(delA)
 
-	require.Equal(t, expectedUnlockedDelSize+lockKeyInfoSize, twb.estimateSize(ba, true))
+	require.Equal(t, expectedUnlockedDelSize+lockKeyInfoSize, twb.estimateSize(ba, cfg))
 
 	ba = &kvpb.BatchRequest{}
 	ba.Add(&kvpb.ScanRequest{
@@ -1462,8 +1469,8 @@ func TestTxnWriteBufferEstimateSize(t *testing.T) {
 	})
 
 	expectedLockedScanSize := int64(len(keyA)) + bufferedWriteStructOverhead + bufferedValueStructOverhead + lockKeyInfoSize
-	require.Equal(t, expectedLockedScanSize, twb.estimateSize(ba, true))
-	require.Equal(t, int64(0), twb.estimateSize(ba, false))
+	require.Equal(t, expectedLockedScanSize, twb.estimateSize(ba, cfg))
+	require.Equal(t, int64(0), twb.estimateSize(ba, noScanCfg))
 
 	ba = &kvpb.BatchRequest{}
 	ba.Add(&kvpb.ReverseScanRequest{
@@ -1473,24 +1480,24 @@ func TestTxnWriteBufferEstimateSize(t *testing.T) {
 	})
 
 	expectedLockedRScanSize := int64(len(keyA)) + bufferedWriteStructOverhead + bufferedValueStructOverhead + lockKeyInfoSize
-	require.Equal(t, expectedLockedRScanSize, twb.estimateSize(ba, true))
-	require.Equal(t, int64(0), twb.estimateSize(ba, false))
+	require.Equal(t, expectedLockedRScanSize, twb.estimateSize(ba, cfg))
+	require.Equal(t, int64(0), twb.estimateSize(ba, noScanCfg))
 
 	ba = &kvpb.BatchRequest{}
 	ba.Add(&kvpb.ScanRequest{
 		KeyLockingStrength:   lock.Exclusive,
 		KeyLockingDurability: lock.Unreplicated,
 		RequestHeader:        kvpb.RequestHeader{Key: keyA, EndKey: keyB, Sequence: txn.Sequence},
 	})
-	require.Equal(t, int64(0), twb.estimateSize(ba, true))
+	require.Equal(t, int64(0), twb.estimateSize(ba, cfg))
 
 	ba = &kvpb.BatchRequest{}
 	ba.Add(&kvpb.ReverseScanRequest{
 		KeyLockingStrength:   lock.Exclusive,
 		KeyLockingDurability: lock.Unreplicated,
 		RequestHeader:        kvpb.RequestHeader{Key: keyA, EndKey: keyB, Sequence: txn.Sequence},
 	})
-	require.Equal(t, int64(0), twb.estimateSize(ba, true))
+	require.Equal(t, int64(0), twb.estimateSize(ba, cfg))
 }
 
 // TestTxnWriteBufferFlushesWhenOverBudget verifies that the txnWriteBuffer