Stream data from Memory functionality

Changelog.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## [1.0.2] - singlestore-nodejs (unreleased)
+
+### Features
+
+- **`infileStream` query option** — stream in-memory bytes for `LOAD DATA LOCAL INFILE` without reading a client file; SQL filename is ignored (aligned with SingleStore stream-based load documentation).
+- **`local_infile` connection option** — explicit enable/disable for `LOAD DATA LOCAL INFILE` (default enabled); clearer client error when disabled.
+- **Prepared statement / pool cluster** — connection-level `infileStreamFactory` is passed through `statement.execute()` and `PoolCluster.query()`.
+
+### Documentation
+
+- README and site docs describe both local-file (`infileStreamFactory`) and in-memory (`infileStream`) loads wherever `LOAD DATA LOCAL INFILE` is documented.
+
 ## [3.14.3](https://github.com/sidorares/node-mysql2/compare/v3.14.2...v3.14.3) (2025-07-29)
 
 

README.md

@@ -22,7 +22,8 @@
 [docs-mysql-server]: https://singlestore-labs.github.io/singlestore-nodejs/docs/documentation/mysql-server
 [docs-promise-wrapper]: https://singlestore-labs.github.io/singlestore-nodejs/documentation/promise-wrapper
 [docs-authentication-switch]: https://singlestore-labs.github.io/singlestore-nodejs/docs/documentation/authentication-switch
-[docs-streams]: https://sidorares.github.io/node-mysql2/docs/documentation/extras
+[docs-load-data]: https://singlestore-labs.github.io/singlestore-nodejs/docs/documentation/extras#load-data-local-infile
+[docs-streams]: https://singlestore-labs.github.io/singlestore-nodejs/docs/documentation/extras#connecting-using-custom-stream
 [docs-typescript-docs]: https://sidorares.github.io/node-mysql2/docs/documentation/typescript-examples
 [docs-qs-pooling]: https://singlestore-labs.github.io/singlestore-nodejs/docs#using-connection-pools
 [docs-qs-first-query]: https://singlestore-labs.github.io/singlestore-nodejs/docs#first-query
@@ -73,6 +74,7 @@ MySQL2 is mostly API compatible with [Node MySQL][node-mysql] and supports major
 - [Promise Wrapper][docs-promise-wrapper]
 - Compression
 - SSL and [Authentication Switch][docs-authentication-switch]
+- [LOAD DATA LOCAL INFILE][docs-load-data] (local files and in-memory streams)
 - [Custom Streams][docs-streams]
 - [Pooling][docs-qs-pooling]
 
