streamingest,bulk: flush SSTs on range boundary, c2c destination side

Before this change, c2c buffered an SST in memory and flushed it to KV, then,
if the SST spans multiple ranges (likely) the addSST fails and therefore
split and rewritten to KV. We see about 20-30 retries per addSST happening
in a roachtest.

The code to split these SSTs on range boundaries already exists, but disabled
without a RangeCache. This pr passes a RangeCache to the SST batcher to enable
flushing on range boundaries.

Epic: CRDB-24777
Informs: #97592

Release note: None

pkg/ccl/streamingccl/streamingest/stream_ingestion_processor.go

@@ -302,9 +302,10 @@ func (sip *streamIngestionProcessor) Start(ctx context.Context) {
 
 	evalCtx := sip.FlowCtx.EvalCtx
 	db := sip.FlowCtx.Cfg.DB
+	rc := sip.FlowCtx.Cfg.RangeCache
 	var err error
 	sip.batcher, err = bulk.MakeStreamSSTBatcher(
-		ctx, db.KV(), evalCtx.Settings, sip.flowCtx.Cfg.BackupMonitor.MakeBoundAccount(),
+		ctx, db.KV(), rc, evalCtx.Settings, sip.flowCtx.Cfg.BackupMonitor.MakeBoundAccount(),
 		sip.flowCtx.Cfg.BulkSenderLimiter, func(batchSummary kvpb.BulkOpSummary) {
 			// OnFlush update the ingested logical and SST byte metrics.
 			sip.metrics.IngestedLogicalBytes.Inc(batchSummary.DataSize)

pkg/kv/bulk/sst_batcher.go

@@ -241,14 +241,15 @@ func MakeSSTBatcher(
 func MakeStreamSSTBatcher(
 	ctx context.Context,
 	db *kv.DB,
+	rc *rangecache.RangeCache,
 	settings *cluster.Settings,
 	mem mon.BoundAccount,
 	sendLimiter limit.ConcurrentRequestLimiter,
 	onFlush func(summary kvpb.BulkOpSummary),
 ) (*SSTBatcher, error) {
 	// A mutex is needed because flushes on range boundaries can Grow and Shrink memory asynchronously.
 	mem.Mu = &syncutil.Mutex{}
-	b := &SSTBatcher{db: db, settings: settings, ingestAll: true, mem: mem, limiter: sendLimiter}
+	b := &SSTBatcher{db: db, rc: rc, settings: settings, ingestAll: true, mem: mem, limiter: sendLimiter}
 	b.SetOnFlush(onFlush)
 	err := b.Reset(ctx)
 	return b, err