feat: External rollup implementation

examples/hn-insights/schema/Events.js

@@ -286,7 +286,8 @@ cube(`Events`, {
       partitionGranularity: `day`,
       refreshKey: {
         sql: `select current_timestamp`
-      }
+      },
+      external: true
     },
     leaderBoard: {
       type: `rollup`,
@@ -319,6 +320,7 @@ cube(`Events`, {
       refreshKey: {
         sql: `select current_timestamp`
       },
+      external: true
     }
   }
 });

packages/cubejs-mysql-driver/driver/MySqlDriver.js

@@ -3,6 +3,10 @@ const genericPool = require('generic-pool');
 const { promisify } = require('util');
 const BaseDriver = require('@cubejs-backend/query-orchestrator/driver/BaseDriver');
 
+const GenericTypeToMySql = {
+  'string': 'varchar(255)'
+};
+
 class MySqlDriver extends BaseDriver {
   constructor(config) {
     super();
@@ -114,6 +118,14 @@ class MySqlDriver extends BaseDriver {
   informationSchemaQuery() {
     return `${super.informationSchemaQuery()} AND columns.table_schema = '${this.config.database}'`
   }
+
+  quoteIdentifier(identifier) {
+    return `\`${identifier}\``;
+  }
+
+  fromGenericType(columnType) {
+    return GenericTypeToMySql[columnType] || super.fromGenericType(columnType);
+  }
 }
 
 module.exports = MySqlDriver;

packages/cubejs-query-orchestrator/driver/BaseDriver.js

@@ -10,6 +10,21 @@ const sortByKeys = (unordered) => {
   return ordered;
 };
 
+const DbTypeToGenericType = {
+  'timestamp without time zone': 'timestamp',
+  'integer': 'int',
+  'character varying': 'text',
+  'varchar': 'text',
+  'text': 'text',
+  'string': 'text',
+  'boolean': 'boolean',
+  'bigint': 'bigint',
+  'time': 'string',
+  'datetime': 'timestamp',
+  'date': 'date',
+  'double precision': 'decimal'
+};
+
 class BaseDriver {
   informationSchemaQuery() {
     return `
@@ -77,12 +92,66 @@ class BaseDriver {
   }
 
   param(/* paramIndex */) {
-    return '?'
+    return '?';
   }
 
   testConnectionTimeout() {
     return 10000;
   }
+
+  async downloadTable(table) {
+    return { rows: await this.query(`SELECT * FROM ${table}`) };
+  }
+
+  async uploadTable(table, columns, tableData) {
+    if (!tableData.rows) {
+      throw new Error(`${this.constructor} driver supports only rows upload`);
+    }
+    await this.createTable(table, columns);
+    for (let i = 0; i < tableData.rows.length; i++) {
+      await this.query(
+        `INSERT INTO ${table}
+        (${columns.map(c => this.quoteIdentifier(c.name)).join(', ')})
+        VALUES (${columns.map((c, paramIndex) => this.param(paramIndex)).join(', ')})`,
+        columns.map(c => tableData.rows[i][c.name])
+      );
+    }
+  }
+
+  async tableColumnTypes(table) {
+    const [schema, name] = table.split('.');
+    const columns = await this.query(
+      `SELECT columns.column_name,
+             columns.table_name,
+             columns.table_schema,
+             columns.data_type
+      FROM information_schema.columns 
+      WHERE table_name = ${this.param(0)} AND table_schema = ${this.param(1)}`,
+      [name, schema]
+    );
+    return columns.map(c => ({ name: c.column_name, type: this.toGenericType(c.data_type) }));
+  }
+
+  createTable(quotedTableName, columns) {
+    return this.query(this.createTableSql(quotedTableName, columns), []);
+  }
+
+  createTableSql(quotedTableName, columns) {
+    columns = columns.map(c => `${this.quoteIdentifier(c.name)} ${this.fromGenericType(c.type)}`);
+    return `CREATE TABLE ${quotedTableName} (${columns.join(', ')})`;
+  }
+
+  toGenericType(columnType) {
+    return DbTypeToGenericType[columnType] || columnType;
+  }
+
+  fromGenericType(columnType) {
+    return columnType;
+  }
+
+  quoteIdentifier(identifier) {
+    return `"${identifier}"`;
+  }
 }
 
 module.exports = BaseDriver;