fix(query-orchestrator): QueryQueue - improve performance for high concurrency setups

Author: ovr

packages/cubejs-cubestore-driver/src/CubeStoreQueueDriver.ts

@@ -136,12 +136,31 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
   }
 
   public async getActiveAndToProcess(): Promise<GetActiveAndToProcessResponse> {
+    const active: QueryKeysTuple[] = [];
+    const toProcess: QueryKeysTuple[] = [];
+
+    const rows = await this.driver.query<CubeStoreListResponse>('QUEUE LIST ?', [
+      this.options.redisQueuePrefix
+    ]);
+    if (rows.length) {
+      for (const row of rows) {
+        if (row.status === 'active') {
+          active.push([
+            row.id as QueryKeyHash,
+            row.queue_id ? parseInt(row.queue_id, 10) : null,
+          ]);
+        } else {
+          toProcess.push([
+            row.id as QueryKeyHash,
+            row.queue_id ? parseInt(row.queue_id, 10) : null,
+          ]);
+        }
+      }
+    }
+
     return [
-      // We don't return active queries, because it's useless
-      // There is only one place where it's used, and it's QueryQueue.reconcileQueueImpl
-      // Cube Store provides strict guarantees that queue item cannot be active & pending in the same time
-      [],
-      await this.getToProcessQueries()
+      active,
+      toProcess,
     ];
   }
 

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

@@ -558,14 +558,19 @@ export class QueryQueue {
       }));
 
       /**
-       * There is a bug somewhere in Redis (maybe in memory too?),
-       * which doesn't remove queue item from pending, while it's in active state
-       *
        * TODO(ovr): Check LocalQueueDriver for strict guarantees that item cannot be in active & pending in the same time
-       * TODO(ovr): Migrate to getToProcessQueries after removal of Redis
        */
       const [active, toProcess] = await queueConnection.getActiveAndToProcess();
 
+      /**
+       * Important notice: Concurrency configuration works per a specific queue, not per node.
+       *
+       * In production clusters where it contains N nodes, it shares the same concurrency. It leads to a point
+       * where every node tries to pick up jobs as much as concurrency is defined for the whole cluster. To minimize
+       * the effect of competition between nodes, it's important to reduce the number of tries to process by active jobs.
+       */
+      const toProcessLimit = active.length >= this.concurrency ? 1 : this.concurrency - active.length;
+
       await Promise.all(
         R.pipe(
           R.filter(([queryKey, _queueId]) => {
@@ -585,7 +590,7 @@ export class QueryQueue {
               return false;
             }
           }),
-          R.take(this.concurrency),
+          R.take(toProcessLimit),
           R.map((([queryKey, queueId]) => this.sendProcessMessageFn(queryKey, queueId)))
         )(toProcess)
       );
@@ -743,8 +748,8 @@ export class QueryQueue {
   }
 
   /**
-   * Processing query specified by the `queryKey`. This method encapsulate most
-   * of the logic related with the queues updates, heartbeat, etc.
+   * Processing query specified by the `queryKey`. This method encapsulates most
+   * of the logic related to the queue updates, heartbeat, etc.
    *
    * @param {QueryKeyHash} queryKeyHashed
    * @param {QueueId | null} queueId Supported by new Cube Store and Memory