kv/bulk: extract fields into batch

This extracts per batch state tracking into a `batch` struct. The intent
is to make it clear what state needs to be reset before accepting writes
for the next batch.

Release note: none
Part of: #146828

Author: jeffswenson

pkg/kv/bulk/buffering_adder.go

@@ -124,10 +124,18 @@ func MakeBulkAdder(
 			settings:               settings,
 			skipDuplicates:         opts.SkipDuplicates,
 			disallowShadowingBelow: opts.DisallowShadowingBelow,
-			batchTS:                opts.BatchTimestamp,
-			writeAtBatchTS:         opts.WriteAtBatchTimestamp,
-			mem:                    bulkMon.MakeConcurrentBoundAccount(),
-			limiter:                sendLimiter,
+			batch: batch{
+				// TODO(jeffswenson): As far as I can tell, setting the batch timestamp here does nothing. The API
+				// is actively misleading.
+				//
+				// It used to do something before #75275 was merged, but now it is always test to hlc.Timestamp{}
+				// by reset before the first flush. I think we should clean this up, but cleaning it up properly
+				// requires cleanups throughout the schema changer code base to replace "WriteAsOf" with "ReadAsOf".
+				ts: opts.BatchTimestamp,
+			},
+			writeAtBatchTS: opts.WriteAtBatchTimestamp,
+			mem:            bulkMon.MakeConcurrentBoundAccount(),
+			limiter:        sendLimiter,
 		},
 		timestamp:      timestamp,
 		maxBufferLimit: opts.MaxBufferSize,
@@ -201,7 +209,7 @@ func (b *BufferingAdder) Add(ctx context.Context, key roachpb.Key, value []byte)
 		return b.curBuf.append(key, value)
 	}
 
-	b.sink.currentStats.FlushesDueToSize++
+	b.sink.batch.stats.FlushesDueToSize++
 	log.VEventf(ctx, 3, "%s adder triggering flush of %s of KVs in %s buffer",
 		b.name, b.curBuf.KVSize(), b.bufferedMemSize())
 
@@ -264,7 +272,7 @@ func (b *BufferingAdder) Flush(ctx context.Context) error {
 }
 
 func (b *BufferingAdder) doFlush(ctx context.Context, forSize bool) error {
-	b.sink.currentStats.FillWait += timeutil.Since(b.lastFlush)
+	b.sink.batch.stats.FillWait += timeutil.Since(b.lastFlush)
 
 	if b.bufferedKeys() == 0 {
 		if b.onFlush != nil {
@@ -275,7 +283,7 @@ func (b *BufferingAdder) doFlush(ctx context.Context, forSize bool) error {
 		return nil
 	}
 	b.sink.Reset(ctx)
-	b.sink.currentStats.BufferFlushes++
+	b.sink.batch.stats.BufferFlushes++
 
 	var before *bulkpb.IngestionPerformanceStats
 	var beforeSize int64
@@ -284,7 +292,7 @@ func (b *BufferingAdder) doFlush(ctx context.Context, forSize bool) error {
 		b.sink.mu.Lock()
 		before = b.sink.mu.totalStats.Identity().(*bulkpb.IngestionPerformanceStats)
 		before.Combine(&b.sink.mu.totalStats)
-		before.Combine(&b.sink.currentStats)
+		before.Combine(&b.sink.batch.stats)
 		beforeSize = b.sink.mu.totalBulkOpSummary.DataSize
 		b.sink.mu.Unlock()
 	}
@@ -297,7 +305,7 @@ func (b *BufferingAdder) doFlush(ctx context.Context, forSize bool) error {
 	mvccKey := storage.MVCCKey{Timestamp: b.timestamp}
 
 	beforeFlush := timeutil.Now()
-	b.sink.currentStats.SortWait += beforeFlush.Sub(beforeSort)
+	b.sink.batch.stats.SortWait += beforeFlush.Sub(beforeSort)
 
 	// If this is the first flush and is due to size, if it was unsorted then
 	// create initial splits if requested before flushing.
@@ -340,14 +348,14 @@ func (b *BufferingAdder) doFlush(ctx context.Context, forSize bool) error {
 		}
 	}
 
-	b.sink.currentStats.FlushWait += timeutil.Since(beforeFlush)
+	b.sink.batch.stats.FlushWait += timeutil.Since(beforeFlush)
 
 	if log.V(3) && before != nil {
 		b.sink.mu.Lock()
 		written := b.sink.mu.totalBulkOpSummary.DataSize - beforeSize
 		afterStats := b.sink.mu.totalStats.Identity().(*bulkpb.IngestionPerformanceStats)
 		afterStats.Combine(&b.sink.mu.totalStats)
-		afterStats.Combine(&b.sink.currentStats)
+		afterStats.Combine(&b.sink.batch.stats)
 		b.sink.mu.Unlock()
 
 		files := afterStats.Batches - before.Batches
@@ -457,24 +465,24 @@ func (b *BufferingAdder) createInitialSplits(ctx context.Context) error {
 	splitsWait := beforeScatters.Sub(beforeSplits)
 	log.Infof(ctx, "%s adder created %d initial splits in %v from %d keys in %s buffer",
 		b.name, len(toScatter), timing(splitsWait), b.curBuf.Len(), b.curBuf.MemSize())
-	b.sink.currentStats.Splits += int64(len(toScatter))
-	b.sink.currentStats.SplitWait += splitsWait
+	b.sink.batch.stats.Splits += int64(len(toScatter))
+	b.sink.batch.stats.SplitWait += splitsWait
 
 	for _, splitKey := range toScatter {
 		resp, err := b.sink.db.AdminScatter(ctx, splitKey, 0 /* maxSize */)
 		if err != nil {
 			log.Warningf(ctx, "failed to scatter: %v", err)
 			continue
 		}
-		b.sink.currentStats.Scatters++
-		b.sink.currentStats.ScatterMoved += resp.ReplicasScatteredBytes
+		b.sink.batch.stats.Scatters++
+		b.sink.batch.stats.ScatterMoved += resp.ReplicasScatteredBytes
 		if resp.ReplicasScatteredBytes > 0 {
 			log.VEventf(ctx, 1, "pre-split scattered %s in non-empty range %s",
 				sz(resp.ReplicasScatteredBytes), resp.RangeInfos[0].Desc.KeySpan().AsRawSpanWithNoLocals())
 		}
 	}
 	scattersWait := timeutil.Since(beforeScatters)
-	b.sink.currentStats.ScatterWait += scattersWait
+	b.sink.batch.stats.ScatterWait += scattersWait
 	log.Infof(ctx, "%s adder scattered %d initial split spans in %v",
 		b.name, len(toScatter), timing(scattersWait))