Merge #144615

144615: kvserver: use WriteBatch instead of NewUnindexedBatch r=tbg a=pav-kv

This PR replaces a few uses of `Engine.NewUnindexedBatch()` with `Engine.NewWriteBatch()`, in places where reading is not needed.

Epic: CRDB-46488

Co-authored-by: Pavel Kalinnikov <[email protected]>

Author: craig[bot]

pkg/kv/kvserver/logstore/logstore.go

@@ -476,7 +476,7 @@ func Compact(
 	prev kvserverpb.RaftTruncatedState,
 	next kvserverpb.RaftTruncatedState,
 	loader StateLoader,
-	readWriter storage.ReadWriter,
+	writer storage.Writer,
 ) error {
 	if next.Index <= prev.Index {
 		// TODO(pav-kv): return an assertion failure error.
@@ -490,7 +490,7 @@ func Compact(
 	if numTruncatedEntries >= raftLogTruncationClearRangeThreshold {
 		start := prefixBuf.RaftLogKey(prev.Index + 1).Clone()
 		end := prefixBuf.RaftLogKey(next.Index + 1).Clone() // end is exclusive
-		if err := readWriter.ClearRawRange(start, end, true, false); err != nil {
+		if err := writer.ClearRawRange(start, end, true, false); err != nil {
 			return errors.Wrapf(err,
 				"unable to clear truncated Raft entries for %+v after index %d",
 				next, prev.Index)
@@ -500,7 +500,7 @@ func Compact(
 		// allocating when constructing Raft log keys (16 bytes).
 		prefix := prefixBuf.RaftLogPrefix()
 		for idx := prev.Index + 1; idx <= next.Index; idx++ {
-			if err := readWriter.ClearUnversioned(
+			if err := writer.ClearUnversioned(
 				keys.RaftLogKeyFromPrefix(prefix, idx),
 				storage.ClearOptions{},
 			); err != nil {
@@ -517,9 +517,8 @@ func Compact(
 	}
 	value.InitChecksum(key)
 
-	if _, err := storage.MVCCPut(
-		ctx, readWriter, key, hlc.Timestamp{}, value,
-		storage.MVCCWriteOptions{Category: fs.ReplicationReadCategory},
+	if _, err := storage.MVCCBlindPut(
+		ctx, writer, key, hlc.Timestamp{}, value, storage.MVCCWriteOptions{},
 	); err != nil {
 		return errors.Wrap(err, "unable to write RaftTruncatedState")
 	}

pkg/kv/kvserver/raft_log_truncator.go

@@ -545,7 +545,7 @@ func (t *raftLogTruncator) tryEnactTruncations(
 	}
 	// Do the truncation of persistent raft entries, specified by enactIndex
 	// (this subsumes all the preceding queued truncations).
-	batch := t.store.getEngine().NewUnindexedBatch()
+	batch := t.store.getEngine().NewWriteBatch()
 	defer batch.Close()
 	if err := handleTruncatedStateBelowRaftPreApply(ctx, truncState,
 		pendingTruncs.mu.truncs[enactIndex].RaftTruncatedState,

pkg/kv/kvserver/raftstorebench/engine.go

@@ -89,7 +89,7 @@ func growMemtable(t T, eng storage.Engine, target int64) {
 	// memtable size, and doubles with each flush. We'll have reached any
 	// reasonable memtable size in well under 20 iterations.
 	for memtableSize := int64(256 * 1024); memtableSize < target; memtableSize *= 2 {
-		b := eng.NewUnindexedBatch()
+		b := eng.NewWriteBatch()
 		require.NoError(t, b.PutUnversioned(roachpb.Key("zoo"), nil))
 		require.NoError(t, b.Commit(false)) // this works even when WAL is off
 
@@ -98,7 +98,7 @@ func growMemtable(t T, eng storage.Engine, target int64) {
 	}
 
 	// Delete the key.
-	b := eng.NewUnindexedBatch()
+	b := eng.NewWriteBatch()
 	require.NoError(t, b.ClearUnversioned(roachpb.Key("zoo"), storage.ClearOptions{}))
 	require.NoError(t, b.Commit(false))
 

pkg/kv/kvserver/raftstorebench/worker.go

@@ -83,7 +83,7 @@ func (w *worker) doTruncForReplica(t T, r *replicaWriteState, truncIndex uint64)
 		logf(t, "r%d: truncIndex %d, nextIndex: %d", r.rangeID, truncIndex,
 			r.nextRaftLogIndex)
 	}
-	raftBatch := w.o.raftEng.NewUnindexedBatch()
+	raftBatch := w.o.raftEng.NewWriteBatch()
 	for i := r.truncatedLogIndex + 1; i <= truncIndex; i++ {
 		key := r.rangeIDPrefixBuf.RaftLogKey(kvpb.RaftIndex(i))
 		if w.o.cfg.SingleDel {

pkg/kv/kvserver/replica_application_state_machine.go

@@ -143,6 +143,9 @@ func (sm *replicaStateMachine) NewBatch() apply.Batch {
 	// make it safer.
 	b.r = r
 	b.applyStats = &sm.applyStats
+	// TODO(#144627): most commands do not need to read. Use NewWriteBatch because
+	// it is more efficient. If there are exceptions, sparingly use NewReader or
+	// NewBatch (if it needs to read its own writes, which is unlikely).
 	b.batch = r.store.TODOEngine().NewBatch()
 	r.mu.RLock()
 	b.state = r.shMu.state

pkg/kv/kvserver/replica_raft.go

@@ -2983,9 +2983,9 @@ func handleTruncatedStateBelowRaftPreApply(
 	prev kvserverpb.RaftTruncatedState,
 	next kvserverpb.RaftTruncatedState,
 	loader logstore.StateLoader,
-	readWriter storage.ReadWriter,
+	writer storage.Writer,
 ) error {
-	return logstore.Compact(ctx, prev, next, loader, readWriter)
+	return logstore.Compact(ctx, prev, next, loader, writer)
 }
 
 // shouldCampaignAfterConfChange returns true if the current replica should