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

This reverts commit a353cf1.

Author: KSDaemon

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

@@ -1,5 +1,4 @@
 import { TableStructure } from '@cubejs-backend/base-driver';
-import { asyncDebounce } from '@cubejs-backend/shared';
 import { DriverFactory } from './DriverFactory';
 import { QueryCache, QueryWithParams } from './QueryCache';
 import {
@@ -17,6 +16,16 @@ 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;
 
@@ -26,6 +35,8 @@ export class PreAggregationLoadCache {
 
   private readonly queryResults: any;
 
+  private queryResultRequests: { [redisKey: string]: { resolve: CallableFunction, reject: CallableFunction }[]} = {};
+
   private readonly externalDriverFactory: any;
 
   private readonly requestId: any;
@@ -45,8 +56,6 @@ export class PreAggregationLoadCache {
 
   private readonly tablePrefixes: string[] | null;
 
-  private readonly cacheQueryResultDebounced: Function;
-
   public constructor(
     clientFactory: DriverFactory,
     queryCache,
@@ -64,7 +73,6 @@ 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) {
@@ -201,26 +209,53 @@ 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] = [];
 
-    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
+    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();
       }
-    );
 
-    return this.queryResults[queryKey];
+      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;
+    }
   }
 
   public hasKeyQueryResult(keyQuery) {