fix: MS SQL has unusual CTAS syntax

Fixes #185

Author: paveltiunov

.circleci/config.yml

@@ -6,12 +6,18 @@ jobs:
         environment:
           TEST_PG_USER: root
           TEST_CLICKHOUSE_HOST: localhost
+          TEST_LOCAL: true
+          TEST_DB_PASSWORD: Test1test
       - image: circleci/redis:5.0.5
       - image: circleci/postgres:9.6.8
         environment:
           POSTGRES_USER: root
           POSTGRES_DB: model_test
-      - image: yandex/clickhouse-server
+      - image: yandex/clickhouse-server:2017-latest
+        environment:
+          ACCEPT_EULA: Y
+          SA_PASSWORD: Test1test
+
 
     working_directory: ~/repo
 

packages/cubejs-schema-compiler/adapter/MssqlQuery.js

@@ -1,19 +1,19 @@
 const moment = require('moment-timezone');
 const R = require('ramda');
 
-var abbrs = {
-  EST : 'Eastern Standard Time',
-  EDT : 'Eastern Standard Time',
-  CST : 'Central Standard Time',
-  CDT : 'Central Standard Time',
-  MST : 'Mountain Standard Time',
-  MDT : 'Mountain Standard Time',
-  PST : 'Pacific Standard Time',
-  PDT : 'Pacific Standard Time',
+const abbrs = {
+  EST: 'Eastern Standard Time',
+  EDT: 'Eastern Standard Time',
+  CST: 'Central Standard Time',
+  CDT: 'Central Standard Time',
+  MST: 'Mountain Standard Time',
+  MDT: 'Mountain Standard Time',
+  PST: 'Pacific Standard Time',
+  PDT: 'Pacific Standard Time',
 };
 
 moment.fn.zoneName = function () {
-  var abbr = this.zoneAbbr();
+  const abbr = this.zoneAbbr();
   return abbrs[abbr] || abbr;
 };
 
@@ -39,9 +39,7 @@ const GRANULARITY_TO_INTERVAL = {
 class MssqlFilter extends BaseFilter {
   // noinspection JSMethodCanBeStatic
   escapeWildcardChars(param) {
-    return typeof param === 'string'
-      ? param.replace(/([_%])/gi, '[$1]')
-      : param;
+    return typeof param === 'string' ? param.replace(/([_%])/gi, '[$1]') : param;
   }
 
   likeIgnoreCase(column, not) {
@@ -84,9 +82,19 @@ class MssqlQuery extends BaseQuery {
 
   groupByClause() {
     const dimensionsForSelect = this.dimensionsForSelect();
-    const dimensionColumns = R.flatten(dimensionsForSelect.map(s => s.selectColumns() && s.dimensionSql())).filter(s => !!s);
+    const dimensionColumns = R.flatten(dimensionsForSelect.map(s => s.selectColumns() && s.dimensionSql()))
+      .filter(s => !!s);
     return dimensionColumns.length ? ` GROUP BY ${dimensionColumns.join(', ')}` : '';
   }
+
+  nowTimestampSql() {
+    return `CURRENT_TIMESTAMP`;
+  }
+
+  preAggregationLoadSql(cube, preAggregation, tableName) {
+    const sqlAndParams = this.preAggregationSql(cube, preAggregation);
+    return [`SELECT * INTO ${tableName} FROM (${sqlAndParams[0]}) AS PreAggregation`, sqlAndParams[1]];
+  }
 }
 
 module.exports = MssqlQuery;

packages/cubejs-schema-compiler/package.json

@@ -30,6 +30,7 @@
     "eslint-plugin-import": "^2.16.0",
     "eslint-plugin-node": "^5.2.1",
     "mocha": "^3.4.2",
+    "mssql": "^5.1.0",
     "pg-promise": "^7.3.2",
     "request": "^2.88.0",
     "request-promise": "^4.2.4",

packages/cubejs-schema-compiler/test/BaseDbRunner.js

@@ -0,0 +1,41 @@
+class BaseDbRunner {
+  testQuery(query, fixture) {
+    return this.testQueries([query], fixture);
+  }
+
+  async testQueries(queries, fixture) {
+    if (!this.container && !process.env.TEST_LOCAL) {
+      console.log(`Starting container`);
+      this.container = await this.containerLazyInit();
+    }
+    if (!this.connection) {
+      console.log(`Initializing connection`);
+      const port = this.container ? this.container.getMappedPort(this.port()) : this.port();
+      this.connection = await this.connectionLazyInit(port);
+    }
+    return this.connection.testQueries(queries, fixture);
+  }
+
+  async tearDown() {
+    if (this.container) {
+      await this.container.stop();
+      this.connection = null;
+      this.container = null;
+    }
+  }
+
+  // eslint-disable-next-line no-unused-vars
+  async connectionLazyInit(port) {
+    throw new Error('Not implemented');
+  }
+
+  async containerLazyInit() {
+    throw new Error('Not implemented');
+  }
+
+  port() {
+    throw new Error('Not implemented');
+  }
+}
+
+module.exports = BaseDbRunner;

packages/cubejs-schema-compiler/test/MSSqlDbRunner.js

@@ -0,0 +1,100 @@
+const { GenericContainer, Wait } = require("testcontainers");
+const sql = require('mssql');
+const BaseDbRunner = require('./BaseDbRunner');
+
+class MSSqlDbRunner extends BaseDbRunner {
+  async connectionLazyInit(port) {
+    return {
+      testQueries: async (queries, fixture) => {
+        const pool = new sql.ConnectionPool({
+          server: 'localhost',
+          port,
+          user: 'sa',
+          password: this.password()
+        });
+
+        await pool.connect();
+
+        try {
+          const tx = new sql.Transaction(pool);
+          await tx.begin();
+          try {
+            await this.prepareFixture(tx, fixture);
+            const result = await queries.map(query => async () => {
+              const request = new sql.Request(tx);
+              (query[1] || []).forEach((v, i) => request.input(`_${i + 1}`, v));
+              return (await request.query(query[0])).recordset;
+            }).reduce((a, b) => a.then(b), Promise.resolve());
+            await tx.commit();
+            return result;
+          } catch (e) {
+            // console.log(e.stack);
+            await tx.rollback();
+            throw e;
+          }
+        } finally {
+          await pool.close();
+        }
+      }
+    };
+  }
+
+  async prepareFixture(tx) {
+    const query = async (q) => {
+      const request = new sql.Request(tx);
+      await request.query(q);
+    };
+    await query('CREATE TABLE ##visitors (id INT, amount INT, created_at datetime, updated_at datetime, status INT, source VARCHAR(MAX), latitude DECIMAL, longitude DECIMAL)');
+    await query('CREATE TABLE ##visitor_checkins (id INT, visitor_id INT, created_at datetime, source VARCHAR(MAX))');
+    await query('CREATE TABLE ##cards (id INT, visitor_id INT, visitor_checkin_id INT)');
+    await query(`
+    INSERT INTO
+    ##visitors
+    (id, amount, created_at, updated_at, status, source, latitude, longitude) VALUES
+    (1, 100, '2017-01-03', '2017-01-30', 1, 'some', 120.120, 40.60),
+    (2, 200, '2017-01-05', '2017-01-15', 1, 'some', 120.120, 58.60),
+    (3, 300, '2017-01-06', '2017-01-20', 2, 'google', 120.120, 70.60),
+    (4, 400, '2017-01-07', '2017-01-25', 2, NULL, 120.120, 10.60),
+    (5, 500, '2017-01-07', '2017-01-25', 2, NULL, 120.120, 58.10),
+    (6, 500, '2016-09-07', '2016-09-07', 2, NULL, 120.120, 58.10)
+    `);
+    await query(`
+    INSERT INTO
+    ##visitor_checkins
+    (id, visitor_id, created_at, source) VALUES
+    (1, 1, '2017-01-03', NULL),
+    (2, 1, '2017-01-04', NULL),
+    (3, 1, '2017-01-05', 'google'),
+    (4, 2, '2017-01-05', NULL),
+    (5, 2, '2017-01-05', NULL),
+    (6, 3, '2017-01-06', NULL)
+    `);
+    await query(`
+    INSERT INTO
+    ##cards
+    (id, visitor_id, visitor_checkin_id) VALUES
+    (1, 1, 1),
+    (2, 1, 2),
+    (3, 3, 6)
+    `);
+  }
+
+  password() {
+    return process.env.TEST_DB_PASSWORD || 'Test1test';
+  }
+
+  async containerLazyInit() {
+    return new GenericContainer("mcr.microsoft.com/mssql/server", '2017-latest')
+      .withEnv("ACCEPT_EULA", "Y")
+      .withEnv("SA_PASSWORD", this.password())
+      .withExposedPorts(this.port())
+      .withWaitStrategy(Wait.forLogMessage("Server is listening on"))
+      .start();
+  }
+
+  port() {
+    return 1433;
+  }
+}
+
+module.exports = new MSSqlDbRunner();