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

# Conflicts:
#	packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.js

# Conflicts:
#	packages/cubejs-query-orchestrator/src/orchestrator/QueryQueue.ts

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.ts

@@ -543,7 +543,16 @@ export class QueryQueue {
         }
       }));
 
-      const [_active, toProcess] = await queueConnection.getActiveAndToProcess();
+      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;
 
       const tasks = toProcess
         .filter(([queryKey, _queueId]) => {
@@ -559,7 +568,7 @@ export class QueryQueue {
             return false;
           }
         })
-        .slice(0, this.concurrency)
+        .slice(0, toProcessLimit)
         .map(([queryKey, queueId]) => this.sendProcessMessageFn(queryKey, queueId));
 
       await Promise.all(tasks);