From 516553d9e1c5773a7e8cb256c26a77ab8b44e164 Mon Sep 17 00:00:00 2001 From: Yahor Yuzefovich Date: Sat, 9 Jul 2022 10:08:44 -0700 Subject: [PATCH] Revert "kvstreamer: reuse incomplete Get requests on resume batches" This reverts commit 21f2390b0ed05150f3bb04cd9bbf769a2a5b9337. Previously, I didn't realize that the KV layer would modify all requests included into the BatchRequest in `txnSeqNumAllocator`, so we cannot reuse even incomplete GetRequests. It is unfortunate, but not a big deal. Release note: None --- pkg/kv/kvclient/kvstreamer/streamer.go | 28 +++++++++++++++----------- 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/pkg/kv/kvclient/kvstreamer/streamer.go b/pkg/kv/kvclient/kvstreamer/streamer.go index 2c19df5070ce..37154fda23b0 100644 --- a/pkg/kv/kvclient/kvstreamer/streamer.go +++ b/pkg/kv/kvclient/kvstreamer/streamer.go @@ -1567,6 +1567,14 @@ func buildResumeSingleRangeBatch( // requests with the ResumeSpans. resumeReq.reqsReservedBytes = fp.resumeReqsMemUsage resumeReq.overheadAccountedFor = req.overheadAccountedFor + // Note that due to limitations of the KV layer (#75452) we cannot reuse + // original requests because the KV doesn't allow mutability (and all + // requests are modified by txnSeqNumAllocator, even if they are not + // evaluated due to TargetBytes limit). + gets := make([]struct { + req roachpb.GetRequest + union roachpb.RequestUnion_Get + }, fp.numIncompleteGets) scans := make([]struct { req roachpb.ScanRequest union roachpb.RequestUnion_Scan @@ -1583,18 +1591,14 @@ func buildResumeSingleRangeBatch( emptyResponse = false continue } - // This Get wasn't completed - include it into the batch again (we - // can just reuse the original request since it hasn't been - // modified which is also asserted below). - if buildutil.CrdbTestBuild { - if origSpan := req.reqs[i].GetInner().Header().Span(); !get.ResumeSpan.Equal(origSpan) { - panic(errors.AssertionFailedf( - "unexpectedly the ResumeSpan %s on the GetResponse is different from the original span %s", - get.ResumeSpan, origSpan, - )) - } - } - resumeReq.reqs[resumeReqIdx] = req.reqs[i] + // This Get wasn't completed - create a new request according to the + // ResumeSpan and include it into the batch. + newGet := gets[0] + gets = gets[1:] + newGet.req.SetSpan(*get.ResumeSpan) + newGet.req.KeyLocking = s.keyLocking + newGet.union.Get = &newGet.req + resumeReq.reqs[resumeReqIdx].Value = &newGet.union resumeReq.positions = append(resumeReq.positions, position) if req.subRequestIdx != nil { resumeReq.subRequestIdx = append(resumeReq.subRequestIdx, req.subRequestIdx[i])