distsql: add missing MoveToDraining calls

I found several places where processors were not checking for error
metadata from their inputs and draining appropriately. The most
egregious of these is in TableReader, which was not calling
MoveToDraining when it encountered an error from the underlying
RowFetcher. This meant that rather than draining it would continue
calling NextRow and generating errors in a tight loop, which caused the
query to hang as well as high CPU and memory usage on the affected node.

Other affected processors are Distinct, SortChunks, and SortTopK.

Fixes cockroachdb#29374
Fixes cockroachdb#29978

Release note: None

pkg/sql/distsqlrun/distinct.go

@@ -215,6 +215,9 @@ func (d *Distinct) Next() (sqlbase.EncDatumRow, *ProducerMetadata) {
 	for d.State == StateRunning {
 		row, meta := d.input.Next()
 		if meta != nil {
+			if meta.Err != nil {
+				d.MoveToDraining(nil /* err */)
+			}
 			return nil, meta
 		}
 		if row == nil {
@@ -282,6 +285,9 @@ func (d *SortedDistinct) Next() (sqlbase.EncDatumRow, *ProducerMetadata) {
 	for d.State == StateRunning {
 		row, meta := d.input.Next()
 		if meta != nil {
+			if meta.Err != nil {
+				d.MoveToDraining(nil /* err */)
+			}
 			return nil, meta
 		}
 		if row == nil {

pkg/sql/distsqlrun/sorter.go

@@ -406,6 +406,10 @@ func (s *sortTopKProcessor) Start(ctx context.Context) context.Context {
 		row, meta := s.input.Next()
 		if meta != nil {
 			s.trailingMeta = append(s.trailingMeta, *meta)
+			if meta.Err != nil {
+				s.MoveToDraining(nil /* err */)
+				break
+			}
 			continue
 		}
 		if row == nil {
@@ -549,6 +553,9 @@ func (s *sortChunksProcessor) fill() (bool, error) {
 
 		if meta != nil {
 			s.trailingMeta = append(s.trailingMeta, *meta)
+			if meta.Err != nil {
+				return false, nil
+			}
 			continue
 		}
 		if nextChunkRow == nil {

pkg/sql/distsqlrun/tablereader.go

@@ -242,6 +242,9 @@ func (tr *tableReader) Next() (sqlbase.EncDatumRow, *ProducerMetadata) {
 		row, meta := tr.input.Next()
 
 		if meta != nil {
+			if meta.Err != nil {
+				tr.MoveToDraining(nil /* err */)
+			}
 			return nil, meta
 		}
 		if row == nil {