kv: permit ExportRequest evaluation on followers

Now that we've fixed cockroachdb#67016, and follower reads correctly synchronize
with concurrent splits, it's safe for us to serve ExportRequests from
followers. This patch permits that.

Closes cockroachdb#88804

Release note: None

pkg/ccl/kvccl/kvfollowerreadsccl/followerreads_test.go

@@ -108,10 +108,12 @@ func TestCanSendToFollower(t *testing.T) {
 		txn.GlobalUncertaintyLimit = ts
 		return txn
 	}
-	batch := func(txn *roachpb.Transaction, req roachpb.Request) *roachpb.BatchRequest {
+	batch := func(txn *roachpb.Transaction, reqs ...roachpb.Request) *roachpb.BatchRequest {
 		ba := &roachpb.BatchRequest{}
 		ba.Txn = txn
-		ba.Add(req)
+		for _, req := range reqs {
+			ba.Add(req)
+		}
 		return ba
 	}
 	withBatchTimestamp := func(ba *roachpb.BatchRequest, ts hlc.Timestamp) *roachpb.BatchRequest {
@@ -146,7 +148,17 @@ func TestCanSendToFollower(t *testing.T) {
 		{
 			name: "stale non-txn export batch",
 			ba:   withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}), stale),
-			exp:  false,
+			exp:  true,
+		},
+		{
+			name: "stale non-txn multiple exports batch",
+			ba:   withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}, &roachpb.ExportRequest{}), stale),
+			exp:  true,
+		},
+		{
+			name: "stale non-txn mixed export-scan batch",
+			ba:   withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}, &roachpb.ScanRequest{}), stale),
+			exp:  true,
 		},
 		{
 			name: "current-time non-txn batch",
@@ -158,6 +170,11 @@ func TestCanSendToFollower(t *testing.T) {
 			ba:   withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}), current),
 			exp:  false,
 		},
+		{
+			name: "current-time non-txn multiple exports batch",
+			ba:   withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}, &roachpb.ExportRequest{}), current),
+			exp:  false,
+		},
 		{
 			name: "future non-txn batch",
 			ba:   withBatchTimestamp(batch(nil, &roachpb.GetRequest{}), future),
@@ -275,7 +292,7 @@ func TestCanSendToFollower(t *testing.T) {
 			name:     "stale non-txn export batch, global reads policy",
 			ba:       withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}), stale),
 			ctPolicy: roachpb.LEAD_FOR_GLOBAL_READS,
-			exp:      false,
+			exp:      true,
 		},
 		{
 			name:     "current-time non-txn batch, global reads policy",
@@ -287,7 +304,7 @@ func TestCanSendToFollower(t *testing.T) {
 			name:     "current-time non-txn export batch, global reads policy",
 			ba:       withBatchTimestamp(batch(nil, &roachpb.ExportRequest{}), current),
 			ctPolicy: roachpb.LEAD_FOR_GLOBAL_READS,
-			exp:      false,
+			exp:      true,
 		},
 		{
 			name:     "future non-txn batch, global reads policy",

pkg/kv/kvserver/replica_follower_read.go

@@ -43,9 +43,9 @@ func BatchCanBeEvaluatedOnFollower(ba *roachpb.BatchRequest) bool {
 	//    and this then allows the follower to evaluate the batch as a follower
 	//    read, then the batch might miss past writes served at higher timestamps
 	//    on the leaseholder.
-	// 2. each request in the batch needs to be "transactional", because those are
-	//    the only ones that have clearly defined semantics when served under the
-	//    closed timestamp.
+	// 2. Each request in the batch needs to have clearly defined semantics when
+	//    served under the closed timestamp. This includes all "transactional"
+	//    requests and ExportRequests (which are non-transactional).
 	// 3. the batch needs to be read-only, because a follower replica cannot
 	//    propose writes to Raft.
 	// 4. the batch needs to be non-locking, because unreplicated locks are only
@@ -54,7 +54,7 @@ func BatchCanBeEvaluatedOnFollower(ba *roachpb.BatchRequest) bool {
 	if tsFromServerClock {
 		return false
 	}
-	return ba.IsAllTransactional() && ba.IsReadOnly() && !ba.IsLocking()
+	return ba.IsAllTransactionalOrExportRequests() && ba.IsReadOnly() && !ba.IsLocking()
 }
 
 // canServeFollowerReadRLocked tests, when a range lease could not be acquired,

pkg/roachpb/batch.go

@@ -192,6 +192,20 @@ func (ba *BatchRequest) IsAllTransactional() bool {
 	return ba.hasFlagForAll(isTxn)
 }
 
+// IsAllTransactionalOrExportRequests returns true iff every request in the
+// BatchRequest can either be part of a transaction or is an ExportRequest.
+func (ba *BatchRequest) IsAllTransactionalOrExportRequests() bool {
+	if len(ba.Requests) == 0 {
+		return false
+	}
+	for _, union := range ba.Requests {
+		if (union.GetInner().flags()&isTxn) == 0 && union.GetInner().Method() != Export {
+			return false
+		}
+	}
+	return true
+}
+
 // IsLocking returns true iff the BatchRequest intends to acquire locks.
 func (ba *BatchRequest) IsLocking() bool {
 	return ba.hasFlag(isLocking)