diff --git a/docs-mintlify/reference/configuration/environment-variables.mdx b/docs-mintlify/reference/configuration/environment-variables.mdx index ed782d1eb9fc5..e9c34965f880f 100644 --- a/docs-mintlify/reference/configuration/environment-variables.mdx +++ b/docs-mintlify/reference/configuration/environment-variables.mdx @@ -1417,6 +1417,31 @@ this variable. | ------------------------- | ---------------------- | --------------------- | | A valid number in seconds | `600` | `600` | +## `CUBEJS_REFRESH_KEY_LOCAL_TIME` + +If `true`, interval and cron based [`refresh_key`](/reference/data-modeling/cube#refresh_key) +values are computed from the Cube instance's own clock instead of being fetched with a +`SELECT FLOOR(...) as refresh_key` query. This removes a database round trip per refresh +key check, which for the default 10 second refresh key means one fewer query on nearly +every request. + +| Possible Values | Default in Development | Default in Production | +| --------------- | ---------------------- | --------------------- | +| `true`, `false` | `false` | `false` | + +Only `every`-based refresh keys are affected. Refresh keys defined with `sql`, and those +using [`incremental`](/reference/data-modeling/pre-aggregations#incremental), still run +against the database because their values cannot be derived from a time interval alone. + + + +Set this variable to the same value on **every** API instance and refresh worker, and keep +their clocks synchronized (for example with NTP). Instances whose clocks straddle an +interval boundary compute different refresh keys, which can cause the same +pre-aggregation to be built more than once. + + + ## `CUBEJS_REFRESH_WORKER` If `true`, this instance of Cube will **only** refresh pre-aggregations. diff --git a/packages/cubejs-backend-shared/src/env.ts b/packages/cubejs-backend-shared/src/env.ts index 158707e6948d4..6ff7c42659c1b 100644 --- a/packages/cubejs-backend-shared/src/env.ts +++ b/packages/cubejs-backend-shared/src/env.ts @@ -368,6 +368,15 @@ const variables: Record any> = { preciseDecimalInCubestore: () => get('CUBEJS_DB_PRECISE_DECIMAL_IN_CUBESTORE') .default('false') .asBoolStrict(), + /** + * Evaluates interval and cron based `refreshKey` values from the API instance + * clock instead of issuing `SELECT FLOOR(...) as refresh_key`. Requires clocks + * to be in sync across all API instances and refresh workers, otherwise nodes + * straddling an interval boundary disagree and rebuild pre-aggregations twice. + */ + refreshKeyLocalTime: () => get('CUBEJS_REFRESH_KEY_LOCAL_TIME') + .default('false') + .asBoolStrict(), /** **************************************************************** * Common db options * diff --git a/packages/cubejs-backend-shared/test/env.test.ts b/packages/cubejs-backend-shared/test/env.test.ts index 65893dda63b8c..bd4fcb8e3b579 100644 --- a/packages/cubejs-backend-shared/test/env.test.ts +++ b/packages/cubejs-backend-shared/test/env.test.ts @@ -107,6 +107,25 @@ describe('getEnv', () => { ).toBe(10); }); + test('refreshKeyLocalTime', () => { + delete process.env.CUBEJS_REFRESH_KEY_LOCAL_TIME; + expect(getEnv('refreshKeyLocalTime')).toBe(false); + + process.env.CUBEJS_REFRESH_KEY_LOCAL_TIME = 'true'; + expect(getEnv('refreshKeyLocalTime')).toBe(true); + + process.env.CUBEJS_REFRESH_KEY_LOCAL_TIME = 'false'; + expect(getEnv('refreshKeyLocalTime')).toBe(false); + }); + + test('refreshKeyLocalTime(exception)', () => { + process.env.CUBEJS_REFRESH_KEY_LOCAL_TIME = 'yes'; + + expect(() => getEnv('refreshKeyLocalTime')).toThrowError(); + + delete process.env.CUBEJS_REFRESH_KEY_LOCAL_TIME; + }); + test('livePreview', () => { expect(getEnv('livePreview')).toBe(true); diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts b/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts index d956215fc50d2..a2b1b8e7433ea 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/QueryCache.ts @@ -25,7 +25,12 @@ import { ContinueWaitError } from './ContinueWaitError'; import { LocalCacheDriver } from './LocalCacheDriver'; import { DriverFactory, DriverFactoryByDataSource } from './DriverFactory'; import { LoadPreAggregationResult, PreAggregationDescription } from './PreAggregations'; -import { getCacheHash, extractRequestUUID } from './utils'; +import { + getCacheHash, + extractRequestUUID, + evaluateLocalRefreshKey, + isValidLocalRefreshKey, +} from './utils'; import { CacheAndQueryDriverType, MetadataOperationType } from './QueryOrchestrator'; export type CacheQueryResultOptions = { @@ -52,12 +57,28 @@ export type CacheQueryResultOptions = { export type RefreshKeyCacheOptions = Pick; +/** + * Everything needed to evaluate an `every` based refreshKey without touching a + * database: `FLOOR((utcOffset + unixTimestamp - dayOffset) / interval)`. + * + * `utcOffset` is frozen at compile time, exactly as it already is when baked into + * the emitted SQL string — the orchestrator must never recompute it, or the local + * and SQL paths would stop agreeing across a DST transition. + */ +export type LocalRefreshKeyDescriptor = { + interval: number; + utcOffset: number; + dayOffset: number; + cron?: boolean; +}; + type QueryOptions = { external?: boolean; renewalThreshold?: number; updateWindowSeconds?: number; renewalThresholdOutsideUpdateWindow?: number; incremental?: boolean; + localRefreshKey?: LocalRefreshKeyDescriptor; }; export type QueryWithParams = [ @@ -169,6 +190,7 @@ export interface QueryCacheOptions { cacheAndQueueDriver: CacheAndQueryDriverType; maxInMemoryCacheEntries?: number; skipExternalCacheAndQueue?: boolean; + localRefreshKey?: boolean; } export class QueryCache { @@ -182,6 +204,10 @@ export class QueryCache { protected static readonly IN_MEMORY_CACHE_DISABLE_PERIOD = 5 * 60 * 1000; + protected readonly localRefreshKeyEnabled: boolean; + + protected localRefreshKeyLogged: boolean = false; + public constructor( protected readonly cachePrefix: string, protected readonly driverFactory: DriverFactoryByDataSource, @@ -208,6 +234,31 @@ export class QueryCache { this.memoryCache = new LRUCache({ max: options.maxInMemoryCacheEntries || 10000 }); + + this.localRefreshKeyEnabled = options.localRefreshKey ?? getEnv('refreshKeyLocalTime'); + } + + /** + * Evaluates an interval based refresh key locally, or returns null when the SQL + * path must be used: the feature is off, the key is one we cannot reproduce + * (`refreshKey.sql`, incremental), or the descriptor is malformed. + */ + public localRefreshKeyResult(queryOptions?: QueryOptions): [{ refresh_key: number }] | null { + if (!this.localRefreshKeyEnabled || !isValidLocalRefreshKey(queryOptions?.localRefreshKey)) { + return null; + } + + // Logged on first use rather than from the constructor, where subclass field + // initialization has not run yet and `this.logger` may not be usable. + if (!this.localRefreshKeyLogged) { + this.localRefreshKeyLogged = true; + this.logger('Local refresh key evaluation enabled', { + warning: 'Interval based refresh keys are evaluated from this instance clock. ' + + 'Clocks must be in sync across all API instances and refresh workers.', + }); + } + + return evaluateLocalRefreshKey(queryOptions?.localRefreshKey); } public getCacheDriver(): CacheDriverInterface { @@ -462,6 +513,18 @@ export class QueryCache { options: RefreshKeyCacheOptions, ) { const [query, values, queryOptions] = sqlQuery; + + // A locally evaluated key is free, so there is nothing to cache and no queue to wait on. + // Short-circuiting here rather than in each caller keeps the identity work below as the + // single owner of the cache key, and leaves `PreAggregationLoadCache`'s per-request memo + // wrapping this call: one load reads the invalidation keys several times, and re-reading + // the clock could straddle an interval boundary and have it look a table up under one + // content version while enqueueing it under another. + const local = this.localRefreshKeyResult(queryOptions); + if (local) { + return local; + } + const cacheKey = QueryCache.refreshKeyIdentity(sqlQuery, options.dataSource); return this.cacheQueryResult(query, values, cacheKey, expiration, { diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/utils.ts b/packages/cubejs-query-orchestrator/src/orchestrator/utils.ts index 9a32c023e3ab5..23c52ece94a58 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/utils.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/utils.ts @@ -2,7 +2,7 @@ import crypto from 'crypto'; import { getProcessUid } from '@cubejs-backend/shared'; import { QueryKey, QueryKeyHash } from '@cubejs-backend/base-driver'; -import { CacheKey } from './QueryCache'; +import { CacheKey, LocalRefreshKeyDescriptor } from './QueryCache'; /** * Unique process ID regexp. @@ -32,6 +32,35 @@ export function getCacheHash(queryKey: QueryKey | CacheKey, processUid?: string) } } +/** + * Evaluates an `every` based refreshKey from the local clock, producing the same + * row shape the equivalent `SELECT FLOOR(...) as refresh_key` would return. + * + * `nowMs` is not floored to whole seconds first: for integer `x`, fractional + * `f` in [0, 1) and `interval >= 1`, floor((x + f) / interval) === floor(x / interval), + * which is also why the fractional seconds of `EXTRACT(EPOCH FROM NOW())` never + * mattered on the SQL path. + */ +export function evaluateLocalRefreshKey( + descriptor: LocalRefreshKeyDescriptor, + nowMs: number = Date.now(), +): [{ refresh_key: number }] { + const { utcOffset, interval, dayOffset } = descriptor; + + return [{ refresh_key: Math.floor((utcOffset + nowMs / 1000 - dayOffset) / interval) }]; +} + +/** + * A malformed descriptor must fall back to the SQL path rather than produce a + * garbage refresh key, which would silently invalidate everything downstream. + */ +export function isValidLocalRefreshKey(descriptor?: LocalRefreshKeyDescriptor): boolean { + return !!descriptor && + Number.isFinite(descriptor.interval) && descriptor.interval > 0 && + Number.isFinite(descriptor.utcOffset) && + Number.isFinite(descriptor.dayOffset); +} + /** * Extracts the UUID prefix from a request ID by stripping the `-span-N` suffix. */ diff --git a/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts b/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts index f1fa0fd82525a..a33f2f430f6f7 100644 --- a/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts +++ b/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts @@ -467,6 +467,110 @@ describe('PreAggregations', () => { }); }); + describe('local refresh key', () => { + const REFRESH_KEY_SQL = 'SELECT FLOOR((UNIX_TIMESTAMP()) / 600) as refresh_key'; + const descriptor = { interval: 600, utcOffset: 0, dayOffset: 0, cron: false }; + + const newLoadCache = (localRefreshKey?: boolean) => { + const cache = new QueryCache( + 'TEST', + mockDriverFactory as any, + // eslint-disable-next-line @typescript-eslint/no-empty-function + () => {}, + { + cacheAndQueueDriver: 'memory', + localRefreshKey, + queueOptions: async () => ({ executionTimeout: 1, concurrency: 2 }), + }, + ); + (cache.getCacheDriver() as LocalCacheDriver).reset(); + + const preAggregations = new PreAggregations( + 'TEST', + mockDriverFactory as any, + // eslint-disable-next-line @typescript-eslint/no-empty-function + () => {}, + cache, + { queueOptions: async () => ({ executionTimeout: 1, concurrency: 2 }) }, + ); + + return new PreAggregationLoadCache( + mockDriverFactory as any, + cache, + preAggregations, + { dataSource: 'default' }, + ); + }; + + test('keyQueryResult evaluates locally without querying the datasource', async () => { + const loadCache = newLoadCache(true); + + const result = await loadCache.keyQueryResult( + [REFRESH_KEY_SQL, [], { external: true, renewalThreshold: 60, localRefreshKey: descriptor }], + false, + 10, + ); + + expect(result).toEqual([{ refresh_key: Math.floor(Date.now() / 1000 / descriptor.interval) }]); + expect(mockDriver!.executedQueries).toEqual([]); + }); + + test('keyQueryResult still queries when the flag is off', async () => { + const loadCache = newLoadCache(false); + + await loadCache.keyQueryResult( + [REFRESH_KEY_SQL, [], { external: false, renewalThreshold: 60, localRefreshKey: descriptor }], + false, + 10, + ); + + expect(mockDriver!.executedQueries).toEqual([REFRESH_KEY_SQL]); + }); + + test('keyQueryResult still queries an incremental key that carries no descriptor', async () => { + const loadCache = newLoadCache(true); + const incrementalSql = 'SELECT CASE WHEN NOW() < $1 THEN FLOOR((UNIX_TIMESTAMP()) / 3600) END as refresh_key'; + + await loadCache.keyQueryResult( + [incrementalSql, [], { + external: false, + renewalThreshold: 300, + incremental: true, + renewalThresholdOutsideUpdateWindow: 86400, + }], + false, + 10, + ); + + expect(mockDriver!.executedQueries).toEqual([incrementalSql]); + }); + + // The per-request memo must pin the value for the life of one load. A single + // load reads the invalidation keys several times (contentVersion, the returned + // refreshKeyValues, the refresh queue key); if the clock were re-read, a load + // crossing an interval boundary would look a table up under one content version + // and enqueue it under another. + test('keyQueryResult is stable across an interval boundary within one load cache', async () => { + const loadCache = newLoadCache(true); + const key: [string, any[], Record] = + [REFRESH_KEY_SQL, [], { external: true, renewalThreshold: 60, localRefreshKey: descriptor }]; + + const nowSpy = jest.spyOn(Date, 'now').mockReturnValue(600_000); + try { + const first = await loadCache.keyQueryResult(key, false, 10); + expect(first).toEqual([{ refresh_key: 1 }]); + + // Advance well past the next boundary. + nowSpy.mockReturnValue(1_200_000); + const second = await loadCache.keyQueryResult(key, false, 10); + + expect(second).toEqual(first); + } finally { + nowSpy.mockRestore(); + } + }); + }); + describe('loadAllPreAggregationsIfNeeded', () => { let preAggregations: PreAggregations | null = null; diff --git a/packages/cubejs-query-orchestrator/test/unit/QueryCache.abstract.ts b/packages/cubejs-query-orchestrator/test/unit/QueryCache.abstract.ts index 178749756aaac..f3e192a81ec15 100644 --- a/packages/cubejs-query-orchestrator/test/unit/QueryCache.abstract.ts +++ b/packages/cubejs-query-orchestrator/test/unit/QueryCache.abstract.ts @@ -470,6 +470,82 @@ export const QueryCacheTest = (name: string, options: QueryCacheTestOptions) => }); }); + describe('local refresh key', () => { + const REFRESH_KEY_SQL = 'SELECT FLOOR((UNIX_TIMESTAMP()) / 600) as refresh_key'; + const descriptor = { interval: 600, utcOffset: 0, dayOffset: 0, cron: false }; + + const newCache = (localRefreshKey?: boolean) => new QueryCacheOpened( + crypto.randomBytes(16).toString('hex'), + () => { + throw new Error('driverFactory is not implemented, mock should be used...'); + }, + jest.fn(), + { ...options, localRefreshKey }, + ); + + const loadRefreshKey = async (localRefreshKey: boolean | undefined, queryOptions: any) => { + const localCache = newCache(localRefreshKey); + const spy = jest.spyOn(localCache, 'queryWithRetryAndRelease') + .mockImplementation(async () => [{ refresh_key: 12345 }]); + + try { + const [result] = await Promise.all( + localCache.loadRefreshKeys( + [[REFRESH_KEY_SQL, [], queryOptions]], + 60, + { dataSource: 'default' }, + ) + ); + + return { result, executed: spy.mock.calls.length }; + } finally { + spy.mockRestore(); + await localCache.cleanup(); + } + }; + + it('evaluates locally without touching the driver', async () => { + const { result, executed } = await loadRefreshKey(true, { + external: true, + renewalThreshold: 60, + localRefreshKey: descriptor, + }); + + expect(executed).toBe(0); + expect(result).toEqual([{ refresh_key: Math.floor(Date.now() / 1000 / 600) }]); + }); + + it('runs the query when the flag is off', async () => { + const { result, executed } = await loadRefreshKey(false, { + external: true, + renewalThreshold: 60, + localRefreshKey: descriptor, + }); + + expect(executed).toBe(1); + expect(result).toEqual([{ refresh_key: 12345 }]); + }); + + it('runs the query when there is no descriptor', async () => { + const { executed } = await loadRefreshKey(true, { + external: false, + renewalThreshold: 10, + }); + + expect(executed).toBe(1); + }); + + it('runs the query when the descriptor is malformed', async () => { + const { executed } = await loadRefreshKey(true, { + external: true, + renewalThreshold: 60, + localRefreshKey: { ...descriptor, interval: 0 }, + }); + + expect(executed).toBe(1); + }); + }); + it('queryCacheKey format', () => { const key1 = QueryCache.queryCacheKey({ query: 'select data', diff --git a/packages/cubejs-query-orchestrator/test/unit/utils.test.ts b/packages/cubejs-query-orchestrator/test/unit/utils.test.ts new file mode 100644 index 0000000000000..26ac79f3ab250 --- /dev/null +++ b/packages/cubejs-query-orchestrator/test/unit/utils.test.ts @@ -0,0 +1,69 @@ +import { evaluateLocalRefreshKey, isValidLocalRefreshKey } from '../../src/orchestrator/utils'; + +describe('evaluateLocalRefreshKey', () => { + const tenMinutes = { interval: 600, utcOffset: 0, dayOffset: 0 }; + + test('returns the same row shape as SELECT FLOOR(...) as refresh_key', () => { + // 600_000ms == 600s == exactly one 10 minute bucket + expect(evaluateLocalRefreshKey(tenMinutes, 600_000)).toEqual([{ refresh_key: 1 }]); + expect(evaluateLocalRefreshKey(tenMinutes, 6_000_000)).toEqual([{ refresh_key: 10 }]); + }); + + test('changes exactly at the interval boundary', () => { + expect(evaluateLocalRefreshKey(tenMinutes, 599_999)).toEqual([{ refresh_key: 0 }]); + expect(evaluateLocalRefreshKey(tenMinutes, 600_000)).toEqual([{ refresh_key: 1 }]); + expect(evaluateLocalRefreshKey(tenMinutes, 1_199_999)).toEqual([{ refresh_key: 1 }]); + expect(evaluateLocalRefreshKey(tenMinutes, 1_200_000)).toEqual([{ refresh_key: 2 }]); + }); + + test('sub-second precision never changes the result', () => { + for (const ms of [0, 1, 250, 500, 999]) { + expect(evaluateLocalRefreshKey(tenMinutes, 600_000 + ms)).toEqual( + evaluateLocalRefreshKey(tenMinutes, 600_000) + ); + } + }); + + test('applies a negative utcOffset', () => { + // America/Los_Angeles in PST: -8 hours + expect(evaluateLocalRefreshKey({ interval: 3600, utcOffset: -28800, dayOffset: 0 }, 28_800_000)) + .toEqual([{ refresh_key: 0 }]); + }); + + test('applies dayOffset for cron based keys', () => { + // every '0 10 * * *' => interval 1 day, dayOffset 10 hours + const daily = { interval: 86400, utcOffset: 0, dayOffset: 36000, cron: true }; + + // 09:59:59 UTC on the epoch day is still before the first fire time + expect(evaluateLocalRefreshKey(daily, 35_999_000)).toEqual([{ refresh_key: -1 }]); + expect(evaluateLocalRefreshKey(daily, 36_000_000)).toEqual([{ refresh_key: 0 }]); + // and the day after + expect(evaluateLocalRefreshKey(daily, 36_000_000 + 86_400_000)).toEqual([{ refresh_key: 1 }]); + }); + + test('defaults to the current clock', () => { + const before = Math.floor(Date.now() / 1000 / tenMinutes.interval); + const [{ refresh_key: value }] = evaluateLocalRefreshKey(tenMinutes); + const after = Math.floor(Date.now() / 1000 / tenMinutes.interval); + + expect(value).toBeGreaterThanOrEqual(before); + expect(value).toBeLessThanOrEqual(after); + }); +}); + +describe('isValidLocalRefreshKey', () => { + test('accepts a well formed descriptor', () => { + expect(isValidLocalRefreshKey({ interval: 600, utcOffset: 0, dayOffset: 0 })).toBe(true); + expect(isValidLocalRefreshKey({ interval: 1, utcOffset: -28800, dayOffset: 36000 })).toBe(true); + }); + + test('rejects anything that would produce a garbage key', () => { + expect(isValidLocalRefreshKey(undefined)).toBe(false); + expect(isValidLocalRefreshKey({ interval: 0, utcOffset: 0, dayOffset: 0 })).toBe(false); + expect(isValidLocalRefreshKey({ interval: -600, utcOffset: 0, dayOffset: 0 })).toBe(false); + expect(isValidLocalRefreshKey({ interval: NaN, utcOffset: 0, dayOffset: 0 })).toBe(false); + expect(isValidLocalRefreshKey({ interval: Infinity, utcOffset: 0, dayOffset: 0 })).toBe(false); + expect(isValidLocalRefreshKey({ interval: 600, utcOffset: NaN, dayOffset: 0 })).toBe(false); + expect(isValidLocalRefreshKey({ interval: 600, utcOffset: 0, dayOffset: NaN })).toBe(false); + }); +}); diff --git a/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js b/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js index 5649c9bc9554a..15fd2608d3bb7 100644 --- a/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js +++ b/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js @@ -287,6 +287,7 @@ export class BaseQuery { memberToAlias: this.options.memberToAlias, expressionParams: this.options.expressionParams, convertTzForRawTimeDimension: this.options.convertTzForRawTimeDimension, + localRefreshKey: this.options.localRefreshKey, from: this.options.from, multiStageQuery: this.options.multiStageQuery, multiStageDimensions: this.options.multiStageDimensions, @@ -362,6 +363,9 @@ export class BaseQuery { // toggled independently. The neverUseSqlPlannerPreaggregation() guard still opts // specific query types (e.g. CubeStoreQuery) out for correctness. this.canUseNativeSqlPlannerPreAggregation = this.useNativeSqlPlanner && !this.neverUseSqlPlannerPreaggregation(); + // Gated at emit time so that with the flag off the refresh key tuples stay + // byte-identical and nothing downstream re-hashes to a new cache key. + this.localRefreshKey = this.options.localRefreshKey ?? getEnv('refreshKeyLocalTime'); this.queryLevelJoinHints = this.options.joinHints ?? []; this.prebuildJoin(); @@ -4227,6 +4231,7 @@ export class BaseQuery { externalQueryClass: this.options.externalQueryClass, queryFactory: this.options.queryFactory, useNativeSqlPlanner: this.options.useNativeSqlPlanner, + localRefreshKey: this.options.localRefreshKey, ...options, }; } @@ -4265,19 +4270,22 @@ export class BaseQuery { this.refreshKeySelect(sql), { external, - renewalThreshold: this.refreshKeyRenewalThresholdForInterval(cubeFromPath.refreshKey) + renewalThreshold: this.refreshKeyRenewalThresholdForInterval(cubeFromPath.refreshKey), + ...this.localRefreshKeyOptions(cubeFromPath.refreshKey, query) }, query ]; } } - const [sql, external, query] = this.everyRefreshKeySql(this.defaultEveryRefreshKey()); + const defaultEveryRefreshKey = this.defaultEveryRefreshKey(); + const [sql, external, query] = this.everyRefreshKeySql(defaultEveryRefreshKey); return [ this.refreshKeySelect(sql), { external, - renewalThreshold: this.defaultRefreshKeyRenewalThreshold() + renewalThreshold: this.defaultRefreshKeyRenewalThreshold(), + ...this.localRefreshKeyOptions(defaultEveryRefreshKey, query) }, query ]; @@ -4832,21 +4840,57 @@ export class BaseQuery { }; } + /** + * The interval arithmetic behind an `every` based refreshKey, with no SQL in it. + * Both the rendered SQL and the descriptor handed to the orchestrator for local + * evaluation derive from this, so they cannot disagree on the formula. + * + * @param {Object} refreshKey + * @return {{ utcOffset: number, interval: number, dayOffset: number, cron: boolean }} + */ + everyRefreshKeyParts(refreshKey) { + const every = refreshKey.every || '1 hour'; + + if (/^(\d+) (second|minute|hour|day|week)s?$/.test(every)) { + return { + utcOffset: this.timezone ? moment.tz(this.timezone).utcOffset() * 60 : 0, + interval: this.parseSecondDuration(every), + dayOffset: 0, + cron: false, + }; + } + + return { ...this.calcIntervalForCronString(refreshKey), cron: true }; + } + + /** + * Refresh key options letting the orchestrator evaluate this key from its own + * clock instead of running the SQL. Returns nothing unless the feature is on, + * which keeps the emitted tuples unchanged by default. + * + * @protected + * @param {Object} refreshKey + * @param {BaseQuery} [query] the instance that rendered the SQL, when it differs from `this` + * @return {Object} + */ + localRefreshKeyOptions(refreshKey, query) { + return this.localRefreshKey + ? { localRefreshKey: (query || this).everyRefreshKeyParts(refreshKey) } + : {}; + } + everyRefreshKeySql(refreshKey, external = false) { if (this.externalQueryClass) { return this.externalQuery().everyRefreshKeySql(refreshKey, true); } - const every = refreshKey.every || '1 hour'; + const { utcOffset, interval, dayOffset, cron } = this.everyRefreshKeyParts(refreshKey); - if (/^(\d+) (second|minute|hour|day|week)s?$/.test(every)) { - const utcOffset = this.timezone ? moment.tz(this.timezone).utcOffset() * 60 : 0; + if (!cron) { const utcOffsetPrefix = utcOffset ? `${utcOffset} + ` : ''; - return [this.floorSql(`(${utcOffsetPrefix}${this.unixTimestampSql()}) / ${this.parseSecondDuration(every)}`), external, this]; + return [this.floorSql(`(${utcOffsetPrefix}${this.unixTimestampSql()}) / ${interval}`), external, this]; } - const { dayOffset, utcOffset, interval } = this.calcIntervalForCronString(refreshKey); - /** * Small explanation how it works for every `0 8 * * *` * 28800 is a $dayOffset @@ -5038,6 +5082,14 @@ export class BaseQuery { } if (preAggregation.refreshKey.every || preAggregation.refreshKey.incremental) { + // An incremental key is wrapped into `CASE WHEN NOW() < ` + // against an allocated partition range param, and its options drive the + // renewalThresholdOutsideUpdateWindow shortening for freshly sealed partitions. + // Neither is reproducible from a time interval, so opt out of local evaluation. + const localRefreshKeyOptions = preAggregation.refreshKey.incremental + ? {} + : this.localRefreshKeyOptions(preAggregation.refreshKey, refreshKeyQuery); + return [ refreshKeyQuery.paramAllocator.buildSqlAndParams(this.refreshKeySelect(refreshKey)).concat({ external: refreshKeyExternal, @@ -5046,7 +5098,8 @@ export class BaseQuery { updateWindowSeconds: preAggregation.refreshKey.updateWindow && this.parseSecondDuration(preAggregation.refreshKey.updateWindow), renewalThresholdOutsideUpdateWindow: preAggregation.refreshKey.incremental && - 24 * 60 * 60 + 24 * 60 * 60, + ...localRefreshKeyOptions }) ]; } @@ -5070,15 +5123,15 @@ export class BaseQuery { () => preAggregationQueryForSql.cacheKeyQueries( (refreshKeyCube, [refreshKeySQL, refreshKeyQueryOptions, refreshKeyQuery]) => { if (!cubeFromPath.refreshKey) { - const [sql, external, query] = this.everyRefreshKeySql({ - every: '1 hour' - }); + const hourlyRefreshKey = { every: '1 hour' }; + const [sql, external, query] = this.everyRefreshKeySql(hourlyRefreshKey); return [ this.refreshKeySelect(sql), { external, renewalThreshold: this.defaultRefreshKeyRenewalThreshold(), + ...this.localRefreshKeyOptions(hourlyRefreshKey, query) }, query ]; diff --git a/packages/cubejs-schema-compiler/test/unit/base-query.test.ts b/packages/cubejs-schema-compiler/test/unit/base-query.test.ts index 10ef064f06814..b702cc4f1b0f9 100644 --- a/packages/cubejs-schema-compiler/test/unit/base-query.test.ts +++ b/packages/cubejs-schema-compiler/test/unit/base-query.test.ts @@ -1367,6 +1367,75 @@ describe('SQL Generation', () => { expect(error).toBeInstanceOf(UserError); } }); + + it('Test for everyRefreshKeyParts', async () => { + await compilers.compiler.compile(); + + const timezone = 'America/Los_Angeles'; + const query = new PostgresQuery(compilers, { + measures: ['cards.count'], + timeDimensions: [], + filters: [], + timezone, + }); + + const utcOffset = moment.tz(timezone).utcOffset() * 60; + + expect(query.everyRefreshKeyParts({ every: '1 hour' })) + .toEqual({ utcOffset, interval: 3600, dayOffset: 0, cron: false }); + expect(query.everyRefreshKeyParts({ every: '10 seconds' })) + .toEqual({ utcOffset, interval: 10, dayOffset: 0, cron: false }); + expect(query.everyRefreshKeyParts({ every: '10 minute' })) + .toEqual({ utcOffset, interval: 600, dayOffset: 0, cron: false }); + + expect(query.everyRefreshKeyParts({ every: '0 * * * *', timezone })) + .toEqual({ utcOffset, interval: 3600, dayOffset: 0, cron: true }); + expect(query.everyRefreshKeyParts({ every: '0 10 * * *', timezone })) + .toEqual({ utcOffset, interval: 86400, dayOffset: 36000, cron: true }); + expect(query.everyRefreshKeyParts({ every: '30 5 * * 5', timezone })) + .toEqual({ utcOffset, interval: 604800, dayOffset: 106200, cron: true }); + }); + + it('everyRefreshKeyParts agrees with the SQL it renders', async () => { + await compilers.compiler.compile(); + + const timezone = 'America/Los_Angeles'; + const query = new PostgresQuery(compilers, { + measures: ['cards.count'], + timeDimensions: [], + filters: [], + timezone, + }); + + // Evaluates the emitted FLOOR(...) expression as JS with the clock pinned to `t`, + // which is what the orchestrator now computes from the descriptor instead. + const evalSql = (sql: string, t: number) => { + const asJs = sql + .split('EXTRACT(EPOCH FROM NOW())').join(String(t)) + .split('FLOOR').join('Math.floor'); + // eslint-disable-next-line no-new-func + return Function(`"use strict"; return (${asJs});`)(); + }; + + const refreshKeys = [ + { every: '10 seconds' }, + { every: '10 minute' }, + { every: '1 hour' }, + { every: '7 day' }, + { every: '0 * * * *', timezone }, + { every: '0 10 * * *', timezone }, + { every: '30 5 * * 5', timezone }, + ]; + + for (const refreshKey of refreshKeys) { + const [sql] = query.everyRefreshKeySql(refreshKey); + const { utcOffset, interval, dayOffset } = query.everyRefreshKeyParts(refreshKey); + + for (const t of [0, 1, 1_500_000_000, 1_767_225_600]) { + expect(Math.floor((utcOffset + t - dayOffset) / interval)).toEqual(evalSql(sql, t)); + } + } + }); }); describe('refreshKey from schema', () => { @@ -1592,6 +1661,188 @@ describe('SQL Generation', () => { }); }); + describe('refreshKey local time evaluation', () => { + const compilers = /** @type Compilers */ prepareJsCompiler( + createCubeSchema({ + name: 'cards', + refreshKey: ` + refreshKey: { + every: '10 minute', + }, + `, + preAggregations: ` + countCreatedAt: { + type: 'rollup', + external: true, + measureReferences: [count], + timeDimensionReference: createdAt, + granularity: \`day\`, + partitionGranularity: \`month\`, + refreshKey: { + every: '1 hour', + }, + scheduledRefresh: true, + }, + maxCreatedAt: { + type: 'rollup', + external: true, + measureReferences: [max], + timeDimensionReference: createdAt, + granularity: \`day\`, + partitionGranularity: \`month\`, + refreshKey: { + sql: 'SELECT MAX(created_at) FROM cards', + }, + scheduledRefresh: true, + }, + minCreatedAt: { + type: 'rollup', + external: false, + measureReferences: [min], + timeDimensionReference: createdAt, + granularity: \`day\`, + partitionGranularity: \`month\`, + refreshKey: { + every: '1 hour', + incremental: true, + }, + scheduledRefresh: true, + }, + ` + }) + ); + + const timezone = 'America/Los_Angeles'; + + const newQuery = (query: any) => new PostgresQuery(compilers, { + timeDimensions: [], + filters: [], + timezone, + localRefreshKey: true, + ...query, + }); + + it('carries the descriptor on a cube refreshKey.every', async () => { + await compilers.compiler.compile(); + + const utcOffset = moment.tz(timezone).utcOffset() * 60; + const query = newQuery({ measures: ['cards.sum'], externalQueryClass: MssqlQuery }); + + expect(query.cacheKeyQueries()).toEqual([ + [ + `SELECT FLOOR((${utcOffset} + DATEDIFF(SECOND,'1970-01-01', GETUTCDATE())) / 600) as refresh_key`, + [], + { + external: true, + renewalThreshold: 60, + localRefreshKey: { utcOffset, interval: 600, dayOffset: 0, cron: false }, + } + ] + ]); + }); + + it('carries the descriptor on a pre-aggregation refreshKey.every', async () => { + await compilers.compiler.compile(); + + const utcOffset = moment.tz(timezone).utcOffset() * 60; + const query = newQuery({ measures: ['cards.count'], externalQueryClass: MssqlQuery }); + + const preAggregations: any = query.newPreAggregations().preAggregationsDescription(); + expect(preAggregations.length).toEqual(1); + expect(preAggregations[0].invalidateKeyQueries).toEqual([ + [ + `SELECT FLOOR((${utcOffset} + DATEDIFF(SECOND,'1970-01-01', GETUTCDATE())) / 3600) as refresh_key`, + [], + { + external: true, + renewalThreshold: 300, + localRefreshKey: { utcOffset, interval: 3600, dayOffset: 0, cron: false }, + } + ] + ]); + }); + + it('leaves refreshKey.sql on the SQL path', async () => { + await compilers.compiler.compile(); + + const query = newQuery({ measures: ['cards.max'], externalQueryClass: MssqlQuery }); + + const preAggregations: any = query.newPreAggregations().preAggregationsDescription(); + expect(preAggregations.length).toEqual(1); + expect(preAggregations[0].invalidateKeyQueries[0][2]).not.toHaveProperty('localRefreshKey'); + }); + + it('leaves an incremental refreshKey on the SQL path', async () => { + await compilers.compiler.compile(); + + const query = newQuery({ + measures: ['cards.min'], + timeDimensions: [{ + dimension: 'cards.createdAt', + granularity: 'day', + dateRange: ['2016-12-30', '2017-01-05'] + }], + externalQueryClass: MssqlQuery, + }); + + const preAggregations: any = query.newPreAggregations().preAggregationsDescription(); + expect(preAggregations.length).toEqual(1); + expect(preAggregations[0].invalidateKeyQueries[0][2]).toMatchObject({ incremental: true }); + expect(preAggregations[0].invalidateKeyQueries[0][2]).not.toHaveProperty('localRefreshKey'); + }); + }); + + describe('refreshKey local time evaluation (cube without refreshKey)', () => { + const compilers = /** @type Compilers */ prepareJsCompiler( + createCubeSchema({ + name: 'cards', + preAggregations: ` + countCreatedAt: { + type: 'rollup', + external: true, + measureReferences: [count], + timeDimensionReference: createdAt, + granularity: \`day\`, + partitionGranularity: \`month\`, + scheduledRefresh: true, + }, + ` + }) + ); + + it('falls back to the hourly default with a descriptor', async () => { + await compilers.compiler.compile(); + + const query = new PostgresQuery(compilers, { + measures: ['cards.count'], + timeDimensions: [], + filters: [], + timezone: 'UTC', + localRefreshKey: true, + }); + + const preAggregations: any = query.newPreAggregations().preAggregationsDescription(); + expect(preAggregations.length).toEqual(1); + expect(preAggregations[0].invalidateKeyQueries[0][2].localRefreshKey) + .toEqual({ utcOffset: 0, interval: 3600, dayOffset: 0, cron: false }); + }); + + it('emits no descriptor when the flag is off', async () => { + await compilers.compiler.compile(); + + const query = new PostgresQuery(compilers, { + measures: ['cards.count'], + timeDimensions: [], + filters: [], + timezone: 'UTC', + }); + + const preAggregations: any = query.newPreAggregations().preAggregationsDescription(); + expect(preAggregations.length).toEqual(1); + expect(preAggregations[0].invalidateKeyQueries[0][2]).not.toHaveProperty('localRefreshKey'); + }); + }); + describe('refreshKey only cube (immutable)', () => { /** @type Compilers */ prepareJsCompiler( createCubeSchema({ diff --git a/packages/cubejs-server-core/src/core/CompilerApi.ts b/packages/cubejs-server-core/src/core/CompilerApi.ts index c6e1481713747..f82feb816ae58 100644 --- a/packages/cubejs-server-core/src/core/CompilerApi.ts +++ b/packages/cubejs-server-core/src/core/CompilerApi.ts @@ -37,6 +37,7 @@ export interface CompilerApiOptions { preAggregationsSchema?: string | ((context: Context) => string | Promise); allowUngroupedWithoutPrimaryKey?: boolean; convertTzForRawTimeDimension?: boolean; + localRefreshKey?: boolean; schemaVersion?: () => string | object | Promise; contextToGroups?: (context: Context) => string[] | Promise; compileContext?: any; @@ -105,6 +106,8 @@ export class CompilerApi { protected readonly convertTzForRawTimeDimension?: boolean; + protected readonly localRefreshKey?: boolean; + public schemaVersion?: () => string | object | Promise; protected readonly contextToGroups?: (context: Context) => string[] | Promise; @@ -146,6 +149,7 @@ export class CompilerApi { this.preAggregationsSchema = this.options.preAggregationsSchema; this.allowUngroupedWithoutPrimaryKey = this.options.allowUngroupedWithoutPrimaryKey; this.convertTzForRawTimeDimension = this.options.convertTzForRawTimeDimension; + this.localRefreshKey = this.options.localRefreshKey; this.schemaVersion = this.options.schemaVersion; this.contextToGroups = this.options.contextToGroups; this.compileContext = options.compileContext; @@ -967,6 +971,7 @@ export class CompilerApi { preAggregationsSchema: this.preAggregationsSchema, allowUngroupedWithoutPrimaryKey: this.allowUngroupedWithoutPrimaryKey, convertTzForRawTimeDimension: this.convertTzForRawTimeDimension, + localRefreshKey: this.localRefreshKey, queryFactory: this.queryFactory, } ); diff --git a/packages/cubejs-server-core/src/core/server.ts b/packages/cubejs-server-core/src/core/server.ts index 48b146c8abb70..d52b43566cbc1 100644 --- a/packages/cubejs-server-core/src/core/server.ts +++ b/packages/cubejs-server-core/src/core/server.ts @@ -738,6 +738,7 @@ export class CubejsServerCore { this.options.allowUngroupedWithoutPrimaryKey || getEnv('allowUngroupedWithoutPrimaryKey'), convertTzForRawTimeDimension: getEnv('convertTzForRawTimeDimension'), + localRefreshKey: getEnv('refreshKeyLocalTime'), compileContext: options.context, dialectClass: options.dialectClass, externalDialectClass: options.externalDialectClass,