use asyncDebounce for caching in pre-agg-load-cache

Author: KSDaemon

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

@@ -1,4 +1,5 @@
 import { TableStructure } from '@cubejs-backend/base-driver';
+import { asyncDebounce } from '@cubejs-backend/shared';
 import { DriverFactory } from './DriverFactory';
 import { QueryCache, QueryWithParams } from './QueryCache';
 import {
@@ -16,16 +17,6 @@ type PreAggregationLoadCacheOptions = {
   tablePrefixes?: string[],
 };
 
-function createDeferred() {
-  let resolve;
-  let reject;
-  const promise = new Promise((res, rej) => {
-    resolve = res;
-    reject = rej;
-  });
-  return { promise, resolve, reject };
-}
-
 export class PreAggregationLoadCache {
   private readonly driverFactory: DriverFactory;
 
@@ -35,8 +26,6 @@ export class PreAggregationLoadCache {
 
   private readonly queryResults: any;
 
-  private queryResultRequests: { [redisKey: string]: { resolve: CallableFunction, reject: CallableFunction }[]} = {};
-
   private readonly externalDriverFactory: any;
 
   private readonly requestId: any;
@@ -56,6 +45,8 @@ export class PreAggregationLoadCache {
 
   private readonly tablePrefixes: string[] | null;
 
+  private readonly cacheQueryResultDebounced: Function;
+
   public constructor(
     clientFactory: DriverFactory,
     queryCache,
@@ -73,6 +64,7 @@ export class PreAggregationLoadCache {
     this.versionEntries = {};
     this.tables = {};
     this.tableColumnTypes = {};
+    this.cacheQueryResultDebounced = asyncDebounce(this.queryCache.cacheQueryResult.bind(this.queryCache));
   }
 
   protected async tablesFromCache(preAggregation, forceRenew: boolean = false) {
@@ -209,53 +201,26 @@ export class PreAggregationLoadCache {
       return this.queryResults[queryKey];
     }
 
-    // There is ongoing request
-    if (this.queryResultRequests[queryKey]) {
-      const { promise, resolve, reject } = createDeferred();
-      this.queryResultRequests[queryKey].push({ resolve, reject });
-
-      return promise;
-    }
-
-    // Making query for a first time
-    this.queryResultRequests[queryKey] = [];
 
-    try {
-      this.queryResults[queryKey] = await this.queryCache.cacheQueryResult(
-        query,
-        values,
-        [query, values],
-        60 * 60,
-        {
-          renewalThreshold: this.queryCache.options.refreshKeyRenewalThreshold
-            || queryOptions?.renewalThreshold || 2 * 60,
-          renewalKey: [query, values],
-          waitForRenew,
-          priority,
-          requestId: this.requestId,
-          dataSource: this.dataSource,
-          useInMemory: true,
-          external: queryOptions?.external
-        }
-      );
-
-      let r = (this.queryResultRequests[queryKey] || []).pop();
-      while (r) {
-        r.resolve(this.queryResults[queryKey]);
-        r = this.queryResultRequests[queryKey].pop();
+    this.queryResults[queryKey] = await this.cacheQueryResultDebounced(
+      query,
+      values,
+      [query, values],
+      60 * 60,
+      {
+        renewalThreshold: this.queryCache.options.refreshKeyRenewalThreshold
+          || queryOptions?.renewalThreshold || 2 * 60,
+        renewalKey: [query, values],
+        waitForRenew,
+        priority,
+        requestId: this.requestId,
+        dataSource: this.dataSource,
+        useInMemory: true,
+        external: queryOptions?.external
       }
+    );
 
-      return this.queryResults[queryKey];
-    } catch (err) {
-      let r = (this.queryResultRequests[queryKey] || []).pop();
-      while (r) {
-        r.reject(err);
-        r = this.queryResultRequests[queryKey].pop();
-      }
-      throw err;
-    } finally {
-      this.queryResultRequests[queryKey] = null;
-    }
+    return this.queryResults[queryKey];
   }
 
   public hasKeyQueryResult(keyQuery) {