colflow: handle error during draining correctly

This commit fixes the way we handle internal errors that occur during
the draining in the `FlowCoordinator`. Previously, with all such errors
we would call `MoveToDraining`, but if the processor is already
draining, then it would panic. This could occur, for example, when
accounting for the metadata footprint and exceeding the limit (which was
added several weeks ago). This is now fixed.

Release justification: bug fix.

Release note: None

pkg/sql/colflow/flow_coordinator.go

@@ -150,7 +150,17 @@ func (f *FlowCoordinator) nextAdapter() {
 // Next is part of the execinfra.RowSource interface.
 func (f *FlowCoordinator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMetadata) {
 	if err := colexecerror.CatchVectorizedRuntimeError(f.nextAdapter); err != nil {
-		f.MoveToDraining(err)
+		if f.State == execinfra.StateRunning {
+			f.MoveToDraining(err)
+		} else {
+			// We have encountered an error during draining, so we will just
+			// return the error as metadata directly. This could occur, for
+			// example, when accounting for the metadata footprint and exceeding
+			// the limit.
+			meta := execinfrapb.GetProducerMeta()
+			meta.Err = err
+			return nil, meta
+		}
 		return nil, f.DrainHelper()
 	}
 	return f.row, f.meta