fix(cubestore-driver): Correct pending count and active keys in queue logs (#6250)

Author: ovr

packages/cubejs-base-driver/src/queue-driver.interface.ts

@@ -8,7 +8,23 @@ export interface QueryKeyHash extends String {
 
 export type AddToQueueResponse = [added: number, _b: any, _c: any, queueSize: number, addedToQueueTime: number];
 export type QueryStageStateResponse = [active: string[], toProcess: string[]] | [active: string[], toProcess: string[], defs: Record<string, QueryDef>];
-export type RetrieveForProcessingResponse = [added: any, removed: any, active: QueryKeyHash[], toProcess: any, def: QueryDef, lockAquired: boolean] | null;
+export type RetrieveForProcessingSuccess = [
+  added: any /** todo(ovr): Remove, useless */,
+  removed: any /** todo(ovr): Remove, useless */,
+  active: QueryKeyHash[],
+  pending: number,
+  def: QueryDef,
+  lockAquired: true
+];
+export type RetrieveForProcessingFail = [
+  added: any /** todo(ovr): Remove, useless */,
+  removed: any /** todo(ovr): Remove, useless */,
+  active: QueryKeyHash[],
+  pending: number,
+  def: null,
+  lockAquired: false
+];
+export type RetrieveForProcessingResponse = RetrieveForProcessingSuccess | RetrieveForProcessingFail | null;
 
 export interface AddToQueueQuery {
   isJob: boolean,

packages/cubejs-cubestore-driver/package.json

@@ -45,6 +45,7 @@
     "@types/flatbuffers": "^1.10.0",
     "@types/generic-pool": "^3.1.9",
     "@types/ws": "^7.4.0",
+    "@types/csv-write-stream": "^2.0.0",
     "jest": "^26.6.3",
     "typescript": "~4.9.5"
   },

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

@@ -256,19 +256,20 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
   }
 
   public async retrieveForProcessing(queryKeyHashed: QueryKeyHash, _processingId: string): Promise<RetrieveForProcessingResponse> {
+    // TODO(ovr): Enable extended
     const rows = await this.driver.query('QUEUE RETRIEVE CONCURRENCY ? ?', [
       this.options.concurrency,
       this.prefixKey(queryKeyHashed),
     ]);
     if (rows && rows.length) {
       const addedCount = 1;
-      const active = [queryKeyHashed];
-      const toProcess = 0;
+      const active = (rows[0].active || queryKeyHashed /* backward compatibility for old Cube Store */).split(',');
+      const pending = parseInt(rows[0].pending || '0' /* backward compatibility for old Cube Store */, 10);
       const lockAcquired = true;
       const def = this.decodeQueryDefFromRow(rows[0], 'retrieveForProcessing');
 
       return [
-        addedCount, null, active, toProcess, def, lockAcquired
+        addedCount, null, active, pending, def, lockAcquired
       ];
     }
 

yarn.lock

@@ -6579,6 +6579,13 @@
   dependencies:
     "@types/node" "*"
 
+"@types/csv-write-stream@^2.0.0":
+  version "2.0.0"
+  resolved "https://registry.yarnpkg.com/@types/csv-write-stream/-/csv-write-stream-2.0.0.tgz#44dcabe9d85c0196c7fd6ffb8033658451ff9dcc"
+  integrity sha512-lAqCgs4614VpX/+U7BuR/SiEHdpet5ufc8cdjuQjhA7VVREJAT4LHKNUvAss9jf6eHH400yHhOEWbFP0RZPYrA==
+  dependencies:
+    "@types/node" "*"
+
 "@types/data-api-client@^1.2.1":
   version "1.2.3"
   resolved "https://registry.yarnpkg.com/@types/data-api-client/-/data-api-client-1.2.3.tgz#31694580015493fac8f3f3528e3a7f0a41c1741d"