kv,sql: explicitly set IsReverse on the Header

When processing batches that touch multiple ranges, the DistSender needs
to know whether to iterate across those ranges in the forward or reverse
manner (i.e. ASC or DESC for range keys). Currently, it only uses the
reverse direction if it finds at least one ReverseScan request in the
batch, and it uses the forward direction otherwise. This goes against
the needs of SQL which might issue point Gets when performing a SQL
revscan operation, and currently in such a scenario we would hit an
error (instead of returning the results in incorrect order).

This commit fixes the issue by introducing `IsReverse` boolean on the
batch header to explicitly indicate the direction for range iteration.
It seems reasonable that the caller should be explicit about this, and
we also add a validation that the boolean is set correctly (meaning that
it should be `false` when only forward range requests are present and
`true` when only reverse range requests are present).

In order to simplify the story a bit, the DistSender will no longer
allow batches that have both forward and reverse range requests.
Previously, this limitation only applied to the batches with limits, but
now it's extended to be unconditional. SQL never issues such batches, so
the limitation seems acceptable. This limitation required some updates
to the existing tests, including KVNemesis to not generate batches that
are now disallowed.

See also 619f395 for some related context.

Given that the new header field is only examined on the KV client, the
change can be backported with no mixed-version concerns.

Release note (bug fix): Previously, CockroachDB could hit an error
`ERROR: span with results after resume span...` when evaluating some
queries with ORDER BY ... DESC in an edge case. The bug has been present
since about v22.1 and is now fixed.

Author: yuzefovich

pkg/kv/batch.go

@@ -365,6 +365,8 @@ func (b *Batch) AddRawRequest(reqs ...kvpb.Request) {
 			*kvpb.IncrementRequest,
 			*kvpb.DeleteRequest:
 			numRows = 1
+		case *kvpb.ReverseScanRequest:
+			b.Header.IsReverse = true
 		}
 		b.appendReqs(args)
 		b.initResult(1 /* calls */, numRows, raw, nil)
