kv: mark AddSSTable requests "unsplittable"

Usually we split span requests automatically in DistSender however
AddSSTable requests are special: their large SST file is an opaque
payload that contains keys that could span the post-split requests.

We currently assert during ingestion that the span of an SST's contents
matches the request's span, as the file is ingested whole, without
modification, meaning that currently a split-and-truncated request
fails.

However we if a request will fail, we can reject it as soon as we'd try
to split it, before sending these large, expensive files around. More
importantly, if communicate back to the caller the actual range bounds,
it can try again, generating re-chunked SSTs with bounds that should be
OK.

Note: we could try to make the recipient able to correctly ingest only
the subset of the unsplit SST by iterating over it and generating a new
file containing just those keys in the request span. However this is
significant complexity to add to a raft command evaluation, and changes
the performance characteristics of `AddSSTable` that make it appealing
in the first place.

Release note: none.

pkg/kv/dist_sender.go

@@ -743,6 +743,11 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 		return resp.reply, resp.pErr
 	}
 
+	if ba.IsUnsplittable() {
+		mismatch := roachpb.NewRangeKeyMismatchError(rs.Key.AsRawKey(), rs.EndKey.AsRawKey(), ri.Desc())
+		return nil, roachpb.NewError(mismatch)
+	}
+
 	// Make an empty slice of responses which will be populated with responses
 	// as they come in via Combine().
 	br = &roachpb.BatchResponse{
@@ -850,6 +855,11 @@ func (ds *DistSender) divideAndSendBatchToRanges(
 				responseCh <- response{pErr: errNo1PCTxn}
 				return
 			}
+			if ba.IsUnsplittable() {
+				mismatch := roachpb.NewRangeKeyMismatchError(rs.Key.AsRawKey(), rs.EndKey.AsRawKey(), ri.Desc())
+				responseCh <- response{pErr: roachpb.NewError(mismatch)}
+				return
+			}
 		}
 
 		// Determine next seek key, taking a potentially sparse batch into

pkg/roachpb/api.go

@@ -103,14 +103,15 @@ func (rc ReadConsistencyType) SupportsBatch(ba BatchRequest) error {
 }
 
 const (
-	isAdmin    = 1 << iota // admin cmds don't go through raft, but run on lease holder
-	isRead                 // read-only cmds don't go through raft, but may run on lease holder
-	isWrite                // write cmds go through raft and must be proposed on lease holder
-	isTxn                  // txn commands may be part of a transaction
-	isTxnWrite             // txn write cmds start heartbeat and are marked for intent resolution
-	isRange                // range commands may span multiple keys
-	isReverse              // reverse commands traverse ranges in descending direction
-	isAlone                // requests which must be alone in a batch
+	isAdmin        = 1 << iota // admin cmds don't go through raft, but run on lease holder
+	isRead                     // read-only cmds don't go through raft, but may run on lease holder
+	isWrite                    // write cmds go through raft and must be proposed on lease holder
+	isTxn                      // txn commands may be part of a transaction
+	isTxnWrite                 // txn write cmds start heartbeat and are marked for intent resolution
+	isRange                    // range commands may span multiple keys
+	isReverse                  // reverse commands traverse ranges in descending direction
+	isAlone                    // requests which must be alone in a batch
+	isUnsplittable             // range command that must not be split during sending
 	// Requests for acquiring a lease skip the (proposal-time) check that the
 	// proposing replica has a valid lease.
 	skipLeaseCheck
@@ -1068,7 +1069,7 @@ func (*WriteBatchRequest) flags() int               { return isWrite | isRange }
 func (*ExportRequest) flags() int                   { return isRead | isRange | updatesReadTSCache }
 func (*ImportRequest) flags() int                   { return isAdmin | isAlone }
 func (*AdminScatterRequest) flags() int             { return isAdmin | isAlone | isRange }
-func (*AddSSTableRequest) flags() int               { return isWrite | isAlone | isRange }
+func (*AddSSTableRequest) flags() int               { return isWrite | isAlone | isRange | isUnsplittable }
 
 // RefreshRequest and RefreshRangeRequest both list
 // updates(Read)TSCache, though they actually update the read or write

pkg/roachpb/batch.go

@@ -117,6 +117,11 @@ func (ba *BatchRequest) IsTransactionWrite() bool {
 	return ba.hasFlag(isTxnWrite)
 }
 
+// IsUnsplittable returns true iff the BatchRequest an un-splittable request.
+func (ba *BatchRequest) IsUnsplittable() bool {
+	return ba.hasFlag(isUnsplittable)
+}
+
 // IsSingleRequest returns true iff the BatchRequest contains a single request.
 func (ba *BatchRequest) IsSingleRequest() bool {
 	return len(ba.Requests) == 1