util/bufalloc: clean up ByteAllocator API

yuzefovich · yuzefovich · commit daad9f29ba8d · 2025-06-27T16:31:20.000-07:00
`bufalloc.ByteAllocator`, as evidenced by a fix in the previous commit,
has confusing API for allocating "extra capacity". This commit cleans it up
by removing the option altogether. The only two spots that utilized the
extra capacity previously now explicitly request combined allocated
upfront and manage the copying behavior.

Release note: None
diff --git a/pkg/BUILD.bazel b/pkg/BUILD.bazel
@@ -711,6 +711,7 @@ ALL_TESTS = [
     "//pkg/util/binfetcher:binfetcher_test",
     "//pkg/util/bitarray:bitarray_test",
     "//pkg/util/bitmap:bitmap_test",
+    "//pkg/util/bufalloc:bufalloc_test",
     "//pkg/util/buildutil:buildutil_test",
     "//pkg/util/cache:cache_test",
     "//pkg/util/caller:caller_test",
@@ -2528,6 +2529,7 @@ GO_TARGETS = [
     "//pkg/util/bitmap:bitmap",
     "//pkg/util/bitmap:bitmap_test",
     "//pkg/util/bufalloc:bufalloc",
+    "//pkg/util/bufalloc:bufalloc_test",
     "//pkg/util/buildutil:buildutil",
     "//pkg/util/buildutil:buildutil_test",
     "//pkg/util/bulk:bulk",
diff --git a/pkg/ccl/changefeedccl/event_processing.go b/pkg/ccl/changefeedccl/event_processing.go
@@ -453,14 +453,14 @@ func (c *kvEventToRowConsumer) encodeAndEmit(
 	if err != nil {
 		return err
 	}
-	c.scratch, keyCopy = c.scratch.Copy(encodedKey, 0 /* extraCap */)
+	c.scratch, keyCopy = c.scratch.Copy(encodedKey)
 	// TODO(yevgeniy): Some refactoring is needed in the encoder: namely, prevRow
 	// might not be available at all when working with changefeed expressions.
 	encodedValue, err := c.encoder.EncodeValue(ctx, evCtx, updatedRow, prevRow)
 	if err != nil {
 		return err
 	}
-	c.scratch, valueCopy = c.scratch.Copy(encodedValue, 0 /* extraCap */)
+	c.scratch, valueCopy = c.scratch.Copy(encodedValue)
 
 	// Since we're done processing/converting this event, and will not use much more
 	// than len(key)+len(bytes) worth of resources, adjust allocation to match.
diff --git a/pkg/ccl/changefeedccl/sink.go b/pkg/ccl/changefeedccl/sink.go
@@ -494,7 +494,7 @@ func (s *bufferSink) EmitResolvedTimestamp(
 	if err != nil {
 		return err
 	}
-	s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)
+	s.scratch, payload = s.scratch.Copy(payload)
 	s.buf.Push(rowenc.EncDatumRow{
 		{Datum: tree.DNull}, // resolved span
 		{Datum: tree.DNull}, // topic
diff --git a/pkg/ccl/changefeedccl/sink_kafka.go b/pkg/ccl/changefeedccl/sink_kafka.go
@@ -424,7 +424,7 @@ func (s *kafkaSink) EmitResolvedTimestamp(
 		if err != nil {
 			return err
 		}
-		s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)
+		s.scratch, payload = s.scratch.Copy(payload)
 
 		// sarama caches this, which is why we have to periodically refresh the
 		// metadata above. Staleness here does not impact correctness. Some new
diff --git a/pkg/ccl/changefeedccl/sink_sql.go b/pkg/ccl/changefeedccl/sink_sql.go
@@ -168,7 +168,7 @@ func (s *sqlSink) EmitResolvedTimestamp(
 		if err != nil {
 			return err
 		}
-		s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)
+		s.scratch, payload = s.scratch.Copy(payload)
 		for partition := int32(0); partition < sqlSinkNumPartitions; partition++ {
 			if err := s.emit(ctx, topic, partition, noKey, noValue, payload); err != nil {
 				return err
diff --git a/pkg/crosscluster/streamclient/randclient/random_stream_client.go b/pkg/crosscluster/streamclient/randclient/random_stream_client.go
@@ -738,7 +738,7 @@ func duplicateEvent(event crosscluster.Event) crosscluster.Event {
 		for i := range kvs {
 			res[i].Key = kvs[i].KeyValue.Key.Clone()
 			res[i].Value.Timestamp = kvs[i].KeyValue.Value.Timestamp
-			a, res[i].Value.RawBytes = a.Copy(kvs[i].KeyValue.Value.RawBytes, 0)
+			a, res[i].Value.RawBytes = a.Copy(kvs[i].KeyValue.Value.RawBytes)
 		}
 		dup = crosscluster.MakeKVEventFromKVs(res)
 	case crosscluster.SSTableEvent:
diff --git a/pkg/kv/kvnemesis/engine.go b/pkg/kv/kvnemesis/engine.go
@@ -93,7 +93,7 @@ func (e *Engine) Get(key roachpb.Key, ts hlc.Timestamp) roachpb.Value {
 	if err != nil {
 		panic(err)
 	}
-	e.b, valCopy = e.b.Copy(v, 0 /* extraCap */)
+	e.b, valCopy = e.b.Copy(v)
 	mvccVal, err := storage.DecodeMVCCValue(valCopy)
 	if err != nil {
 		panic(err)
@@ -137,12 +137,12 @@ func (e *Engine) Iterate(
 	for iter.First(); iter.Valid(); iter.Next() {
 		hasPoint, _ := iter.HasPointAndRange()
 		var keyCopy, valCopy []byte
-		e.b, keyCopy = e.b.Copy(iter.Key(), 0 /* extraCap */)
+		e.b, keyCopy = e.b.Copy(iter.Key())
 		v, err := iter.ValueAndErr()
 		if err != nil {
 			fn(nil, nil, hlc.Timestamp{}, nil, err)
 		}
-		e.b, valCopy = e.b.Copy(v, 0 /* extraCap */)
+		e.b, valCopy = e.b.Copy(v)
 		if hasPoint {
 			key, err := storage.DecodeMVCCKey(keyCopy)
 			if err != nil {
@@ -153,8 +153,8 @@ func (e *Engine) Iterate(
 		}
 		if iter.RangeKeyChanged() {
 			keyCopy, endKeyCopy := iter.RangeBounds()
-			e.b, keyCopy = e.b.Copy(keyCopy, 0 /* extraCap */)
-			e.b, endKeyCopy = e.b.Copy(endKeyCopy, 0 /* extraCap */)
+			e.b, keyCopy = e.b.Copy(keyCopy)
+			e.b, endKeyCopy = e.b.Copy(endKeyCopy)
 			for _, rk := range iter.RangeKeys() {
 				ts, err := mvccencoding.DecodeMVCCTimestampSuffix(rk.Suffix)
 				if err != nil {
@@ -170,7 +170,7 @@ func (e *Engine) Iterate(
 					fn(nil, nil, hlc.Timestamp{}, nil, errors.Errorf("invalid key %q", endKeyCopy))
 				}
 
-				e.b, rk.Value = e.b.Copy(rk.Value, 0)
+				e.b, rk.Value = e.b.Copy(rk.Value)
 
 				fn(engineKey.Key, engineEndKey.Key, ts, rk.Value, nil)
 			}
diff --git a/pkg/kv/kvserver/gc/gc.go b/pkg/kv/kvserver/gc/gc.go
@@ -809,7 +809,7 @@ func (b *gcKeyBatcher) foundGarbage(
 		// Whenever new key is started or new batch is started with the same key in
 		// it, record key value using batches' allocator.
 		if b.prevWasNewest || len(b.pointsBatches[i].batchGCKeys) == 0 {
-			b.pointsBatches[i].alloc, key = b.pointsBatches[i].alloc.Copy(cur.key.Key, 0)
+			b.pointsBatches[i].alloc, key = b.pointsBatches[i].alloc.Copy(cur.key.Key)
 			b.pointsBatches[i].batchGCKeys = append(b.pointsBatches[i].batchGCKeys,
 				kvpb.GCRequest_GCKey{Key: key, Timestamp: cur.key.Timestamp})
 			keyMemUsed := len(key) + hlcTimestampSize
@@ -1065,7 +1065,7 @@ func (b *intentBatcher) addAndMaybeFlushIntents(
 	// We need to register passed intent regardless of flushing operation result
 	// so that batcher is left in consistent state and don't miss any keys if
 	// caller resumes batching.
-	b.alloc, key = b.alloc.Copy(key, 0)
+	b.alloc, key = b.alloc.Copy(key)
 	b.pendingLocks = append(b.pendingLocks, roachpb.MakeLock(meta.Txn, key, str))
 	b.collectedIntentBytes += int64(len(key))
 	b.pendingTxns[txnID] = true
diff --git a/pkg/kv/kvserver/gc/gc_iterator.go b/pkg/kv/kvserver/gc/gc_iterator.go
@@ -237,15 +237,12 @@ func (b *gcIteratorRingBuf) pushBack(
 	}
 	i := (b.head + b.len) % gcIteratorRingBufSize
 	b.allocs[i] = b.allocs[i].Truncate()
-	b.allocs[i], k.Key = b.allocs[i].Copy(k.Key, len(metaValue))
-	if len(metaValue) > 0 {
-		// k.Key now contains the capacity for copy of metaValue - separate it
-		// out.
-		buf := k.Key[len(k.Key):cap(k.Key)]
-		k.Key = k.Key[:len(k.Key):len(k.Key)]
-		copy(buf, metaValue)
-		metaValue = buf
-	}
+	var buf []byte
+	b.allocs[i], buf = b.allocs[i].Alloc(len(k.Key) + len(metaValue))
+	copy(buf, k.Key)
+	k.Key, buf = buf[:len(k.Key):len(k.Key)], buf[len(k.Key):]
+	copy(buf, metaValue)
+	metaValue = buf
 	b.buf[i] = mvccKeyValue{
 		key:                  k,
 		mvccValueLen:         mvccValueLen,
diff --git a/pkg/kv/kvserver/rangefeed/catchup_scan.go b/pkg/kv/kvserver/rangefeed/catchup_scan.go
@@ -198,8 +198,8 @@ func (i *CatchUpIterator) CatchUpScan(
 				rangeKeys := i.RangeKeys()
 				for j := rangeKeys.Len() - 1; j >= 0; j-- {
 					var span roachpb.Span
-					a, span.Key = a.Copy(rangeKeys.Bounds.Key, 0)
-					a, span.EndKey = a.Copy(rangeKeys.Bounds.EndKey, 0)
+					a, span.Key = a.Copy(rangeKeys.Bounds.Key)
+					a, span.EndKey = a.Copy(rangeKeys.Bounds.EndKey)
 					ts := rangeKeys.Versions[j].Timestamp
 					err := outputFn(&kvpb.RangeFeedEvent{
 						DeleteRange: &kvpb.RangeFeedDeleteRange{
@@ -298,7 +298,7 @@ func (i *CatchUpIterator) CatchUpScan(
 			if err := outputEvents(); err != nil {
 				return err
 			}
-			a, lastKey = a.Copy(unsafeKey.Key, 0)
+			a, lastKey = a.Copy(unsafeKey.Key)
 		}
 		key := lastKey
 
@@ -316,7 +316,7 @@ func (i *CatchUpIterator) CatchUpScan(
 		//   value.
 		if !ignore || (withDiff && len(reorderBuf) > 0) {
 			var val []byte
-			a, val = a.Copy(unsafeVal, 0)
+			a, val = a.Copy(unsafeVal)
 			if withDiff {
 				// Update the last version with its previous value (this version).
 				if l := len(reorderBuf) - 1; l >= 0 {
diff --git a/pkg/sql/rowcontainer/disk_row_container.go b/pkg/sql/rowcontainer/disk_row_container.go
@@ -523,10 +523,10 @@ func (r *diskRowIterator) EncRow() (rowenc.EncDatumRow, error) {
 	// directly into the EncDatum, so we need to make a copy here. We cannot
 	// reuse the same byte slice across EncRow() calls because it would lead to
 	// modification of the EncDatums (which is not allowed).
-	r.rowBuf, k = r.rowBuf.Copy(k, len(v))
-	// k now contains the capacity for copy of v - separate it out.
-	buf := k[len(k):cap(k)]
-	k = k[:len(k):len(k)]
+	var buf []byte
+	r.rowBuf, buf = r.rowBuf.Alloc(len(k) + len(v))
+	copy(buf, k)
+	k, buf = buf[:len(k):len(k)], buf[len(k):]
 	copy(buf, v)
 	v = buf
 
diff --git a/pkg/storage/mvcc.go b/pkg/storage/mvcc.go
@@ -4033,7 +4033,7 @@ func MVCCPredicateDeleteRange(
 			if runSize < rangeTombstoneThreshold {
 				// Only buffer keys if there's a possibility of issuing point tombstones.
 				var keyCopy roachpb.Key
-				keyAlloc, keyCopy = keyAlloc.Copy(runEnd, 0)
+				keyAlloc, keyCopy = keyAlloc.Copy(runEnd)
 				buf = append(buf, keyCopy)
 			}
 
diff --git a/pkg/storage/mvcc_logical_ops.go b/pkg/storage/mvcc_logical_ops.go
@@ -103,7 +103,7 @@ func (ol *OpLoggerBatch) LogLogicalOpOnly(op MVCCLogicalOpType, details MVCCLogi
 		}
 
 		if !details.Safe {
-			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key, 0)
+			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key)
 		}
 
 		ol.recordOp(&enginepb.MVCCWriteValueOp{
@@ -112,7 +112,7 @@ func (ol *OpLoggerBatch) LogLogicalOpOnly(op MVCCLogicalOpType, details MVCCLogi
 		})
 	case MVCCWriteIntentOpType:
 		if !details.Safe {
-			ol.opsAlloc, details.Txn.Key = ol.opsAlloc.Copy(details.Txn.Key, 0)
+			ol.opsAlloc, details.Txn.Key = ol.opsAlloc.Copy(details.Txn.Key)
 		}
 
 		ol.recordOp(&enginepb.MVCCWriteIntentOp{
@@ -129,7 +129,7 @@ func (ol *OpLoggerBatch) LogLogicalOpOnly(op MVCCLogicalOpType, details MVCCLogi
 		})
 	case MVCCCommitIntentOpType:
 		if !details.Safe {
-			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key, 0)
+			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key)
 		}
 
 		ol.recordOp(&enginepb.MVCCCommitIntentOp{
@@ -143,8 +143,8 @@ func (ol *OpLoggerBatch) LogLogicalOpOnly(op MVCCLogicalOpType, details MVCCLogi
 		})
 	case MVCCDeleteRangeOpType:
 		if !details.Safe {
-			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key, 0)
-			ol.opsAlloc, details.EndKey = ol.opsAlloc.Copy(details.EndKey, 0)
+			ol.opsAlloc, details.Key = ol.opsAlloc.Copy(details.Key)
+			ol.opsAlloc, details.EndKey = ol.opsAlloc.Copy(details.EndKey)
 		}
 		ol.recordOp(&enginepb.MVCCDeleteRangeOp{
 			StartKey:  details.Key,
diff --git a/pkg/util/bufalloc/BUILD.bazel b/pkg/util/bufalloc/BUILD.bazel
@@ -1,8 +1,15 @@
-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "bufalloc",
     srcs = ["byte_allocator.go"],
     importpath = "github.com/cockroachdb/cockroach/pkg/util/bufalloc",
     visibility = ["//visibility:public"],
 )
+
+go_test(
+    name = "bufalloc_test",
+    srcs = ["byte_allocator_test.go"],
+    embed = [":bufalloc"],
+    deps = ["@com_github_stretchr_testify//require"],
+)
diff --git a/pkg/util/bufalloc/byte_allocator.go b/pkg/util/bufalloc/byte_allocator.go
@@ -33,25 +33,21 @@ func (a ByteAllocator) reserve(n int) ByteAllocator {
 	return a
 }
 
-// Alloc allocates a new chunk of memory with the specified length. extraCap
-// indicates additional zero bytes that will be present in the returned []byte,
-// but not part of the length.
-func (a ByteAllocator) Alloc(n int, extraCap int) (ByteAllocator, []byte) {
-	if cap(a.b)-len(a.b) < n+extraCap {
-		a = a.reserve(n + extraCap)
+// Alloc allocates a new chunk of memory with the specified length.
+func (a ByteAllocator) Alloc(n int) (ByteAllocator, []byte) {
+	if cap(a.b)-len(a.b) < n {
+		a = a.reserve(n)
 	}
 	p := len(a.b)
-	r := a.b[p : p+n : p+n+extraCap]
-	a.b = a.b[:p+n+extraCap]
+	r := a.b[p : p+n : p+n]
+	a.b = a.b[:p+n]
 	return a, r
 }
 
 // Copy allocates a new chunk of memory, initializing it from src.
-// extraCap indicates additional zero bytes that will be present in the returned
-// []byte, but not part of the length.
-func (a ByteAllocator) Copy(src []byte, extraCap int) (ByteAllocator, []byte) {
+func (a ByteAllocator) Copy(src []byte) (ByteAllocator, []byte) {
 	var alloc []byte
-	a, alloc = a.Alloc(len(src), extraCap)
+	a, alloc = a.Alloc(len(src))
 	copy(alloc, src)
 	return a, alloc
 }
diff --git a/pkg/util/bufalloc/byte_allocator_test.go b/pkg/util/bufalloc/byte_allocator_test.go
@@ -0,0 +1,41 @@
+// Copyright 2025 The Cockroach Authors.
+//
+// Use of this software is governed by the CockroachDB Software License
+// included in the /LICENSE file.
+
+package bufalloc
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestByteAllocator(t *testing.T) {
+	var a ByteAllocator
+	// Must always allocate at least chunkAllocMinSize.
+	a, _ = a.Copy([]byte{1})
+	require.Equal(t, chunkAllocMinSize, cap(a.b))
+	// Must use already allocated capacity.
+	a, _ = a.Copy([]byte{2})
+	require.Equal(t, 2, len(a.b))
+	require.Equal(t, chunkAllocMinSize, cap(a.b))
+	// Allocated capacity grows exponentially.
+	a, _ = a.Copy(make([]byte, chunkAllocMinSize))
+	require.Equal(t, 2*chunkAllocMinSize, cap(a.b))
+	// The requested capacity is always allocated, skiping some exponential
+	// steps if necessary.
+	a, _ = a.Copy(make([]byte, chunkAllocMaxSize))
+	require.Equal(t, chunkAllocMaxSize, cap(a.b))
+	// Once the maximum allocation size is reached, we never allocate less.
+	a, _ = a.Copy(make([]byte, 1))
+	require.Equal(t, chunkAllocMaxSize, cap(a.b))
+	// Allocate space that exceeds the max size.
+	var buf []byte
+	a, buf = a.Alloc(2 * chunkAllocMaxSize)
+	require.Equal(t, 2*chunkAllocMaxSize, cap(a.b))
+	require.Equal(t, 2*chunkAllocMaxSize, len(buf))
+	// We reduce allocation size if the next allocation is below max chunk size.
+	a, _ = a.Copy(make([]byte, 1))
+	require.Equal(t, chunkAllocMaxSize, cap(a.b))
+}
diff --git a/pkg/workload/bank/bank.go b/pkg/workload/bank/bank.go
@@ -151,7 +151,7 @@ func (b *bank) Tables() []workload.Table {
 				payloadCol.Reset()
 				for rowIdx := rowBegin; rowIdx < rowEnd; rowIdx++ {
 					var payload []byte
-					*a, payload = a.Alloc(b.payloadBytes, 0 /* extraCap */)
+					*a, payload = a.Alloc(b.payloadBytes)
 					randStringLetters(rng, payload)
 
 					rowOffset := rowIdx - rowBegin
diff --git a/pkg/workload/bulkingest/bulkingest.go b/pkg/workload/bulkingest/bulkingest.go
@@ -157,7 +157,7 @@ func (w *bulkingest) Tables() []workload.Table {
 
 				rng := rand.New(rand.NewSource(RandomSeed.Seed() + int64(ab)))
 				var payload []byte
-				*alloc, payload = alloc.Alloc(w.cCount*w.payloadBytes, 0 /* extraCap */)
+				*alloc, payload = alloc.Alloc(w.cCount * w.payloadBytes)
 				randutil.ReadTestdataBytes(rng, payload)
 				payloadCol.Reset()
 				for rowIdx := 0; rowIdx < w.cCount; rowIdx++ {
diff --git a/pkg/workload/kv/kv.go b/pkg/workload/kv/kv.go
@@ -439,7 +439,7 @@ func (w *kv) Tables() []workload.Table {
 					rowOffset := rowIdx - rowBegin
 					var payload []byte
 					blockSize, uniqueSize := w.randBlockSize(rndBlock)
-					*a, payload = a.Alloc(blockSize, 0 /* extraCap */)
+					*a, payload = a.Alloc(blockSize)
 					w.randFillBlock(rndBlock, payload, uniqueSize)
 					valCol.Set(rowOffset, payload)
 				}
diff --git a/pkg/workload/tpcc/generate.go b/pkg/workload/tpcc/generate.go
@@ -428,7 +428,7 @@ func (w *tpcc) tpccHistoryInitialRowBatch(rowIdx int, cb coldata.Batch, a *bufal
 	historyRowCount := numHistoryPerWarehouse * w.warehouses
 	l.uuidAlloc.DeterministicV4(uint64(rowIdx), uint64(historyRowCount))
 	var rowID []byte
-	*a, rowID = a.Alloc(36, 0 /* extraCap */)
+	*a, rowID = a.Alloc(36)
 	l.uuidAlloc.StringBytes(rowID)
 
 	cID := (rowIdx % numCustomersPerDistrict) + 1
diff --git a/pkg/workload/tpcc/random.go b/pkg/workload/tpcc/random.go
diff --git a/pkg/workload/tpch/random.go b/pkg/workload/tpch/random.go

Original file line number	Diff line number	Diff line change
`@@ -494,7 +494,7 @@ func (s *bufferSink) EmitResolvedTimestamp(`
`494`	`494`	`if err != nil {`
`495`	`495`	`return err`
`496`	`496`	`}`
`497`		`- s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)`
	`497`	`+ s.scratch, payload = s.scratch.Copy(payload)`
`498`	`498`	`s.buf.Push(rowenc.EncDatumRow{`
`499`	`499`	`{Datum: tree.DNull}, // resolved span`
`500`	`500`	`{Datum: tree.DNull}, // topic`
Original file line number	Diff line number	Diff line change
`@@ -424,7 +424,7 @@ func (s *kafkaSink) EmitResolvedTimestamp(`
`424`	`424`	`if err != nil {`
`425`	`425`	`return err`
`426`	`426`	`}`
`427`		`- s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)`
	`427`	`+ s.scratch, payload = s.scratch.Copy(payload)`
`428`	`428`
`429`	`429`	`// sarama caches this, which is why we have to periodically refresh the`
`430`	`430`	`// metadata above. Staleness here does not impact correctness. Some new`
Original file line number	Diff line number	Diff line change
`@@ -168,7 +168,7 @@ func (s *sqlSink) EmitResolvedTimestamp(`
`168`	`168`	`if err != nil {`
`169`	`169`	`return err`
`170`	`170`	`}`
`171`		`- s.scratch, payload = s.scratch.Copy(payload, 0 /* extraCap */)`
	`171`	`+ s.scratch, payload = s.scratch.Copy(payload)`
`172`	`172`	`for partition := int32(0); partition < sqlSinkNumPartitions; partition++ {`
`173`	`173`	`if err := s.emit(ctx, topic, partition, noKey, noValue, payload); err != nil {`
`174`	`174`	`return err`
Original file line number	Diff line number	Diff line change
`@@ -738,7 +738,7 @@ func duplicateEvent(event crosscluster.Event) crosscluster.Event {`
`738`	`738`	`for i := range kvs {`
`739`	`739`	`res[i].Key = kvs[i].KeyValue.Key.Clone()`
`740`	`740`	`res[i].Value.Timestamp = kvs[i].KeyValue.Value.Timestamp`
`741`		`- a, res[i].Value.RawBytes = a.Copy(kvs[i].KeyValue.Value.RawBytes, 0)`
	`741`	`+ a, res[i].Value.RawBytes = a.Copy(kvs[i].KeyValue.Value.RawBytes)`
`742`	`742`	`}`
`743`	`743`	`dup = crosscluster.MakeKVEventFromKVs(res)`
`744`	`744`	`case crosscluster.SSTableEvent:`