@@ -86,6 +88,7 @@ SingleStore Node.js Driver is mostly API compatible with [node-mysql][node-mysql
 - **Consistent Data Type Handling**: Handles SingleStore data types more consistently than generic MySQL drivers
 - **Configurable Connection Behavior**: Allows you to configure connection behavior by setting session variables
 - **MySQL Compatibility**: Maintains full compatibility with MySQL wire protocol
+- **Bulk load**: `LOAD DATA LOCAL INFILE` from client files (`infileStreamFactory`) or in-memory streams (`infileStream`)
 
 For more information about SingleStore, visit the [official documentation][singlestore-docs].
 
@@ -100,7 +103,7 @@ If you are using TypeScript, you will need to install `@types/node`.
 ## Quick Start
 
 ```javascript
-const singlestore = require('singlestore_nodejs');
+const singlestore = require('singlestore-nodejs');
 
 // Create connection
 const connection = singlestore.createConnection({
@@ -140,12 +143,23 @@ const mysql = require('mysql2');
 const connection = mysql.createConnection({...});
 
 // After (SingleStore Node.js Driver)
-const singlestore = require('singlestore_nodejs');
+const singlestore = require('singlestore-nodejs');
 const connection = singlestore.createConnection({...});
 
 // All existing queries work as-is!
 ```
 
+## Load data from local files and in-memory streams
+
+Use `LOAD DATA LOCAL INFILE` to bulk-load CSV (or other `LOAD DATA` formats) into SingleStore. The driver reads data on the **client** (your Node process), not from the database server disk.
+
+- **Local file:** set `infileStreamFactory` on the connection or query so the driver opens the path the server requests (for example `fs.createReadStream(path)`).
+- **In-memory stream:** pass `infileStream` on the query (a Node.js `Readable`). The filename in SQL is ignored; use a placeholder such as `':stream:'`. This matches [SingleStore stream-based load](https://docs.singlestore.com/db/v9.1/load-data/load-data-from-files/load-data-from-local-files/).
+
+`local_infile` defaults to `true`. Set `local_infile: false` to refuse file and stream loads on that connection.
+
+See [LOAD DATA LOCAL INFILE][docs-load-data] for examples.
+
 ## Documentation
 
 📚 **Full documentation:** [SingleStore Nodejs Driver][docs-base]

lib/commands/command.js

@@ -2,6 +2,7 @@
 
 const EventEmitter = require('events').EventEmitter;
 const Timers = require('timers');
+const { rewriteLocalInfileError } = require('./local_infile.js');
 
 class Command extends EventEmitter {
   constructor() {
@@ -28,6 +29,7 @@ class Command extends EventEmitter {
     if (packet && packet.isError()) {
       const err = packet.asError(connection.clientEncoding);
       err.sql = this.sql || this.query;
+      rewriteLocalInfileError(err, connection, err.sql);
       if (this.queryTimeout) {
         Timers.clearTimeout(this.queryTimeout);
         this.queryTimeout = null;

lib/commands/execute.js

@@ -1,6 +1,7 @@
 'use strict';
 
 const Command = require('./command.js');
+const { assignLocalInfileOptions } = require('./local_infile.js');
 const Query = require('./query.js');
 const Packets = require('../packets/index.js');
 
@@ -27,7 +28,7 @@ class Execute extends Command {
     this._resultIndex = 0;
     this._localStream = null;
     this._unpipeStream = function () {};
-    this._streamFactory = options.infileStreamFactory;
+    assignLocalInfileOptions(this, options);
     this._connection = null;
   }
 

lib/commands/local_infile.js

@@ -0,0 +1,61 @@
+'use strict';
+
+const LOCAL_INFILE_DISABLED_MESSAGE =
+  'Local infile is disabled by connector. Enable `local_infile` to allow local infile commands';
+
+const LOCAL_INFILE_DISABLED_ERRNOS = new Set([1148, 3948, 4166]);
+
+function localInfileAllowed(connection) {
+  return (
+    !connection || !connection.config || connection.config.localInfile !== false
+  );
+}
+
+function localInfileDisabledError() {
+  const err = new Error(LOCAL_INFILE_DISABLED_MESSAGE);
+  err.code = 'ER_CLIENT_LOCAL_FILES_DISABLED';
+  err.errno = 3948;
+  return err;
+}
+
+function assignLocalInfileOptions(command, options) {
+  const source = options || {};
+  command._streamFactory = source.infileStreamFactory;
+  command._infileStream = null;
+  if (source.infileStream !== undefined && source.infileStream !== null) {
+    if (
+      typeof source.infileStream.read !== 'function' ||
+      typeof source.infileStream.on !== 'function'
+    ) {
+      throw new TypeError('infileStream must be a Readable stream');
+    }
+    command._infileStream = source.infileStream;
+  }
+}
+
+function rewriteLocalInfileError(err, connection, sql) {
+  if (!err || localInfileAllowed(connection)) {
+    return err;
+  }
+  if (!LOCAL_INFILE_DISABLED_ERRNOS.has(err.errno)) {
+    return err;
+  }
+  // 1148 is a generic "command not allowed" code. Only rewrite it for LOCAL INFILE.
+  if (
+    err.errno === 1148 &&
+    !/load\s+(data\s+)?local\s+infile/i.test(sql || '')
+  ) {
+    return err;
+  }
+  err.message = LOCAL_INFILE_DISABLED_MESSAGE;
+  err.sqlMessage = LOCAL_INFILE_DISABLED_MESSAGE;
+  return err;
+}
+
+module.exports = {
+  LOCAL_INFILE_DISABLED_MESSAGE,
+  assignLocalInfileOptions,
+  localInfileAllowed,
+  localInfileDisabledError,
+  rewriteLocalInfileError,
+};

lib/commands/prepare.js

@@ -24,8 +24,17 @@ class PreparedStatementInfo {
       callback = parameters;
       parameters = [];
     }
-    return this._connection.addCommand(
-      new Execute({ statement: this, values: parameters }, callback)
+    const connection = this._connection;
+    return connection.addCommand(
+      new Execute(
+        {
+          statement: this,
+          sql: this.query,
+          values: parameters,
+          infileStreamFactory: connection.config.infileStreamFactory,
+        },
+        callback
+      )
     );
   }
 }

lib/commands/query.js

@@ -6,6 +6,11 @@ const Timers = require('timers');
 const Readable = require('stream').Readable;
 
 const Command = require('./command.js');
+const {
+  assignLocalInfileOptions,
+  localInfileAllowed,
+  localInfileDisabledError,
+} = require('./local_infile.js');
 const Packets = require('../packets/index.js');
 const getTextParser = require('../parsers/text_parser.js');
 const staticParser = require('../parsers/static_text_parser.js');
@@ -32,7 +37,7 @@ class Query extends Command {
     this._resultIndex = 0;
     this._localStream = null;
     this._unpipeStream = function () {};
-    this._streamFactory = options.infileStreamFactory;
+    assignLocalInfileOptions(this, options);
     this._connection = null;
   }
 
@@ -137,9 +142,29 @@ class Query extends Command {
   }
 
   _streamLocalInfile(connection, path) {
-    if (this._streamFactory) {
-      this._localStream = this._streamFactory(path);
+    if (!localInfileAllowed(connection)) {
+      const err = localInfileDisabledError();
+      err.sql = this.sql;
+      this._localStreamError = err;
+      connection.writePacket(EmptyPacket);
+      return this.infileOk;
+    }
+
+    // A direct stream ignores the filename the server asked for.
+    if (this._infileStream) {
+      this._localStream = this._infileStream;
     } else {
+      const config = connection.config || {};
+      const factory =
+        this._streamFactory === undefined
+          ? config.infileStreamFactory
+          : this._streamFactory;
+      if (factory) {
+        this._localStream = factory(path);
+      }
+    }
+
+    if (!this._localStream) {
       this._localStreamError = new Error(
         `As a result of LOCAL INFILE command server wants to read ${path} file, but as of v2.0 you must provide streamFactory option returning ReadStream.`
       );

lib/connection_config.js

@@ -32,6 +32,7 @@ const validOptions = {
   host: 1,
   insecureAuth: 1,
   infileStreamFactory: 1,
+  local_infile: 1,
   isServer: 1,
   keepAliveInitialDelay: 1,
   localAddress: 1,
@@ -179,6 +180,18 @@ class ConnectionConfig {
       ConnectionConfig.getDefaultFlags(options),
       options.flags || ''
     );
+    if (options.local_infile !== undefined) {
+      this.localInfile = ConnectionConfig.resolveLocalInfile(
+        options.local_infile
+      );
+      if (this.localInfile) {
+        this.clientFlags |= ClientConstants.LOCAL_FILES;
+      } else {
+        this.clientFlags &= ~ClientConstants.LOCAL_FILES;
+      }
+    } else {
+      this.localInfile = (this.clientFlags & ClientConstants.LOCAL_FILES) !== 0;
+    }
     // Default connection attributes
     // https://dev.mysql.com/doc/refman/8.0/en/performance-schema-connection-attribute-tables.html
     const defaultConnectAttributes = {
@@ -242,6 +255,22 @@ class ConnectionConfig {
     return flags;
   }
 
+  static resolveLocalInfile(value) {
+    if (typeof value === 'string') {
+      const normalized = value.toLowerCase();
+      if (
+        normalized === 'false' ||
+        normalized === '0' ||
+        normalized === 'off' ||
+        normalized === 'no'
+      ) {
+        return false;
+      }
+      return true;
+    }
+    return value !== false && value !== 0;
+  }
+
   static getDefaultFlags(options) {
     const defaultFlags = [
       'LONG_PASSWORD',

lib/pool_cluster.js

@@ -103,6 +103,13 @@ class PoolNamespace {
         }
         return;
       }
+      if (
+        query._streamFactory === undefined &&
+        conn.config &&
+        conn.config.infileStreamFactory
+      ) {
+        query._streamFactory = conn.config.infileStreamFactory;
+      }
       try {
         conn.query(query).once('end', () => {
           conn.release();

test/integration/connection/test-load-infile.test.cjs

@@ -3,7 +3,12 @@
 const common = require('../../common.test.cjs');
 const { assert } = require('poku');
 const fs = require('node:fs');
+const { Readable } = require('node:stream');
 const process = require('node:process');
+const mysql = require('../../../index.js');
+
+const LOCAL_INFILE_DISABLED_MESSAGE =
+  'Local infile is disabled by connector. Enable `local_infile` to allow local infile commands';
 
 if (`${process.env.MYSQL_CONNECTION_URL}`.includes('pscale_pw_')) {
   console.log('skipping test for planetscale');
@@ -76,6 +81,8 @@ const createMyStream = function () {
 };
 
 let streamResult;
+let memoryResult;
+let disabledErr;
 connection.query(
   {
     sql: sql,
@@ -87,7 +94,56 @@ connection.query(
       throw err;
     }
     streamResult = result;
-    connection.end();
+    connection.query(
+      {
+        sql: `LOAD DATA LOCAL INFILE ':stream:' INTO TABLE ${table} FIELDS TERMINATED BY ',' (id, title)`,
+        infileStream: Readable.from([
+          '31,From Memory\n',
+          '32,Still',
+          ' Memory\n',
+        ]),
+        infileStreamFactory: () => {
+          throw new Error(
+            'infileStreamFactory must not run when infileStream is set'
+          );
+        },
+      },
+      (memErr, memResult) => {
+        if (memErr) {
+          throw memErr;
+        }
+        memoryResult = memResult;
+        const disabled = mysql.createConnection({
+          ...common.config,
+          local_infile: false,
+        });
+        disabled.query(
+          [
+            `CREATE TEMPORARY TABLE \`${table}\` (`,
+            '`id` int(11) unsigned NOT NULL,',
+            '`title` varchar(255)',
+            ')',
+          ].join('\n'),
+          (createErr) => {
+            if (createErr) {
+              throw createErr;
+            }
+            disabled.query(
+              {
+                sql: `LOAD DATA LOCAL INFILE ':stream:' INTO TABLE ${table} FIELDS TERMINATED BY ',' (id, title)`,
+                infileStream: Readable.from(['99,should-not-load\n']),
+              },
+              (loadError) => {
+                disabledErr = loadError;
+                disabled.end(() => {
+                  connection.end();
+                });
+              }
+            );
+          }
+        );
+      }
+    );
   }
 );
 
@@ -104,4 +160,6 @@ process.on('exit', () => {
   assert.equal(loadResult.affectedRows, 0);
 
   assert.equal(streamResult.affectedRows, 2);
+  assert.equal(memoryResult.affectedRows, 2);
+  assert.equal(disabledErr.message, LOCAL_INFILE_DISABLED_MESSAGE);
 });