sqlite stream

plugins/dbgate-plugin-sqlite/package.json

@@ -25,6 +25,7 @@
       "dbgate-tools": "^4.1.1",
       "dbgate-plugin-tools": "^1.0.4",
       "byline": "^5.0.0",
+      "sql-query-identifier": "^2.1.0",
       "webpack": "^4.42.0",
       "webpack-cli": "^3.3.11"
     }

plugins/dbgate-plugin-sqlite/src/backend/driver.js

@@ -2,9 +2,54 @@ const _ = require('lodash');
 const stream = require('stream');
 const driverBase = require('../frontend/driver');
 const Analyser = require('./Analyser');
+const { identify } = require('sql-query-identifier');
 
 let Database;
 
+async function runStreamItem(client, sql, options) {
+  return new Promise((resolve, reject) => {
+    try {
+      const stmt = client.prepare(sql);
+      if (stmt.reader) {
+        const columns = stmt.columns();
+        // const rows = stmt.all();
+
+        options.recordset(
+          columns.map((col) => ({
+            columnName: col.name,
+            dataType: col.type,
+          }))
+        );
+
+        for (const row of stmt.iterate()) {
+          options.row(row);
+        }
+
+        resolve();
+      } else {
+        const info = stmt.run();
+        options.info({
+          message: `${info.changes} rows affected`,
+          time: new Date(),
+          severity: 'info',
+        });
+        resolve();
+      }
+    } catch (error) {
+      console.log('ERROR', error);
+      const { message, lineNumber, procName } = error;
+      options.info({
+        message,
+        line: lineNumber,
+        procedure: procName,
+        time: new Date(),
+        severity: 'error',
+      });
+      resolve();
+    }
+  });
+}
+
 /** @type {import('dbgate-types').EngineDriver} */
 const driver = {
   ...driverBase,
@@ -15,7 +60,6 @@ const driver = {
   },
   // @ts-ignore
   async query(pool, sql) {
-    console.log('SQLITE SQL', sql);
     const stmt = pool.prepare(sql);
     // stmt.raw();
     const columns = stmt.columns();
@@ -28,8 +72,15 @@ const driver = {
       })),
     };
   },
-  async stream(pool, sql, options) {
-    return null;
+  async stream(client, sql, options) {
+    const sqlSplitted = identify(sql, { dialect: 'sqlite' });
+
+    for (const sqlItem of sqlSplitted) {
+      await runStreamItem(client, sqlItem.text, options);
+    }
+
+    options.done();
+    // return stream;
   },
   async readQuery(pool, sql, structure) {
     const pass = new stream.PassThrough({