Implement background refresh

Author: KSDaemon

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

@@ -101,6 +101,30 @@ export type QueryBody = {
   [key: string]: any;
 };
 
+/*
+
+stale-if-slow (default) — equivalent to current renewQuery: false
+  If refresh keys are up-to-date, returns the value from cache
+  If refresh keys are expired, tries to return the value from the database
+    Returns fresh value from the database if the query executed in the database until the first “Continue wait” interval is reached
+    Returns stale value from cache otherwise
+
+stale-while-revalidate — AKA “backgroundRefresh”
+  If refresh keys are up-to-date, returns the value from cache
+  If refresh keys are expired, returns stale data from cache
+  Updates the cache in background
+
+must-revalidate — equivalent to current renewQuery: true
+  If refresh keys are up-to-date, returns the value from cache
+  If refresh keys are expired, tries to return the value from the database
+    Returns fresh value from the database even if it takes minutes and many “Continue wait” intervals
+
+no-cache — AKA “forceRefresh”
+  Skips refresh key checks
+  Returns fresh data from the database, even if it takes minutes and many ”Continue wait” intervals
+
+*/
+
 /**
  * Temp (partition/lambda) table definition.
  */
@@ -760,6 +784,42 @@ export class QueryCache {
     });
   }
 
+  public async startBackgroundRefreshForQuery(queryBody: QueryBody, preAggregationsTablesToTempTables: PreAggTableToTempTable[]) {
+    const replacePreAggregationTableNames =
+      (queryAndParams: string | QueryWithParams) => (
+        QueryCache.replacePreAggregationTableNames(
+          queryAndParams,
+          preAggregationsTablesToTempTables,
+        )
+      );
+
+    const query = replacePreAggregationTableNames(queryBody.query);
+    const { values } = queryBody;
+
+    const cacheKeyQueries = this
+      .cacheKeyQueriesFrom(queryBody)
+      .map(replacePreAggregationTableNames);
+
+    const renewalThreshold = queryBody.cacheKeyQueries?.renewalThreshold;
+    const expireSecs = this.getExpireSecs(queryBody);
+    const cacheKey = QueryCache.queryCacheKey(queryBody);
+
+    this.startRenewCycle(
+      query,
+      values,
+      cacheKeyQueries,
+      expireSecs,
+      cacheKey,
+      renewalThreshold,
+      {
+        external: queryBody.external,
+        requestId: queryBody.requestId,
+        dataSource: queryBody.dataSource,
+        persistent: false, // We don't need stream back as there will be no consumer
+      }
+    );
+  }
+
   public renewQuery(
     query: string | QueryWithParams,
     values: string[],

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

@@ -143,6 +143,29 @@ export class QueryOrchestrator {
     return this.preAggregations;
   }
 
+  /**
+   * Starts background refresh cycle for a query.
+   */
+  public async startBackgroundRefresh(queryBody: QueryBody): Promise<void> {
+    if (!queryBody.query) {
+      return;
+    }
+
+    const {
+      preAggregationsTablesToTempTables,
+      values,
+    } = await this.preAggregations.loadAllPreAggregationsIfNeeded(queryBody);
+
+    if (values) {
+      queryBody = {
+        ...queryBody,
+        values
+      };
+    }
+
+    await this.queryCache.startBackgroundRefreshForQuery(queryBody, preAggregationsTablesToTempTables);
+  }
+
   /**
    * Force reconcile queue logic to be executed.
    */

packages/cubejs-server-core/src/core/OrchestratorApi.ts

@@ -155,7 +155,15 @@ export class OrchestratorApi {
         }
 
         if (query.cache === 'stale-while-revalidate' && fromCache) {
-          // TODO: Run background refresh
+          // Start background refresh
+          this.orchestrator.startBackgroundRefresh(query).catch(e => {
+            this.logger('Error starting background refresh', {
+              query: queryForLog,
+              requestId: query.requestId,
+              error: ((e as Error).stack || e)
+            });
+          });
+
           return {
             ...fromCache,
             slowQuery: true