feat(server-core, query-orchestrator): Use the fastest xxhash for cache keys and model versions

packages/cubejs-backend-shared/package.json

@@ -35,6 +35,7 @@
     "typescript": "~5.2.2"
   },
   "dependencies": {
+    "@node-rs/xxhash": "^1.7.6",
     "@oclif/color": "^0.1.2",
     "bytes": "^3.1.2",
     "cli-progress": "^3.9.0",

packages/cubejs-backend-shared/src/promises.ts

@@ -1,6 +1,5 @@
 /* eslint-disable arrow-body-style,no-restricted-syntax */
-import crypto from 'crypto';
-
+import { xxh3 } from '@node-rs/xxhash';
 import { Optional } from './type-helpers';
 
 export type PromiseLock = {
@@ -273,9 +272,7 @@ export const asyncDebounce = <Ret, Arguments>(
   const cache = new Map<string, Promise<Ret>>();
 
   return async (...args: Arguments[]) => {
-    const key = crypto.createHash('md5')
-      .update(args.map((v) => JSON.stringify(v)).join(','))
-      .digest('hex');
+    const key = xxh3.xxh64(args.map((v) => JSON.stringify(v)).join(',')).toString(16);
 
     if (cache.has(key)) {
       return <Promise<Ret>>cache.get(key);

packages/cubejs-cubestore-driver/package.json

@@ -30,6 +30,7 @@
     "@cubejs-backend/cubestore": "1.2.30",
     "@cubejs-backend/native": "1.2.30",
     "@cubejs-backend/shared": "1.2.30",
+    "@node-rs/xxhash": "^1.7.6",
     "csv-write-stream": "^2.0.0",
     "flatbuffers": "23.3.3",
     "fs-extra": "^9.1.0",

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

@@ -1,4 +1,4 @@
-import crypto from 'crypto';
+import { xxh3 } from '@node-rs/xxhash';
 import {
   QueueDriverInterface,
   QueueDriverConnectionInterface,
@@ -22,7 +22,7 @@ import { CubeStoreDriver } from './CubeStoreDriver';
 
 function hashQueryKey(queryKey: QueryKey, processUid?: string): QueryKeyHash {
   processUid = processUid || getProcessUid();
-  const hash = crypto.createHash('md5').update(JSON.stringify(queryKey)).digest('hex');
+  const hash = xxh3.xxh128(JSON.stringify(queryKey)).toString(16);
 
   if (typeof queryKey === 'object' && queryKey.persistent) {
     return `${hash}@${processUid}` as any;
@@ -83,7 +83,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
     values.push(JSON.stringify(data));
 
     const rows = await this.driver.query(`QUEUE ADD PRIORITY ?${options.orphanedTimeout ? ' ORPHANED ?' : ''} ? ?`, values);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return [
         rows[0].added === 'true' ? 1 : 0,
         rows[0].id ? parseInt(rows[0].id, 10) : null,
@@ -104,7 +104,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
       // queryKeyHash as compatibility fallback
       queueId || this.prefixKey(hash),
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return this.decodeQueryDefFromRow(rows[0], 'cancelQuery');
     }
 
@@ -149,7 +149,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
     const rows = await this.driver.query('CACHE INCR ?', [
       `${this.options.redisQueuePrefix}:PROCESSING_COUNTER`
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return rows[0].value;
     }
 
@@ -186,7 +186,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
     const rows = await this.driver.query('QUEUE RESULT ?', [
       this.prefixKey(this.redisHash(queryKey)),
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return this.decodeQueryDefFromRow(rows[0], 'getResult');
     }
 
@@ -245,7 +245,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
     const rows = await this.driver.query('QUEUE GET ?', [
       queueId || this.prefixKey(hash),
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return this.decodeQueryDefFromRow(rows[0], 'getQueryDef');
     }
 
@@ -271,7 +271,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
       this.options.concurrency,
       this.prefixKey(hash),
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       const active = rows[0].active ? (rows[0].active).split(',') as unknown as QueryKeyHash[] : [];
       const pending = parseInt(rows[0].pending, 10);
 
@@ -302,7 +302,7 @@ class CubestoreQueueDriverConnection implements QueueDriverConnectionInterface {
       // queryKeyHash as compatibility fallback
       queueId || this.prefixKey(hash),
     ]);
-    if (rows && rows.length) {
+    if (rows?.length) {
       return this.decodeQueryDefFromRow(rows[0], 'getResultBlocking');
     }
 
@@ -348,8 +348,8 @@ export class CubeStoreQueueDriver implements QueueDriverInterface {
       return this.connection;
     }
 
-    // eslint-disable-next-line no-return-assign
-    return this.connection = await this.driverFactory();
+    this.connection = await this.driverFactory();
+    return this.connection;
   }
 
   public async createConnection(): Promise<CubestoreQueueDriverConnection> {

packages/cubejs-query-orchestrator/package.json

@@ -32,6 +32,7 @@
     "@cubejs-backend/base-driver": "1.2.30",
     "@cubejs-backend/cubestore-driver": "1.2.30",
     "@cubejs-backend/shared": "1.2.30",
+    "@node-rs/xxhash": "^1.7.6",
     "csv-write-stream": "^2.0.0",
     "generic-pool": "^3.8.2",
     "lru-cache": "^5.1.1",

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

@@ -29,6 +29,7 @@ export function version(cacheKey) {
   let result = '';
 
   const hashCharset = 'abcdefghijklmnopqrstuvwxyz012345';
+  // TODO: switch to use '@node-rs/xxhash' instead of crypto
   const digestBuffer = crypto.createHash('md5').update(JSON.stringify(cacheKey)).digest();
 
   let residue = 0;

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

@@ -1,4 +1,4 @@
-import crypto from 'crypto';
+import { xxh3 } from '@node-rs/xxhash';
 
 import { getProcessUid } from '@cubejs-backend/shared';
 import { QueryKey, QueryKeyHash } from '@cubejs-backend/base-driver';
@@ -18,16 +18,11 @@ export function getCacheHash(queryKey: QueryKey | CacheKey, processUid?: string)
     return queryKey as any;
   }
 
+  const hash = xxh3.xxh128(JSON.stringify(queryKey)).toString(16);
+
   if (typeof queryKey === 'object' && 'persistent' in queryKey && queryKey.persistent) {
-    return `${crypto
-      .createHash('md5')
-      .update(JSON.stringify(queryKey))
-      .digest('hex')
-    }@${processUid}` as any;
+    return `${hash}@${processUid}` as any;
   } else {
-    return crypto
-      .createHash('md5')
-      .update(JSON.stringify(queryKey))
-      .digest('hex') as any;
+    return hash as any;
   }
 }

packages/cubejs-query-orchestrator/test/unit/QueryQueue.abstract.ts

@@ -323,7 +323,7 @@ export const QueryQueueTest = (name: string, options: QueryQueueTestOptions = {}
       expect(processUidRE.test(key3.split('@')[1])).toBeTruthy();
 
       if (options.cacheAndQueueDriver === 'cubestore') {
-        expect(queue.redisHash('string')).toBe('095d71cf12556b9d5e330ad575b3df5d');
+        expect(queue.redisHash('string')).toBe('7aeaa28c6474df82129b9fe0d65c3ae4');
       } else {
         expect(queue.redisHash('string')).toBe('string');
       }

packages/cubejs-server-core/package.json

@@ -37,6 +37,7 @@
     "@cubejs-backend/schema-compiler": "1.2.30",
     "@cubejs-backend/shared": "1.2.30",
     "@cubejs-backend/templates": "1.2.30",
+    "@node-rs/xxhash": "^1.7.6",
     "codesandbox-import-utils": "^2.1.12",
     "cross-spawn": "^7.0.1",
     "fs-extra": "^8.1.0",

packages/cubejs-server-core/src/core/CompilerApi.js

@@ -1,4 +1,4 @@
-import crypto from 'crypto';
+import { xxh3 } from '@node-rs/xxhash';
 import R from 'ramda';
 import { createQuery, compile, queryClass, PreAggregations, QueryFactory } from '@cubejs-backend/schema-compiler';
 import { v4 as uuidv4, parse as uuidParse, stringify as uuidStringify } from 'uuid';
@@ -55,7 +55,7 @@
 
     if (this.options.devServer || this.options.fastReload) {
       const files = await this.repository.dataSchemaFiles();
-      compilerVersion += `_${crypto.createHash('md5').update(JSON.stringify(files)).digest('hex')}`;
+      compilerVersion += `_${xxh3.xxh64(JSON.stringify(files)).toString(16)}`;
     }
 
     if (!this.compilers || this.compilerVersion !== compilerVersion) {
@@ -224,7 +224,7 @@
 
   hashRequestContext(context) {
     if (!context.__hash) {
-      context.__hash = crypto.createHash('md5').update(JSON.stringify(context)).digest('hex');
+      context.__hash = xxh3.xxh64(JSON.stringify(context)).toString(16);
     }
     return context.__hash;
   }
@@ -505,7 +505,7 @@
     const visibiliyMask = JSON.stringify(isMemberVisibleInContext, Object.keys(isMemberVisibleInContext).sort());
     // This hash will be returned along the modified meta config and can be used
     // to distinguish between different "schema versions" after DAP visibility is applied
-    const visibilityMaskHash = crypto.createHash('sha256').update(visibiliyMask).digest('hex');
+    const visibilityMaskHash = xxh3.xxh64(visibiliyMask).toString(16);
 
     return {
       cubes: cubes
@@ -523,10 +523,7 @@
   }
 
   mixInVisibilityMaskHash(compilerId, visibilityMaskHash) {
-    const uuidBytes = uuidParse(compilerId);
-    const hashBytes = Buffer.from(visibilityMaskHash, 'hex');
-    return uuidv4({ random: crypto.createHash('sha256').update(uuidBytes).update(hashBytes).digest()
-      .subarray(0, 16) });
+    return uuidv4({ random: Buffer.from(xxh3.xxh64(`${compilerId}${visibilityMaskHash}`).toString()) });
   }
 
   async metaConfig(requestContext, options = {}) {
@@ -542,8 +539,8 @@
       return {
         cubes: patchedCubes,
         // This compilerId is primarily used by the cubejs-backend-native or caching purposes.
-        // By default it doesn't account for member visibility changes introduced above by DAP.
-        // Here we're modifying the originila compilerId in a way that it's distinct for
+        // By default, it doesn't account for member visibility changes introduced above by DAP.
+        // Here we're modifying the original compilerId in a way that it's distinct for
         // distinct schema versions while still being a valid UUID.
         compilerId: visibilityMaskHash ? this.mixInVisibilityMaskHash(compilers.compilerId, visibilityMaskHash) : compilers.compilerId,
       };

packages/cubejs-server-core/src/core/RefreshScheduler.ts

@@ -1,7 +1,7 @@
 import R from 'ramda';
 import pLimit from 'p-limit';
 import { v4 as uuidv4 } from 'uuid';
-import crypto from 'crypto';
+import { xxh3 } from '@node-rs/xxhash';
 import { Required } from '@cubejs-backend/shared';
 import {
   PreAggregationDescription,
@@ -127,10 +127,7 @@ function getPreAggsJobsList(
  * Returns MD5 hash token of the job object.
  */
 function getPreAggJobToken(job: PreAggJob) {
-  return crypto
-    .createHash('md5')
-    .update(JSON.stringify(job))
-    .digest('hex');
+  return xxh3.xxh64(JSON.stringify(job)).toString(16);
 }
 
 export class RefreshScheduler {

packages/cubejs-testing/test/smoke-cubesql.test.ts

@@ -179,23 +179,23 @@ describe('SQL API', () => {
       }
 
       it('regular query', async () => {
-        expect(await generateSql(`SELECT SUM(totalAmount) AS total FROM Orders;`)).toMatchSnapshot();
+        expect(await generateSql('SELECT SUM(totalAmount) AS total FROM Orders;')).toMatchSnapshot();
       });
 
       it('regular query with missing column', async () => {
-        expect(await generateSql(`SELECT SUM(foobar) AS total FROM Orders;`)).toMatchSnapshot();
+        expect(await generateSql('SELECT SUM(foobar) AS total FROM Orders;')).toMatchSnapshot();
       });
 
       it('regular query with parameters', async () => {
-        expect(await generateSql(`SELECT SUM(totalAmount) AS total FROM Orders WHERE status = 'foo';`)).toMatchSnapshot();
+        expect(await generateSql('SELECT SUM(totalAmount) AS total FROM Orders WHERE status = \'foo\';')).toMatchSnapshot();
       });
 
       it('strictly post-processing', async () => {
-        expect(await generateSql(`SELECT version();`)).toMatchSnapshot();
+        expect(await generateSql('SELECT version();')).toMatchSnapshot();
       });
 
       it('strictly post-processing with disabled post-processing', async () => {
-        expect(await generateSql(`SELECT version();`, true)).toMatchSnapshot();
+        expect(await generateSql('SELECT version();', true)).toMatchSnapshot();
       });
 
       it('double aggregation post-processing', async () => {

packages/cubejs-testing/test/smoke-duckdb.test.ts

@@ -90,7 +90,7 @@ describe('duckdb', () => {
       ]
     });
 
-    // There are 2 'processed' orders 
+    // There are 2 'processed' orders
     expect(response.rawData()[0]['Orders.count']).toBe('2');
   });