Merge #61037 #61197 #61198

61037: streamingest: add AOST to stream ingestion job r=adityamaru,dt a=pbardea

To be rebased on the PR that introduces the core changefeed stream client.


Release justification: low-risk (very experimental feature), high reward
(AOST stream ingestion)

Release note: None

61197: execinfra: fix starting of processors r=yuzefovich a=yuzefovich

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

61198: storage: add a scan benchmark with resolved intents r=sumeerbhola a=sumeerbhola

It stresses the case where many versions were written
transactionally and all were resolved.

This is the benchmark mentioned in
cockroachdb/pebble#1067
to justify the Pebble seek noop optimization.

Release justification: Non-production code change that
adds a benchmark.

Release note: None

Co-authored-by: Paul Bardea <[email protected]>
Co-authored-by: Yahor Yuzefovich <[email protected]>
Co-authored-by: sumeerbhola <[email protected]>

docs/generated/sql/bnf/restore.bnf

@@ -23,4 +23,5 @@ restore_stmt ::=
 	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' subdirectory 'IN' full_backup_location ( | partitioned_backup_location ( ',' partitioned_backup_location )*)  'WITH' restore_options_list
 	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' subdirectory 'IN' full_backup_location ( | partitioned_backup_location ( ',' partitioned_backup_location )*)  'WITH' 'OPTIONS' '(' restore_options_list ')'
 	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' subdirectory 'IN' full_backup_location ( | partitioned_backup_location ( ',' partitioned_backup_location )*)  
-	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' 'REPLICATION' 'STREAM' 'FROM' subdirectory_opt_list
+	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' 'REPLICATION' 'STREAM' 'FROM' subdirectory_opt_list 'AS' 'OF' 'SYSTEM' 'TIME' timestamp
+	| 'RESTORE' ( ( 'TABLE' | ) table_pattern ( ( ',' table_pattern ) )* | 'DATABASE' database_name ( ( ',' database_name ) )* ) 'FROM' 'REPLICATION' 'STREAM' 'FROM' subdirectory_opt_list 

docs/generated/sql/bnf/stmt_block.bnf

@@ -180,7 +180,7 @@ restore_stmt ::=
 	| 'RESTORE' 'FROM' string_or_placeholder 'IN' list_of_string_or_placeholder_opt_list opt_as_of_clause opt_with_restore_options
 	| 'RESTORE' targets 'FROM' list_of_string_or_placeholder_opt_list opt_as_of_clause opt_with_restore_options
 	| 'RESTORE' targets 'FROM' string_or_placeholder 'IN' list_of_string_or_placeholder_opt_list opt_as_of_clause opt_with_restore_options
-	| 'RESTORE' targets 'FROM' 'REPLICATION' 'STREAM' 'FROM' string_or_placeholder_opt_list
+	| 'RESTORE' targets 'FROM' 'REPLICATION' 'STREAM' 'FROM' string_or_placeholder_opt_list opt_as_of_clause
 
 resume_stmt ::=
 	resume_jobs_stmt

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/ccl/streamingccl/streamingest/stream_ingestion_job.go

