diff --git a/packages/server/__mocks__/mysql.js b/packages/server/__mocks__/mysql.js
index 2b4df3e44b..da0126e88b 100644
--- a/packages/server/__mocks__/mysql.js
+++ b/packages/server/__mocks__/mysql.js
@@ -2,7 +2,9 @@ const mysql = {}
 
 const client = {
   connect: jest.fn(),
-  query: jest.fn(),
+  query: jest.fn((query, fn) => {
+    fn(null, [])
+  }),
 }
 
 mysql.createConnection = jest.fn(() => client)
diff --git a/packages/server/src/api/controllers/datasource.js b/packages/server/src/api/controllers/datasource.js
index 346a15ab89..75a0762d97 100644
--- a/packages/server/src/api/controllers/datasource.js
+++ b/packages/server/src/api/controllers/datasource.js
@@ -6,7 +6,7 @@ const {
 } = require("../../db/utils")
 const { integrations } = require("../../integrations")
 const plusIntegrations = require("../../integrations/plus")
-const PostgresConnector = require("../../integrations/plus/Postgres")
+const PostgresConnector = require("../../integrations/plus/postgres")
 
 exports.fetch = async function (ctx) {
   const database = new CouchDB(ctx.appId)
@@ -63,6 +63,11 @@ exports.find = async function (ctx) {
   ctx.body = await database.get(ctx.params.datasourceId)
 }
 
+// dynamic query functionality
+exports.query = async function (ctx) {
+
+}
+
 // TODO: merge endpoint with main datasource endpoint
 exports.plus = async function (ctx) {
   const db = new CouchDB(ctx.appId)
diff --git a/packages/server/src/integrations/base/sql.js b/packages/server/src/integrations/base/sql.js
index a0fe9bd0eb..0d18f04e66 100644
--- a/packages/server/src/integrations/base/sql.js
+++ b/packages/server/src/integrations/base/sql.js
@@ -102,12 +102,17 @@ class SqlQueryBuilder {
     this._limit = limit
   }
 
+  _operation(json) {
+    if (!json || !json.endpoint) {
+      return null
+    }
+    return json.endpoint.operation
+  }
+
   _query(json) {
-    const { endpoint } = json
     const knex = require("knex")({ client: this._client })
-    const operation = endpoint.operation
     let query
-    switch (operation) {
+    switch (this._operation(json)) {
       case Operation.CREATE:
         query = buildCreate(knex, json)
         break
diff --git a/packages/server/src/integrations/microsoftSqlServer.js b/packages/server/src/integrations/microsoftSqlServer.js
index bece6678a3..8228006b5d 100644
--- a/packages/server/src/integrations/microsoftSqlServer.js
+++ b/packages/server/src/integrations/microsoftSqlServer.js
@@ -51,6 +51,15 @@ const SCHEMA = {
   },
 }
 
+async function internalQuery(client, sql) {
+  try {
+    return await client.query(sql)
+  } catch (err) {
+    throw new Error(err)
+  }
+}
+
+
 class SqlServerIntegration extends Sql {
   static pool
 
@@ -67,52 +76,43 @@ class SqlServerIntegration extends Sql {
   }
 
   async connect() {
-    const client = await this.pool.connect()
-    this.client = client.request()
+    try {
+      const client = await this.pool.connect()
+      this.client = client.request()
+    } catch (err) {
+      throw new Error(err)
+    }
   }
 
   async read(query) {
-    try {
-      await this.connect()
-      const response = await this.client.query(query.sql)
-      return response.recordset
-    } catch (err) {
-      console.error("Error querying MS SQL Server", err)
-      throw err
-    }
+    await this.connect()
+    const response = await internalQuery(this.client, query.sql)
+    return response.recordset
   }
 
   async create(query) {
-    try {
-      await this.connect()
-      const response = await this.client.query(query.sql)
-      return response.recordset || [{ created: true }]
-    } catch (err) {
-      console.error("Error querying MS SQL Server", err)
-      throw err
-    }
+    await this.connect()
+    const response = await internalQuery(this.client, query.sql)
+    return response.recordset || [{ created: true }]
   }
 
   async update(query) {
-    try {
-      await this.connect()
-      const response = await this.client.query(query.sql)
-      return response.recordset
-    } catch (err) {
-      console.error("Error querying MS SQL Server", err)
-      throw err
-    }
+    await this.connect()
+    const response = await internalQuery(this.client, query.sql)
+    return response.recordset || [{ updated: true }]
   }
 
   async delete(query) {
-    try {
-      await this.connect()
-      const response = await this.client.query(query.sql)
-      return response.recordset
-    } catch (err) {
-      console.error("Error querying MS SQL Server", err)
-      throw err
-    }
+    await this.connect()
+    const response = await internalQuery(this.client, query.sql)
+    return response.recordset || [{ deleted: true }]
+  }
+
+  async query(json) {
+    const operation = this._operation(json).toLowerCase()
+    const input = this._query(json)
+    const response = await internalQuery(this.client, input)
+    return response.recordset ? response.recordset : [{ [operation]: true }]
   }
 }
 
diff --git a/packages/server/src/integrations/mysql.js b/packages/server/src/integrations/mysql.js
index 6c72e05a12..b54aedd7fa 100644
--- a/packages/server/src/integrations/mysql.js
+++ b/packages/server/src/integrations/mysql.js
@@ -53,6 +53,21 @@ const SCHEMA = {
   },
 }
 
+function internalQuery(client, query) {
+  // Node MySQL is callback based, so we must wrap our call in a promise
+  return new Promise((resolve, reject) => {
+    client.connect()
+    return client.query(query, (error, results) => {
+      if (error) {
+        reject(error)
+      } else {
+        resolve(results)
+        client.end()
+      }
+    })
+  })
+}
+
 class MySQLIntegration extends Sql {
   constructor(config) {
     super("mysql")
@@ -63,36 +78,31 @@ class MySQLIntegration extends Sql {
     this.client = mysql.createConnection(config)
   }
 
-  query(query) {
-    // Node MySQL is callback based, so we must wrap our call in a promise
-    return new Promise((resolve, reject) => {
-      this.client.connect()
-      return this.client.query(query.sql, (error, results) => {
-        if (error) return reject(error)
-        resolve(results)
-        this.client.end()
-      })
-    })
-  }
-
   async create(query) {
-    const results = await this.query(query)
+    const results = await internalQuery(this.client, query.sql)
     return results.length ? results : [{ created: true }]
   }
 
   read(query) {
-    return this.query(query)
+    return internalQuery(this.client, query.sql)
   }
 
   async update(query) {
-    const results = await this.query(query)
+    const results = await internalQuery(this.client, query.sql)
     return results.length ? results : [{ updated: true }]
   }
 
   async delete(query) {
-    const results = await this.query(query)
+    const results = await internalQuery(this.client, query.sql)
     return results.length ? results : [{ deleted: true }]
   }
+
+  async query(json) {
+    const operation = this._operation(json).toLowerCase()
+    const input = this._query(json)
+    const results = await internalQuery(this.client, input)
+    return results.length ? results : [{ [operation]: true }]
+  }
 }
 
 module.exports = {
diff --git a/packages/server/src/integrations/postgres.js b/packages/server/src/integrations/postgres.js
index 2a39a91c2d..d9ac63b431 100644
--- a/packages/server/src/integrations/postgres.js
+++ b/packages/server/src/integrations/postgres.js
@@ -55,6 +55,15 @@ const SCHEMA = {
   },
 }
 
+async function internalQuery(client, sql) {
+  try {
+    return await client.query(sql)
+  } catch (err) {
+    throw new Error(err)
+  }
+}
+
+
 class PostgresIntegration extends Sql {
   static pool
 
@@ -74,33 +83,32 @@ class PostgresIntegration extends Sql {
     this.client = this.pool
   }
 
-  async query(sql) {
-    try {
-      return await this.client.query(sql)
-    } catch (err) {
-      throw new Error(err)
-    }
-  }
-
   async create({ sql }) {
-    const response = await this.query(sql)
+    const response = await internalQuery(this.client, sql)
     return response.rows.length ? response.rows : [{ created: true }]
   }
 
   async read({ sql }) {
-    const response = await this.query(sql)
+    const response = await internalQuery(this.client, sql)
     return response.rows
   }
 
   async update({ sql }) {
-    const response = await this.query(sql)
+    const response = await internalQuery(this.client, sql)
     return response.rows.length ? response.rows : [{ updated: true }]
   }
 
   async delete({ sql }) {
-    const response = await this.query(sql)
+    const response = await internalQuery(this.client, sql)
     return response.rows.length ? response.rows : [{ deleted: true }]
   }
+
+  async query(json) {
+    const operation = this._operation(json).toLowerCase()
+    const input = this._query(json)
+    const response = await internalQuery(this.client, input)
+    return response.rows.length ? response.rows : [{ [operation]: true }]
+  }
 }
 
 module.exports = {
diff --git a/packages/server/src/integrations/tests/mysql.spec.js b/packages/server/src/integrations/tests/mysql.spec.js
index eca3e523b0..71b828a4e6 100644
--- a/packages/server/src/integrations/tests/mysql.spec.js
+++ b/packages/server/src/integrations/tests/mysql.spec.js
@@ -1,12 +1,9 @@
-const pg = require("mysql")
 const MySQLIntegration = require("../mysql")
 jest.mock("mysql")
 
 class TestConfiguration {
   constructor(config = { ssl: {} }) {
-    this.integration = new MySQLIntegration.integration(config) 
-    this.query = jest.fn(() => [{ id: 1 }])
-    this.integration.query = this.query
+    this.integration = new MySQLIntegration.integration(config)
   }
 }
 
@@ -19,43 +16,37 @@ describe("MySQL Integration", () => {
 
   it("calls the create method with the correct params", async () => {
     const sql = "insert into users (name, age) values ('Joe', 123);"
-    const response = await config.integration.create({ 
+    await config.integration.create({
       sql
     })
-    expect(config.query).toHaveBeenCalledWith({ sql })
+    expect(config.integration.client.query).toHaveBeenCalledWith(sql, expect.any(Function))
   })
 
   it("calls the read method with the correct params", async () => {
     const sql = "select * from users;"
-    const response = await config.integration.read({ 
-      sql
-    })
-    expect(config.query).toHaveBeenCalledWith({
+    await config.integration.read({
       sql
     })
+    expect(config.integration.client.query).toHaveBeenCalledWith(sql, expect.any(Function))
   })
 
   it("calls the update method with the correct params", async () => {
     const sql = "update table users set name = 'test';"
-    const response = await config.integration.update({ 
+    await config.integration.update({
       sql
     })
-    expect(config.query).toHaveBeenCalledWith({ sql })
+    expect(config.integration.client.query).toHaveBeenCalledWith(sql, expect.any(Function))
   })
 
   it("calls the delete method with the correct params", async () => {
     const sql = "delete from users where name = 'todelete';"
-    const response = await config.integration.delete({ 
+    await config.integration.delete({
       sql
     })
-    expect(config.query).toHaveBeenCalledWith({ sql })
+    expect(config.integration.client.query).toHaveBeenCalledWith(sql, expect.any(Function))
   })
 
   describe("no rows returned", () => {
-    beforeEach(() => {
-      config.query.mockImplementation(() => [])
-    })
-
     it("returns the correct response when the create response has no rows", async () => {
     const sql = "insert into users (name, age) values ('Joe', 123);"
       const response = await config.integration.create({