Merge #143334

143334: kvserver: add kv excise command r=iskettaneh a=iskettaneh

kvserver: add kv excise command

This commit adds a KV command called Excise, which ends up calling Pebble's excise command. The command deletes all the data overlapping with the provided key span.

This command will be used by the OnlineRestore job to clean up the state where an external SSTable is pointing to a deleted file.

These are the remaining items:

1. Perform split operations after the file is removed.
2. Add more detailed Rangefeed/Changefeed tests.
3. Add a test to excise part of the deleted span only (after cockroachdb/pebble#4417 is fixed)
4. Understand if merges require to read the SSTable if the range contains estimates.
5. Add a small test to verify that SSTables aren't read during writeBatch commit.
6. Defensive patches: Handle FileNotFound error from code locations that didn't fail in tests, but that could in theory fail under some circumstances.

References: #143135

Release note: None

Co-authored-by: Ibrahim Kettaneh <[email protected]>

Author: craig[bot]

docs/generated/metrics/metrics.html

@@ -641,6 +641,7 @@
 <tr><td>STORAGE</td><td>rpc.method.delete.recv</td><td>Number of Delete requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>rpc.method.deleterange.recv</td><td>Number of DeleteRange requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>rpc.method.endtxn.recv</td><td>Number of EndTxn requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>STORAGE</td><td>rpc.method.excise.recv</td><td>Number of Excise requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>rpc.method.export.recv</td><td>Number of Export requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>rpc.method.gc.recv</td><td>Number of GC requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>STORAGE</td><td>rpc.method.get.recv</td><td>Number of Get requests processed</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
@@ -1105,6 +1106,7 @@
 <tr><td>APPLICATION</td><td>distsender.rpc.err.unsupportedrequesterrtype</td><td>Number of UnsupportedRequestErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.writeintenterrtype</td><td>Number of WriteIntentErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.writetooolderrtype</td><td>Number of WriteTooOldErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>distsender.rpc.excise.sent</td><td>Number of Excise requests processed.<br/><br/>This counts the requests in batches handed to DistSender, not the RPCs<br/>sent to individual Ranges as a result.</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.export.sent</td><td>Number of Export requests processed.<br/><br/>This counts the requests in batches handed to DistSender, not the RPCs<br/>sent to individual Ranges as a result.</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.gc.sent</td><td>Number of GC requests processed.<br/><br/>This counts the requests in batches handed to DistSender, not the RPCs<br/>sent to individual Ranges as a result.</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.get.sent</td><td>Number of Get requests processed.<br/><br/>This counts the requests in batches handed to DistSender, not the RPCs<br/>sent to individual Ranges as a result.</td><td>RPCs</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>

pkg/kv/batch.go

@@ -301,6 +301,7 @@ func (b *Batch) fillResults(ctx context.Context) {
 			case *kvpb.QueryResolvedTimestampRequest:
 			case *kvpb.BarrierRequest:
 			case *kvpb.LinkExternalSSTableRequest:
+			case *kvpb.ExciseRequest:
 			default:
 				if result.Err == nil {
 					result.Err = errors.Errorf("unsupported reply: %T for %T",

pkg/kv/kvpb/api.go

@@ -720,6 +720,19 @@ func (r *IsSpanEmptyResponse) combine(_ context.Context, c combinable, _ *BatchR
 
 var _ combinable = &IsSpanEmptyResponse{}
 
+// combine implements the combinable interface.
+func (r *ExciseResponse) combine(_ context.Context, c combinable, _ *BatchRequest) error {
+	otherDR := c.(*ExciseResponse)
+	if r != nil {
+		if err := r.ResponseHeader.combine(otherDR.Header()); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+var _ combinable = &ExciseResponse{}
+
 // Header implements the Request interface.
 func (rh RequestHeader) Header() RequestHeader {
 	return rh
@@ -970,6 +983,9 @@ func (*AddSSTableRequest) Method() Method { return AddSSTable }
 // Method implements the Request interface.
 func (*LinkExternalSSTableRequest) Method() Method { return LinkExternalSSTable }
 
+// Method implements the Request interface.
+func (*ExciseRequest) Method() Method { return Excise }
+
 // Method implements the Request interface.
 func (*MigrateRequest) Method() Method { return Migrate }
 
@@ -1236,6 +1252,12 @@ func (r *LinkExternalSSTableRequest) ShallowCopy() Request {
 	return &shallowCopy
 }
 
+// ShallowCopy implements the Request interface.
+func (r *ExciseRequest) ShallowCopy() Request {
+	shallowCopy := *r
+	return &shallowCopy
+}
+
 // ShallowCopy implements the Request interface.
 func (r *MigrateRequest) ShallowCopy() Request {
 	shallowCopy := *r
@@ -1528,6 +1550,12 @@ func (r *LinkExternalSSTableResponse) ShallowCopy() Response {
 	return &shallowCopy
 }
 
+// ShallowCopy implements the Response interface.
+func (r *ExciseResponse) ShallowCopy() Response {
+	shallowCopy := *r
+	return &shallowCopy
+}
+
 // ShallowCopy implements the Response interface.
 func (r *MigrateResponse) ShallowCopy() Response {
 	shallowCopy := *r
@@ -2100,6 +2128,11 @@ func (r *LinkExternalSSTableRequest) flags() flag {
 	}
 	return flags
 }
+
+func (r *ExciseRequest) flags() flag {
+	return isWrite | isRange | isAlone | bypassesReplicaCircuitBreaker
+}
+
 func (*MigrateRequest) flags() flag { return isWrite | isRange | isAlone }
 
 // RefreshRequest and RefreshRangeRequest both determine which timestamp cache

pkg/kv/kvpb/api.proto

@@ -698,6 +698,25 @@ message ScanResponse {
   ColBatches col_batches = 5 [(gogoproto.nullable) = false];
 }
 
+// ExciseRequest is the argument to the Excise() method, which ends up calling
+// the storage Excise() method. It is used to atomically remove all data in the
+// specified key span.
+// It's used as a recovery tool in the event of data loss. It's a non-MVCC
+// command that will destroy MVCC history, and can result in transactional
+// anomalies like lost portions of transactions, uncommiting committed
+// transactions, and possible other anomalies. It allows for inconsistent
+// follower reads.
+// It can not be applied to non-user spans.
+message ExciseRequest {
+  reserved 2, 3;
+  RequestHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+}
+
+// ExciseResponse is the return value from the Excise() method.
+message ExciseResponse {
+  ResponseHeader header = 1 [(gogoproto.nullable) = false, (gogoproto.embed) = true];
+}
+
 // A ReverseScanRequest is the argument to the ReverseScan() method. It specifies the
 // start and end keys for a descending scan of [start,end) and the maximum
 // number of results (unbounded if zero).
@@ -2536,6 +2555,7 @@ message RequestUnion {
     ProbeRequest probe = 54;
     IsSpanEmptyRequest is_span_empty = 56;
     LinkExternalSSTableRequest link_external_sstable = 57;
+    ExciseRequest excise = 58;
   }
   reserved 8, 15, 23, 25, 27, 31, 34, 52;
 }
@@ -2592,6 +2612,7 @@ message ResponseUnion {
     ProbeResponse probe = 54;
     IsSpanEmptyResponse is_span_empty = 56;
     LinkExternalSSTableResponse link_external_sstable = 57;
+    ExciseResponse excise = 58;
   }
   reserved 8, 15, 23, 25, 27, 28, 31, 34, 52;
 }

pkg/kv/kvpb/batch_generated.go

@@ -171,6 +171,9 @@ const (
 	// IsSpanEmpty is a non-transaction read request used to determine whether
 	// a span contains any keys whatsoever (garbage or otherwise).
 	IsSpanEmpty
+	// Excise is a non-MVCC command that destroys all data in a user MVCC key
+	// span. See ExciseRequest for details.
+	Excise
 	// MaxMethod is the maximum method.
 	MaxMethod Method = iota - 1
 	// NumMethods represents the total number of API methods.

pkg/kv/kvpb/method.go

@@ -306,6 +306,7 @@ go_test(
         "closed_timestamp_test.go",
         "consistency_queue_test.go",
         "debug_print_test.go",
+        "deleted_external_sstable_test.go",
         "errors_test.go",
         "flow_control_integration_test.go",
         "flow_control_raft_transport_test.go",

pkg/kv/kvpb/method_string.go

@@ -198,5 +198,15 @@ func (b *appBatch) runPostAddTriggers(
 			cmd.Index(),
 			*res.LinkExternalSSTable)
 	}
+
+	if res.Excise != nil {
+		if err := env.eng.Excise(ctx, res.Excise.Span); err != nil {
+			return errors.Wrapf(err, "error while excising span: %v", res.Excise.Span)
+		}
+		if err := env.eng.Excise(ctx, res.Excise.LockTableSpan); err != nil {
+			return errors.Wrapf(err, "error while excising span: %v", res.Excise.LockTableSpan)
+		}
+	}
+
 	return nil
 }

pkg/kv/kvserver/BUILD.bazel

@@ -11,6 +11,7 @@ go_library(
         "cmd_delete.go",
         "cmd_delete_range.go",
         "cmd_end_transaction.go",
+        "cmd_excise.go",
         "cmd_export.go",
         "cmd_gc.go",
         "cmd_get.go",