fix(schema-compiler): only build pre-agg partitions needed for bounded rolling-window queries

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<typeof parseSqlInterval>;
+
 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 `<func>(<column>)`
   // 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(

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',
+    ]);
+  });
+});