changed query workflow for duckdb

packages/api/src/controllers/databaseConnections.js

@@ -39,6 +39,8 @@ const axios = require('axios');
 const { callTextToSqlApi, callCompleteOnCursorApi, callRefactorSqlQueryApi } = require('../utility/authProxy');
 const { decryptConnection } = require('../utility/crypting');
 const { getSshTunnel } = require('../utility/sshTunnel');
+const sessions = require('./sessions');
+const jsldata = require('./jsldata');
 
 const logger = getLogger('databaseConnections');
 
@@ -96,6 +98,33 @@ module.exports = {
 
   handle_ping() {},
 
+  // session event handlers
+
+  handle_info(conid, database, props) {
+    const { sesid, info } = props;
+    sessions.dispatchMessage(sesid, info);
+  },
+
+  handle_done(conid, database, props) {
+    const { sesid } = props;
+    socket.emit(`session-done-${sesid}`);
+    sessions.dispatchMessage(sesid, 'Query execution finished');
+  },
+
+  handle_recordset(conid, database, props) {
+    const { jslid, resultIndex } = props;
+    socket.emit(`session-recordset-${props.sesid}`, { jslid, resultIndex });
+  },
+
+  handle_stats(conid, database, stats) {
+    jsldata.notifyChangedStats(stats);
+  },
+
+  handle_initializeFile(conid, database, props) {
+    const { jslid } = props;
+    socket.emit(`session-initialize-file-${jslid}`);
+  },
+
   async ensureOpened(conid, database) {
     const existing = this.opened.find(x => x.conid == conid && x.database == database);
     if (existing) return existing;
@@ -763,4 +792,15 @@ module.exports = {
       commandLine: this.commandArgsToCommandLine(commandArgs),
     };
   },
+
+  executeSessionQuery_meta: true,
+  async executeSessionQuery({ sesid, conid, database, sql }) {
+    logger.info({ sesid, sql }, 'Processing query');
+    sessions.dispatchMessage(sesid, 'Query execution started');
+
+    const opened = await this.ensureOpened(conid, database);
+    opened.subprocess.send({ msgtype: 'executeSessionQuery', sql, sesid });
+
+    return { state: 'ok' };
+  },
 };

packages/api/src/proc/databaseConnectionProcess.js

@@ -16,6 +16,7 @@ const { handleProcessCommunication } = require('../utility/processComm');
 const { SqlGenerator } = require('dbgate-tools');
 const generateDeploySql = require('../shell/generateDeploySql');
 const { dumpSqlSelect } = require('dbgate-sqltree');
+const { allowExecuteCustomScript, handleQueryStream } = require('../utility/handleQueryStream');
 
 const logger = getLogger('dbconnProcess');
 
@@ -375,6 +376,36 @@ async function handleGenerateDeploySql({ msgid, modelFolder }) {
   }
 }
 
+async function handleExecuteSessionQuery({ sesid, sql }) {
+  await waitConnected();
+  const driver = requireEngineDriver(storedConnection);
+
+  if (!allowExecuteCustomScript(storedConnection, driver)) {
+    process.send({
+      msgtype: 'info',
+      info: {
+        message: 'Connection without read-only sessions is read only',
+        severity: 'error',
+      },
+      sesid,
+    });
+    process.send({ msgtype: 'done', sesid, skipFinishedMessage: true });
+    return;
+    //process.send({ msgtype: 'error', error: e.message });
+  }
+
+  const resultIndexHolder = {
+    value: 0,
+  };
+  for (const sqlItem of splitQuery(sql, {
+    ...driver.getQuerySplitterOptions('stream'),
+    returnRichInfo: true,
+  })) {
+    await handleQueryStream(dbhan, driver, resultIndexHolder, sqlItem, sesid);
+  }
+  process.send({ msgtype: 'done', sesid });
+}
+
 // async function handleRunCommand({ msgid, sql }) {
 //   await waitConnected();
 //   const driver = engines(storedConnection);
@@ -405,6 +436,7 @@ const messageHandlers = {
   sqlSelect: handleSqlSelect,
   exportKeys: handleExportKeys,
   schemaList: handleSchemaList,
+  executeSessionQuery: handleExecuteSessionQuery,
   // runCommand: handleRunCommand,
 };
 

packages/api/src/proc/sessionProcess.js

@@ -11,7 +11,7 @@ const { decryptConnection } = require('../utility/crypting');
 const { connectUtility } = require('../utility/connectUtility');
 const { handleProcessCommunication } = require('../utility/processComm');
 const { getLogger, extractIntSettingsValue, extractBoolSettingsValue } = require('dbgate-tools');
-const { handleQueryStream, QueryStreamTableWriter } = require('../utility/handleQueryStream');
+const { handleQueryStream, QueryStreamTableWriter, allowExecuteCustomScript } = require('../utility/handleQueryStream');
 
 const logger = getLogger('sessionProcess');
 
@@ -24,17 +24,6 @@ let lastActivity = null;
 let currentProfiler = null;
 let executingScripts = 0;
 
