batcheval: pass in split trigger information as a struct

Instead of creating a stateloader and reading information about the
LHS's lease, gcThrehsold etc. in the splitTriggerHelper, this patch
lifts that two levels up. In reference to @pav-kv's comment on #149714.

Epic: none

Release note: None

Author: arulajmani

pkg/kv/kvserver/batcheval/cmd_end_transaction.go

@@ -882,45 +882,74 @@ func RunCommitTrigger(
 			"commit wait. Was its timestamp bumped after acquiring latches?", txn, ct.Kind())
 	}
 
+	// Used by both splits and merges.
+	maybeWrapReplicaCorruptionError := func(ctx context.Context, err error) error {
+		if err == nil {
+			log.KvExec.Fatalf(ctx, "unexpected nil error")
+		}
+		if info := pebble.ExtractDataCorruptionInfo(err); info != nil {
+			// Data corruption errors due to external SSTable references getting
+			// deleted should not be wrapped in replica corruption errors. This
+			// ensures that we simply fail the split or merge and propagate the error,
+			// but don't crash the process. In such cases, an excise command should be
+			// used to get out of this data corruption situation.
+			return err
+		}
+		// Otherwise, fail the split or merge with a critical error that crashes the
+		// process. Reporting a replica corruption error ensures this. See
+		// setCorruptRaftMuLocked.
+		return kvpb.MaybeWrapReplicaCorruptionError(ctx, err)
+	}
+
 	// Stage the commit trigger's side-effects so that they will go into effect on
 	// each Replica when the corresponding Raft log entry is applied. Only one
 	// commit trigger can be set.
 	if ct.GetSplitTrigger() != nil {
+		sl := MakeStateLoader(rec)
+		lhsLease, err := sl.LoadLease(ctx, batch)
+		if err != nil {
+			return result.Result{}, maybeWrapReplicaCorruptionError(
+				ctx, errors.Wrap(err, "unable to load lease"),
+			)
+		}
+		gcThreshold, err := sl.LoadGCThreshold(ctx, batch)
+		if err != nil {
+			return result.Result{}, maybeWrapReplicaCorruptionError(
+				ctx, errors.Wrap(err, "unable to load GCThreshold"),
+			)
+		}
+		gcHint, err := sl.LoadGCHint(ctx, batch)
+		if err != nil {
+			return result.Result{}, maybeWrapReplicaCorruptionError(
+				ctx, errors.Wrap(err, "unable to load GCHint"),
+			)
+		}
+		replicaVersion, err := sl.LoadVersion(ctx, batch)
+		if err != nil {
+			return result.Result{}, maybeWrapReplicaCorruptionError(
+				ctx, errors.Wrap(err, "unable to load replica version"),
+			)
+		}
+		in := splitTriggerHelperInput{
+			leftLease:      lhsLease,
+			gcThreshold:    gcThreshold,
+			gcHint:         gcHint,
+			replicaVersion: replicaVersion,
+		}
+
 		newMS, res, err := splitTrigger(
-			ctx, rec, batch, *ms, ct.SplitTrigger, txn.WriteTimestamp,
+			ctx, rec, batch, *ms, ct.SplitTrigger, in, txn.WriteTimestamp,
 		)
 		if err != nil {
-			if info := pebble.ExtractDataCorruptionInfo(err); info != nil {
-				// We want to handle the data corruption error here because it's possible
-				// that a file that an external SSTable references got deleted. We want to
-				// fail the split and propagate the error, but we don't want to crash the
-				// process. An excise command could be used to get out of this data
-				// corruption.
-				return result.Result{}, err
-			} else {
-				// Otherwise, failing the split is a critical error. We should crash
-				// the process and report a replica corruption.
-				return result.Result{}, kvpb.MaybeWrapReplicaCorruptionError(ctx, err)
-			}
+			return result.Result{}, maybeWrapReplicaCorruptionError(ctx, err)
 		}
 		*ms = newMS
 		return res, nil
 	}
 	if mt := ct.GetMergeTrigger(); mt != nil {
 		res, err := mergeTrigger(ctx, rec, batch, ms, mt, txn.WriteTimestamp)
 		if err != nil {
-			if info := pebble.ExtractDataCorruptionInfo(err); info != nil {
-				// We want to handle the data corruption error here because it's
-				// possible that a file that an external SSTable references got deleted.
-				// We want to fail the merge and propagate the error, but we don't want
-				// to crash the process. An excise command could be used to get out of
-				// this data corruption.
-				return result.Result{}, err
-			} else {
-				// Otherwise, failing the merge is a critical error. We should crash
-				// the process and report a replica corruption.
-				return result.Result{}, kvpb.MaybeWrapReplicaCorruptionError(ctx, err)
-			}
+			return result.Result{}, maybeWrapReplicaCorruptionError(ctx, err)
 		}
 		return res, nil
 	}
@@ -1135,6 +1164,7 @@ func splitTrigger(
 	batch storage.Batch,
 	bothDeltaMS enginepb.MVCCStats,
 	split *roachpb.SplitTrigger,
+	in splitTriggerHelperInput,
 	ts hlc.Timestamp,
 ) (enginepb.MVCCStats, result.Result, error) {
 	desc := rec.Desc()
@@ -1210,7 +1240,7 @@ func splitTrigger(
 		MaxBytesDiff:             MaxMVCCStatBytesDiff.Get(&rec.ClusterSettings().SV),
 		UseEstimatesBecauseExternalBytesArePresent: split.UseEstimatesBecauseExternalBytesArePresent,
 	}
-	return splitTriggerHelper(ctx, rec, batch, h, split, ts)
+	return splitTriggerHelper(ctx, rec, batch, in, h, split, ts)
 }
 
 // splitScansRightForStatsFirst controls whether the left hand side or the right
