dev

Author: MikeNitsenko

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

@@ -8,7 +8,7 @@ import {
   QueryColumnsResult,
   QueryKey } from '@cubejs-backend/base-driver';
 
-import { QueryCache, QueryBody, TempTable, PreAggTableToTempTable } from './QueryCache';
+import { QueryCache, QueryBody, TempTable, PreAggTableToTempTable, QueryWithParams, CacheKey } from './QueryCache';
 import { PreAggregations, PreAggregationDescription, getLastUpdatedAtTimestamp } from './PreAggregations';
 import { DriverFactory, DriverFactoryByDataSource } from './DriverFactory';
 import { QueryStream } from './QueryStream';
@@ -439,6 +439,14 @@ export class QueryOrchestrator {
     return this.preAggregations.updateRefreshEndReached();
   }
 
+  private createMetadataQuery(operation: string, params: Record<string, any>): QueryWithParams {
+    return [
+      `METADATA:${operation}`,
+      [JSON.stringify(params)],
+      { external: false, renewalThreshold: 24 * 60 * 60 }
+    ];
+  }
+
   private async queryDataSourceMetadata<T>(
     operation: MetadataOperationType,
     params: Record<string, any>,
@@ -452,74 +460,35 @@ export class QueryOrchestrator {
   ): Promise<T> {
     const {
       requestId,
-      forceRefresh,
-      expiration
+      forceRefresh = false,
+      renewalThreshold = 24 * 60 * 60,
+      expiration = 7 * 24 * 60 * 60,
     } = options;
 
-    // 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) });
-      }
-    }
-
-    // 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;
+    const metadataQuery = this.createMetadataQuery(operation, params);
+    const cacheKey: CacheKey = [`METADATA:${operation}`, metadataQuery, dataSource];
 
-    // Get queue for the given datasource
-    const queue = await this.queryCache.getQueue(dataSource);
-
-    // Execute metadata request through the queue
-    const result = await queue.executeInQueue(
-      'metadata',
-      queueKey,
+    const renewalKey = forceRefresh ? undefined : [
+      `METADATA_RENEWAL:${operation}`,
+      dataSource,
+      Math.floor(Date.now() / (renewalThreshold * 1000))
+    ];
+
+    return this.queryCache.cacheQueryResult(
+      metadataQuery,
+      [],
+      cacheKey,
+      expiration,
       {
-        ...metadataRequest,
+        renewalThreshold,
+        renewalKey,
+        forceNoCache: forceRefresh,
+        requestId,
         dataSource,
-        requestId
-      },
-      10, // priority
-      {
-        stageQueryKey: queueKey,
-        requestId
+        useInMemory: true,
+        waitForRenew: true,
       }
     );
-
-    // 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

@@ -25,7 +25,23 @@ class MockDriver {
   query(query) {
     this.executedQueries.push(query);
 
-    // Handle metadata operations using the new approach
+    // Handle metadata operations - check if query is an array with metadata operation
+    if (Array.isArray(query) && query.length > 0 && typeof query[0] === 'string') {
+      const operation = query[0];
+      if (operation === 'METADATA:GET_SCHEMAS') {
+        return this.getSchemas();
+      } else if (operation === 'METADATA:GET_TABLES_FOR_SCHEMAS') {
+        // Parse parameters from the query array
+        const params = query[1] && query[1].length > 0 ? JSON.parse(query[1][0]) : {};
+        return this.getTablesForSpecificSchemas(params.schemas || []);
+      } else if (operation === 'METADATA:GET_COLUMNS_FOR_TABLES') {
+        // Parse parameters from the query array
+        const params = query[1] && query[1].length > 0 ? JSON.parse(query[1][0]) : {};
+        return this.getColumnsForSpecificTables(params.tables || []);
+      }
+    }
+
+    // Handle metadata operations using the new approach (legacy format)
     if (query && typeof query === 'object' && query.type === 'metadata') {
       const { operation, params = {} } = query;
       if (operation === 'GET_SCHEMAS') {
@@ -38,7 +54,11 @@ class MockDriver {
       return Promise.resolve([]);
     }
 
-    // Handle regular SQL queries
+    // Handle regular SQL queries - ensure query is a string
+    if (typeof query !== 'string') {
+      return Promise.resolve([]);
+    }
+
     let promise = Promise.resolve([query]);
     if (query.match('orders_too_big')) {
       promise = promise.then((res) => new Promise(resolve => setTimeout(() => resolve(res), 3000)));