@@ -735,6 +737,7 @@ func (b *Batch) scan(
 	str kvpb.KeyLockingStrengthType,
 	dur kvpb.KeyLockingDurabilityType,
 ) {
+	b.Header.IsReverse = isReverse
 	begin, err := marshalKey(s)
 	if err != nil {
 		b.initResult(0, 0, notRaw, err)

pkg/kv/db.go

@@ -499,6 +499,7 @@ func (db *DB) scan(
 	if maxRows > 0 {
 		b.Header.MaxSpanRequestKeys = maxRows
 	}
+	b.Header.IsReverse = isReverse
 	b.scan(begin, end, isReverse, str, dur)
 	r, err := getOneResult(db.Run(ctx, b), b)
 	return r.Rows, err

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -1079,35 +1079,46 @@ func (ds *DistSender) initAndVerifyBatch(ctx context.Context, ba *kvpb.BatchRequ
 		return kvpb.NewErrorf("empty batch")
 	}
 
-	if ba.MaxSpanRequestKeys != 0 || ba.TargetBytes != 0 {
-		// Verify that the batch contains only specific requests. Verify that a
-		// batch with a ReverseScan only contains ReverseScan range requests.
-		var foundForward, foundReverse bool
-		for _, req := range ba.Requests {
-			inner := req.GetInner()
-			switch inner.(type) {
-			case *kvpb.ScanRequest, *kvpb.ResolveIntentRangeRequest,
-				*kvpb.DeleteRangeRequest, *kvpb.RevertRangeRequest,
-				*kvpb.ExportRequest, *kvpb.QueryLocksRequest, *kvpb.IsSpanEmptyRequest:
-				// Accepted forward range requests.
-				foundForward = true
+	// Verify that forward and reverse range requests are never in the same
+	// batch. Also verify that the batch with limits contains only specific
+	// requests.
+	var foundForward, foundReverse bool
+	var disallowedReq string
+	for _, req := range ba.Requests {
+		inner := req.GetInner()
+		switch inner.(type) {
+		case *kvpb.ScanRequest, *kvpb.ResolveIntentRangeRequest,
+			*kvpb.DeleteRangeRequest, *kvpb.RevertRangeRequest,
+			*kvpb.ExportRequest, *kvpb.QueryLocksRequest, *kvpb.IsSpanEmptyRequest:
+			// Accepted forward range requests.
+			foundForward = true
 
-			case *kvpb.ReverseScanRequest:
-				// Accepted reverse range requests.
-				foundReverse = true
+		case *kvpb.ReverseScanRequest:
+			// Accepted reverse range requests.
+			foundReverse = true
 
-			case *kvpb.QueryIntentRequest, *kvpb.EndTxnRequest,
-				*kvpb.GetRequest, *kvpb.ResolveIntentRequest, *kvpb.DeleteRequest, *kvpb.PutRequest:
-				// Accepted point requests that can be in batches with limit.
+		case *kvpb.QueryIntentRequest, *kvpb.EndTxnRequest,
+			*kvpb.GetRequest, *kvpb.ResolveIntentRequest, *kvpb.DeleteRequest, *kvpb.PutRequest:
+			// Accepted point requests that can be in batches with limit. No
+			// need to set disallowedReq.
 
-			default:
-				return kvpb.NewErrorf("batch with limit contains %s request", inner.Method())
-			}
-		}
-		if foundForward && foundReverse {
-			return kvpb.NewErrorf("batch with limit contains both forward and reverse scans")
+		default:
+			disallowedReq = inner.Method().String()
 		}
 	}
+	if foundForward && foundReverse {
+		return kvpb.NewErrorf("batch contains both forward and reverse requests")
+	}
+	if (ba.MaxSpanRequestKeys != 0 || ba.TargetBytes != 0) && disallowedReq != "" {
+		return kvpb.NewErrorf("batch with limit contains %s request", disallowedReq)
+	}
+	// Also verify that IsReverse is set accordingly on the batch header.
+	if foundForward && ba.Header.IsReverse {
+		return kvpb.NewErrorf("batch contains forward requests but IsReverse is set")
+	}
+	if foundReverse && !ba.Header.IsReverse {
+		return kvpb.NewErrorf("batch contains reverse requests but IsReverse is not set")
+	}
 
 	switch ba.WaitPolicy {
 	case lock.WaitPolicy_Block, lock.WaitPolicy_Error:
@@ -1268,7 +1279,6 @@ func (ds *DistSender) Send(
 		if err != nil {
 			return nil, kvpb.NewError(err)
 		}
-		isReverse := ba.IsReverse()
 
 		// Determine whether this part of the BatchRequest contains a committing
 		// EndTxn request.
@@ -1282,9 +1292,9 @@ func (ds *DistSender) Send(
 		var rpl *kvpb.BatchResponse
 		var pErr *kvpb.Error
 		if withParallelCommit {
-			rpl, pErr = ds.divideAndSendParallelCommit(ctx, ba, rs, isReverse, 0 /* batchIdx */)
+			rpl, pErr = ds.divideAndSendParallelCommit(ctx, ba, rs, 0 /* batchIdx */)
 		} else {
-			rpl, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, withCommit, 0 /* batchIdx */)
+			rpl, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, withCommit, 0 /* batchIdx */)
 		}
 
 		if pErr == errNo1PCTxn {
@@ -1475,7 +1485,7 @@ type response struct {
 // method is never invoked recursively, but it is exposed to maintain symmetry
 // with divideAndSendBatchToRanges.
 func (ds *DistSender) divideAndSendParallelCommit(
-	ctx context.Context, ba *kvpb.BatchRequest, rs roachpb.RSpan, isReverse bool, batchIdx int,
+	ctx context.Context, ba *kvpb.BatchRequest, rs roachpb.RSpan, batchIdx int,
 ) (br *kvpb.BatchResponse, pErr *kvpb.Error) {
 	// Search backwards, looking for the first pre-commit QueryIntent.
 	swapIdx := -1
@@ -1491,7 +1501,7 @@ func (ds *DistSender) divideAndSendParallelCommit(
 	if swapIdx == -1 {
 		// No pre-commit QueryIntents. Nothing to split.
 		log.VEvent(ctx, 3, "no pre-commit QueryIntents found, sending batch as-is")
-		return ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, true /* withCommit */, batchIdx)
+		return ds.divideAndSendBatchToRanges(ctx, ba, rs, true /* withCommit */, batchIdx)
 	}
 
 	// Swap the EndTxn request and the first pre-commit QueryIntent. This
@@ -1518,7 +1528,8 @@ func (ds *DistSender) divideAndSendParallelCommit(
 	if err != nil {
 		return br, kvpb.NewError(err)
 	}
-	qiIsReverse := false // QueryIntentRequests do not carry the isReverse flag
+	// No need to process QueryIntentRequests in the reverse order.
+	qiBa.IsReverse = false
 	qiBatchIdx := batchIdx + 1
 	qiResponseCh := make(chan response, 1)
 
@@ -1544,7 +1555,7 @@ func (ds *DistSender) divideAndSendParallelCommit(
 
 		// Send the batch with withCommit=true since it will be inflight
 		// concurrently with the EndTxn batch below.
-		reply, pErr := ds.divideAndSendBatchToRanges(ctx, qiBa, qiRS, qiIsReverse, true /* withCommit */, qiBatchIdx)
+		reply, pErr := ds.divideAndSendBatchToRanges(ctx, qiBa, qiRS, true /* withCommit */, qiBatchIdx)
 		qiResponseCh <- response{reply: reply, positions: positions, pErr: pErr}
 	}
 
@@ -1576,10 +1587,7 @@ func (ds *DistSender) divideAndSendParallelCommit(
 	if err != nil {
 		return nil, kvpb.NewError(err)
 	}
-	// Note that we don't need to recompute isReverse for the updated batch
-	// since we only separated out QueryIntentRequests which don't carry the
-	// isReverse flag.
-	br, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, true /* withCommit */, batchIdx)
+	br, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, true /* withCommit */, batchIdx)
 
 	// Wait for the QueryIntent-only batch to complete and stitch
 	// the responses together.
@@ -1748,12 +1756,6 @@ func mergeErrors(pErr1, pErr2 *kvpb.Error) *kvpb.Error {
 // is trimmed against each range which is part of the span and sent
 // either serially or in parallel, if possible.
 //
-// isReverse indicates the direction that the provided span should be
-// iterated over while sending requests. It is passed in by callers
-// instead of being recomputed based on the requests in the batch to
-// prevent the iteration direction from switching midway through a
-// batch, in cases where partial batches recurse into this function.
-//
 // withCommit indicates that the batch contains a transaction commit
 // or that a transaction commit is being run concurrently with this
 // batch. Either way, if this is true then sendToReplicas will need
@@ -1763,12 +1765,7 @@ func mergeErrors(pErr1, pErr2 *kvpb.Error) *kvpb.Error {
 // being processed by this method. It's specified as non-zero when
 // this method is invoked recursively.
 func (ds *DistSender) divideAndSendBatchToRanges(
-	ctx context.Context,
-	ba *kvpb.BatchRequest,
-	rs roachpb.RSpan,
-	isReverse bool,
-	withCommit bool,
-	batchIdx int,
+	ctx context.Context, ba *kvpb.BatchRequest, rs roachpb.RSpan, withCommit bool, batchIdx int,
 ) (br *kvpb.BatchResponse, pErr *kvpb.Error) {
 	// Clone the BatchRequest's transaction so that future mutations to the
 	// proto don't affect the proto in this batch.
@@ -1778,7 +1775,7 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 	// Get initial seek key depending on direction of iteration.
 	var scanDir ScanDirection
 	var seekKey roachpb.RKey
-	if !isReverse {
+	if !ba.IsReverse {
 		scanDir = Ascending
 		seekKey = rs.Key
 	} else {
@@ -1793,7 +1790,7 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 	// Take the fast path if this batch fits within a single range.
 	if !ri.NeedAnother(rs) {
 		resp := ds.sendPartialBatch(
-			ctx, ba, rs, isReverse, withCommit, batchIdx, ri.Token(),
+			ctx, ba, rs, withCommit, batchIdx, ri.Token(),
 		)
 		// resp.positions remains nil since the original batch is fully
 		// contained within a single range.
@@ -1915,7 +1912,7 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 		}
 
 		if pErr == nil && couldHaveSkippedResponses {
-			fillSkippedResponses(ba, br, seekKey, resumeReason, isReverse)
+			fillSkippedResponses(ba, br, seekKey, resumeReason)
 		}
 	}()
 
@@ -1994,7 +1991,7 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 		// If we can reserve one of the limited goroutines available for parallel
 		// batch RPCs, send asynchronously.
 		if canParallelize && !lastRange && !ds.disableParallelBatches {
-			if ds.sendPartialBatchAsync(ctx, curRangeBatch, curRangeRS, isReverse, withCommit, batchIdx, ri.Token(), responseCh, positions) {
+			if ds.sendPartialBatchAsync(ctx, curRangeBatch, curRangeRS, withCommit, batchIdx, ri.Token(), responseCh, positions) {
 				asyncSent = true
 			} else {
 				asyncThrottled = true
@@ -2009,7 +2006,7 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 					}()
 				}
 				return ds.sendPartialBatch(
-					ctx, curRangeBatch, curRangeRS, isReverse, withCommit, batchIdx, ri.Token(),
+					ctx, curRangeBatch, curRangeRS, withCommit, batchIdx, ri.Token(),
 				)
 			}()
 			resp.positions = positions
@@ -2089,7 +2086,6 @@ func (ds *DistSender) sendPartialBatchAsync(
 	ctx context.Context,
 	ba *kvpb.BatchRequest,
 	rs roachpb.RSpan,
-	isReverse bool,
 	withCommit bool,
 	batchIdx int,
 	routing rangecache.EvictionToken,
@@ -2111,7 +2107,7 @@ func (ds *DistSender) sendPartialBatchAsync(
 		ds.metrics.AsyncSentCount.Inc(1)
 		ds.metrics.AsyncInProgress.Inc(1)
 		defer ds.metrics.AsyncInProgress.Dec(1)
-		resp := ds.sendPartialBatch(ctx, ba, rs, isReverse, withCommit, batchIdx, routing)
+		resp := ds.sendPartialBatch(ctx, ba, rs, withCommit, batchIdx, routing)
 		resp.positions = positions
 		responseCh <- resp
 	}(ctx)
@@ -2175,7 +2171,6 @@ func (ds *DistSender) sendPartialBatch(
 	ctx context.Context,
 	ba *kvpb.BatchRequest,
 	rs roachpb.RSpan,
-	isReverse bool,
 	withCommit bool,
 	batchIdx int,
 	routingTok rangecache.EvictionToken,
@@ -2203,7 +2198,7 @@ func (ds *DistSender) sendPartialBatch(
 
 		if !routingTok.Valid() {
 			var descKey roachpb.RKey
-			if isReverse {
+			if ba.IsReverse {
 				descKey = rs.EndKey
 			} else {
 				descKey = rs.Key
@@ -2219,7 +2214,7 @@ func (ds *DistSender) sendPartialBatch(
 			// replica, while detecting hazardous cases where the follower does
 			// not have the latest information and the current descriptor did
 			// not result in a successful send.
-			routingTok, err = ds.getRoutingInfo(ctx, descKey, prevTok, isReverse)
+			routingTok, err = ds.getRoutingInfo(ctx, descKey, prevTok, ba.IsReverse)
 			if err != nil {
 				log.VErrEventf(ctx, 1, "range descriptor re-lookup failed: %s", err)
 				// We set pErr if we encountered an error getting the descriptor in
@@ -2242,7 +2237,7 @@ func (ds *DistSender) sendPartialBatch(
 			}
 			if !intersection.Equal(rs) {
 				log.Eventf(ctx, "range shrunk; sub-dividing the request")
-				reply, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, withCommit, batchIdx)
+				reply, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, withCommit, batchIdx)
 				return response{reply: reply, pErr: pErr}
 			}
 		}
@@ -2344,7 +2339,7 @@ func (ds *DistSender) sendPartialBatch(
 			// batch here would give a potentially larger response slice
 			// with unknown mapping to our truncated reply).
 			log.VEventf(ctx, 1, "likely split; will resend. Got new descriptors: %s", tErr.Ranges)
-			reply, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, isReverse, withCommit, batchIdx)
+			reply, pErr = ds.divideAndSendBatchToRanges(ctx, ba, rs, withCommit, batchIdx)
 			return response{reply: reply, pErr: pErr}
 		}
 		break
@@ -2393,7 +2388,6 @@ func fillSkippedResponses(
 	br *kvpb.BatchResponse,
 	nextKey roachpb.RKey,
 	resumeReason kvpb.ResumeReason,
-	isReverse bool,
 ) {
 	// Some requests might have no response at all if we used a batch-wide
 	// limit; simply create trivial responses for those. Note that any type
@@ -2423,7 +2417,7 @@ func fillSkippedResponses(
 	for i, resp := range br.Responses {
 		req := ba.Requests[i].GetInner()
 		hdr := resp.GetInner().Header()
-		maybeSetResumeSpan(req, &hdr, nextKey, isReverse)
+		maybeSetResumeSpan(req, &hdr, nextKey, ba.IsReverse)
 		if hdr.ResumeSpan != nil {
 			hdr.ResumeReason = resumeReason
 		}

pkg/kv/kvclient/kvcoord/dist_sender_server_test.go

@@ -809,11 +809,10 @@ func TestMultiRequestBatchWithFwdAndReverseRequests(t *testing.T) {
 		t.Fatal(err)
 	}
 	b := &kv.Batch{}
-	b.Header.MaxSpanRequestKeys = 100
 	b.Scan("a", "b")
 	b.ReverseScan("a", "b")
 	if err := db.Run(ctx, b); !testutils.IsError(
-		err, "batch with limit contains both forward and reverse scans",
+		err, "batch contains both forward and reverse requests",
 	) {
 		t.Fatal(err)
 	}

pkg/kv/kvclient/kvstreamer/streamer.go

@@ -1398,6 +1398,7 @@ func (w *workerCoordinator) performRequestAsync(
 		ba.Header.TargetBytes = targetBytes
 		ba.Header.AllowEmpty = !headOfLine
 		ba.Header.WholeRowsOfSize = w.s.maxKeysPerRow
+		ba.Header.IsReverse = w.s.reverse
 		// TODO(yuzefovich): consider setting MaxSpanRequestKeys whenever
 		// applicable (#67885).
 		ba.AdmissionHeader = w.requestAdmissionHeader

pkg/kv/kvnemesis/applier_test.go

@@ -252,10 +252,16 @@ func TestApplier(t *testing.T) {
 			"txn-si-err", step(closureTxn(ClosureTxnType_Commit, isolation.Snapshot, delRange(k2, k4, 1))),
 		},
 		{
-			"batch-mixed", step(batch(put(k2, 2), get(k1), del(k2, 1), del(k3, 1), scan(k1, k3), reverseScanForUpdate(k1, k5))),
+			"batch-mixed-fwd", step(batch(put(k2, 2), get(k1), del(k2, 1), del(k3, 1), scan(k1, k3))),
 		},
 		{
-			"batch-mixed-err", step(batch(put(k2, 2), getForUpdate(k1), scanForUpdate(k1, k3), reverseScan(k1, k3))),
+			"batch-mixed-rev", step(batch(put(k2, 2), get(k1), del(k2, 1), del(k3, 1), reverseScanForUpdate(k1, k5))),
+		},
+		{
+			"batch-mixed-err-fwd", step(batch(put(k2, 2), getForUpdate(k1), scanForUpdate(k1, k3))),
+		},
+		{
+			"batch-mixed-err-rev", step(batch(put(k2, 2), getForUpdate(k1), reverseScan(k1, k3))),
 		},
 		{
 			"txn-ssi-commit-mixed", step(closureTxn(ClosureTxnType_Commit, isolation.Serializable, put(k5, 5), batch(put(k6, 6), delRange(k3, k5, 1)))),

pkg/kv/kvnemesis/generator.go

@@ -1517,8 +1517,28 @@ func makeRandBatch(c *ClientOperationConfig) opGenFunc {
 		g.registerClientOps(&allowed, c)
 		numOps := rng.Intn(4)
 		ops := make([]Operation, numOps)
-		for i := range ops {
+		var addedForwardScan, addedReverseScan bool
+		for i := 0; i < numOps; i++ {
 			ops[i] = g.selectOp(rng, allowed)
+			if ops[i].Scan != nil {
+				if !ops[i].Scan.Reverse {
+					if addedReverseScan {
+						// We cannot include the forward scan into the batch
+						// that already contains the reverse scan.
+						i--
+						continue
+					}
+					addedForwardScan = true
+				} else {
+					if addedForwardScan {
+						// We cannot include the reverse scan into the batch
+						// that already contains the forward scan.
+						i--
+						continue
+					}
+					addedReverseScan = true
+				}
+			}
 		}
 		return batch(ops...)
 	}