@@ -1246,13 +1276,23 @@ func makeScanStatsFn(
 	}
 }
 
+// splitTriggerHelperInput contains metadata needed by the RHS when running the
+// splitTriggerHelper.
+type splitTriggerHelperInput struct {
+	leftLease      roachpb.Lease
+	gcThreshold    *hlc.Timestamp
+	gcHint         *roachpb.GCHint
+	replicaVersion roachpb.Version
+}
+
 // splitTriggerHelper continues the work begun by splitTrigger, but has a
 // reduced scope that has all stats-related concerns bundled into a
 // splitStatsHelper.
 func splitTriggerHelper(
 	ctx context.Context,
 	rec EvalContext,
 	batch storage.Batch,
+	in splitTriggerHelperInput,
 	statsInput splitStatsHelperInput,
 	split *roachpb.SplitTrigger,
 	ts hlc.Timestamp,
@@ -1394,27 +1434,22 @@ func splitTriggerHelper(
 		// - node two becomes the lease holder for [c,e). Its timestamp cache does
 		//   not know about the read at 'd' which happened at the beginning.
 		// - node two can illegally propose a write to 'd' at a lower timestamp.
-		sl := MakeStateLoader(rec)
-		leftLease, err := sl.LoadLease(ctx, batch)
-		if err != nil {
-			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load lease")
-		}
-		if leftLease.Empty() {
+		if in.leftLease.Empty() {
 			log.KvExec.Fatalf(ctx, "LHS of split has no lease")
 		}
 
 		// Copy the lease from the left-hand side of the split over to the
 		// right-hand side so that it can immediately start serving requests.
 		// When doing so, we need to make a few modifications.
-		rightLease := leftLease
+		rightLease := in.leftLease
 		// Rebind the lease to the existing leaseholder store's replica from the
 		// right-hand side's descriptor.
 		var ok bool
-		rightLease.Replica, ok = split.RightDesc.GetReplicaDescriptor(leftLease.Replica.StoreID)
+		rightLease.Replica, ok = split.RightDesc.GetReplicaDescriptor(in.leftLease.Replica.StoreID)
 		if !ok {
 			return enginepb.MVCCStats{}, result.Result{}, errors.Errorf(
 				"pre-split lease holder %+v not found in post-split descriptor %+v",
-				leftLease.Replica, split.RightDesc,
+				in.leftLease.Replica, split.RightDesc,
 			)
 		}
 		// Convert leader leases into expiration-based leases. A leader lease is
@@ -1429,18 +1464,9 @@ func splitTriggerHelper(
 			rightLease.Term = 0
 			rightLease.MinExpiration = hlc.Timestamp{}
 		}
-
-		gcThreshold, err := sl.LoadGCThreshold(ctx, batch)
-		if err != nil {
-			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load GCThreshold")
-		}
-		if gcThreshold.IsEmpty() {
+		if in.gcThreshold.IsEmpty() {
 			log.VEventf(ctx, 1, "LHS's GCThreshold of split is not set")
 		}
-		gcHint, err := sl.LoadGCHint(ctx, batch)
-		if err != nil {
-			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load GCHint")
-		}
 
 		// Writing the initial state is subtle since this also seeds the Raft
 		// group. It becomes more subtle due to proposer-evaluated Raft.
@@ -1471,13 +1497,9 @@ func splitTriggerHelper(
 		// HardState via a call to synthesizeRaftState. Here, we only call
 		// writeInitialReplicaState which essentially writes a ReplicaState
 		// only.
-		replicaVersion, err := sl.LoadVersion(ctx, batch)
-		if err != nil {
-			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to load replica version")
-		}
 		if *h.AbsPostSplitRight(), err = stateloader.WriteInitialReplicaState(
 			ctx, batch, *h.AbsPostSplitRight(), split.RightDesc, rightLease,
-			*gcThreshold, *gcHint, replicaVersion,
+			*in.gcThreshold, *in.gcHint, in.replicaVersion,
 		); err != nil {
 			return enginepb.MVCCStats{}, result.Result{}, errors.Wrap(err, "unable to write initial Replica state")
 		}

pkg/kv/kvserver/batcheval/cmd_end_transaction_test.go

@@ -2203,9 +2203,16 @@ func TestSplitTriggerWritesInitialReplicaState(t *testing.T) {
 	err = sl.SetVersion(ctx, batch, nil, &version)
 	require.NoError(t, err)
 
+	in := splitTriggerHelperInput{
+		leftLease:      lease,
+		gcThreshold:    &gcThreshold,
+		gcHint:         &gcHint,
+		replicaVersion: version,
+	}
+
 	// Run the split trigger, which is normally run as a subset of EndTxn request
 	// evaluation.
-	_, _, err = splitTrigger(ctx, rec, batch, enginepb.MVCCStats{}, split, hlc.Timestamp{})
+	_, _, err = splitTrigger(ctx, rec, batch, enginepb.MVCCStats{}, split, in, hlc.Timestamp{})
 	require.NoError(t, err)
 
 	// Verify that range state was migrated to the right-hand side properly.