feat(server-core,api-gateway): Allow manual rebuilding pre-aggregation partitions within date Range (#9342)

* spelling

* add dateRange to PreAggsJobsRequest

* pass dateRange to preAggregations(filter)

* a bit of code polishment

* add preAggsJobsRequestSchema validator

* fix PreAggsJobsRequest types

* specify type for parseLocalDate()

* refresh only matched partitions

* add html reporters

* some code polishment

* fix/add some tests to dateParser()

* code polishment

* add another check

* add tests for api-gw preagg jobs endpoint

* align refactor some types

* code polish

* refresh only matched partitions fix

* code polish

* add tests for pre-agg jobs

* jest reporters

* add tests for refreshScheduler.getCachedBuildJobs()

* improve caching in pre-agg-load-cache

* increase delay in tests to pass

* use asyncDebounce for caching in pre-agg-load-cache

* Revert "use asyncDebounce for caching in pre-agg-load-cache"

This reverts commit a353cf1.

* Revert "improve caching in pre-agg-load-cache"

This reverts commit f8f640c.

* set coverageReporters

* add docs

* fix regexp

* fix dateParser + tests

* fix preAggsJobsRequestSchema validator

* fix error message

Co-authored-by: Igor Lukanin <[email protected]>

* simplify regexp in dateParser

* tests polish

* little fix

---------

Co-authored-by: Igor Lukanin <[email protected]>

Author: KSDaemon

docs/pages/product/apis-integrations/rest-api/reference.mdx

@@ -273,6 +273,7 @@ Trigger pre-aggregation build jobs or retrieve statuses of such jobs.
 | `selector.datasources`     | Array of data source names which have pre-aggregations defined | ❌       |
 | `selector.cubes`           | Array of cube names which contain pre-aggregations             | ❌       |
 | `selector.preAggregations` | Array of pre-aggregation names                                 | ❌       |
+| `selector.dateRange`       | Date Range tuple ['range-date-start', 'range-date-end']        | ❌       |
 
 To trigger pre-aggregation builds, send a `POST` request with a payload
 including `post` as the `action` and `selector` properties. The response will
@@ -334,6 +335,27 @@ curl \
   https://localhost:4000/cubejs-api/v1/pre-aggregations/jobs
 ```
 
+Example request triggering builds of the `main` pre-aggregation defined in the
+`orders` cube within date range with some security context data
+and an `America/Los_Angeles` timezone:
+
+```bash{outputLines: 2-13}
+curl \
+  -d '{
+    "action": "post",
+    "selector": {
+      "contexts": [{ "securityContext": { "tenantId": "tenant1" } }],
+      "timezones": ["America/Los_Angeles"],
+      "preAggregations": ["orders.main"],
+      "dateRange": ["2020-01-01", "2020-02-01"]
+    }
+  }' \
+  -H "Authorization: EXAMPLE-API-TOKEN" \
+  -H "Content-Type: application/json" \
+  -X POST \
+  https://localhost:4000/cubejs-api/v1/pre-aggregations/jobs
+```
+
 Example response:
 
 ```json

packages/cubejs-api-gateway/package.json

