Added support for a fast text protocol row parser and integrated it with query logic.

Author: Mark Dierolf

lib/commands/query.js

@@ -9,6 +9,7 @@ const Command = require('./command.js');
 const Packets = require('../packets/index.js');
 const getTextParser = require('../parsers/text_parser.js');
 const staticParser = require('../parsers/static_text_parser.js');
+const getFastTextParser = require('../parsers/fast_text_parser.js');
 const ServerStatus = require('../constants/server_status.js');
 
 const EmptyPacket = new Packets.Packet(0, Buffer.allocUnsafe(4), 0, 4);
@@ -31,7 +32,7 @@ class Query extends Command {
     this._receivedFieldsCount = 0;
     this._resultIndex = 0;
     this._localStream = null;
-    this._unpipeStream = function () {};
+    this._unpipeStream = function() {};
     this._streamFactory = options.infileStreamFactory;
     this._connection = null;
   }
@@ -156,7 +157,7 @@ class Query extends Command {
     const onPause = () => {
       this._localStream.pause();
     };
-    const onData = function (data) {
+    const onData = function(data) {
       const dataWithHeader = Buffer.allocUnsafe(data.length + 4);
       data.copy(dataWithHeader, 4);
       connection.writePacket(
@@ -213,7 +214,13 @@ class Query extends Command {
     if (this._receivedFieldsCount === this._fieldCount) {
       const fields = this._fields[this._resultIndex];
       this.emit('fields', fields);
-      if (this.options.disableEval) {
+      if (this.options.useFastTextParser) {
+        this._rowParser = new (getFastTextParser(
+          fields,
+          this.options,
+          connection.config
+        ))(fields);
+      } else if (this.options.disableEval) {
         this._rowParser = staticParser(fields, this.options, connection.config);
       } else {
         this._rowParser = new (getTextParser(
@@ -288,6 +295,7 @@ class Query extends Command {
     });
     this.on('end', () => {
       stream.push(null); // pushing null, indicating EOF
+      stream.emit('close'); // notify readers that query has completed
     });
     this.on('fields', (fields) => {
       stream.emit('fields', fields); // replicate old emitter
@@ -301,7 +309,10 @@ class Query extends Command {
   _setTimeout() {
     if (this.timeout) {
       const timeoutHandler = this._handleTimeoutError.bind(this);
-      this.queryTimeout = Timers.setTimeout(timeoutHandler, this.timeout);
+      this.queryTimeout = Timers.setTimeout(
+        timeoutHandler,
+        this.timeout
+      );
     }
   }
 

lib/parsers/fast_text_parser.js

@@ -0,0 +1,117 @@
+'use strict';
+
+const Types = require('../constants/types.js');
+const Charsets = require('../constants/charsets.js');
+const helpers = require('../helpers');
+const genFunc = require('generate-function');
+const parserCache = require('./parser_cache.js');
+
+const typeNames = [];
+for (const t in Types) {
+  typeNames[Types[t]] = t;
+}
+
+function readCodeFor(type, charset, encodingExpr, config, options) {
+  const supportBigNumbers =
+    options.supportBigNumbers || config.supportBigNumbers;
+  const bigNumberStrings = options.bigNumberStrings || config.bigNumberStrings;
+  const timezone = options.timezone || config.timezone;
+  const dateStrings = options.dateStrings || config.dateStrings;
+
+  switch (type) {
+    case Types.TINY:
+    case Types.SHORT:
+    case Types.LONG:
+    case Types.INT24:
+    case Types.YEAR:
+      return 'packet.parseLengthCodedIntNoBigCheck()';
+    case Types.LONGLONG:
+      if (supportBigNumbers && bigNumberStrings) {
+        return 'packet.parseLengthCodedIntString()';
+      }
+      return `packet.parseLengthCodedInt(${supportBigNumbers})`;
+    case Types.FLOAT:
+    case Types.DOUBLE:
+      return 'packet.parseLengthCodedFloat()';
+    case Types.NULL:
+      return 'packet.readLengthCodedNumber()';
+    case Types.DECIMAL:
+    case Types.NEWDECIMAL:
+      if (config.decimalNumbers) {
+        return 'packet.parseLengthCodedFloat()';
+      }
+      return 'packet.readLengthCodedString("ascii")';
+    case Types.DATE:
+      if (helpers.typeMatch(type, dateStrings, Types)) {
+        return 'packet.readLengthCodedString("ascii")';
+      }
+      return `packet.parseDate('${timezone}')`;
+    case Types.DATETIME:
+    case Types.TIMESTAMP:
+      if (helpers.typeMatch(type, dateStrings, Types)) {
+        return 'packet.readLengthCodedString("ascii")';
+      }
+      return `packet.parseDateTime('${timezone}')`;
+    case Types.TIME:
+      return 'packet.readLengthCodedString("ascii")';
+    case Types.GEOMETRY:
+      return 'packet.parseGeometryValue()';
+    case Types.JSON:
+      // Since for JSON columns mysql always returns charset 63 (BINARY),
+      // we have to handle it according to JSON specs and use "utf8",
+      // see https://github.com/sidorares/node-mysql2/issues/409
+      return 'JSON.parse(packet.readLengthCodedString("utf8"))';
+    default:
+      if (charset === Charsets.BINARY) {
+        return 'packet.readLengthCodedBuffer()';
+      }
+      return `packet.readLengthCodedString(${encodingExpr})`;
+  }
+}
+
+function compile(fields, options, config) {
+  const parserFn = genFunc();
+
+  /* eslint-disable no-trailing-spaces */
+  /* eslint-disable no-spaced-func */
+  /* eslint-disable no-unexpected-multiline */
+  parserFn('(function () {')('return class TextRow {');
+
+  // next method
+  parserFn('next(packet, fields, options) {');
+  parserFn('return [');
+
+  for (let i = 0; i < fields.length; i++) {
+    const encodingExpr = `fields[${i}].encoding`;
+    const readCode = readCodeFor(
+      fields[i].columnType,
+      fields[i].characterSet,
+      encodingExpr,
+      config,
+      options
+    );
+    parserFn(`${readCode},`);
+  }
+
+  parserFn('];');
+  parserFn('}');
+  parserFn('};')('})()');
+
+  /* eslint-enable no-trailing-spaces */
+  /* eslint-enable no-spaced-func */
+  /* eslint-enable no-unexpected-multiline */
+
+  if (config.debug) {
+    helpers.printDebugWithCode(
+      'Compiled text protocol row parser',
+      parserFn.toString()
+    );
+  }
+  return parserFn.toFunction();
+}
+
+function getTextParser(fields, options, config) {
+  return parserCache.getParser('fast', fields, options, config, compile);
+}
+
+module.exports = getTextParser;