dev

Author: MikeNitsenko

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

@@ -1,5 +1,4 @@
 import * as stream from 'stream';
-import * as crypto from 'crypto';
 import R from 'ramda';
 import { getEnv } from '@cubejs-backend/shared';
 import { CubeStoreDriver } from '@cubejs-backend/cubestore-driver';
@@ -440,57 +439,6 @@ export class QueryOrchestrator {
     return this.preAggregations.updateRefreshEndReached();
   }
 
-  private createMetadataHash(operation: MetadataOperationType, params: Record<string, any>): string {
-    if (!params || Object.keys(params).length === 0) {
-      return 'empty';
-    }
-
-    const hashData: string[] = [];
-
-    switch (operation) {
-      case MetadataOperationType.GET_SCHEMAS:
-        return 'all_schemas';
-
-      case MetadataOperationType.GET_TABLES_FOR_SCHEMAS:
-        if (params.schemas && Array.isArray(params.schemas)) {
-          hashData.push(...params.schemas.map(schema => schema.schema_name).sort());
-        }
-        break;
-
-      case MetadataOperationType.GET_COLUMNS_FOR_TABLES:
-        if (params.tables && Array.isArray(params.tables)) {
-          hashData.push(...params.tables.map(table => `${table.schema_name}.${table.table_name}`).sort());
-        }
-        break;
-
-      default:
-        return crypto
-          .createHash('sha256')
-          .update(JSON.stringify(params))
-          .digest('hex')
-          .substring(0, 16);
-    }
-
-    if (hashData.length === 0) {
-      return 'empty';
-    }
-
-    return crypto
-      .createHash('sha256')
-      .update(hashData.join('|'))
-      .digest('hex')
-      .substring(0, 16);
-  }
-
-  private createMetadataRequest(operation: string, params: Record<string, any>) {
-    return {
-      operation,
-      params,
-      external: false,
-      type: 'metadata'
-    };
-  }
-
   private async queryDataSourceMetadata<T>(
     operation: MetadataOperationType,
     params: Record<string, any>,
@@ -504,32 +452,74 @@ export class QueryOrchestrator {
   ): Promise<T> {
     const {
       requestId,
+      forceRefresh,
+      expiration
     } = options;
 
-    const paramsHash = this.createMetadataHash(operation, params);
+    // Create a unique cache key for this metadata request
+    const cacheKey = `METADATA:${operation}:${dataSource}:${JSON.stringify(params)}`;
+    const cacheDriver = this.queryCache.getCacheDriver();
+    
+    // Check cache first (unless forceRefresh is true)
+    if (!forceRefresh) {
+      try {
+        const cachedResult = await cacheDriver.get(cacheKey);
+        if (cachedResult && cachedResult.result) {
+          this.logger('Found cached metadata result', { cacheKey, operation, dataSource });
+          return cachedResult.result;
+        }
+      } catch (e) {
+        this.logger('Error reading from cache', { cacheKey, error: e instanceof Error ? e.message : String(e) });
+      }
+    }
 
-    const metadataRequest = this.createMetadataRequest(operation, params);
-    // Create a unique string key for this metadata request
-    const cacheKey = `METADATA:${operation}:${paramsHash}:${dataSource}`;
+    // If not in cache or forceRefresh, execute through queue
+    const metadataRequest = {
+      operation,
+      params,
+      external: false,
+      type: 'metadata'
+    };
+
+    // Create unique queue key for forceRefresh to avoid conflicts
+    const queueKey = forceRefresh
+      ? `${cacheKey}:${new Date().getTime()}`
+      : cacheKey;
 
     // Get queue for the given datasource
     const queue = await this.queryCache.getQueue(dataSource);
 
     // Execute metadata request through the queue
-    return queue.executeInQueue(
+    const result = await queue.executeInQueue(
       'metadata',
-      cacheKey,
+      queueKey,
       {
         ...metadataRequest,
         dataSource,
         requestId
       },
       10, // priority
       {
-        stageQueryKey: cacheKey,
+        stageQueryKey: queueKey,
         requestId
       }
     );
+
+    // Store result in cache driver (unless forceRefresh)
+    if (!forceRefresh && result) {
+      try {
+        const cacheValue = {
+          time: new Date().getTime(),
+          result
+        };
+        await cacheDriver.set(cacheKey, cacheValue, expiration || (24 * 60 * 60)); // Default 24 hours
+        this.logger('Stored metadata result in cache', { cacheKey, operation, dataSource });
+      } catch (e) {
+        this.logger('Error storing in cache', { cacheKey, error: e instanceof Error ? e.message : String(e) });
+      }
+    }
+
+    return result;
   }
 
   /**

packages/cubejs-query-orchestrator/test/unit/QueryOrchestrator.test.js

@@ -1996,17 +1996,30 @@ describe('QueryOrchestrator', () => {
       test('should cache results based on schema list', async () => {
         const schemas = [{ schema_name: 'public' }];
 
-        // First call
+        // First call - will execute and store in cache
         await metadataOrchestrator.queryTablesForSchemas(schemas);
+
+        // Add a delay to ensure the first query has completed and cached its result
+        await new Promise(resolve => setTimeout(resolve, 100));
+
+        // Clear the mock calls
+        metadataMockDriver.getTablesForSpecificSchemas.mockClear();
+
+        // Create equivalent but different object instance
+        // Our hash function should handle this correctly
+        const schemas2 = [{ schema_name: 'public' }];
+
         // Second call should use cache
-        const result = await metadataOrchestrator.queryTablesForSchemas(schemas);
+        const result = await metadataOrchestrator.queryTablesForSchemas(schemas2);
 
         expect(result).toEqual([
           { schema_name: 'public', table_name: 'users' },
           { schema_name: 'public', table_name: 'orders' },
           { schema_name: 'public', table_name: 'products' }
         ]);
-        expect(metadataMockDriver.getTablesForSpecificSchemas).toHaveBeenCalledTimes(1);
+
+        // Verify driver wasn't called again
+        expect(metadataMockDriver.getTablesForSpecificSchemas).not.toHaveBeenCalled();
       });
 
       test('should handle empty schema list', async () => {
@@ -2086,10 +2099,21 @@ describe('QueryOrchestrator', () => {
       test('should cache results based on table list', async () => {
         const tables = [{ schema_name: 'public', table_name: 'users' }];
 
-        // First call
+        // First call - will execute and store in cache
         await metadataOrchestrator.queryColumnsForTables(tables);
+
+        // Add a delay to ensure the first query has completed and cached its result
+        await new Promise(resolve => setTimeout(resolve, 100));
+
+        // Clear the mock calls
+        metadataMockDriver.getColumnsForSpecificTables.mockClear();
+
+        // Create equivalent but different object instance
+        // Our hash function should handle this correctly
+        const tables2 = [{ schema_name: 'public', table_name: 'users' }];
+
         // Second call should use cache
-        const result = await metadataOrchestrator.queryColumnsForTables(tables);
+        const result = await metadataOrchestrator.queryColumnsForTables(tables2);
 
         expect(result).toEqual([
           {
@@ -2114,7 +2138,9 @@ describe('QueryOrchestrator', () => {
             attributes: ['UNIQUE']
           }
         ]);
-        expect(metadataMockDriver.getColumnsForSpecificTables).toHaveBeenCalledTimes(1);
+
+        // Verify driver wasn't called again
+        expect(metadataMockDriver.getColumnsForSpecificTables).not.toHaveBeenCalled();
       });
 
       test('should handle empty table list', async () => {