execinfra: fix starting of processors

Almost all processors embed `ProcessorBase` and call `StartInternal` on
it. That function derives a new context and starts a tracing span if the
tracing is enabled on the parent ctx and the derived context is
returned. Previously, in all callsites the inputs to the processors
would get `Start`ed before `StartInternal` is called, but this doesn't
make sense: the consumer's ctx and span should be encompassing the
producer's (input's) ctx and span, so this commit switches all callsites
to the following layout
```
  ctx = p.StartInternal(ctx)
  input.Start(ctx) // if there are any inputs
```

Release justification: bug fix.

Release note: None

pkg/ccl/backupccl/restore_data_processor.go

@@ -80,12 +80,8 @@ func newRestoreDataProcessor(
 
 // Start is part of the RowSource interface.
 func (rd *restoreDataProcessor) Start(ctx context.Context) {
-	rd.input.Start(ctx)
 	ctx = rd.StartInternal(ctx, restoreDataProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	rd.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -906,11 +906,10 @@ func newChangeFrontierProcessor(
 
 // Start is part of the RowSource interface.
 func (cf *changeFrontier) Start(ctx context.Context) {
-	cf.input.Start(ctx)
-
 	// StartInternal called at the beginning of the function because there are
 	// early returns if errors are detected.
 	ctx = cf.StartInternal(ctx, changeFrontierProcName)
+	cf.input.Start(ctx)
 
 	// Pass a nil oracle because this sink is only used to emit resolved timestamps
 	// but the oracle is only used when emitting row updates.

pkg/ccl/streamingccl/streamingest/stream_ingestion_frontier_processor.go

@@ -88,12 +88,8 @@ func newStreamIngestionFrontierProcessor(
 
 // Start is part of the RowSource interface.
 func (sf *streamIngestionFrontier) Start(ctx context.Context) {
-	sf.input.Start(ctx)
 	ctx = sf.StartInternal(ctx, streamIngestionFrontierProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	sf.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/colexec/materializer.go

@@ -233,12 +233,8 @@ func (m *Materializer) Child(nth int, verbose bool) execinfra.OpNode {
 
 // Start is part of the execinfra.RowSource interface.
 func (m *Materializer) Start(ctx context.Context) {
-	m.drainHelper.Start(ctx)
 	ctx = m.ProcessorBase.StartInternal(ctx, materializerProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	m.drainHelper.Start(ctx)
 	// We can encounter an expected error during Init (e.g. an operator
 	// attempts to allocate a batch, but the memory budget limit has been
 	// reached), so we need to wrap it with a catcher.

pkg/sql/execinfra/metadata_test_receiver.go

@@ -140,12 +140,8 @@ func (mtr *MetadataTestReceiver) checkRowNumMetadata() *execinfrapb.ProducerMeta
 
 // Start is part of the RowSource interface.
 func (mtr *MetadataTestReceiver) Start(ctx context.Context) {
-	mtr.input.Start(ctx)
 	ctx = mtr.StartInternal(ctx, metadataTestReceiverProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	mtr.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/execinfra/metadata_test_sender.go

@@ -75,12 +75,8 @@ func NewMetadataTestSender(
 
 // Start is part of the RowSource interface.
 func (mts *MetadataTestSender) Start(ctx context.Context) {
-	mts.input.Start(ctx)
 	ctx = mts.StartInternal(ctx, metadataTestSenderProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	mts.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/execinfra/processorsbase.go

@@ -870,6 +870,7 @@ func ProcessorSpan(ctx context.Context, name string) (context.Context, *tracing.
 // It is likely that this method is called from RowSource.Start implementation,
 // and the recommended layout is the following:
 //   ctx = pb.StartInternal(ctx, name)
+//   < inputs >.Start(ctx) // if there are any inputs-RowSources to pb
 //   < other initialization >
 // so that the caller doesn't mistakenly use old ctx object.
 func (pb *ProcessorBase) StartInternal(ctx context.Context, name string) context.Context {

pkg/sql/rowexec/aggregator.go

@@ -337,8 +337,8 @@ func (ag *orderedAggregator) Start(ctx context.Context) {
 }
 
 func (ag *aggregatorBase) start(ctx context.Context, procName string) {
-	ag.input.Start(ctx)
 	ctx = ag.StartInternal(ctx, procName)
+	ag.input.Start(ctx)
 	ag.cancelChecker = cancelchecker.NewCancelChecker(ctx)
 	ag.runningState = aggAccumulating
 }

pkg/sql/rowexec/bulk_row_writer.go

@@ -76,8 +76,8 @@ func newBulkRowWriterProcessor(
 
 // Start is part of the RowSource interface.
 func (sp *bulkRowWriter) Start(ctx context.Context) {
-	sp.input.Start(ctx)
 	ctx = sp.StartInternal(ctx, "bulkRowWriter")
+	sp.input.Start(ctx)
 	err := sp.work(ctx)
 	sp.MoveToDraining(err)
 }

pkg/sql/rowexec/countrows.go

@@ -68,12 +68,8 @@ func newCountAggregator(
 }
 
 func (ag *countAggregator) Start(ctx context.Context) {
-	ag.input.Start(ctx)
 	ctx = ag.StartInternal(ctx, countRowsProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	ag.input.Start(ctx)
 }
 
 func (ag *countAggregator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadata) {

pkg/sql/rowexec/distinct.go

@@ -142,22 +142,14 @@ func newDistinct(
 
 // Start is part of the RowSource interface.
 func (d *distinct) Start(ctx context.Context) {
-	d.input.Start(ctx)
 	ctx = d.StartInternal(ctx, distinctProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	d.input.Start(ctx)
 }
 
 // Start is part of the RowSource interface.
 func (d *sortedDistinct) Start(ctx context.Context) {
-	d.input.Start(ctx)
 	ctx = d.StartInternal(ctx, sortedDistinctProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	d.input.Start(ctx)
 }
 
 func (d *distinct) matchLastGroupKey(row rowenc.EncDatumRow) (bool, error) {

pkg/sql/rowexec/filterer.go

@@ -71,12 +71,8 @@ func newFiltererProcessor(
 
 // Start is part of the RowSource interface.
 func (f *filtererProcessor) Start(ctx context.Context) {
-	f.input.Start(ctx)
 	ctx = f.StartInternal(ctx, filtererProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	f.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/rowexec/hashjoiner.go

@@ -158,9 +158,9 @@ func newHashJoiner(
 
 // Start is part of the RowSource interface.
 func (h *hashJoiner) Start(ctx context.Context) {
+	ctx = h.StartInternal(ctx, hashJoinerProcName)
 	h.leftSource.Start(ctx)
 	h.rightSource.Start(ctx)
-	ctx = h.StartInternal(ctx, hashJoinerProcName)
 	h.cancelChecker = cancelchecker.NewCancelChecker(ctx)
 	h.runningState = hjBuilding
 }

pkg/sql/rowexec/inverted_filterer.go

@@ -288,12 +288,8 @@ func (ifr *invertedFilterer) emitRow() (
 
 // Start is part of the RowSource interface.
 func (ifr *invertedFilterer) Start(ctx context.Context) {
-	ifr.input.Start(ctx)
 	ctx = ifr.StartInternal(ctx, invertedFiltererProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	ifr.input.Start(ctx)
 	ifr.runningState = ifrReadingInput
 }
 

pkg/sql/rowexec/inverted_joiner.go

@@ -734,12 +734,8 @@ func (ij *invertedJoiner) transformToTableRow(indexRow rowenc.EncDatumRow) {
 
 // Start is part of the RowSource interface.
 func (ij *invertedJoiner) Start(ctx context.Context) {
-	ij.input.Start(ctx)
 	ctx = ij.StartInternal(ctx, invertedJoinerProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	ij.input.Start(ctx)
 	ij.runningState = ijReadingInput
 }
 

pkg/sql/rowexec/joinreader.go

@@ -700,12 +700,8 @@ func (jr *joinReader) emitRow() (
 
 // Start is part of the RowSource interface.
 func (jr *joinReader) Start(ctx context.Context) {
-	jr.input.Start(ctx)
 	ctx = jr.StartInternal(ctx, joinReaderProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	jr.input.Start(ctx)
 	jr.runningState = jrReadingInput
 }
 

pkg/sql/rowexec/mergejoiner.go

@@ -115,8 +115,8 @@ func newMergeJoiner(
 
 // Start is part of the RowSource interface.
 func (m *mergeJoiner) Start(ctx context.Context) {
-	m.streamMerger.start(ctx)
 	ctx = m.StartInternal(ctx, mergeJoinerProcName)
+	m.streamMerger.start(ctx)
 	m.cancelChecker = cancelchecker.NewCancelChecker(ctx)
 }
 

pkg/sql/rowexec/noop.go

@@ -64,12 +64,8 @@ func newNoopProcessor(
 
 // Start is part of the RowSource interface.
 func (n *noopProcessor) Start(ctx context.Context) {
-	n.input.Start(ctx)
 	ctx = n.StartInternal(ctx, noopProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	n.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/rowexec/ordinality.go

@@ -76,12 +76,8 @@ func newOrdinalityProcessor(
 
 // Start is part of the RowSource interface.
 func (o *ordinalityProcessor) Start(ctx context.Context) {
-	o.input.Start(ctx)
 	ctx = o.StartInternal(ctx, ordinalityProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	o.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/rowexec/project_set.go

@@ -116,8 +116,8 @@ func newProjectSetProcessor(
 
 // Start is part of the RowSource interface.
 func (ps *projectSetProcessor) Start(ctx context.Context) {
-	ps.input.Start(ctx)
 	ctx = ps.StartInternal(ctx, projectSetProcName)
+	ps.input.Start(ctx)
 	ps.cancelChecker = cancelchecker.NewCancelChecker(ctx)
 }
 

pkg/sql/rowexec/sample_aggregator.go

@@ -181,18 +181,14 @@ func (s *sampleAggregator) pushTrailingMeta(ctx context.Context) {
 
 // Run is part of the Processor interface.
 func (s *sampleAggregator) Run(ctx context.Context) {
-	s.input.Start(ctx)
 	ctx = s.StartInternal(ctx, sampleAggregatorProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	s.input.Start(ctx)
 
-	earlyExit, err := s.mainLoop(s.Ctx)
+	earlyExit, err := s.mainLoop(ctx)
 	if err != nil {
-		execinfra.DrainAndClose(s.Ctx, s.Out.Output(), err, s.pushTrailingMeta, s.input)
+		execinfra.DrainAndClose(ctx, s.Out.Output(), err, s.pushTrailingMeta, s.input)
 	} else if !earlyExit {
-		s.pushTrailingMeta(s.Ctx)
+		s.pushTrailingMeta(ctx)
 		s.input.ConsumerClosed()
 		s.Out.Close()
 	}

pkg/sql/rowexec/sampler.go

@@ -218,18 +218,14 @@ func (s *samplerProcessor) pushTrailingMeta(ctx context.Context) {
 
 // Run is part of the Processor interface.
 func (s *samplerProcessor) Run(ctx context.Context) {
-	s.input.Start(ctx)
 	ctx = s.StartInternal(ctx, samplerProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	s.input.Start(ctx)
 
-	earlyExit, err := s.mainLoop(s.Ctx)
+	earlyExit, err := s.mainLoop(ctx)
 	if err != nil {
-		execinfra.DrainAndClose(s.Ctx, s.Out.Output(), err, s.pushTrailingMeta, s.input)
+		execinfra.DrainAndClose(ctx, s.Out.Output(), err, s.pushTrailingMeta, s.input)
 	} else if !earlyExit {
-		s.pushTrailingMeta(s.Ctx)
+		s.pushTrailingMeta(ctx)
 		s.input.ConsumerClosed()
 		s.Out.Close()
 	}

pkg/sql/rowexec/sorter.go

@@ -227,12 +227,8 @@ func newSortAllProcessor(
 
 // Start is part of the RowSource interface.
 func (s *sortAllProcessor) Start(ctx context.Context) {
-	s.input.Start(ctx)
 	ctx = s.StartInternal(ctx, sortAllProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	s.input.Start(ctx)
 
 	valid, err := s.fill()
 	if !valid || err != nil {
@@ -343,8 +339,8 @@ func newSortTopKProcessor(
 
 // Start is part of the RowSource interface.
 func (s *sortTopKProcessor) Start(ctx context.Context) {
-	s.input.Start(ctx)
 	ctx = s.StartInternal(ctx, sortTopKProcName)
+	s.input.Start(ctx)
 
 	// The execution loop for the SortTopK processor is similar to that of the
 	// SortAll processor; the difference is that we push rows into a max-heap
@@ -526,12 +522,8 @@ func (s *sortChunksProcessor) fill() (bool, error) {
 
 // Start is part of the RowSource interface.
 func (s *sortChunksProcessor) Start(ctx context.Context) {
-	s.input.Start(ctx)
 	ctx = s.StartInternal(ctx, sortChunksProcName)
-	// Go around "this value of ctx is never used" linter error. We do it this
-	// way instead of omitting the assignment to ctx above so that if in the
-	// future other initialization is added, the correct ctx is used.
-	_ = ctx
+	s.input.Start(ctx)
 }
 
 // Next is part of the RowSource interface.

pkg/sql/rowexec/windower.go

@@ -205,8 +205,8 @@ func newWindower(
 
 // Start is part of the RowSource interface.
 func (w *windower) Start(ctx context.Context) {
-	w.input.Start(ctx)
 	ctx = w.StartInternal(ctx, windowerProcName)
+	w.input.Start(ctx)
 	w.cancelChecker = cancelchecker.NewCancelChecker(ctx)
 	w.runningState = windowerAccumulating
 }