kvcoord: track a write buffer transformation for all requests

Previously, the transaction write buffer kept a transformation for many
requests: truly transformed requests (e.g. CPut), and other requests,
like Get, which were not actually transformed but needed to be tracked
in order to correctly be able to read your own writes. Scan requests,
specifically, were not tracked as transformations, but should be, just
like Gets (#146103).

This commit tracks all requests as transformations regardless of whether
they were actually transformed. For most requests, this is needed for
correctness; for requests like QueryLocksRequest and LeaseInfoRequest,
it is not necessary, but it is harmless and simplifies the code.

Fixes: #146103

Release note: None

Author: miraradeva

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer.go

@@ -675,13 +675,25 @@ func (twb *txnWriteBuffer) applyTransformations(
 	var ts transformations
 	for i, ru := range ba.Requests {
 		req := ru.GetInner()
+		// Track a transformation for the request regardless of the type, and
+		// regardless of whether it was served from the buffer or not. For truly
+		// transformed requests (e.g. CPut) this is expected. For Gets and Scans, we
+		// need to track a transformation because we haven't buffered any writes
+		// from our current batch in the buffer yet, so checking the buffer here, at
+		// request time, isn't sufficient to determine whether the request needs to
+		// serve a read from the buffer before returning a response or not.
+		//
+		// Only QueryLocksRequest and LeaseInfoRequest don't require a tracking
+		// transformation, but it's harmless to add one, and it simplifies the code.
+		//
+		// The stripped field will be set below for specific stripped requests.
+		tr := transformation{
+			stripped:    false,
+			index:       i,
+			origRequest: req,
+		}
 		switch t := req.(type) {
 		case *kvpb.ConditionalPutRequest:
-			ts = append(ts, transformation{
-				stripped:    false,
-				index:       i,
-				origRequest: req,
-			})
 			// NB: Regardless of whether there is already a buffered write on
 			// this key or not, we need to send a locking Get to the KV layer to
 			// acquire a lock. However, if we had knowledge of what locks the
@@ -720,11 +732,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 				})
 				baRemote.Requests = append(baRemote.Requests, getReqU)
 			}
-			ts = append(ts, transformation{
-				stripped:    !t.MustAcquireExclusiveLock,
-				index:       i,
-				origRequest: req,
-			})
+			tr.stripped = !t.MustAcquireExclusiveLock
 
 		case *kvpb.DeleteRequest:
 			// If MustAcquireExclusiveLock flag is set on the DeleteRequest,
@@ -744,11 +752,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 				})
 				baRemote.Requests = append(baRemote.Requests, getReqU)
 			}
-			ts = append(ts, transformation{
-				stripped:    !t.MustAcquireExclusiveLock,
-				index:       i,
-				origRequest: req,
-			})
+			tr.stripped = !t.MustAcquireExclusiveLock
 
 		case *kvpb.GetRequest:
 			// If the key is in the buffer, we must serve the read from the buffer.
@@ -779,45 +783,14 @@ func (twb *txnWriteBuffer) applyTransformations(
 				// Wasn't served locally; send the request to the KV layer.
 				baRemote.Requests = append(baRemote.Requests, ru)
 			}
-			// Even if the request wasn't served from the buffer here, we still track
-			// a transformation for it. That's because we haven't buffered any writes
-			// from our current batch in the buffer yet, so checking the buffer above
-			// isn't sufficient to determine whether the request needs to serve a read
-			// from the buffer before returning a response or not.
-			ts = append(ts, transformation{
-				stripped:    stripped,
-				index:       i,
-				origRequest: req,
-			})
+			tr.stripped = stripped
 
-		case *kvpb.ScanRequest:
-			overlaps := twb.scanOverlaps(t.Key, t.EndKey)
-			if overlaps {
-				ts = append(ts, transformation{
-					stripped:    false,
-					index:       i,
-					origRequest: req,
-				})
-			}
+		case *kvpb.ScanRequest, *kvpb.ReverseScanRequest:
 			// 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.
 			baRemote.Requests = append(baRemote.Requests, ru)
 
-		case *kvpb.ReverseScanRequest:
-			overlaps := twb.scanOverlaps(t.Key, t.EndKey)
-			if overlaps {
-				ts = append(ts, transformation{
-					stripped:    false,
-					index:       i,
-					origRequest: req,
-				})
-			}
-			// Similar to the reasoning above, regardless of whether the reverse
-			// scan overlaps with any writes in the buffer or not, we must send
-			// the request to the KV layer.
-			baRemote.Requests = append(baRemote.Requests, ru)
-
 		case *kvpb.QueryLocksRequest, *kvpb.LeaseInfoRequest:
 			// These requests don't interact with buffered writes, so we simply
 			// let them through.
@@ -826,6 +799,7 @@ func (twb *txnWriteBuffer) applyTransformations(
 		default:
 			return nil, nil, kvpb.NewError(unsupportedMethodError(t.Method()))
 		}
+		ts = append(ts, tr)
 	}
 	return baRemote, ts, nil
 }
@@ -873,14 +847,6 @@ func (twb *txnWriteBuffer) maybeServeRead(
 	return nil, false
 }
 
-// scanOverlaps returns whether the given key range overlaps with any buffered
-// write.
-func (twb *txnWriteBuffer) scanOverlaps(key roachpb.Key, endKey roachpb.Key) bool {
-	it := twb.buffer.MakeIter()
-	it.FirstOverlap(twb.seekItemForSpan(key, endKey))
-	return it.Valid()
-}
-
 // mergeWithScanResp takes a ScanRequest, that was sent to the KV layer, and the
 // response returned by the KV layer, and merges it with any writes that were
 // buffered by the transaction to correctly uphold read-your-own-write

pkg/kv/kvclient/kvcoord/txn_interceptor_write_buffer_test.go

@@ -292,10 +292,10 @@ func TestTxnWriteBufferBlindWritesIncludingOtherRequests(t *testing.T) {
 	// Expect 4 responses, even though only 2 KV requests were sent. Moreover,
 	// ensure that the responses are in the correct order.
 	require.Len(t, br.Responses, 4)
-	require.Equal(t, br.Responses[0].GetInner(), &kvpb.PutResponse{})
-	require.Equal(t, br.Responses[1].GetInner(), &kvpb.GetResponse{})
-	require.Equal(t, br.Responses[2].GetInner(), &kvpb.DeleteResponse{})
-	require.Equal(t, br.Responses[3].GetInner(), &kvpb.ScanResponse{})
+	require.IsType(t, &kvpb.PutResponse{}, br.Responses[0].GetInner())
+	require.IsType(t, &kvpb.GetResponse{}, br.Responses[1].GetInner())
+	require.IsType(t, &kvpb.DeleteResponse{}, br.Responses[2].GetInner())
+	require.IsType(t, &kvpb.ScanResponse{}, br.Responses[3].GetInner())
 
 	// Verify the writes were buffered correctly.
 	expBufferedWrites := []bufferedWrite{

pkg/kv/kvclient/kvcoord/txn_test.go

@@ -2308,12 +2308,10 @@ func TestTxnBufferedWriteReadYourOwnWrites(t *testing.T) {
 				b.Scan(keyB, keyC)
 				return b
 			},
-			// The Scans should see the values written preceding in the same batch,
-			// but currently they don't because of a bug (#146103) that doesn't add
-			// always add a transformation for a Scan request.
+			// The Scans should see the values written preceding in the same batch.
 			expected: map[int32]map[string][]byte{
 				0: {"keyB": value22},
-				2: {"keyB": value22},
+				2: {"keyB": value3},
 			},
 		},
 		// TODO(mira): See if we need more test coverage for other request types