From 1b3b5a85b01f7202da425ca43c84209df6dfb154 Mon Sep 17 00:00:00 2001 From: Dmitry Patsura Date: Thu, 20 Aug 2026 18:41:14 +0200 Subject: [PATCH] feat(query-orchestrator): evaluate interval refresh keys from local time behind a flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Interval and cron based `refreshKey` values can now be computed from the API instance clock instead of via `SELECT FLOOR(...) as refresh_key`, gated by `CUBEJS_REFRESH_KEY_LOCAL_TIME` (default `false`). The compiler already derived every input in JS — `interval`, `dayOffset` and `utcOffset` come from `parseSecondDuration`/`calcIntervalForCronString`. SQL only contributed `now()`. So `BaseQuery.everyRefreshKeyParts()` is extracted as the single source of truth that both the rendered SQL and a serializable descriptor derive from, and when the flag is on that descriptor rides in the refresh-key tuple's options element. The orchestrator short-circuits it in exactly one place, `QueryCache.cacheRefreshKeyResult`. #11607 had just made that the sole owner of a refresh key's cache key, renewal key and threshold, so both consumers — the scheduler's `loadRefreshKey` and the loader's `keyQueryResult` — are covered without either being touched. `PreAggregationLoadCache` keeps its per-request memo wrapping the call, which matters: one pre-aggregation load reads the invalidation keys several times (`contentVersion`, the returned `refreshKeyValues`, the refresh queue key), and re-reading the clock could straddle an interval boundary and have the load look a table up under one content version while enqueueing it under another. A test pins that by stubbing `Date.now` across a boundary. Why: the default `{ every: '10 seconds' }` cube refresh key has `renewalThreshold: 10`, so it is re-issued to Cube Store on essentially every request. That query has no `TableScan`, so `is_data_select_query` is false and it takes the `QueryPlan::Meta` path — correctly bypassing `SqlResultCache`, but still paying parser + plan + optimize + `collect`, plus `MetaStoreSchemaProvider::new(get_tables_with_path(false))` on every call, plus a WebSocket round trip. All to learn the wall clock. Scope: cube-level `cacheKeyQueries` and pre-aggregation `invalidateKeyQueries`, including the `10 seconds` and `1 hour` defaults. Excluded are `refreshKey.sql` and `incremental` keys — the latter are wrapped in `CASE WHEN NOW() < ` against an allocated partition-range param, and their options drive `renewalThresholdOutsideUpdateWindow` shortening for freshly sealed partitions. Flag off is byte-identical. The gate is applied at emit time as well as consume time, so the tuples and every hash derived from them are unchanged and no persisted cache is invalidated on upgrade. The existing `everyRefreshKeySql` assertions pass unedited, which is what proves the emitted SQL did not move. Clock skew is why this is flagged. Two nodes straddling an interval boundary produce different `contentVersion`s, so the same pre-aggregation gets built twice, recurring at each boundary. `externalRefresh` bounds the blast radius — non-builder API instances never run these queries — so a single refresh worker is safe; multiple workers or `CUBEJS_PRE_AGGREGATIONS_BUILDER=true` API instances are not. Flipping the flag also forces one pre-aggregation rebuild each way, since `pg` returns `numeric` as a string and Cube Store as a number. Table names are unaffected; `getStructureVersion` excludes invalidation keys. One finding worth recording: `preAggregationInvalidateKeyQueries` is memoized against a `queryCache` whose key did not include the flag, so a flag-on query's compiled refresh keys leaked to a flag-off query. Fixed by adding `localRefreshKey` to the key list next to the analogous `convertTzForRawTimeDimension`. A test caught this, not review. | Suite | Result | | --- | --- | | `cubejs-backend-shared` `test/env.test.ts` | 25/25 | | `cubejs-query-orchestrator` `test/unit` | 129/129, 6 suites | | `cubejs-schema-compiler` `dist/test/unit` | 826/831, 39/42 suites | The five schema-compiler failures are `error-reporter`, `FILTER_PARAMS` and `pre-agg-interpolated-cube-refs`; all five reproduce on an unmodified master, verified by reverting this branch's two schema-compiler files and re-running. Co-Authored-By: Claude Opus 5 (1M context) --- .../configuration/environment-variables.mdx | 25 ++ packages/cubejs-backend-shared/src/env.ts | 9 + .../cubejs-backend-shared/test/env.test.ts | 19 ++ .../src/orchestrator/QueryCache.ts | 65 ++++- .../src/orchestrator/utils.ts | 31 ++- .../test/unit/PreAggregations.test.ts | 104 ++++++++ .../test/unit/QueryCache.abstract.ts | 76 ++++++ .../test/unit/utils.test.ts | 69 +++++ .../src/adapter/BaseQuery.js | 79 +++++- .../test/unit/base-query.test.ts | 251 ++++++++++++++++++ .../src/core/CompilerApi.ts | 5 + .../cubejs-server-core/src/core/server.ts | 1 + 12 files changed, 719 insertions(+), 15 deletions(-) create mode 100644 packages/cubejs-query-orchestrator/test/unit/utils.test.ts 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,