@@ -73,6 +73,7 @@
   "jest": {
     "testEnvironment": "node",
     "collectCoverage": false,
+    "coverageReporters": ["text", "html"],
     "coverageDirectory": "coverage/",
     "collectCoverageFrom": [
       "dist/src/**/*.js",

packages/cubejs-api-gateway/src/dateParser.js

@@ -65,18 +65,19 @@ export function dateParser(dateString, timezone, now = new Date()) {
       moment.tz(timezone).endOf('day').add(1, 'day')
     ];
   } else if (dateString.match(/^from (.*) to (.*)$/)) {
-    // eslint-disable-next-line no-unused-vars,@typescript-eslint/no-unused-vars
-    const [all, from, to] = dateString.match(/^from (.*) to (.*)$/);
+    let [, from, to] = dateString.match(/^from(.{0,50})to(.{0,50})$/);
+    from = from.trim();
+    to = to.trim();
 
     const current = moment(now).tz(timezone);
-    const fromResults = parse(from, new Date(current.format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
-    const toResults = parse(to, new Date(current.format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
+    const fromResults = parse(from.trim(), new Date(current.format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
+    const toResults = parse(to.trim(), new Date(current.format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
 
     if (!Array.isArray(fromResults) || !fromResults.length) {
       throw new UserError(`Can't parse date: '${from}'`);
     }
 
-    if (!Array.isArray(fromResults) || !fromResults.length) {
+    if (!Array.isArray(toResults) || !toResults.length) {
       throw new UserError(`Can't parse date: '${to}'`);
     }
 
@@ -88,8 +89,10 @@ export function dateParser(dateString, timezone, now = new Date()) {
 
     momentRange = [momentRange[0].startOf(exactGranularity), momentRange[1].endOf(exactGranularity)];
   } else {
-    const results = parse(dateString, new Date(moment().tz(timezone).format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
-    if (!results || !results.length) {
+    const current = moment(now).tz(timezone);
+    const results = parse(dateString, new Date(current.format(moment.HTML5_FMT.DATETIME_LOCAL_MS)));
+
+    if (!results?.length) {
       throw new UserError(`Can't parse date: '${dateString}'`);
     }
 

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

@@ -8,6 +8,7 @@ import structuredClone from '@ungap/structured-clone';
 import {
   getEnv,
   getRealType,
+  parseLocalDate,
   QueryAlias,
 } from '@cubejs-backend/shared';
 import {
@@ -81,7 +82,9 @@ import {
   normalizeQuery,
   normalizeQueryCancelPreAggregations,
   normalizeQueryPreAggregationPreview,
-  normalizeQueryPreAggregations, remapToQueryAdapterFormat,
+  normalizeQueryPreAggregations,
+  preAggsJobsRequestSchema,
+  remapToQueryAdapterFormat,
 } from './query';
 import { cachedHandler } from './cached-handler';
 import { createJWKsFetcher } from './jwk';
@@ -770,8 +773,8 @@ class ApiGateway {
             preAggregations: [{ id: preAggregationId }]
           }
         );
-      const { partitions } = (preAggregationPartitions && preAggregationPartitions[0] || {});
-      const preAggregationPartition = partitions && partitions.find(p => p?.tableName === versionEntry.table_name);
+      const { partitions } = (preAggregationPartitions?.[0] || {});
+      const preAggregationPartition = partitions?.find(p => p?.tableName === versionEntry.table_name);
 
       res({
         preview: preAggregationPartition && await orchestratorApi.getPreAggregationPreview(
@@ -845,7 +848,6 @@ class ApiGateway {
    *   ]
    * }
    * ```
-   * TODO (buntarb): selector object validator.
    */
   private async preAggregationsJobs(req: Request, res: ExpressResponse) {
     const response = this.resToResultFn(res);
@@ -855,49 +857,26 @@ class ApiGateway {
     let result;
     try {
       await this.assertApiScope('jobs', req?.context?.securityContext);
+
+      if (!query || Object.keys(query).length === 0) {
+        throw new UserError('No job description provided');
+      }
+
+      const { error } = preAggsJobsRequestSchema.validate(query);
+      if (error) {
+        throw new UserError(`Invalid Job query format: ${error.message || error.toString()}`);
+      }
+
       switch (query.action) {
         case 'post':
-          if (
-            !(<PreAggsSelector>query.selector).timezones ||
-            (<PreAggsSelector>query.selector).timezones.length === 0
-          ) {
-            throw new UserError(
-              'A user\'s selector must contain at least one time zone.'
-            );
-          }
-          if (
-            !(<PreAggsSelector>query.selector).contexts ||
-            (
-              <{securityContext: any}[]>(
-                <PreAggsSelector>query.selector
-              ).contexts
-            ).length === 0
-          ) {
-            throw new UserError(
-              'A user\'s selector must contain at least one context element.'
-            );
-          } else {
-            let e = false;
-            (<{securityContext: any}[]>(
-              <PreAggsSelector>query.selector
-            ).contexts).forEach((c) => {
-              if (!c.securityContext) e = true;
-            });
-            if (e) {
-              throw new UserError(
-                'Every context element must contain the ' +
-                '\'securityContext\' property.'
-              );
-            }
-          }
           result = await this.preAggregationsJobsPOST(
             context,
             <PreAggsSelector>query.selector
           );
           if (result.length === 0) {
             throw new UserError(
               'A user\'s selector doesn\'t match any of the ' +
-              'pre-aggregations described by the Cube schemas.'
+              'pre-aggregations defined in the data model.'
             );
           }
           break;
@@ -928,30 +907,38 @@ class ApiGateway {
     selector: PreAggsSelector,
   ): Promise<string[]> {
     let jobs: string[] = [];
-    if (!selector.contexts?.length) {
-      jobs = await this.postPreAggregationsBuildJobs(
-        context,
-        selector,
-      );
-    } else {
-      const promise = Promise.all(
-        selector.contexts.map(async (config) => {
-          const ctx = <RequestContext>{
-            ...context,
-            ...config,
-          };
-          const _jobs = await this.postPreAggregationsBuildJobs(
-            ctx,
-            selector,
-          );
-          return _jobs;
-        })
-      );
-      const resolve = await promise;
-      resolve.forEach((_jobs) => {
-        jobs = jobs.concat(_jobs);
-      });
+
+    // There might be a few contexts but dateRange if present is still the same
+    // so let's normalize it only once.
+    // It's expected that selector.dateRange is provided in local time (without timezone)
+    // At the same time it is ok to get timestamps with `Z` (in UTC).
+    if (selector.dateRange) {
+      const start = parseLocalDate([{ val: selector.dateRange[0] }], 'UTC');
+      const end = parseLocalDate([{ val: selector.dateRange[1] }], 'UTC');
+      if (!start || !end) {
+        throw new UserError(`Cannot parse selector date range ${selector.dateRange}`);
+      }
+      selector.dateRange = [start, end];
     }
+
+    const promise = Promise.all(
+      selector.contexts.map(async (config) => {
+        const ctx = <RequestContext>{
+          ...context,
+          ...config,
+        };
+        const _jobs = await this.postPreAggregationsBuildJobs(
+          ctx,
+          selector,
+        );
+        return _jobs;
+      })
+    );
+    const resolve = await promise;
+    resolve.forEach((_jobs) => {
+      jobs = jobs.concat(_jobs);
+    });
+
     return jobs;
   }
 
@@ -963,7 +950,7 @@ class ApiGateway {
     selector: PreAggsSelector
   ): Promise<string[]> {
     const compiler = await this.getCompilerApi(context);
-    const { timezones } = selector;
+    const { timezones, dateRange } = selector;
     const preaggs = await compiler.preAggregations({
       dataSources: selector.dataSources,
       cubes: selector.cubes,
@@ -979,12 +966,13 @@ class ApiGateway {
           {
             metadata: undefined,
             timezones,
+            dateRange,
             preAggregations: preaggs.map(p => ({
               id: p.id,
-              cacheOnly: undefined, // boolean
+              cacheOnly: false,
               partitions: undefined, // string[]
             })),
-            forceBuildPreAggregations: undefined,
+            forceBuildPreAggregations: false,
             throwErrors: false,
           }
         );

packages/cubejs-api-gateway/src/query.js

@@ -6,9 +6,10 @@ import { getEnv } from '@cubejs-backend/shared';
 import { UserError } from './UserError';
 import { dateParser } from './dateParser';
 import { QueryType } from './types/enums';
+import { PreAggsJobsRequest } from "./types/request";
 
 const getQueryGranularity = (queries) => R.pipe(
-  R.map(({ timeDimensions }) => timeDimensions[0] && timeDimensions[0].granularity || null),
+  R.map(({ timeDimensions }) => timeDimensions[0]?.granularity),
   R.filter(Boolean),
   R.uniq
 )(queries);
@@ -145,6 +146,36 @@ const normalizeQueryOrder = order => {
   return result;
 };
 
+export const preAggsJobsRequestSchema = Joi.object({
+  action: Joi.string().valid('post', 'get').required(),
+  selector: Joi.when('action', {
+    is: 'post',
+    then: Joi.object({
+      contexts: Joi.array().items(
+        Joi.object({
+          securityContext: Joi.required(),
+        })
+      ).min(1).required(),
+      timezones: Joi.array().items(Joi.string()).min(1).required(),
+      dataSources: Joi.array().items(Joi.string()),
+      cubes: Joi.array().items(Joi.string()),
+      preAggregations: Joi.array().items(Joi.string()),
+      dateRange: Joi.array().length(2).items(Joi.string()),
+    }).optional(),
+    otherwise: Joi.forbidden(),
+  }),
+  tokens: Joi.when('action', {
+    is: 'get',
+    then: Joi.array().items(Joi.string()).min(1).required(),
+    otherwise: Joi.forbidden(),
+  }),
+  resType: Joi.when('action', {
+    is: 'get',
+    then: Joi.string().valid('object').optional(),
+    otherwise: Joi.forbidden(),
+  }),
+});
+
 const DateRegex = /^\d\d\d\d-\d\d-\d\d$/;
 
 const normalizeQueryFilters = (filter) => (
@@ -196,9 +227,9 @@ const normalizeQuery = (query, persistent) => {
   if (error) {
     throw new UserError(`Invalid query format: ${error.message || error.toString()}`);
   }
-  const validQuery = query.measures && query.measures.length ||
-    query.dimensions && query.dimensions.length ||
-    query.timeDimensions && query.timeDimensions.filter(td => !!td.granularity).length;
+  const validQuery = query.measures?.length ||
+    query.dimensions?.length ||
+    query.timeDimensions?.filter(td => !!td.granularity).length;
   if (!validQuery) {
     throw new UserError(
       'Query should contain either measures, dimensions or timeDimensions with granularities in order to be valid'

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

@@ -108,7 +108,7 @@ type ResponseResultFn =
   (
     message: (Record<string, any> | Record<string, any>[]) | DataResult | ErrorResponse,
     extra?: { status: number }
-  ) => void;
+  ) => void | Promise<void>;
 
 /**
  * Base HTTP request parameters map data type.
@@ -148,11 +148,12 @@ type SqlApiRequest = BaseRequest & {
  * Pre-aggregations selector object.
  */
 type PreAggsSelector = {
-  contexts?: {securityContext: any}[],
+  contexts: {securityContext: any}[],
   timezones: string[],
   dataSources?: string[],
   cubes?: string[],
   preAggregations?: string[],
+  dateRange?: [string, string], // We expect only single date Range for rebuilding
 };
 
 /**
@@ -177,7 +178,7 @@ type PreAggJob = {
  * The `/cubejs-system/v1/pre-aggregations/jobs` endpoint object type.
  */
 type PreAggsJobsRequest = {
-  action: 'post' | 'get' | 'delete',
+  action: 'post' | 'get',
   selector?: PreAggsSelector,
   tokens?: string[]
   resType?: 'object' | 'array'

packages/cubejs-api-gateway/test/dateParser.test.js

@@ -168,4 +168,28 @@ describe('dateParser', () => {
 
     Date.now.mockRestore();
   });
+
+  test('throws error on from invalid date to date', () => {
+    expect(() => dateParser('from invalid to 2020-02-02', 'UTC')).toThrow(
+      'Can\'t parse date: \'invalid\''
+    );
+  });
+
+  test('throws error on from date to invalid date', () => {
+    expect(() => dateParser('from 2020-02-02 to invalid', 'UTC')).toThrow(
+      'Can\'t parse date: \'invalid\''
+    );
+  });
+
+  test('from 12AM till now by hour', () => {
+    Date.now = jest.fn().mockReturnValue(new Date(2021, 2, 5, 13, 0, 0, 0));
+    expect(dateParser('2 weeks ago by hour', 'UTC', new Date(Date.UTC(2021, 2, 5, 13, 0, 0, 0)))).toStrictEqual(
+      [
+        '2021-02-19T13:00:00.000',
+        '2021-02-19T13:59:59.999'
+      ]
+    );
+
+    Date.now.mockRestore();
+  });
 });