feat: Rollup mode (#843) Thanks to @jcw-!

* Add option for rollup only mode

* Add option for externalRefresh mode

* Fix lint warnings and errors in query-orchestrator

* Add additional config validation

* Start a unit test file for PreAggregations

* Update docs

* Empty commit to trigger a CI rerun

Author: jcw-

docs/Cube.js-Backend/@cubejs-backend-server-core.md

@@ -351,7 +351,7 @@ CubejsServerCore.create({
 });
 ```
 
-Best practice is to run `scheduledRefreshTimer` in a separate worker Cube.js instance. 
+Best practice is to run `scheduledRefreshTimer` in a separate worker Cube.js instance.
 For serverless deployments [REST API](rest-api#api-reference-v-1-run-scheduled-refresh) should be used instead of timer.
 
 ### extendContext
@@ -397,10 +397,14 @@ _Please note that this is advanced configuration._
 | Option | Description | Default Value |
 | ------ | ----------- | ------------- |
 | redisPrefix | Prefix to be set an all Redis keys | `STANDALONE` |
+| rollupOnlyMode | When enabled, an error will be thrown if a query can't be served from a pre-aggregation (rollup) | `false`
 | queryCacheOptions | Query cache options for DB queries | `{}`
 | queryCacheOptions.refreshKeyRenewalThreshold | Time in seconds to cache the result of [refreshKey](cube#parameters-refresh-key) check | `defined by DB dialect`
 | queryCacheOptions.backgroundRenew | Controls whether to wait in foreground for refreshed query data if `refreshKey` value has been changed. Refresh key queries or pre-aggregations are never awaited in foreground and always processed in background unless cache is empty. If `true` it immediately returns values from cache if available without [refreshKey](cube#parameters-refresh-key) check to renew in foreground. Default value before 0.15.0 was `true` | `false`
+| queryCacheOptions.queueOptions | Query queue options for DB queries | `{}`
 | preAggregationsOptions | Query cache options for pre-aggregations | `{}`
+| preAggregationsOptions.queueOptions | Query queue options for pre-aggregations | `{}`
+| preAggregationsOptions.externalRefresh | When running a separate instance of Cube.js to refresh pre-aggregations in the background, this option can be set on the API instance to prevent it from trying to check for rollup data being current - it won't try to create or refresh them when this option is `true` | `false`
 
 To set options for `queryCache` and `preAggregations`, set an object with key queueOptions. `queryCacheOptions` are used while querying database tables, while `preAggregationsOptions` settings are used to query pre-aggregated tables.
 

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

@@ -4,7 +4,7 @@ class BaseQueueDriver {
   redisHash(queryKey) {
     return typeof queryKey === 'string' && queryKey.length < 256 ?
       queryKey :
-      crypto.createHash('md5').update(JSON.stringify(queryKey)).digest("hex");
+      crypto.createHash('md5').update(JSON.stringify(queryKey)).digest('hex');
   }
 }
 

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

@@ -78,7 +78,7 @@ class PreAggregationLoadCache {
 
   async fetchTables(preAggregation) {
     if (preAggregation.external && !this.externalDriverFactory) {
-      throw new Error(`externalDriverFactory should be set in order to use external pre-aggregations`);
+      throw new Error('externalDriverFactory should be set in order to use external pre-aggregations');
     }
     const client = preAggregation.external ?
       await this.externalDriverFactory() :
@@ -180,6 +180,10 @@ class PreAggregationLoader {
     this.externalDriverFactory = preAggregations.externalDriverFactory;
     this.requestId = options.requestId;
     this.structureVersionPersistTime = preAggregations.structureVersionPersistTime;
+    this.externalRefresh = options.externalRefresh;
+    if (this.externalRefresh && this.waitForRenew) {
+      throw new Error("Invalid configuration - when externalRefresh is true, it will not perform a renew, therefore you cannot wait for it using waitForRenew.");
+    }
   }
 
   async loadPreAggregation() {
@@ -199,7 +203,19 @@ class PreAggregationLoader {
       const versionEntryByStructureVersion = versionEntries.find(
         v => v.table_name === this.preAggregation.tableName && v.structure_version === structureVersion
       );
+      if (this.externalRefresh) {
+        if (!versionEntryByStructureVersion) {
+          throw new Error('One or more pre-aggregation tables could not be found to satisfy that query');
+        }
+
+        // the rollups are being maintained independently of this instance of cube.js,
+        // immediately return the latest data it already has
+        return this.targetTableName(versionEntryByStructureVersion);
+      }
+
       if (versionEntryByStructureVersion) {
+        // this triggers an asyncronous/background load of the pre-aggregation but immediately
+        // returns the latest data it already has
         this.loadPreAggregationWithKeys().catch(e => {
           if (!(e instanceof ContinueWaitError)) {
             this.logger('Error loading pre-aggregation', {
@@ -211,9 +227,12 @@ class PreAggregationLoader {
         });
         return this.targetTableName(versionEntryByStructureVersion);
       } else {
+        // no rollup has been built yet - build it syncronously as part of responding to this request
         return this.loadPreAggregationWithKeys();
       }
     } else {
+      // either we have no data cached for this rollup or waitForRenew is true, either way,
+      // syncronously renew what data is needed so that the most current data will be returned for the current request
       return {
         targetTableName: await this.loadPreAggregationWithKeys(),
         refreshKeyValues: await this.getInvalidationKeyValues()
@@ -477,7 +496,7 @@ class PreAggregationLoader {
     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()`);
