Merge #148182

148182: sql: minor cleanup of readers r=yuzefovich a=yuzefovich

**rowexec: remove a couple testing knobs from processor protos**

This commit removes TableReaderSpec.BatchBytesLimit and
JoinReaderSpec.LookupBatchBytesLimit that are only used in tests. Given
that we have access to the testing knobs on each node that creates the
necessary processor, we can just consult that directly. This removal
avoids the possible confusion for how and when these fields are used.

**sql: clarify fetcher parallelization**

I find "limit batches" terminology a bit confusing, so this commit
switches places where we use to "parallelize" which indicates that the
DistSender-level cross-range parallelism should be used, which in turn
means that TargetBytes limit cannot be set (i.e. "should _not_ limit
batches").

Epic: None
Release note: None

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

Author: craig[bot]

pkg/sql/colfetcher/cfetcher.go

@@ -586,15 +586,15 @@ func cFetcherFirstBatchLimit(limitHint rowinfra.RowLimit, maxKeysPerRow uint32)
 func (cf *cFetcher) StartScan(
 	ctx context.Context,
 	spans roachpb.Spans,
-	limitBatches bool,
+	parallelize bool,
 	batchBytesLimit rowinfra.BytesLimit,
 	limitHint rowinfra.RowLimit,
 ) error {
 	if len(spans) == 0 {
 		return errors.AssertionFailedf("no spans")
 	}
-	if !limitBatches && batchBytesLimit != rowinfra.NoBytesLimit {
-		return errors.AssertionFailedf("batchBytesLimit set without limitBatches")
+	if parallelize && batchBytesLimit != rowinfra.NoBytesLimit {
+		return errors.AssertionFailedf("TargetBytes limit requested with parallelize=true")
 	}
 
 	firstBatchLimit := cFetcherFirstBatchLimit(limitHint, cf.table.spec.MaxKeysPerRow)

pkg/sql/colfetcher/colbatch_scan.go

@@ -169,14 +169,14 @@ func newColBatchScanBase(
 		s.MakeSpansCopy()
 	}
 
-	if spec.LimitHint > 0 || spec.BatchBytesLimit > 0 {
+	if spec.LimitHint > 0 {
 		// Parallelize shouldn't be set when there's a limit hint, but double-check
 		// just in case.
 		spec.Parallelize = false
 	}
 	var batchBytesLimit rowinfra.BytesLimit
 	if !spec.Parallelize {
-		batchBytesLimit = rowinfra.BytesLimit(spec.BatchBytesLimit)
+		batchBytesLimit = rowinfra.BytesLimit(flowCtx.Cfg.TestingKnobs.TableReaderBatchBytesLimit)
 		if batchBytesLimit == 0 {
 			batchBytesLimit = rowinfra.GetDefaultBatchBytesLimit(flowCtx.EvalCtx.TestingKnobs.ForceProductionValues)
 		}
@@ -221,11 +221,10 @@ func (s *ColBatchScan) Init(ctx context.Context) {
 		s.Ctx, s.flowCtx, "colbatchscan", s.processorID,
 		&s.ContentionEventsListener, &s.ScanStatsListener, &s.TenantConsumptionListener,
 	)
-	limitBatches := !s.parallelize
 	if err := s.cf.StartScan(
 		s.Ctx,
 		s.Spans,
-		limitBatches,
+		s.parallelize,
 		s.batchBytesLimit,
 		s.limitHint,
 	); err != nil {

pkg/sql/colfetcher/index_join.go

@@ -215,7 +215,7 @@ func (s *ColIndexJoin) Next() coldata.Batch {
 			if err := s.cf.StartScan(
 				s.Ctx,
 				spans,
-				false, /* limitBatches */
+				true, /* parallelize */
 				rowinfra.NoBytesLimit,
 				rowinfra.NoRowLimit,
 			); err != nil {

pkg/sql/distsql_physical_planner.go

@@ -2467,9 +2467,6 @@ func (dsp *DistSQLPlanner) planTableReaders(
 		}
 
 		tr.Parallelize = info.parallelize
-		if !tr.Parallelize {
-			tr.BatchBytesLimit = dsp.distSQLSrv.TestingKnobs.TableReaderBatchBytesLimit
-		}
 		tr.IgnoreMisplannedRanges = ignoreMisplannedRanges
 		p.TotalEstimatedScannedRows += info.estimatedRowCount
 
@@ -3444,7 +3441,6 @@ func (dsp *DistSQLPlanner) planLookupJoin(
 		MaintainLookupOrdering:            maintainLookupOrdering,
 		LeftJoinWithPairedJoiner:          planInfo.isSecondJoinInPairedJoiner,
 		OutputGroupContinuationForLeftRow: planInfo.isFirstJoinInPairedJoiner,
-		LookupBatchBytesLimit:             dsp.distSQLSrv.TestingKnobs.JoinReaderBatchBytesLimit,
 		LimitHint:                         planInfo.limitHint,
 		RemoteOnlyLookups:                 planInfo.remoteOnlyLookups,
 		ReverseScans:                      planInfo.reverseScans,

pkg/sql/execinfrapb/processors_sql.proto

@@ -74,11 +74,6 @@ message TableReaderSpec {
   // limit hint.
   optional bool parallelize = 12 [(gogoproto.nullable) = false];
 
-  // batch_bytes_limit, if non-zero, controls the TargetBytes limits that the
-  // TableReader will use for its scans. If zero, then the server-side default
-  // is used. If parallelize is set, this cannot be set.
-  optional int64 batch_bytes_limit = 17 [(gogoproto.nullable) = false];
-
   // If non-zero, this enables inconsistent historical scanning where different
   // batches can be read with different timestamps. This is used for
   // long-running table statistics which may outlive the TTL. Using this setting
@@ -124,7 +119,7 @@ message TableReaderSpec {
   // leaseholder of the beginning of the key spans to be scanned).
   optional bool ignore_misplanned_ranges = 22 [(gogoproto.nullable) = false];
 
-  reserved 1, 2, 4, 6, 7, 8, 13, 14, 15, 16, 19;
+  reserved 1, 2, 4, 6, 7, 8, 13, 14, 15, 16, 17, 19;
 }
 
 // FiltererSpec is the specification for a processor that filters input rows
@@ -303,18 +298,10 @@ message JoinReaderSpec {
   // variables @(N+1) to @(N+M) refer to fetched columns.
   optional Expression on_expr = 4 [(gogoproto.nullable) = false];
 
-  // This used to be used for an extra index filter expression. It was removed
-  // in DistSQL version 24.
-  reserved 5;
-
   // For lookup joins. Only JoinType_INNER and JoinType_LEFT_OUTER are
   // supported.
   optional sqlbase.JoinType type = 6 [(gogoproto.nullable) = false];
 
-  // This field used to be a visibility level of the columns that should be
-  // produced. We now produce the columns in the FetchSpec.
-  reserved 7;
-
   // Indicates the row-level locking strength to be used by the join. If set to
   // FOR_NONE, no row-level locking should be performed.
   optional sqlbase.ScanLockingStrength locking_strength = 9 [(gogoproto.nullable) = false];
@@ -335,8 +322,6 @@ message JoinReaderSpec {
   // optimizations.
   optional bool maintain_ordering = 11 [(gogoproto.nullable) = false];
 
-  reserved 12, 13;
-
   // LeftJoinWithPairedJoiner is used when a left {outer,anti,semi} join is
   // being achieved by pairing two joins, and this is the second join. See
   // the comment above.
@@ -349,13 +334,6 @@ message JoinReaderSpec {
   // be true.
   optional bool output_group_continuation_for_left_row = 15 [(gogoproto.nullable) = false];
 
-  // lookup_batch_bytes_limit, if non-zero, controls the TargetBytes limits that
-  // the joiner will use for its lookups. If zero, then the server-side default
-  // is used. Note that, regardless of this setting, bytes limits are not always
-  // used for lookups - it depends on whether the joiner decides it wants
-  // DistSender-parallelism or not.
-  optional int64 lookup_batch_bytes_limit = 18 [(gogoproto.nullable) = false];
-
   // A hint for how many rows the consumer of the join reader output might
   // need. This is used to size the initial batches of input rows to try to
   // avoid reading many more rows than needed by the processor receiving the
@@ -384,6 +362,8 @@ message JoinReaderSpec {
   // reverse order. This is only useful if a lookup can return more than one
   // row.
   optional bool reverse_scans = 25 [(gogoproto.nullable) = false];
+
+  reserved 5, 7, 12, 13, 18;
 }
 
 // SorterSpec is the specification for a "sorting aggregator". A sorting

pkg/sql/rowexec/joinreader.go

@@ -115,11 +115,15 @@ type joinReader struct {
 
 	// fetcher wraps the row.Fetcher used to perform lookups. This enables the
 	// joinReader to wrap the fetcher with a stat collector when necessary.
-	fetcher            rowFetcher
-	alloc              tree.DatumAlloc
-	rowAlloc           rowenc.EncDatumRowAlloc
-	shouldLimitBatches bool
-	readerType         joinReaderType
+	fetcher  rowFetcher
+	alloc    tree.DatumAlloc
+	rowAlloc rowenc.EncDatumRowAlloc
+	// parallelize, if true, indicates that the KV lookups will be parallelized
+	// across ranges when using the DistSender API. It has no influence on the
+	// behavior when using the Streamer API (when the lookups are always
+	// parallelized).
+	parallelize bool
+	readerType  joinReaderType
 
 	// txn is the transaction used by the join reader.
 	txn *kv.Txn
@@ -238,11 +242,6 @@ type joinReader struct {
 	// and requires that the spec has MaintainOrdering set to true.
 	outputGroupContinuationForLeftRow bool
 
-	// lookupBatchBytesLimit controls the TargetBytes of lookup requests. If 0, a
-	// default will be used. Regardless of this value, bytes limits aren't always
-	// used.
-	lookupBatchBytesLimit rowinfra.BytesLimit
-
 	// limitHintHelper is used in limiting batches of input rows in the presence
 	// of hard and soft limits.
 	limitHintHelper execinfra.LimitHintHelper
@@ -331,18 +330,19 @@ func newJoinReader(
 	// in case of indexJoinReaderType, we know that there's exactly one lookup
 	// row for each input row. Similarly, in case of spec.LookupColumnsAreKey,
 	// we know that there's at most one lookup row per input row. In other
-	// cases, we use limits.
-	shouldLimitBatches := !spec.LookupColumnsAreKey && readerType == lookupJoinReaderType
+	// cases, we disable parallelism and use the TargetBytes limit.
+	parallelize := spec.LookupColumnsAreKey || readerType == indexJoinReaderType
 	if flowCtx.EvalCtx.SessionData().ParallelizeMultiKeyLookupJoinsEnabled {
-		shouldLimitBatches = false
+		parallelize = true
 	}
 	if spec.MaintainLookupOrdering {
-		// MaintainLookupOrdering indicates the output of the lookup joiner should
-		// be sorted by <inputCols>, <lookupCols>. It doesn't make sense for
-		// MaintainLookupOrdering to be true when MaintainOrdering is not.
-		// Additionally, we need to disable parallelism for the traditional fetcher
-		// in order to ensure the lookups are ordered, so set shouldLimitBatches.
-		spec.MaintainOrdering, shouldLimitBatches = true, true
+		// MaintainLookupOrdering indicates the output of the lookup joiner
+		// should be sorted by <inputCols>, <lookupCols>. It doesn't make sense
+		// for MaintainLookupOrdering to be true when MaintainOrdering is not.
+		//
+		// Additionally, we need to disable parallelism for the traditional
+		// fetcher in order to ensure the lookups are ordered.
+		spec.MaintainOrdering, parallelize = true, false
 	}
 	useStreamer, txn, err := flowCtx.UseStreamer(ctx)
 	if err != nil {
@@ -359,11 +359,10 @@ func newJoinReader(
 		input:                               input,
 		lookupCols:                          lookupCols,
 		outputGroupContinuationForLeftRow:   spec.OutputGroupContinuationForLeftRow,
-		shouldLimitBatches:                  shouldLimitBatches,
+		parallelize:                         parallelize,
 		readerType:                          readerType,
 		txn:                                 txn,
 		usesStreamer:                        useStreamer,
-		lookupBatchBytesLimit:               rowinfra.BytesLimit(spec.LookupBatchBytesLimit),
 		limitHintHelper:                     execinfra.MakeLimitHintHelper(spec.LimitHint, post),
 		errorOnLookup:                       errorOnLookup,
 		allowEnforceHomeRegionFollowerReads: flowCtx.EvalCtx.SessionData().EnforceHomeRegionFollowerReadsEnabled,
@@ -868,16 +867,15 @@ func (jr *joinReader) getBatchBytesLimit() rowinfra.BytesLimit {
 		// BatchRequests.
 		return rowinfra.NoBytesLimit
 	}
-	if !jr.shouldLimitBatches {
-		// We deem it safe to not limit the batches in order to get the
+	if jr.parallelize {
+		// We deem it safe to not use the TargetBytes limit in order to get the
 		// DistSender-level parallelism.
 		return rowinfra.NoBytesLimit
 	}
-	bytesLimit := jr.lookupBatchBytesLimit
-	if bytesLimit == 0 {
-		bytesLimit = rowinfra.GetDefaultBatchBytesLimit(jr.FlowCtx.EvalCtx.TestingKnobs.ForceProductionValues)
+	if testingLimit := jr.FlowCtx.Cfg.TestingKnobs.JoinReaderBatchBytesLimit; testingLimit != 0 {
+		return rowinfra.BytesLimit(testingLimit)
 	}
-	return bytesLimit
+	return rowinfra.GetDefaultBatchBytesLimit(jr.FlowCtx.EvalCtx.TestingKnobs.ForceProductionValues)
 }
 
 // readInput reads the next batch of input rows and starts an index scan, which
@@ -1054,11 +1052,13 @@ func (jr *joinReader) readInput() (
 	//    fetcher only accepts a limit if the spans are sorted), and
 	// b) Pebble has various optimizations for Seeks in sorted order.
 	if jr.readerType == indexJoinReaderType && jr.maintainOrdering {
-		// Assert that the index join doesn't have shouldLimitBatches set. Since we
-		// didn't sort above, the fetcher doesn't support a limit.
-		if jr.shouldLimitBatches {
+		// Assert that the index join has 'parallelize=true' set. Since we
+		// didn't sort above, the fetcher doesn't support the TargetBytes limit
+		// (which would be set via getBatchBytesLimit() if 'parallelize' was
+		// false).
+		if !jr.parallelize {
 			err := errors.AssertionFailedf("index join configured with both maintainOrdering and " +
-				"shouldLimitBatched; this shouldn't have happened as the implementation doesn't support it")
+				"parallelize=false; this shouldn't have happened as the implementation doesn't support it")
 			jr.MoveToDraining(err)
 			return jrStateUnknown, nil, jr.DrainHelper()
 		}

pkg/sql/rowexec/tablereader.go

@@ -34,9 +34,8 @@ type tableReader struct {
 	execinfra.ProcessorBase
 	execinfra.SpansWithCopy
 
-	limitHint       rowinfra.RowLimit
-	parallelize     bool
-	batchBytesLimit rowinfra.BytesLimit
+	limitHint   rowinfra.RowLimit
+	parallelize bool
 
 	scanStarted bool
 
@@ -83,24 +82,16 @@ func newTableReader(
 		return nil, errors.AssertionFailedf("attempting to create a tableReader with uninitialized NodeID")
 	}
 
-	if spec.LimitHint > 0 || spec.BatchBytesLimit > 0 {
+	if spec.LimitHint > 0 {
 		// Parallelize shouldn't be set when there's a limit hint, but double-check
 		// just in case.
 		spec.Parallelize = false
 	}
-	var batchBytesLimit rowinfra.BytesLimit
-	if !spec.Parallelize {
-		batchBytesLimit = rowinfra.BytesLimit(spec.BatchBytesLimit)
-		if batchBytesLimit == 0 {
-			batchBytesLimit = rowinfra.GetDefaultBatchBytesLimit(flowCtx.EvalCtx.TestingKnobs.ForceProductionValues)
-		}
-	}
 
 	tr := trPool.Get().(*tableReader)
 
 	tr.limitHint = rowinfra.RowLimit(execinfra.LimitHint(spec.LimitHint, post))
 	tr.parallelize = spec.Parallelize
-	tr.batchBytesLimit = batchBytesLimit
 	tr.maxTimestampAge = time.Duration(spec.MaxTimestampAgeNanos)
 
 	// Make sure the key column types are hydrated. The fetched column types
@@ -208,14 +199,14 @@ func (tr *tableReader) startScan(ctx context.Context) error {
 	if cb := tr.FlowCtx.Cfg.TestingKnobs.TableReaderStartScanCb; cb != nil {
 		cb()
 	}
-	limitBatches := !tr.parallelize
-	var bytesLimit rowinfra.BytesLimit
-	if !limitBatches {
-		bytesLimit = rowinfra.NoBytesLimit
-	} else {
-		bytesLimit = tr.batchBytesLimit
+	bytesLimit := rowinfra.NoBytesLimit
+	if !tr.parallelize {
+		bytesLimit = rowinfra.BytesLimit(tr.FlowCtx.Cfg.TestingKnobs.TableReaderBatchBytesLimit)
+		if bytesLimit == 0 {
+			bytesLimit = rowinfra.GetDefaultBatchBytesLimit(tr.FlowCtx.EvalCtx.TestingKnobs.ForceProductionValues)
+		}
 	}
-	log.VEventf(ctx, 1, "starting scan with limitBatches %t", limitBatches)
+	log.VEventf(ctx, 1, "starting scan with parallelize=%t", tr.parallelize)
 	var err error
 	if tr.maxTimestampAge == 0 {
 		err = tr.fetcher.StartScan(