[pull] master from cube-js:master

docs-mintlify/reference/configuration/environment-variables.mdx

@@ -1975,6 +1975,23 @@ The logging level for Cube Store.
 
 See also [`CUBEJS_LOG_LEVEL`](/reference/configuration/environment-variables#cubejs_log_level).
 
+## `CUBESTORE_MAX_WS_CONNECTIONS_PER_USER`
+
+The maximum number of concurrent WebSocket connections a single authenticated
+user may hold. At the limit, that user's oldest connection is closed to admit the
+new one: a client that still needs it reconnects, while one that had forgotten
+about it simply loses it. The limit is counted per Cube Store node, so a user
+connecting to several nodes may hold up to this many on each.
+
+Use it to keep one client that leaks connections from exhausting the node's file
+descriptors for everyone else. Set it comfortably above the number of connections
+a client legitimately keeps open at once, or working connections will be recycled.
+`0` disables the limit.
+
+| Possible Values           | Default in Development | Default in Production |
+| ------------------------- | ---------------------- | --------------------- |
+| A non-negative integer    | `0`                    | `0`                   |
+
 ## `CUBESTORE_META_ADDR`
 
 The address/port pair for the Cube Store **router** node in the cluster.

packages/cubejs-api-gateway/src/gateway.ts

@@ -1133,8 +1133,6 @@ class ApiGateway {
       .refreshScheduler()
       .getCachedBuildJobs(context, tokens);
 
-    const metaCache: Map<string, any> = new Map();
-
     const response: PreAggJobStatusItem[] = await Promise.all(
       jobs.map(async ({ job, token }) => {
         if (!job) {
@@ -1147,12 +1145,15 @@ class ApiGateway {
         const ctx = { ...context, ...job.context };
         const orchestrator = await this.getAdapterApi(ctx);
         const compiler = await this.getCompilerApi(ctx);
+        // TODO(1.8): drop the fallback, no job posted by 1.7 can still be in the cache.
+        const dataSource = job.dataSource || (await compiler.preAggregations())
+          .find(pa => pa.id === job.preagg)?.dataSource;
         const selector: PreAggsSelector = {
           cubes: [job.preagg.split('.')[0]],
           preAggregations: [job.preagg],
           contexts: [job.context],
           timezones: [job.timezone],
-          dataSources: [job.dataSource],
+          dataSources: [dataSource],
         };
         if (
           job.status.indexOf('done') === 0 ||
@@ -1170,6 +1171,7 @@ class ApiGateway {
           const status = await this.getPreAggJobQueueStatus(
             orchestrator,
             job,
+            dataSource,
           );
           if (status) {
             // returning queued status
@@ -1180,18 +1182,13 @@ class ApiGateway {
               selector,
             };
           } else {
-            const metaCacheKey = JSON.stringify(ctx);
-            if (!metaCache.has(metaCacheKey)) {
-              metaCache.set(metaCacheKey, await compiler.metaConfigExtended(context, ctx));
-            }
-
             // checking and fetching result status
             const s = await this.getPreAggJobResultStatus(
               ctx.requestId,
               orchestrator,
               compiler,
-              metaCache.get(metaCacheKey),
               job,
+              dataSource,
               token,
             );
 
@@ -1225,10 +1222,11 @@ class ApiGateway {
   private async getPreAggJobQueueStatus(
     orchestrator: any,
     job: PreAggJob,
+    dataSource?: string,
   ): Promise<false | string> {
     let inQueue = false;
     let status: string = 'n/a';
-    const queuedList = await orchestrator.getPreAggregationQueueStates(job.dataSource);
+    const queuedList = await orchestrator.getPreAggregationQueueStates(dataSource);
     queuedList.forEach((item) => {
       if (
         item.queryHandler &&
@@ -1264,19 +1262,18 @@ class ApiGateway {
     requestId: string,
     orchestrator: any,
     compiler: any,
-    metadata: any,
     job: PreAggJob,
+    dataSource: string | undefined,
     token: string,
   ): Promise<string> {
     const preaggs = await compiler.preAggregations();
     const preagg = preaggs.find(pa => pa.id === job.preagg);
     if (preagg) {
-      const cube = metadata.cubeDefinitions[preagg.cube];
       const [, status]: [boolean, string] =
         await orchestrator.isPartitionExist(
           requestId,
           preagg.preAggregation.external,
-          cube.dataSource,
+          dataSource,
           compiler.preAggregationsSchema,
           job.target,
           job.key,

packages/cubejs-api-gateway/test/index.test.ts

@@ -1231,11 +1231,112 @@ describe('API Gateway', () => {
         }),
       };
 
-      const status = await (apiGateway as any).getPreAggJobQueueStatus(orchestrator, job);
+      const status = await (apiGateway as any).getPreAggJobQueueStatus(orchestrator, job, job.dataSource);
 
       expect(orchestrator.getPreAggregationQueueStates).toHaveBeenCalledWith(job.dataSource);
       expect(status).toEqual('processing');
     });
+
+    // https://github.com/cube-js/cube/issues/11615
+    describe('job result status', () => {
+      const jobFor = (overrides: Partial<PreAggJob> = {}): PreAggJob => ({
+        request: 'request-id',
+        context: { securityContext: {} },
+        preagg: 'orders_test.main',
+        table: 'orders_test_main',
+        target: 'orders_test_main_20200101',
+        structure: 'structure-version',
+        content: 'content-version',
+        updated: 1,
+        key: [],
+        status: 'posted',
+        timezone: 'UTC',
+        dataSource: 'test_ds',
+        ...overrides,
+      });
+
+      const compiler = {
+        preAggregations: async () => ([
+          { id: 'orders_dep.main', cube: 'orders_dep', dataSource: 'dep_ds', preAggregation: { external: false } },
+          { id: 'orders_test.main', cube: 'orders_test', dataSource: 'model_ds', preAggregation: { external: true } },
+        ]),
+        preAggregationsSchema: 'stb_pre_aggregations',
+      };
+
+      const resultStatus = async (job: PreAggJob, orchestrator: any) => {
+        const apiGateway = Object.create(ApiGateway.prototype);
+        return (apiGateway as any).getPreAggJobResultStatus(
+          job.request,
+          orchestrator,
+          compiler,
+          job,
+          job.dataSource,
+          'job-token',
+        );
+      };
+
+      test('is checked on the data source resolved for the job', async () => {
+        const orchestrator = { isPartitionExist: jest.fn(async () => [true, 'done']) };
+        const job = jobFor();
+
+        await expect(resultStatus(job, orchestrator)).resolves.toEqual('done');
+
+        expect(orchestrator.isPartitionExist).toHaveBeenCalledWith(
+          job.request,
+          true,
+          'test_ds',
+          compiler.preAggregationsSchema,
+          job.target,
+          job.key,
+          'job-token',
+        );
+      });
+
+      test('reports a pre-aggregation the model no longer has', async () => {
+        const orchestrator = { isPartitionExist: jest.fn() };
+
+        await expect(
+          resultStatus(jobFor({ preagg: 'orders_test.dropped' }), orchestrator)
+        ).resolves.toEqual('pre_agg_not_found');
+
+        expect(orchestrator.isPartitionExist).not.toHaveBeenCalled();
+      });
+
+      // Everywhere, because a queue lookup left on the default data source finds nothing
+      // and a build that is still scheduled then reads as a missing partition.
+      // TODO(1.8): goes away with the fallback in preAggregationsJobsGET.
+      test('a job with no recorded data source resolves it from the model everywhere', async () => {
+        const apiGateway = Object.create(ApiGateway.prototype);
+        const job = jobFor({ dataSource: undefined as any, status: 'scheduled' });
+        const orchestrator = {
+          getPreAggregationQueueStates: jest.fn(async () => []),
+          isPartitionExist: jest.fn(async () => [false, 'missing_partition']),
+        };
+        apiGateway.refreshScheduler = () => ({
+          getCachedBuildJobs: async () => [{ job, token: 'job-token' }],
+        });
+        apiGateway.getAdapterApi = async () => orchestrator;
+        apiGateway.getCompilerApi = async () => compiler;
+
+        const [item] = await (apiGateway as any).preAggregationsJobsGET(
+          { requestId: 'request-id' },
+          ['job-token'],
+        );
+
+        expect(item.status).toEqual('missing_partition');
+        expect(item.selector.dataSources).toEqual(['model_ds']);
+        expect(orchestrator.getPreAggregationQueueStates).toHaveBeenCalledWith('model_ds');
+        expect(orchestrator.isPartitionExist).toHaveBeenCalledWith(
+          'request-id',
+          true,
+          'model_ds',
+          compiler.preAggregationsSchema,
+          job.target,
+          job.key,
+          'job-token',
+        );
+      });
+    });
   });
 
   describe('healtchecks', () => {

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

@@ -150,6 +150,10 @@ export type LoadPreAggregationResult = {
   partitionRange?: QueryDateRange;
   isMultiTableUnion?: boolean;
   usageTargetTableNames?: Record<string, string>;
+  type?: 'rollup' | 'originalSql';
+  preAggregationId?: string;
+  dataSource?: string;
+  timezone?: string;
 };
 
 export type PreAggregationTableToTempTable = [string, LoadPreAggregationResult];
@@ -574,6 +578,9 @@ export class PreAggregations {
           const usedPreAggregation = {
             ...loadResult,
             type: p.type,
+            preAggregationId: p.preAggregationId,
+            dataSource: p.dataSource || 'default',
+            timezone: p.timezone,
           };
           if (!usedPreAggregation.isMultiTableUnion) {
             await this.addTableUsed(usedPreAggregation.targetTableName);

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

@@ -252,7 +252,7 @@ export class QueryOrchestrator {
       // /cubejs-system/v1/pre-aggregations/jobs endpoint).
       if (queryBody.isJob) {
         return preAggregationsTablesToTempTables.map((pa) => ({
-          preAggregation: queryBody.preAggregations[0].preAggregationId,
+          preAggregation: pa[1].preAggregationId || queryBody.preAggregations[0].preAggregationId,
           tableName: pa[0],
           ...pa[1],
         }));

packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts

@@ -337,6 +337,34 @@ describe('PreAggregations', () => {
       expect(result[0][1].targetTableName).toMatch(/stb_pre_aggregations.orders_number_and_count20191101_kjypcoio_5yftl5il/);
       expect(result[0][1].lastUpdatedAt).toEqual(12345000);
     });
+
+    // A jobed build gets back a flat list of entries and has to tell them apart.
+    // https://github.com/cube-js/cube/issues/11615
+    test('each entry carries the identity of the descriptor it was built from', async () => {
+      const { preAggregationsTablesToTempTables: result } = await preAggregations!.loadAllPreAggregationsIfNeeded(
+        createBasicQuery({
+          cacheMode: 'must-revalidate',
+          preAggregations: [{
+            ...basicQuery.preAggregations[0],
+            preAggregationId: 'Orders.numberAndCount',
+            dataSource: 'orders_ds',
+            timezone: 'America/Los_Angeles',
+          }],
+        })
+      );
+
+      expect(result[0][1]).toMatchObject({
+        preAggregationId: 'Orders.numberAndCount',
+        dataSource: 'orders_ds',
+        timezone: 'America/Los_Angeles',
+      });
+    });
+
+    test('an entry built without a named data source falls back to the default one', async () => {
+      const { preAggregationsTablesToTempTables: result } = await preAggregations!.loadAllPreAggregationsIfNeeded(basicQueryWithRenew);
+
+      expect(result[0][1].dataSource).toEqual('default');
+    });
   });
 
   describe('loadAllPreAggregationsIfNeeded with external rollup and writable source', () => {

packages/cubejs-query-orchestrator/test/unit/QueryOrchestrator.jobs.test.ts

@@ -0,0 +1,54 @@
+/* eslint-disable @typescript-eslint/no-explicit-any */
+import { QueryOrchestrator } from '../../src';
+
+// A jobed build returns one entry per built pre-aggregation — the requested partition plus
+// every dependency it had to build first — and each entry becomes its own polling token.
+// https://github.com/cube-js/cube/issues/11615
+describe('QueryOrchestrator jobed build', () => {
+  const entry = (tableName: string, overrides: Record<string, any> = {}) => ([
+    tableName,
+    {
+      targetTableName: `${tableName}_kjypcoio_5yftl5il_1593709044209`,
+      refreshKeyValues: [],
+      lastUpdatedAt: 1593709044209,
+      ...overrides,
+    },
+  ]);
+
+  const fetchJob = async (preAggregationsTablesToTempTables: any[], preAggregations: any[]) => {
+    const orchestrator = Object.create(QueryOrchestrator.prototype);
+    orchestrator.rollupOnlyMode = false;
+    orchestrator.preAggregations = {
+      loadAllPreAggregationsIfNeeded: async () => ({
+        preAggregationsTablesToTempTables,
+        values: null,
+      }),
+    };
+
+    return orchestrator.fetchQuery({ isJob: true, preAggregations });
+  };
+
+  test('labels every entry with its own pre-aggregation and data source', async () => {
+    const job = await fetchJob(
+      [
+        entry('stb_pre_aggregations.orders_main', { preAggregationId: 'Orders.main', dataSource: 'orders_ds' }),
+        entry('stb_pre_aggregations.orders_rollup', { preAggregationId: 'Orders.rollup', dataSource: 'default', timezone: 'UTC' }),
+      ],
+      [{ preAggregationId: 'Orders.main' }, { preAggregationId: 'Orders.rollup' }],
+    );
+
+    expect(job).toMatchObject([
+      { preAggregation: 'Orders.main', tableName: 'stb_pre_aggregations.orders_main', dataSource: 'orders_ds' },
+      { preAggregation: 'Orders.rollup', tableName: 'stb_pre_aggregations.orders_rollup', dataSource: 'default', timezone: 'UTC' },
+    ]);
+  });
+
+  test('falls back to the requested pre-aggregation for an entry without an id', async () => {
+    const job = await fetchJob(
+      [entry('stb_pre_aggregations.orders_rollup')],
+      [{ preAggregationId: 'Orders.rollup' }],
+    );
+
+    expect(job[0].preAggregation).toEqual('Orders.rollup');
+  });
+});

packages/cubejs-schema-compiler/src/compiler/CubeEvaluator.ts

@@ -160,6 +160,7 @@ export type PreAggregationInfo = {
   preAggregationName: string,
   preAggregation: any,
   cube: string,
+  dataSource: string,
   references: PreAggregationReferences,
   refreshKey: unknown,
   indexesReferences: unknown,
@@ -963,6 +964,7 @@ export class CubeEvaluator extends CubeSymbols {
               preAggregationName,
               preAggregation: preAggregations[preAggregationName],
               cube,
+              dataSource: this.evaluatedCubes[cube].dataSource || 'default',
               references: this.evaluatePreAggregationReferences(cube, preAggregations[preAggregationName]),
               refreshKey,
               indexesReferences: indexes && Object.keys(indexes).reduce((obj, indexName) => {

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

@@ -790,8 +790,11 @@ export class RefreshScheduler {
                       metadata: queryingOptions.metadata,
                       isJob: true,
                     });
-                    job[0].dataSource = partition.dataSource;
-                    job[0].timezone = partition.timezone;
+                    job.forEach((j: JobedPreAggregation) => {
+                      j.dataSource = j.dataSource || partition.dataSource;
+                      j.timezone = j.timezone || partition.timezone;
+                    });
+
                     return job;
                   }
                 )

packages/cubejs-server-core/test/unit/RefreshScheduler.test.ts

@@ -683,6 +683,7 @@ describe('Refresh Scheduler', () => {
                 external: false,
               },
               cube: 'Foo',
+              dataSource: 'default',
               references: {
                 dimensions: [],
                 measures: ['Foo.count'],
@@ -902,6 +903,13 @@ describe('Refresh Scheduler', () => {
       const buildJobs = await refreshScheduler.getCachedBuildJobs(ctx, jobs);
       const allTokensExist = jobs.every(token => buildJobs.some(job => job.token === token));
       expect(allTokensExist).toBeTruthy();
+
+      // Not only the first entry: every entry of a posted job is its own poll token.
+      // https://github.com/cube-js/cube/issues/11615
+      buildJobs.forEach(({ job }) => {
+        expect(job?.dataSource).toEqual('default');
+        expect(['UTC', 'America/Los_Angeles']).toContain(job?.timezone);
+      });
     });
 
     test('Only `first` pre-aggregation', async () => {