diff --git a/packages/cubejs-schema-compiler/src/adapter/PreAggregations.ts b/packages/cubejs-schema-compiler/src/adapter/PreAggregations.ts index 5fa4174ebbb44..000fba135e343 100644 --- a/packages/cubejs-schema-compiler/src/adapter/PreAggregations.ts +++ b/packages/cubejs-schema-compiler/src/adapter/PreAggregations.ts @@ -1,5 +1,6 @@ import R from 'ramda'; - +import moment from 'moment-timezone'; +import { parseSqlInterval, subtractInterval } from '@cubejs-backend/shared'; import { CubeSymbols, PreAggregationDefinition } from '../compiler/CubeSymbols'; import { FinishedJoinTree, JoinEdge } from '../compiler/JoinGraph'; import { UserError } from '../compiler/UserError'; @@ -16,6 +17,8 @@ import { BaseGroupFilter } from './BaseGroupFilter'; import { BaseDimension } from './BaseDimension'; import { BaseSegment } from './BaseSegment'; +type ParsedInterval = ReturnType; + export type PartitionTimeDimension = { dimension: string; dateRange: [string, string]; @@ -116,6 +119,10 @@ export class PreAggregations { private hasCumulativeMeasuresValue: boolean = false; + private cumulativeMeasuresTrailingIntervalValue: ParsedInterval | null | undefined; + + private cumulativeMeasuresTrailingIntervalComputed: boolean = false; + public preAggregationForQuery: PreAggregationForQuery | undefined = undefined; public preAggregationUsageInfos: PreAggregationUsageInfo[] | undefined = undefined; @@ -291,6 +298,91 @@ export class PreAggregations { return this.hasCumulativeMeasuresValue; } + /** + * For a query with cumulative (rolling window) measures, decide how the matched + * time dimension date range can be narrowed for partition selection. + * + * A rolling window needs data preceding the requested range (the trailing window), + * so the range can only be narrowed if every cumulative measure has a bounded + * trailing window and no leading window. In that case we return the largest + * trailing interval so the caller can shift the matched range start back by it. + * + * Returns `null` when narrowing is not safe (an unbounded trailing or any leading + * window is present) — the caller then keeps the previous behaviour of building + * the whole build range. Returns `undefined` when there are no cumulative measures. + */ + private cumulativeMeasuresTrailingInterval(): ParsedInterval | null | undefined { + // Memoized like hasCumulativeMeasures(): computeCumulativeMeasuresTrailingInterval() + // walks the leaf-measure graph, and preAggregationDescriptionFor() calls this once + // per pre-aggregation. + if (!this.cumulativeMeasuresTrailingIntervalComputed) { + this.cumulativeMeasuresTrailingIntervalValue = this.computeCumulativeMeasuresTrailingInterval(); + this.cumulativeMeasuresTrailingIntervalComputed = true; + } + return this.cumulativeMeasuresTrailingIntervalValue; + } + + private computeCumulativeMeasuresTrailingInterval(): ParsedInterval | null | undefined { + // Mirror hasCumulativeMeasures(): inspect the leaf measures so composed + // measures whose cumulative-ness lives in a leaf are handled too. + const measures = [...this.query.measures, ...this.query.measureFilters]; + const collectLeafMeasures = this.query.collectLeafMeasures.bind(this.query); + const leafMeasurePaths = R.uniq( + R.unnest(measures.map(m => this.query.collectFrom([m], collectLeafMeasures, 'collectLeafMeasures'))) + ) as string[]; + const cumulativeMeasures = leafMeasurePaths + .map(p => this.query.newMeasure(p)) + .filter(m => m.isCumulative()); + + if (!cumulativeMeasures.length) { + return undefined; + } + + let maxTrailing: ParsedInterval | undefined; + let maxTrailingSeconds = -1; + + for (const measure of cumulativeMeasures) { + const { rollingWindow } = measure.measureDefinition(); + // Non-fixed windows (to_date/year_to_date/etc.) or running totals have a + // variable trailing extent — don't narrow. + if (!rollingWindow || (rollingWindow.type && rollingWindow.type !== 'fixed')) { + return null; + } + // A leading window needs data after the requested range, and an unbounded + // trailing window has no fixed lookback — neither can be narrowed here. + if (rollingWindow.leading || rollingWindow.trailing === 'unbounded') { + return null; + } + if (rollingWindow.trailing) { + const parsed = parseSqlInterval(rollingWindow.trailing); + const seconds = moment.duration(parsed).asSeconds(); + if (seconds > maxTrailingSeconds) { + maxTrailingSeconds = seconds; + maxTrailing = parsed; + } + } + } + + return maxTrailing; + } + + /** + * Shift the start of a matched time dimension range back by the trailing window + * so partitions covering the rolling window's lookback are also built. + * Leaves the range untouched when there's no range or no trailing interval. + */ + private expandRangeByTrailingWindow( + range: [string, string] | undefined, + trailingInterval: ParsedInterval | null | undefined + ): [string, string] | undefined { + if (!range || !trailingInterval) { + return range; + } + const format = `YYYY-MM-DDTHH:mm:ss.${'S'.repeat(this.query.timestampPrecision())}`; + const start = subtractInterval(moment(range[0], format), trailingInterval).format(format); + return [start, range[1]]; + } + // Return array of `aggregations` columns descriptions in form `()` // Aggregations used in CubeStore create table for describe measures in CubeStore side public aggregationsColumns(cube: string, preAggregation: PreAggregationDefinition): string[] { @@ -326,7 +418,15 @@ export class PreAggregations { let matchedTimeDimension: BaseTimeDimension | undefined; - if (preAggregation.partitionGranularity && !this.hasCumulativeMeasures()) { + // For cumulative (rolling window) measures the matched range can only narrow + // partition selection when every window has a bounded trailing extent and no + // leading window. `undefined` => no cumulative measures (regular narrowing); + // a ParsedInterval => narrow but shift the start back by that trailing window; + // `null` => can't narrow safely, keep building the whole range. + const trailingInterval = this.cumulativeMeasuresTrailingInterval(); + const canMatchTimeDimension = trailingInterval !== null; + + if (preAggregation.partitionGranularity && canMatchTimeDimension) { matchedTimeDimension = this.query.timeDimensions.find(td => { if (!td.dateRange) { return false; @@ -351,7 +451,7 @@ export class PreAggregations { let filters: BaseFilter[] | undefined; - if (preAggregation.partitionGranularity) { + if (preAggregation.partitionGranularity && canMatchTimeDimension) { filters = this.query.filters?.filter((td): td is BaseFilter => { // TODO support all date operators if (td.isDateOperator() && 'camelizeOperator' in td && td.camelizeOperator === 'inDateRange') { @@ -404,9 +504,10 @@ export class PreAggregations { (preAggregation.partitionGranularity || references.timeDimensions[0]?.granularity) && this.refreshRangeQuery(cube).preAggregationStartEndQueries(cube, preAggregation), matchedTimeDimensionDateRange: - preAggregation.partitionGranularity && ( + preAggregation.partitionGranularity && this.expandRangeByTrailingWindow( matchedTimeDimension?.boundaryDateRangeFormatted() || - filters?.[0]?.formattedDateRange() // TODO intersect all date ranges + filters?.[0]?.formattedDateRange(), // TODO intersect all date ranges + trailingInterval ), indexesSql: Object.keys(preAggregation.indexes || {}) .map( diff --git a/packages/cubejs-schema-compiler/test/unit/core-164-rolling-window-partitions.test.ts b/packages/cubejs-schema-compiler/test/unit/core-164-rolling-window-partitions.test.ts new file mode 100644 index 0000000000000..9ff132a0e8b66 --- /dev/null +++ b/packages/cubejs-schema-compiler/test/unit/core-164-rolling-window-partitions.test.ts @@ -0,0 +1,122 @@ +import { PostgresQuery } from '../../src/adapter/PostgresQuery'; +import { BigqueryQuery } from '../../src/adapter/BigqueryQuery'; +import { prepareYamlCompiler } from './PrepareCompiler'; + +// CORE-164: a query with a narrow dateRange against a partitioned pre-aggregation +// should only cause the partitions needed to serve that timeframe to be built. +// +// For cumulative (rolling window) measures the pre-aggregation description used +// to omit `matchedTimeDimensionDateRange` entirely, which made the query +// orchestrator fall back to building the pre-aggregation's whole build range +// (every partition). For a BOUNDED trailing window only the requested range, +// expanded backwards by the trailing window, is actually needed. + +const SCHEMA = ` +cubes: + - name: rent + sql: > + SELECT 1 AS id, '2020-01-01'::timestamp AS date, 100 AS in_place_rent + measures: + - name: count + type: count + - name: rolling_7d + sql: in_place_rent + type: sum + rollingWindow: + trailing: 7 day + - name: rolling_unbounded + sql: in_place_rent + type: sum + rollingWindow: + trailing: unbounded + dimensions: + - name: id + sql: id + type: number + primaryKey: true + - name: date + sql: date + type: time + preAggregations: + - name: rolling_7d_pa + measures: + - rolling_7d + timeDimension: date + granularity: day + partitionGranularity: week + - name: rolling_unbounded_pa + measures: + - rolling_unbounded + timeDimension: date + granularity: day + partitionGranularity: week +`; + +describe('CORE-164 rolling-window pre-aggregation partition scope', () => { + async function preAggDescription(query, QueryClass: any = PostgresQuery) { + const { compiler, cubeEvaluator, joinGraph } = prepareYamlCompiler(SCHEMA); + await compiler.compile(); + const q = new QueryClass({ joinGraph, cubeEvaluator, compiler }, query); + q.buildSqlAndParams(); + return q.preAggregations?.preAggregationsDescription(); + } + + it('bounded trailing window narrows to the requested range expanded by the window', async () => { + const desc: any = await preAggDescription({ + measures: ['rent.rolling_7d'], + timeDimensions: [{ + dimension: 'rent.date', + granularity: 'day', + dateRange: ['2024-06-10', '2024-06-10'], + }], + timezone: 'UTC', + }); + + expect(desc).toHaveLength(1); + expect(desc[0].preAggregationId).toEqual('rent.rolling_7d_pa'); + // The requested day is 2024-06-10; a 7-day trailing window needs data back + // to 2024-06-03, so the matched range must start no later than 2024-06-03 + // and end at the requested day — NOT be undefined (which builds everything). + expect(desc[0].matchedTimeDimensionDateRange).toEqual([ + '2024-06-03T00:00:00.000', + '2024-06-10T23:59:59.999', + ]); + }); + + it('unbounded trailing window still builds the whole range (unchanged)', async () => { + const desc: any = await preAggDescription({ + measures: ['rent.rolling_unbounded'], + timeDimensions: [{ + dimension: 'rent.date', + granularity: 'day', + dateRange: ['2024-06-10', '2024-06-10'], + }], + timezone: 'UTC', + }); + + expect(desc).toHaveLength(1); + expect(desc[0].preAggregationId).toEqual('rent.rolling_unbounded_pa'); + // Unbounded trailing genuinely needs all history => no narrowing. + expect(desc[0].matchedTimeDimensionDateRange).toBeUndefined(); + }); + + it('expanded range keeps the query timestamp precision (microseconds)', async () => { + // BigQuery uses microsecond precision. The expanded start/end must be emitted + // at that same precision — mixing 3-digit and 6-digit strings produced + // malformed ranges (`...,...999999`) that the partition loader rejected. + const desc: any = await preAggDescription({ + measures: ['rent.rolling_7d'], + timeDimensions: [{ + dimension: 'rent.date', + granularity: 'day', + dateRange: ['2024-06-10', '2024-06-10'], + }], + timezone: 'UTC', + }, BigqueryQuery); + + expect(desc[0].matchedTimeDimensionDateRange).toEqual([ + '2024-06-03T00:00:00.000000', + '2024-06-10T23:59:59.999999', + ]); + }); +});