func distBackup(
	ctx context.Context,
	phs sql.PlanHookState,
	spans roachpb.Spans,
	introducedSpans roachpb.Spans,
	pkIDs map[uint64]bool,
	defaultURI string,
	urisByLocalityKV map[string]string,
	encryption *jobspb.BackupEncryptionOptions,
	mvccFilter roachpb.MVCCFilter,
	startTime, endTime hlc.Timestamp,
	progCh chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
) error {
	ctx = logtags.AddTag(ctx, "backup-distsql", nil)
	var noTxn *kv.Txn

	dsp := phs.DistSQLPlanner()
	evalCtx := phs.ExtendedEvalContext()

	planCtx, _, err := dsp.SetupAllNodesPlanning(ctx, evalCtx, phs.ExecCfg())
	if err != nil {
		return err
	}

	backupSpecs, err := makeBackupDataProcessorSpecs(
		planCtx,
		dsp,
		spans,
		introducedSpans,
		pkIDs,
		defaultURI,
		urisByLocalityKV,
		mvccFilter,
		encryption,
		startTime, endTime,
		phs.User(),
	)
	if err != nil {
		return err
	}

	if len(backupSpecs) == 0 {
		close(progCh)
		return nil
	}

	gatewayNodeID, err := evalCtx.ExecCfg.NodeID.OptionalNodeIDErr(47970)
	if err != nil {
		return err
	}
	p := sql.MakePhysicalPlan(gatewayNodeID)

	// Setup a one-stage plan with one proc per input spec.
	corePlacement := make([]physicalplan.ProcessorCorePlacement, len(backupSpecs))
	i := 0
	for node, spec := range backupSpecs {
		corePlacement[i].NodeID = node
		corePlacement[i].Core.BackupData = spec
		i++
	}

	// All of the progress information is sent through the metadata stream, so we
	// have an empty result stream.
	p.AddNoInputStage(corePlacement, execinfrapb.PostProcessSpec{}, []*types.T{}, execinfrapb.Ordering{})
	p.PlanToStreamColMap = []int{}

	dsp.FinalizePlan(planCtx, &p)

	metaFn := func(_ context.Context, meta *execinfrapb.ProducerMetadata) error {
		if meta.BulkProcessorProgress != nil {
			// Send the progress up a level to be written to the manifest.
			progCh <- meta.BulkProcessorProgress
		}
		return nil
	}

	rowResultWriter := sql.NewRowResultWriter(nil)

	recv := sql.MakeDistSQLReceiver(
		ctx,
		sql.NewMetadataCallbackWriter(rowResultWriter, metaFn),
		tree.Rows,
		nil,   /* rangeCache */
		noTxn, /* txn - the flow does not read or write the database */
		func(ts hlc.Timestamp) {},
		evalCtx.Tracing,
	)
	defer recv.Release()

	defer close(progCh)
	// Copy the evalCtx, as dsp.Run() might change it.
	evalCtxCopy := *evalCtx
	dsp.Run(planCtx, noTxn, &p, recv, &evalCtxCopy, nil /* finishedSetupFn */)()
	return rowResultWriter.Err()
}
// distBackup is used to plan the processors for a distributed backup. It
// streams back progress updates over progCh, which is used to incrementally
// build up the BulkOpSummary.
distBackup is used to plan the processors for a distributed backup. It
streams back progress updates over progCh, which is used to incrementally
build up the BulkOpSummary.
// distBackup is used to plan the processors for a distributed backup . It // streams back progress updates over progCh which is used to incrementally // build up the BulkOpSummary .
func distBackup(
	ctx context.Context,
	phs sql.PlanHookState,
	spans roachpb.Spans,
	introducedSpans roachpb.Spans,
	pkIDs map[uint64]bool,
	defaultURI string,
	urisByLocalityKV map[string]string,
	encryption *jobspb.BackupEncryptionOptions,
	mvccFilter roachpb.MVCCFilter,
	startTime, endTime hlc.Timestamp,
	progCh chan *execinfrapb.RemoteProducerMetadata_BulkProcessorProgress,
) error {
	ctx = logtags.AddTag(ctx, "backup-distsql", nil)
	var noTxn *kv.Txn
	dsp := phs.DistSQLPlanner()
	evalCtx := phs.ExtendedEvalContext()
	planCtx, _, err := dsp.SetupAllNodesPlanning(ctx, evalCtx, phs.ExecCfg())
	if err != nil {
		return err
	}
	backupSpecs, err := makeBackupDataProcessorSpecs(
		planCtx,
		dsp,
		spans,
		introducedSpans,
		pkIDs,
		defaultURI,
		urisByLocalityKV,
		mvccFilter,
		encryption,
		startTime, endTime,
		phs.User(),
	)
	if err != nil {
		return err
	}
	if len(backupSpecs) == 0 {
		close(progCh)
		return nil
	}
	gatewayNodeID, err := evalCtx.ExecCfg.NodeID.OptionalNodeIDErr(47970)
	if err != nil {
		return err
	}
	p := sql.MakePhysicalPlan(gatewayNodeID)
	corePlacement := make([]physicalplan.ProcessorCorePlacement, len(backupSpecs))
	i := 0
	for node, spec := range backupSpecs {
		corePlacement[i].NodeID = node
		corePlacement[i].Core.BackupData = spec
		i++
	}
	p.AddNoInputStage(corePlacement, execinfrapb.PostProcessSpec{}, []*types.T{}, execinfrapb.Ordering{})
	p.PlanToStreamColMap = []int{}
	dsp.FinalizePlan(planCtx, &p)
	metaFn := func(_ context.Context, meta *execinfrapb.ProducerMetadata) error {
		if meta.BulkProcessorProgress != nil {
			progCh <- meta.BulkProcessorProgress
		}
		return nil
	}
	rowResultWriter := sql.NewRowResultWriter(nil)
	recv := sql.MakeDistSQLReceiver(
		ctx,
		sql.NewMetadataCallbackWriter(rowResultWriter, metaFn),
		tree.Rows,
		nil,   
		noTxn, 
		func(ts hlc.Timestamp) {},
		evalCtx.Tracing,
	)
	defer recv.Release()
	defer close(progCh)
	evalCtxCopy := *evalCtx
	dsp.Run(planCtx, noTxn, &p, recv, &evalCtxCopy, nil )()
	return rowResultWriter.Err()
}
func distBackup ( ctx context . Context , phs sql . PlanHookState , spans roachpb . Spans , introducedSpans roachpb . Spans , pkIDs map [ uint64 ] bool , defaultURI string , urisByLocalityKV map [ string ] string , encryption * jobspb . BackupEncryptionOptions , mvccFilter roachpb . MVCCFilter , startTime , endTime hlc . Timestamp , progCh chan * execinfrapb . RemoteProducerMetadata_BulkProcessorProgress , ) error { ctx = logtags . AddTag ( ctx , " " , nil ) var noTxn * kv . Txn dsp := phs . DistSQLPlanner ( ) evalCtx := phs . ExtendedEvalContext ( ) planCtx , _ , err := dsp . SetupAllNodesPlanning ( ctx , evalCtx , phs . ExecCfg ( ) ) if err != nil { return err } backupSpecs , err := makeBackupDataProcessorSpecs ( planCtx , dsp , spans , introducedSpans , pkIDs , defaultURI , urisByLocalityKV , mvccFilter , encryption , startTime , endTime , phs . User ( ) , ) if err != nil { return err } if len ( backupSpecs ) == 0 { close ( progCh ) return nil } gatewayNodeID , err := evalCtx . ExecCfg . NodeID . OptionalNodeIDErr ( 47970 ) if err != nil { return err } p := sql . MakePhysicalPlan ( gatewayNodeID ) corePlacement := make ( [ ] physicalplan . ProcessorCorePlacement , len ( backupSpecs ) ) i := 0 for node , spec := range backupSpecs { corePlacement [ i ] . NodeID = node corePlacement [ i ] . Core . BackupData = spec i ++ } p . AddNoInputStage ( corePlacement , execinfrapb . PostProcessSpec { } , [ ] * types . T { } , execinfrapb . Ordering { } ) p . PlanToStreamColMap = [ ] int { } dsp . FinalizePlan ( planCtx , & p ) metaFn := func ( _ context . Context , meta * execinfrapb . ProducerMetadata ) error { if meta . BulkProcessorProgress != nil { progCh <- meta . BulkProcessorProgress } return nil } rowResultWriter := sql . NewRowResultWriter ( nil ) recv := sql . MakeDistSQLReceiver ( ctx , sql . NewMetadataCallbackWriter ( rowResultWriter , metaFn ) , tree . Rows , nil , noTxn , func ( ts hlc . Timestamp ) { } , evalCtx . Tracing , ) defer recv . Release ( ) defer close ( progCh ) evalCtxCopy := * evalCtx dsp . Run ( planCtx , noTxn , & p , recv , & evalCtxCopy , nil ) ( ) return rowResultWriter . Err ( ) }
distBackup is used to plan the processors for a distributed backup.
distBackup is used to plan the processors for a distributed backup .
// Setup a one-stage plan with one proc per input spec. // All of the progress information is sent through the metadata stream, so we // have an empty result stream. // Send the progress up a level to be written to the manifest. /* rangeCache */ /* txn - the flow does not read or write the database */ // Copy the evalCtx, as dsp.Run() might change it. /* finishedSetupFn */