+      throw new Error('Can\'t load external pre-aggregation: source driver doesn\'t support downloadQueryResults()');
     }
 
     this.logExecutingSql(invalidationKeys, sql, params, this.targetTableName(newVersionEntry), newVersionEntry);
@@ -496,7 +515,7 @@ class PreAggregationLoader {
 
   async downloadTempExternalPreAggregation(client, newVersionEntry, saveCancelFn) {
     if (!client.downloadTable) {
-      throw new Error(`Can't load external pre-aggregation: source driver doesn't support downloadTable()`);
+      throw new Error('Can\'t load external pre-aggregation: source driver doesn\'t support downloadTable()');
     }
     const table = this.targetTableName(newVersionEntry);
     this.logger('Downloading external pre-aggregation', {
@@ -512,7 +531,7 @@ class PreAggregationLoader {
     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()`);
+      throw new Error('Can\'t load external pre-aggregation: destination driver doesn\'t support uploadTable()');
     }
     this.logger('Uploading external pre-aggregation', {
       preAggregation: this.preAggregation,
@@ -599,6 +618,7 @@ class PreAggregations {
     this.externalDriverFactory = options.externalDriverFactory;
     this.structureVersionPersistTime = options.structureVersionPersistTime || 60 * 60 * 24 * 30;
     this.usedTablePersistTime = options.usedTablePersistTime || 600;
+    this.externalRefresh = options.externalRefresh;
   }
 
   tablesUsedRedisKey(tableName) {
@@ -629,7 +649,7 @@ class PreAggregations {
         p,
         preAggregationsTablesToTempTables,
         loadCache,
-        { waitForRenew: queryBody.renewQuery, requestId: queryBody.requestId }
+        { waitForRenew: queryBody.renewQuery, requestId: queryBody.requestId, externalRefresh: this.externalRefresh }
       );
       const preAggregationPromise = () => loader.loadPreAggregation().then(async targetTableName => {
         const usedPreAggregation = typeof targetTableName === 'string' ? { targetTableName } : targetTableName;
@@ -655,7 +675,7 @@ class PreAggregations {
           preAggregation,
           preAggregationsTablesToTempTables,
           new PreAggregationLoadCache(this.redisPrefix, this.driverFactory, this.queryCache, this, { requestId }),
-          { requestId }
+          { requestId, externalRefresh: this.externalRefresh }
         );
         return loader.refresh(newVersionEntry, invalidationKeys)(client);
       }, {

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

@@ -111,9 +111,9 @@ class QueryCache {
   }
 
   static replaceAll(replaceThis, withThis, inThis) {
-    withThis = withThis.replace(/\$/g, "$$$$");
+    withThis = withThis.replace(/\$/g, '$$$$');
     return inThis.replace(
-      new RegExp(replaceThis.replace(/([/,!\\^${}[\]().*+?|<>\-&])/g, "\\$&"), "g"),
+      new RegExp(replaceThis.replace(/([/,!\\^${}[\]().*+?|<>\-&])/g, '\\$&'), 'g'),
       withThis
     );
   }
@@ -377,7 +377,7 @@ class QueryCache {
   }
 
   queryRedisKey(cacheKey) {
-    return `SQL_QUERY_RESULT_${this.redisPrefix}_${crypto.createHash('md5').update(JSON.stringify(cacheKey)).digest("hex")}`;
+    return `SQL_QUERY_RESULT_${this.redisPrefix}_${crypto.createHash('md5').update(JSON.stringify(cacheKey)).digest('hex')}`;
   }
 }
 

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

@@ -14,7 +14,7 @@ class QueryOrchestrator {
       process.env.NODE_ENV === 'production' || process.env.REDIS_URL ? 'redis' : 'memory'
     );
     if (cacheAndQueueDriver !== 'redis' && cacheAndQueueDriver !== 'memory') {
-      throw new Error(`Only 'redis' or 'memory' are supported for cacheAndQueueDriver option`);
+      throw new Error('Only \'redis\' or \'memory\' are supported for cacheAndQueueDriver option');
     }
     const redisPool = cacheAndQueueDriver === 'redis' ? new RedisPool() : undefined;
 
@@ -35,12 +35,16 @@ class QueryOrchestrator {
         ...options.preAggregationsOptions
       }
     );
+    this.rollupOnlyMode = options.rollupOnlyMode;
   }
 
   async fetchQuery(queryBody) {
     return this.preAggregations.loadAllPreAggregationsIfNeeded(queryBody)
       .then(async preAggregationsTablesToTempTables => {
         const usedPreAggregations = R.fromPairs(preAggregationsTablesToTempTables);
+        if (this.rollupOnlyMode && Object.keys(usedPreAggregations).length === 0) {
+          throw new Error('No pre-aggregation exists for that query');
+        }
         if (!queryBody.query) {
           return {
             usedPreAggregations

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

@@ -247,7 +247,7 @@ class QueryQueue {
                   try {
                     return redisClient.optimisticQueryUpdate(queryKey, { cancelHandler }, processingId);
                   } catch (e) {
-                    this.logger(`Error while query update`, {
+                    this.logger('Error while query update', {
                       queryKey: query.queryKey,
                       error: e.stack || e,
                       queuePrefix: this.redisQueuePrefix,
@@ -301,7 +301,7 @@ class QueryQueue {
         if (!(await redisClient.setResultAndRemoveQuery(queryKey, executionResult, processingId))) {
           this.logger('Orphaned execution result', {
             processingId,
-            warn: `Result for query was not set due to processing lock wasn't acquired`,
+            warn: 'Result for query was not set due to processing lock wasn\'t acquired',
             queryKey: query.queryKey,
             queuePrefix: this.redisQueuePrefix,
             requestId: query.requestId
@@ -342,7 +342,7 @@ class QueryQueue {
       }
       await this.cancelHandlers[queryHandler](query);
     } catch (e) {
-      this.logger(`Error while cancel`, {
+      this.logger('Error while cancel', {
         queryKey: query.queryKey,
         error: e.stack || e,
         queuePrefix: this.redisQueuePrefix,

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

@@ -0,0 +1,150 @@
+/* globals describe, beforeAll, afterAll, beforeEach, test, expect */
+const crypto = require('crypto');
+
+class MockDriver {
+  constructor() {
+    this.tables = [];
+    this.executedQueries = [];
+    this.cancelledQueries = [];
+  }
+
+  query(query) {
+    this.executedQueries.push(query);
+    let promise = Promise.resolve([query]);
+    if (query.match(`orders_too_big`)) {
+      promise = promise.then((res) => new Promise(resolve => setTimeout(() => resolve(res), 3000)));
+    }
+    promise.cancel = async () => {
+      this.cancelledQueries.push(query);
+    };
+    return promise;
+  }
+
+  async getTablesQuery(schema) {
+    return this.tables.map(t => ({ table_name: t.replace(`${schema}.`, '') }));
+  }
+
+  async createSchemaIfNotExists(schema) {
+    this.schema = schema;
+    return null;
+  }
+
+  loadPreAggregationIntoTable(preAggregationTableName, loadSql) {
+    this.tables.push(preAggregationTableName.substring(0, 100));
+    return this.query(loadSql);
+  }
+
+  async dropTable(tableName) {
+    this.tables = this.tables.filter(t => t !== tableName);
+    return this.query(`DROP TABLE ${tableName}`);
+  }
+}
+
+describe('PreAggregations', () => {
+  let mockDriver = null;
+  let queryCache = null;
+  const basicQuery = {
+    query: "SELECT \"orders__created_at_week\" \"orders__created_at_week\", sum(\"orders__count\") \"orders__count\" FROM (SELECT * FROM stb_pre_aggregations.orders_number_and_count20191101) as partition_union  WHERE (\"orders__created_at_week\" >= ($1::timestamptz::timestamptz AT TIME ZONE 'UTC') AND \"orders__created_at_week\" <= ($2::timestamptz::timestamptz AT TIME ZONE 'UTC')) GROUP BY 1 ORDER BY 1 ASC LIMIT 10000",
+    values: ["2019-11-01T00:00:00Z", "2019-11-30T23:59:59Z"],
+    cacheKeyQueries: {
+      renewalThreshold: 21600,
+      queries: [["SELECT date_trunc('hour', (NOW()::timestamptz AT TIME ZONE 'UTC')) as current_hour", []]]
+    },
+    preAggregations: [{
+      preAggregationsSchema: "stb_pre_aggregations",
+      tableName: "stb_pre_aggregations.orders_number_and_count20191101",
+      loadSql: ["CREATE TABLE stb_pre_aggregations.orders_number_and_count20191101 AS SELECT\n      date_trunc('week', (\"orders\".created_at::timestamptz AT TIME ZONE 'UTC')) \"orders__created_at_week\", count(\"orders\".id) \"orders__count\", sum(\"orders\".number) \"orders__number\"\n    FROM\n      public.orders AS \"orders\"\n  WHERE (\"orders\".created_at >= $1::timestamptz AND \"orders\".created_at <= $2::timestamptz) GROUP BY 1", ["2019-11-01T00:00:00Z", "2019-11-30T23:59:59Z"]],
+      invalidateKeyQueries: [["SELECT date_trunc('hour', (NOW()::timestamptz AT TIME ZONE 'UTC')) as current_hour", []]]
+    }],
+    requestId: 'basic'
+  };
+  const basicQueryWithRenew = {
+    query: "SELECT \"orders__created_at_week\" \"orders__created_at_week\", sum(\"orders__count\") \"orders__count\" FROM (SELECT * FROM stb_pre_aggregations.orders_number_and_count20191101) as partition_union  WHERE (\"orders__created_at_week\" >= ($1::timestamptz::timestamptz AT TIME ZONE 'UTC') AND \"orders__created_at_week\" <= ($2::timestamptz::timestamptz AT TIME ZONE 'UTC')) GROUP BY 1 ORDER BY 1 ASC LIMIT 10000",
+    values: ["2019-11-01T00:00:00Z", "2019-11-30T23:59:59Z"],
+    cacheKeyQueries: {
+      renewalThreshold: 21600,
+      queries: [["SELECT date_trunc('hour', (NOW()::timestamptz AT TIME ZONE 'UTC')) as current_hour", []]]
+    },
+    preAggregations: [{
+      preAggregationsSchema: "stb_pre_aggregations",
+      tableName: "stb_pre_aggregations.orders_number_and_count20191101",
+      loadSql: ["CREATE TABLE stb_pre_aggregations.orders_number_and_count20191101 AS SELECT\n      date_trunc('week', (\"orders\".created_at::timestamptz AT TIME ZONE 'UTC')) \"orders__created_at_week\", count(\"orders\".id) \"orders__count\", sum(\"orders\".number) \"orders__number\"\n    FROM\n      public.orders AS \"orders\"\n  WHERE (\"orders\".created_at >= $1::timestamptz AND \"orders\".created_at <= $2::timestamptz) GROUP BY 1", ["2019-11-01T00:00:00Z", "2019-11-30T23:59:59Z"]],
+      invalidateKeyQueries: [["SELECT date_trunc('hour', (NOW()::timestamptz AT TIME ZONE 'UTC')) as current_hour", []]]
+    }],
+    renewQuery: true,
+    requestId: 'basic'
+  };
+
+  beforeEach(() => {
+    mockDriver = new MockDriver();
+
+    jest.resetModules();
+    const QueryCache = require('../orchestrator/QueryCache');
+    queryCache = new QueryCache(
+      "TEST",
+      async () => mockDriver,
+      (msg, params) => {},
+      {
+        queueOptions: {
+          executionTimeout: 1
+        },
+      },
+    );
+  });
+
+  describe('loadAllPreAggregationsIfNeeded', () => {
+    let preAggregations = null;
+
+    beforeEach(async () => {
+      const PreAggregations = require('../orchestrator/PreAggregations');
+      preAggregations = new PreAggregations(
+        "TEST",
+        async () => mockDriver,
+        (msg, params) => {},
+        queryCache,
+        {
+          queueOptions: {
+            executionTimeout: 1
+          },
+        },
+      );
+    });
+
+    test('syncronously create rollup from scratch', async () => {
+      const result = await preAggregations.loadAllPreAggregationsIfNeeded(basicQueryWithRenew);
+      expect(result[0][1].targetTableName).toMatch(/stb_pre_aggregations.orders_number_and_count20191101_kjypcoio_5yftl5il/);
+    });
+  });
+
+  describe(`loadAllPreAggregationsIfNeeded with externalRefresh true`, () => {
+    let preAggregations = null;
+
+    beforeEach(async () => {
+      const PreAggregations = require('../orchestrator/PreAggregations');
+      preAggregations = new PreAggregations(
+        "TEST",
+        async () => mockDriver,
+        (msg, params) => {},
+        queryCache,
+        {
+          queueOptions: {
+            executionTimeout: 1
+          },
+          externalRefresh: true,
+        },
+      );
+    });
+
+    test('fail if waitForRenew is also specified', async () => {
+      await expect(async () => {
+        await preAggregations.loadAllPreAggregationsIfNeeded(basicQueryWithRenew);
+      }).rejects.toThrowError(/Invalid configuration/);
+    });
+
+    test('fail if rollup doesn\'t already exist', async () => {
+      await expect(async () => {
+        await preAggregations.loadAllPreAggregationsIfNeeded(basicQuery);
+      }).rejects.toThrowError(/One or more pre-aggregation tables could not be found to satisfy that query/);
+    });
+  });
+});

packages/cubejs-server-core/core/index.d.ts

@@ -43,6 +43,7 @@ declare module "@cubejs-backend/server-core" {
     redisPrefix?: string;
     queryCacheOptions?: QueryCacheOptions;
     preAggregationsOptions?: PreAggregationsOptions;
+    rollupOnlyMode?: boolean;
   }
 
   export interface QueryCacheOptions {
@@ -53,6 +54,7 @@ declare module "@cubejs-backend/server-core" {
 
   export interface PreAggregationsOptions {
     queueOptions?: QueueOptions;
+    externalRefresh?: boolean;
   }
 
   export interface QueueOptions {