feat: Support external rollups from readonly source (#395)

* Add downloadQueryResults & readOnly to base driver

* Support including field type info in pg driver

* Add support for download-only refresh strategy

* Fix linting

* Fix missing parameter

* Describe both the create table and select query

* Use correct sql

* Update docs

* Address review feedback

Author: jcw-

docs/Schema/pre-aggregations.md

@@ -97,7 +97,7 @@ cube(`Orders`, {
       sql: `amount`,
       type: `sum`
     },
-    
+
     averageRevenue: {
       sql: `${revenue} / ${count}`,
       type: `number`
@@ -271,6 +271,28 @@ cube(`Orders`, {
 In order to make external pre-aggregations work you should set
 [externalDriverFactory](@cubejs-backend-server-core#external-driver-factory) and [externalDbType](@cubejs-backend-server-core#external-db-type) params while creating your server instance.
 
+Note that by default, Cube.js materializes the pre-aggregration query results as new tables in the source database. For external pre-aggregations, these source tables are temporary - once downloaded and uploaded to the external database, they are cleaned-up.
+
+However, it may not be possible to stage pre-aggregation query results in materialized tables in the source database like this - for example, if the driver doesn't support it, or if your source database is read-only. To fallback to a strategy where the pre-aggreation query results are downloaded without first being materialized, set the `readOnly` param of [driverFactory](@cubejs-backend-server-core#driver-factory) in your configuration:
+
+```javascript
+const CubejsServer = require('@cubejs-backend/server');
+const PostgresDriver = require('@cubejs-backend/postgres-driver');
+
+const options = {
+  driverFactory: () => new PostgresDriver({
+    readOnly: true
+  }),
+  externalDbType: 'postgres',
+  externalDriverFactory: () => new PostgresDriver({
+    host: 'my_host',
+    database: 'my_db',
+    user: 'my_user',
+    password: 'my_pw'
+  })
+};
+```
+
 ## refreshKey
 
 Cube.js also takes care of keeping pre-aggregations up to date.
@@ -320,7 +342,7 @@ In case of partitioned rollups incremental `refreshKey` can be used as follows:
 ```javascript
 cube(`Orders`, {
   sql: `select * from orders`,
-  
+
   // ...
 
   preAggregations: {
@@ -358,7 +380,7 @@ Example usage:
 ```javascript
 cube(`Orders`, {
   sql: `select * from orders`,
-  
+
   // ...
 
   preAggregations: {
@@ -373,4 +395,3 @@ cube(`Orders`, {
   }
 });
 ```
-

packages/cubejs-postgres-driver/driver/PostgresDriver.js

@@ -9,6 +9,12 @@ const GenericTypeToPostgres = {
   string: 'text'
 };
 
+const DataTypeMapping = {};
+Object.entries(types.builtins).forEach(pair => {
+  const [key, value] = pair;
+  DataTypeMapping[value] = key;
+});
+
 const timestampDataTypes = [1114, 1184];
 
 const timestampTypeParser = val => moment.utc(val).format(moment.HTML5_FMT.DATETIME_LOCAL_MS);
@@ -44,7 +50,7 @@ class PostgresDriver extends BaseDriver {
     }
   }
 
-  async query(query, values) {
+  async queryResponse(query, values) {
     const client = await this.pool.connect();
     try {
       await client.query(`SET TIME ZONE '${this.config.storeTimezone || 'UTC'}'`);
@@ -65,12 +71,31 @@ class PostgresDriver extends BaseDriver {
           },
         },
       });
-      return res && res.rows;
+      return res;
     } finally {
       await client.release();
     }
   }
 
+  async query(query, values) {
+    return (await this.queryResponse(query, values)).rows;
+  }
+
+  async downloadQueryResults(query, values) {
+    const res = await this.queryResponse(query, values);
+    return {
+      rows: res.rows,
+      types: res.fields.map(f => ({
+        name: f.name,
+        type: this.toGenericType(DataTypeMapping[f.dataTypeID].toLowerCase())
+      })),
+    };
+  }
+
+  readOnly() {
+    return !!this.config.readOnly;
+  }
+
   async uploadTable(table, columns, tableData) {
     if (!tableData.rows) {
       throw new Error(`${this.constructor} driver supports only rows upload`);

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

@@ -38,11 +38,19 @@ class BaseDriver {
   }
 
   testConnection() {
-    throw 'Not implemented';
+    throw new Error('Not implemented');
   }
 
   query() {
-    throw 'Not implemented';
+    throw new Error('Not implemented');
+  }
+
+  downloadQueryResults() {
+    throw new Error('Not implemented');
+  }
+
+  readOnly() {
+    return false;
   }
 
   tablesSchema() {
@@ -134,7 +142,7 @@ class BaseDriver {
              columns.table_name,
              columns.table_schema,
              columns.data_type
-      FROM information_schema.columns 
+      FROM information_schema.columns
       WHERE table_name = ${this.param(0)} AND table_schema = ${this.param(1)}`,
       [name, schema]
     );

packages/cubejs-query-orchestrator/orchestrator/PreAggregations.js

@@ -387,42 +387,70 @@ class PreAggregationLoader {
 
   refresh(newVersionEntry) {
     return (client) => {
-      const [loadSql, params] =
-        Array.isArray(this.preAggregation.loadSql) ? this.preAggregation.loadSql : [this.preAggregation.loadSql, []];
-      let queryPromise = null;
-      const refreshImpl = async () => {
-        if (this.preAggregation.external) { // TODO optimize
-          await client.createSchemaIfNotExists(this.preAggregation.preAggregationsSchema);
-        }
-        queryPromise = client.loadPreAggregationIntoTable(
-          this.targetTableName(newVersionEntry),
-          QueryCache.replacePreAggregationTableNames(loadSql, this.preAggregationsTablesToTempTables)
-            .replace(
-              this.preAggregation.tableName,
-              this.targetTableName(newVersionEntry)
-            ),
-          params
-        );
-        await queryPromise;
-        if (this.preAggregation.external) {
-          await this.loadExternalPreAggregation(client, newVersionEntry);
-        } else {
-          await this.createIndexes(client, newVersionEntry);
-        }
-        await this.loadCache.reset(this.preAggregation);
-        await this.dropOrphanedTables(client, this.targetTableName(newVersionEntry));
-        if (!this.preAggregation.external) {
-          await this.loadCache.reset(this.preAggregation);
-        }
-      };
-
-      const resultPromise = refreshImpl();
-      resultPromise.cancel = () => queryPromise.cancel(); // TODO cancel for external upload
+      let refreshStrategy = this.refreshImplStoreInSourceStrategy;
+      if (this.preAggregation.external) {
+        refreshStrategy = client.readOnly() ? this.refreshImplStreamExternalStrategy : this.refreshImplTempTableExternalStrategy;
+      }
+      const resultPromise = refreshStrategy.bind(this)(client, newVersionEntry);
+      resultPromise.cancel = () => {} // TODO implement cancel (loading rollup into table and external upload)
       return resultPromise;
     };
   }
 
-  async loadExternalPreAggregation(client, newVersionEntry) {
+  async refreshImplStoreInSourceStrategy(client, newVersionEntry) {
+    const [loadSql, params] =
+        Array.isArray(this.preAggregation.loadSql) ? this.preAggregation.loadSql : [this.preAggregation.loadSql, []];
+    await client.loadPreAggregationIntoTable(
+      this.targetTableName(newVersionEntry),
+      QueryCache.replacePreAggregationTableNames(loadSql, this.preAggregationsTablesToTempTables)
+        .replace(
+          this.preAggregation.tableName,
+          this.targetTableName(newVersionEntry)
+        ),
+      params
+    );
+    await this.createIndexes(client, newVersionEntry);
+    await this.loadCache.reset(this.preAggregation);
+    await this.dropOrphanedTables(client, this.targetTableName(newVersionEntry));
+    await this.loadCache.reset(this.preAggregation);
+  }
+
+  async refreshImplTempTableExternalStrategy(client, newVersionEntry) {
+    const [loadSql, params] =
+        Array.isArray(this.preAggregation.loadSql) ? this.preAggregation.loadSql : [this.preAggregation.loadSql, []];
+    await client.createSchemaIfNotExists(this.preAggregation.preAggregationsSchema);
+    await client.loadPreAggregationIntoTable(
+      this.targetTableName(newVersionEntry),
+      QueryCache.replacePreAggregationTableNames(loadSql, this.preAggregationsTablesToTempTables)
+        .replace(
+          this.preAggregation.tableName,
+          this.targetTableName(newVersionEntry)
+        ),
+      params
+    );
+    const tableData = await this.downloadTempExternalPreAggregation(client, newVersionEntry);
+    await this.uploadExternalPreAggregation(tableData, newVersionEntry);
+    await this.loadCache.reset(this.preAggregation);
+    await this.dropOrphanedTables(client, this.targetTableName(newVersionEntry));
+  }
+
+  async refreshImplStreamExternalStrategy(client, newVersionEntry) {
+    const [sql, params] =
+        Array.isArray(this.preAggregation.sql) ? this.preAggregation.sql : [this.preAggregation.sql, []];
+    if (!client.downloadQueryResults) {
+      throw new Error(`Can't load external pre-aggregation: source driver doesn't support downloadQueryResults()`);
+    }
+
+    this.logger('Downloading external pre-aggregation via query', {
+      preAggregation: this.preAggregation,
+      requestId: this.requestId
+    });
+    const tableData = await client.downloadQueryResults(sql, params);
+    await this.uploadExternalPreAggregation(tableData, newVersionEntry);
+    await this.loadCache.reset(this.preAggregation);
+  }
+
+  async downloadTempExternalPreAggregation(client, newVersionEntry) {
     if (!client.downloadTable) {
       throw new Error(`Can't load external pre-aggregation: source driver doesn't support downloadTable()`);
     }
@@ -432,7 +460,12 @@ class PreAggregationLoader {
       requestId: this.requestId
     });
     const tableData = await client.downloadTable(table);
-    const columns = await client.tableColumnTypes(table);
+    tableData.types = await client.tableColumnTypes(table);
+    return tableData;
+  }
+
+  async uploadExternalPreAggregation(tableData, newVersionEntry) {
+    const table = this.targetTableName(newVersionEntry);
     const externalDriver = await this.externalDriverFactory();
     if (!externalDriver.uploadTable) {
       throw new Error(`Can't load external pre-aggregation: destination driver doesn't support uploadTable()`);
@@ -441,7 +474,7 @@ class PreAggregationLoader {
       preAggregation: this.preAggregation,
       requestId: this.requestId
     });
-    await externalDriver.uploadTable(table, columns, tableData);
+    await externalDriver.uploadTable(table, tableData.types, tableData);
     await this.createIndexes(externalDriver, newVersionEntry);
     await this.loadCache.reset(this.preAggregation);
     await this.dropOrphanedTables(externalDriver, table);

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

@@ -87,6 +87,7 @@ class PreAggregations {
       preAggregationsSchema: this.query.preAggregationSchema(),
       tableName,
       loadSql: this.query.preAggregationLoadSql(cube, preAggregation, tableName),
+      sql: this.query.preAggregationSql(cube, preAggregation),
       invalidateKeyQueries: refreshKeyQueries.queries,
       refreshKeyRenewalThresholds: refreshKeyQueries.refreshKeyRenewalThresholds,
       external: preAggregation.external,
@@ -462,7 +463,7 @@ class PreAggregations {
       preAggregationForQuery.preAggregation.measures :
       this.evaluateAllReferences(preAggregationForQuery.cube, preAggregationForQuery.preAggregation).measures
     );
-    
+
     const rollupGranularity = this.castGranularity(preAggregationForQuery.preAggregation.granularity) || 'day';
 
     return this.query.evaluateSymbolSqlWithContext(