-function allowExecuteCustomScript(driver) {
-  if (driver.readOnlySessions) {
-    return true;
-  }
-  if (storedConnection.isReadOnly) {
-    return false;
-    // throw new Error('Connection is read only');
-  }
-  return true;
-}
-
 async function handleConnect(connection) {
   storedConnection = connection;
 
@@ -65,7 +54,7 @@ async function handleStartProfiler({ jslid }) {
   await waitConnected();
   const driver = requireEngineDriver(storedConnection);
 
-  if (!allowExecuteCustomScript(driver)) {
+  if (!allowExecuteCustomScript(storedConnection, driver)) {
     process.send({ msgtype: 'done' });
     return;
   }
@@ -94,7 +83,7 @@ async function handleExecuteControlCommand({ command }) {
   await waitConnected();
   const driver = requireEngineDriver(storedConnection);
 
-  if (command == 'commitTransaction' && !allowExecuteCustomScript(driver)) {
+  if (command == 'commitTransaction' && !allowExecuteCustomScript(storedConnection, driver)) {
     process.send({
       msgtype: 'info',
       info: {
@@ -134,7 +123,7 @@ async function handleExecuteQuery({ sql, autoCommit }) {
   await waitConnected();
   const driver = requireEngineDriver(storedConnection);
 
-  if (!allowExecuteCustomScript(driver)) {
+  if (!allowExecuteCustomScript(storedConnection, driver)) {
     process.send({
       msgtype: 'info',
       info: {
@@ -178,7 +167,7 @@ async function handleExecuteReader({ jslid, sql, fileName }) {
   if (fileName) {
     sql = fs.readFileSync(fileName, 'utf-8');
   } else {
-    if (!allowExecuteCustomScript(driver)) {
+    if (!allowExecuteCustomScript(storedConnection, driver)) {
       process.send({ msgtype: 'done' });
       return;
     }

packages/api/src/utility/handleQueryStream.js

@@ -6,10 +6,11 @@ const _ = require('lodash');
 const { jsldir } = require('../utility/directories');
 
 class QueryStreamTableWriter {
-  constructor() {
+  constructor(sesid = undefined) {
     this.currentRowCount = 0;
     this.currentChangeIndex = 1;
     this.initializedFile = false;
+    this.sesid = sesid;
   }
 
   initializeFromQuery(structure, resultIndex) {
@@ -26,7 +27,7 @@ class QueryStreamTableWriter {
     this.writeCurrentStats(false, false);
     this.resultIndex = resultIndex;
     this.initializedFile = true;
-    process.send({ msgtype: 'recordset', jslid: this.jslid, resultIndex });
+    process.send({ msgtype: 'recordset', jslid: this.jslid, resultIndex, sesid: this.sesid });
   }
 
   initializeFromReader(jslid) {
@@ -52,7 +53,7 @@ class QueryStreamTableWriter {
 
   rowFromReader(row) {
     if (!this.initializedFile) {
-      process.send({ msgtype: 'initializeFile', jslid: this.jslid });
+      process.send({ msgtype: 'initializeFile', jslid: this.jslid, sesid: this.sesid });
       this.initializedFile = true;
 
       fs.writeFileSync(this.currentFile, JSON.stringify(row) + '\n');
@@ -75,7 +76,7 @@ class QueryStreamTableWriter {
     fs.writeFileSync(`${this.currentFile}.stats`, JSON.stringify(stats));
     this.currentChangeIndex += 1;
     if (emitEvent) {
-      process.send({ msgtype: 'stats', ...stats });
+      process.send({ msgtype: 'stats', sesid: this.sesid, ...stats });
     }
   }
 
@@ -90,9 +91,10 @@ class QueryStreamTableWriter {
 }
 
 class StreamHandler {
-  constructor(resultIndexHolder, resolve, startLine) {
+  constructor(resultIndexHolder, resolve, startLine, sesid = undefined) {
     this.recordset = this.recordset.bind(this);
     this.startLine = startLine;
+    this.sesid = sesid;
     this.row = this.row.bind(this);
     // this.error = this.error.bind(this);
     this.done = this.done.bind(this);
@@ -116,7 +118,7 @@ class StreamHandler {
 
   recordset(columns) {
     this.closeCurrentWriter();
-    this.currentWriter = new QueryStreamTableWriter();
+    this.currentWriter = new QueryStreamTableWriter(this.sesid);
     this.currentWriter.initializeFromQuery(
       Array.isArray(columns) ? { columns } : columns,
       this.resultIndexHolder.value
@@ -133,7 +135,7 @@ class StreamHandler {
   }
   row(row) {
     if (this.currentWriter) this.currentWriter.row(row);
-    else if (row.message) process.send({ msgtype: 'info', info: { message: row.message } });
+    else if (row.message) process.send({ msgtype: 'info', info: { message: row.message }, sesid: this.sesid });
     // this.onRow(this.jslid);
   }
   // error(error) {
@@ -151,19 +153,31 @@ class StreamHandler {
         line: this.startLine + info.line,
       };
     }
-    process.send({ msgtype: 'info', info });
+    process.send({ msgtype: 'info', info, sesid: this.sesid });
   }
 }
 
-function handleQueryStream(dbhan, driver, resultIndexHolder, sqlItem) {
+function handleQueryStream(dbhan, driver, resultIndexHolder, sqlItem, sesid = undefined) {
   return new Promise((resolve, reject) => {
     const start = sqlItem.trimStart || sqlItem.start;
-    const handler = new StreamHandler(resultIndexHolder, resolve, start && start.line);
+    const handler = new StreamHandler(resultIndexHolder, resolve, start && start.line, sesid);
     driver.stream(dbhan, sqlItem.text, handler);
   });
 }
 
+function allowExecuteCustomScript(storedConnection, driver) {
+  if (driver.readOnlySessions) {
+    return true;
+  }
+  if (storedConnection.isReadOnly) {
+    return false;
+    // throw new Error('Connection is read only');
+  }
+  return true;
+}
+
 module.exports = {
   handleQueryStream,
   QueryStreamTableWriter,
+  allowExecuteCustomScript,
 };