@@ -30,6 +30,7 @@ type streamIngestionResumer struct {
 func ingest(
 	ctx context.Context,
 	execCtx sql.JobExecContext,
+	startTime hlc.Timestamp,
 	streamAddress streamingccl.StreamAddress,
 	progress jobspb.Progress,
 	jobID jobspb.JobID,
@@ -49,7 +50,7 @@ func ingest(
 	// KVs. We can skip to ingesting after this resolved ts. Plumb the
 	// initialHighwatermark to the ingestion processor spec based on what we read
 	// from the job progress.
-	var initialHighWater hlc.Timestamp
+	initialHighWater := startTime
 	if h := progress.GetHighWater(); h != nil && !h.IsEmpty() {
 		initialHighWater = *h
 	}
@@ -80,7 +81,7 @@ func (s *streamIngestionResumer) Resume(resumeCtx context.Context, execCtx inter
 	p := execCtx.(sql.JobExecContext)
 
 	// Start ingesting KVs from the replication stream.
-	err := ingest(resumeCtx, p, details.StreamAddress, s.job.Progress(), s.job.ID())
+	err := ingest(resumeCtx, p, details.StartTime, details.StreamAddress, s.job.Progress(), s.job.ID())
 	if err != nil {
 		return err
 	}

pkg/ccl/streamingccl/streamingest/stream_ingestion_planning.go

@@ -24,6 +24,7 @@ import (
 	"github.com/cockroachdb/cockroach/pkg/sql/pgwire/pgerror"
 	"github.com/cockroachdb/cockroach/pkg/sql/sem/tree"
 	"github.com/cockroachdb/cockroach/pkg/util/hlc"
+	"github.com/cockroachdb/cockroach/pkg/util/timeutil"
 	"github.com/cockroachdb/cockroach/pkg/util/tracing"
 	"github.com/cockroachdb/errors"
 )
@@ -92,11 +93,19 @@ func ingestionPlanHook(
 		// TODO(adityamaru): Add privileges checks. Probably the same as RESTORE.
 
 		prefix := keys.MakeTenantPrefix(ingestionStmt.Targets.Tenant)
+		startTime := hlc.Timestamp{WallTime: timeutil.Now().UnixNano()}
+		if ingestionStmt.AsOf.Expr != nil {
+			var err error
+			startTime, err = p.EvalAsOfTimestamp(ctx, ingestionStmt.AsOf)
+			if err != nil {
+				return err
+			}
+		}
+
 		streamIngestionDetails := jobspb.StreamIngestionDetails{
 			StreamAddress: streamingccl.StreamAddress(from[0]),
 			Span:          roachpb.Span{Key: prefix, EndKey: prefix.PrefixEnd()},
-			// TODO: Figure out what the initial ts should be.
-			StartTime: hlc.Timestamp{},
+			StartTime:     startTime,
 		}
 
 		jobDescription, err := streamIngestionJobDescription(p, ingestionStmt)

pkg/ccl/streamingccl/streamingest/stream_ingestion_processor_planning.go

@@ -45,6 +45,7 @@ func distStreamIngestionPlanSpecs(
 		if i < len(nodes) {
 			spec := &execinfrapb.StreamIngestionDataSpec{
 				JobID:              int64(jobID),
+				StartTime:          initialHighWater,
 				StreamAddress:      streamAddress,
 				PartitionAddresses: make([]streamingccl.PartitionAddress, 0),
 			}

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/parser/parse_test.go

@@ -1729,6 +1729,8 @@ func TestParse(t *testing.T) {
 
 		{`RESTORE TENANT 123 FROM REPLICATION STREAM FROM 'bar'`},
 		{`RESTORE TENANT 123 FROM REPLICATION STREAM FROM $1`},
+		{`RESTORE TENANT 123 FROM REPLICATION STREAM FROM 'bar' AS OF SYSTEM TIME '1'`},
+		{`RESTORE TENANT 123 FROM REPLICATION STREAM FROM $1 AS OF SYSTEM TIME '1'`},
 
 		// Currently, we only support TENANT as a target. We have grammar rules for
 		// all targets supported by RESTORE but these will error out during the

pkg/sql/parser/sql.y

@@ -2639,11 +2639,12 @@ restore_stmt:
       Options: *($8.restoreOptions()),
     }
   }
-| RESTORE targets FROM REPLICATION STREAM FROM string_or_placeholder_opt_list
+| RESTORE targets FROM REPLICATION STREAM FROM string_or_placeholder_opt_list opt_as_of_clause
   {
    $$.val = &tree.StreamIngestion{
      Targets: $2.targetList(),
      From: $7.stringOrPlaceholderOptList(),
+     AsOf: $8.asOfClause(),
    }
   }
 | RESTORE error // SHOW HELP: RESTORE

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()
 	}