feat: persistent queue (#5793)

Author: buntarb

packages/cubejs-api-gateway/src/gateway.ts

@@ -1208,7 +1208,7 @@ class ApiGateway {
       renewQuery: normalizedQuery.renewQuery,
       requestId: context.requestId,
       context,
-      persistent: apiType === 'sql',
+      persistent: false, // apiType === 'sql',
     }];
     if (normalizedQuery.total) {
       const normalizedTotal = structuredClone(normalizedQuery);

packages/cubejs-query-orchestrator/package.json

@@ -41,7 +41,8 @@
     "moment-range": "^4.0.2",
     "moment-timezone": "^0.5.33",
     "ramda": "^0.27.2",
-    "redis": "^3.0.2"
+    "redis": "^3.0.2",
+    "uuid": "^8.3.2"
   },
   "devDependencies": {
     "@cubejs-backend/linter": "^0.31.0",
@@ -51,6 +52,7 @@
     "@types/node": "^12",
     "@types/ramda": "^0.27.32",
     "@types/redis": "^2.8.28",
+    "@types/uuid": "^8.3.0",
     "jest": "^26.6.3",
     "typescript": "~4.1.5"
   },

packages/cubejs-query-orchestrator/src/orchestrator/BaseQueueDriver.ts

@@ -1,9 +1,7 @@
-import crypto from 'crypto';
+import { getCacheHash } from './utils';
 
 export abstract class BaseQueueDriver {
   public redisHash(queryKey) {
-    return typeof queryKey === 'string' && queryKey.length < 256 ?
-      queryKey :
-      crypto.createHash('md5').update(JSON.stringify(queryKey)).digest('hex');
+    return getCacheHash(queryKey);
   }
 }

packages/cubejs-query-orchestrator/src/orchestrator/LocalQueueDriver.js

@@ -49,6 +49,13 @@ export class LocalQueueDriverConnection {
     return res;
   }
 
+  /**
+   * Returns promise wich will be resolved with the specified by the
+   * queryKey query result or null if query was not added to the
+   * processing.
+   * @param {*} queryKey
+   * @returns {Promise<null | *>}
+   */
   async getResult(queryKey) {
     const resultListKey = this.resultListKey(queryKey);
     if (this.resultPromises[resultListKey] && this.resultPromises[resultListKey].resolved) {
@@ -66,6 +73,21 @@ export class LocalQueueDriverConnection {
     )(queueObj);
   }
 
+  /**
+   * Adds specified by the queryKey query to the queue, returns tuple
+   * with the operation result.
+   *
+   * @typedef {[added: number, _b: null, _c: null, toProcessLength: number, addedTime: number]} AddedTuple
+   *
+   * @param {number} keyScore
+   * @param {*} queryKey
+   * @param {number} orphanedTime
+   * @param {string} queryHandler (for the regular query is eq to 'query')
+   * @param {*} query
+   * @param {number} priority
+   * @param {*} options
+   * @returns {AddedTuple}
+   */
   addToQueue(keyScore, queryKey, orphanedTime, queryHandler, query, priority, options) {
     const queryQueueObj = {
       queryHandler,
@@ -185,6 +207,11 @@ export class LocalQueueDriverConnection {
     return this.queryDef[this.redisHash(queryKey)];
   }
 
+  /**
+   * Updates heart beat for the processing query by its `queryKey`.
+   *
+   * @param {string} queryKey
+   */
   updateHeartBeat(queryKey) {
     const key = this.redisHash(queryKey);
     if (this.heartBeat[key]) {
@@ -245,14 +272,31 @@ export class LocalQueueDriverConnection {
   release() {
   }
 
+  /**
+   * Returns cache key to the specified by the queryKey query and the
+   * specified by the suffix query state.
+   * @param {*} queryKey
+   * @param {string} suffix
+   * @returns {string}
+   */
   queryRedisKey(queryKey, suffix) {
     return `${this.redisQueuePrefix}_${this.redisHash(queryKey)}_${suffix}`;
   }
 
+  /**
+   * Returns cache key to the cached query result.
+   * @param {*} queryKey
+   * @returns {string}
+   */
   resultListKey(queryKey) {
     return this.queryRedisKey(queryKey, 'RESULT');
   }
 
+  /**
+   * Returns hash sum of the query specified by the queryKey.
+   * @param {*} queryKey
+   * @returns {string}
+   */
   redisHash(queryKey) {
     return this.driver.redisHash(queryKey);
   }

packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts

@@ -704,7 +704,16 @@ export class PreAggregationLoader {
       if (this.isJob) {
         // We don't want to wait for the jobed build query result. So we run the
         // executeInQueue method and immediately return the LoadPreAggregationResult object.
-        this.executeInQueue(invalidationKeys, this.priority(10), newVersionEntry);
+        this
+          .executeInQueue(invalidationKeys, this.priority(10), newVersionEntry)
+          .catch((e: any) => {
+            this.logger('Pre-aggregations build job error', {
+              preAggregation: this.preAggregation,
+              requestId: this.requestId,
+              newVersionEntry,
+              error: (e.stack || e),
+            });
+          });
         const targetTableName = this.targetTableName(newVersionEntry);
         this.updateLastTouch(targetTableName);
         return {

packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts

@@ -1,4 +1,3 @@
-import crypto from 'crypto';
 import csvWriter from 'csv-write-stream';
 import LRUCache from 'lru-cache';
 import { MaybeCancelablePromise, streamToArray } from '@cubejs-backend/shared';
@@ -11,6 +10,7 @@ import { LocalCacheDriver } from './LocalCacheDriver';
 import { CacheDriverInterface } from './cache-driver.interface';
 import { DriverFactory, DriverFactoryByDataSource } from './DriverFactory';
 import { PreAggregationDescription } from './PreAggregations';
+import { getCacheHash } from './utils';
 
 type QueryOptions = {
   external?: boolean;
@@ -335,6 +335,8 @@ export class QueryCache {
     if (queryBody.invalidate) {
       key.push(queryBody.invalidate);
     }
+    // @ts-ignore
+    key.persistent = queryBody.persistent;
     return <CacheKey>key;
   }
 
@@ -387,7 +389,6 @@ export class QueryCache {
       useCsvQuery?: boolean,
     }
   ) {
-    persistent = persistent || false;
     const queue = external
       ? this.getExternalQueue()
       : await this.getQueue(dataSource);
@@ -399,14 +400,14 @@ export class QueryCache {
         query,
         values,
         requestId,
-        persistent,
         inlineTables,
         useCsvQuery,
       },
       priority,
       {
         stageQueryKey: cacheKey,
         requestId,
+        persistent,
       }
     );
   }
@@ -840,7 +841,7 @@ export class QueryCache {
   }
 
   public queryRedisKey(cacheKey) {
-    return `SQL_QUERY_RESULT_${this.redisPrefix}_${crypto.createHash('md5').update(JSON.stringify(cacheKey)).digest('hex')}`;
+    return `SQL_QUERY_RESULT_${this.redisPrefix}_${getCacheHash(cacheKey)}`;
   }
 
   public async cleanup() {