fix(schema-compiler,query-orchestrator): bound the rollupLambda source query #11708
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -165,6 +165,11 @@ export type LambdaOptions = { | |
| export type LambdaQuery = { | ||
| sqlAndParams: QueryWithParams, | ||
| cacheKeyQueries: any[], | ||
| /** | ||
| * Limit actually rendered into `sqlAndParams`. Undefined for SQL that instead carries the | ||
| * MAX_SOURCE_ROW_LIMIT placeholder as a param. | ||
| */ | ||
| maxSourceRowLimit?: number, | ||
| }; | ||
|
|
||
| export type PreAggregationDescription = { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -796,6 +796,111 @@ describe('PreAggregations', () => { | |
| }); | ||
| }); | ||
|
|
||
| // @link https://github.com/cube-js/cube/issues/11682 | ||
| describe('lambda source query loading', () => { | ||
| const buildRangeEnd = '2024-01-02T23:59:59.999'; | ||
|
|
||
| // The PreAggregationLoader.prototype spy below would otherwise leak into later tests. | ||
| afterEach(() => { | ||
| jest.restoreAllMocks(); | ||
| }); | ||
|
|
||
| const createLambdaLoader = (matchedTimeDimensionDateRange?: [string, string]) => { | ||
| const loader = new PreAggregationPartitionRangeLoader( | ||
| {} as any, // driverFactory | ||
| // eslint-disable-next-line @typescript-eslint/no-empty-function | ||
| () => {}, // logger | ||
| { options: {} } as any, // queryCache | ||
| {} as any, // preAggregations | ||
| mockPreAggregation({ | ||
| preAggregationId: 'Orders.d', | ||
| rollupLambdaId: 'Orders.d_lambda', | ||
| lastRollupLambda: true, | ||
| unionWithSourceData: true, | ||
| matchedTimeDimensionDateRange, | ||
| }) as any, | ||
| [], // preAggregationsTablesToTempTables | ||
| { getTableColumnTypes: jest.fn().mockResolvedValue([{ name: 'ts', type: 'timestamp' }]) } as any, | ||
| { | ||
| lambdaQuery: { | ||
| sqlAndParams: ['SELECT * FROM public.orders WHERE ts > ?', [FROM_PARTITION_RANGE]], | ||
| cacheKeyQueries: [], | ||
| }, | ||
| } as any, | ||
| ); | ||
|
|
||
| jest.spyOn(loader as any, 'partitionRanges').mockResolvedValue({ | ||
| buildRange: ['2024-01-01T00:00:00.000', buildRangeEnd], | ||
| partitionRanges: [['2024-01-02T00:00:00.000', buildRangeEnd]], | ||
| }); | ||
| jest.spyOn(PreAggregationLoader.prototype, 'loadPreAggregation').mockResolvedValue({ | ||
| targetTableName: 'stb_pre_aggregations.orders_d20240102_abc_def', | ||
| refreshKeyValues: [], | ||
| lastUpdatedAt: 1, | ||
| buildRangeEnd, | ||
| } as any); | ||
| const downloadLambdaTable = jest.spyOn(loader as any, 'downloadLambdaTable').mockResolvedValue({ | ||
| name: 'lambda_stb_pre_aggregations_orders_d', | ||
| columns: [], | ||
| csvRows: '', | ||
| }); | ||
|
|
||
| return { loader, downloadLambdaTable }; | ||
| }; | ||
|
|
||
| test('skips the source query when the requested range is inside the built range', async () => { | ||
| const { loader, downloadLambdaTable } = createLambdaLoader(['2024-01-01T00:00:00.000', buildRangeEnd]); | ||
|
|
||
| const result: any = await loader.loadPreAggregations(); | ||
|
|
||
| expect(downloadLambdaTable).not.toHaveBeenCalled(); | ||
| expect(result.lambdaTable).toBeUndefined(); | ||
| expect(result.targetTableName).toEqual('stb_pre_aggregations.orders_d20240102_abc_def'); | ||
| }); | ||
|
|
||
| test('runs the source query when the requested range extends past the built range', async () => { | ||
| const { loader, downloadLambdaTable } = createLambdaLoader(['2024-01-01T00:00:00.000', '2024-01-05T23:59:59.999']); | ||
|
|
||
| const result: any = await loader.loadPreAggregations(); | ||
|
|
||
| expect(downloadLambdaTable).toHaveBeenCalledWith(buildRangeEnd, [{ name: 'ts', type: 'timestamp' }]); | ||
| expect(result.lambdaTable?.name).toEqual('lambda_stb_pre_aggregations_orders_d'); | ||
| expect(result.targetTableName).toMatch(/UNION ALL SELECT \* FROM lambda_stb_pre_aggregations_orders_d/); | ||
| }); | ||
|
|
||
| test('runs the source query when no date range was requested', async () => { | ||
| const { loader, downloadLambdaTable } = createLambdaLoader(undefined); | ||
|
|
||
| await loader.loadPreAggregations(); | ||
|
|
||
| expect(downloadLambdaTable).toHaveBeenCalled(); | ||
| }); | ||
| }); | ||
|
|
||
| describe('lambdaSourceDataCovered', () => { | ||
| const covered = (matchedTimeDimensionDateRange: any, buildRangeEnd: any) => ( | ||
| createLoader({ matchedTimeDimensionDateRange }) as any | ||
| ).lambdaSourceDataCovered(buildRangeEnd); | ||
|
|
||
| test('covered when the requested range ends within the built range', () => { | ||
| expect(covered(['2024-01-01T00:00:00.000', '2024-01-02T23:59:59.999'], '2024-01-02T23:59:59.999')).toBe(true); | ||
| expect(covered(['2024-01-01T00:00:00.000', '2024-01-02T23:59:59.999'], '2024-01-03T23:59:59.999')).toBe(true); | ||
| }); | ||
|
|
||
| test('not covered when the requested range extends past the built range', () => { | ||
| expect(covered(['2024-01-01T00:00:00.000', '2024-01-05T23:59:59.999'], '2024-01-03T23:59:59.999')).toBe(false); | ||
| }); | ||
|
|
||
| test('normalizes a buildRangeEnd read back from the DB with a Z suffix', () => { | ||
| expect(covered(['2024-01-01T00:00:00.000', '2024-01-02T23:59:59.999'], '2024-01-02T23:59:59.999Z')).toBe(true); | ||
| }); | ||
|
|
||
| test('keeps the source query when either bound is unknown', () => { | ||
| expect(covered(undefined, '2024-01-02T23:59:59.999')).toBe(false); | ||
| expect(covered(['2024-01-01T00:00:00.000', '2024-01-02T23:59:59.999'], undefined)).toBe(false); | ||
| }); | ||
| }); | ||
|
|
||
| describe('partitionTableName', () => { | ||
| test('should generate correct table names for different granularities', () => { | ||
| const testDateRange: [string, string] = ['2024-01-05T12:34:56.789', '2024-01-05T23:59:59.999']; | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -1093,26 +1093,41 @@ export class BaseQuery { | |
| const lambdaPreAgg = preAggForQuery.referencedPreAggregations[preAggForQuery.referencedPreAggregations.length - 1]; | ||
| // TODO(cristipp) Use source query instead of preaggregation references. | ||
| const references = this.cubeEvaluator.evaluatePreAggregationReferences(lambdaPreAgg.cube, lambdaPreAgg.preAggregation); | ||
| const [timeDimension] = references.timeDimensions; | ||
| // @see https://github.com/cube-js/cube/issues/11682 | ||
| const sourceDateRange = timeDimension && | ||
| this.preAggregations.lambdaSourceDateRange(lambdaPreAgg, preAggForQuery); | ||
| // Tesseract renders LIMIT as an inline number and can't carry the MAX_SOURCE_ROW_LIMIT | ||
| // placeholder (it parses to None, emitting no LIMIT at all), so resolve it here. | ||
| const resolvedRowLimit = this.options.maxSourceRowLimit ?? getEnv('maxSourceRowLimit'); | ||
| const maxSourceRowLimit = typeof resolvedRowLimit === 'number' && resolvedRowLimit > 0 | ||
| ? resolvedRowLimit | ||
|
Comment on lines
+1100
to
+1104
Contributor
There was a problem hiding this comment. Baking the limit here silently drops support for the Two things line up badly:
So for a user with Either plumb |
||
| : undefined; | ||
| const lambdaQuery = this.newSubQuery( | ||
| { | ||
| measures: references.measures, | ||
| dimensions: references.dimensions, | ||
| timeDimensions: references.timeDimensions, | ||
| filters: [ | ||
| ...this.options.filters ?? [], | ||
| ...(timeDimension ? [{ | ||
| member: timeDimension.dimension, | ||
| operator: 'afterDate', | ||
| values: [FROM_PARTITION_RANGE] | ||
| }] : []), | ||
| // Kept separate from the afterDate filter on purpose: inDateRange's lower bound is | ||
| // inclusive and would double count rows sitting exactly at the partition end. | ||
| ...(sourceDateRange ? [{ | ||
| member: timeDimension.dimension, | ||
| operator: 'inDateRange', | ||
| values: sourceDateRange | ||
| }] : []), | ||
| ], | ||
| segments: this.options.segments, | ||
| order: [], | ||
| limit: undefined, | ||
| offset: undefined, | ||
| rowLimit: maxSourceRowLimit ?? MAX_SOURCE_ROW_LIMIT, | ||
| preAggregationQuery: true, | ||
| } | ||
| ); | ||
|
|
@@ -1121,7 +1136,11 @@ export class BaseQuery { | |
| () => this.cacheKeyQueries(), | ||
| { preAggregationQuery: true } | ||
| ); | ||
| result[this.preAggregations.preAggregationId(lambdaPreAgg)] = { | ||
| sqlAndParams, | ||
| cacheKeyQueries, | ||
| maxSourceRowLimit, | ||
| }; | ||
| } | ||
| return result; | ||
| } | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -313,20 +313,24 @@ export class PreAggregations { | |
| return []; | ||
| } | ||
|
|
||
| /** | ||
| * Requested time dimension range this pre-aggregation was matched by, as local | ||
| * (timezone naked) ISO strings, e.g. ['2024-02-01T00:00:00.000', '2024-02-29T23:59:59.999']. | ||
| * Drives partition selection, and bounds the rollupLambda source query so both halves of | ||
| * the union agree. | ||
| */ | ||
| public matchedTimeDimensionDateRangeFor(foundPreAggregation: PreAggregationForQuery): [string, string] | undefined { | ||
| const { preAggregation } = foundPreAggregation; | ||
|
|
||
| if (!preAggregation.partitionGranularity) { | ||
| return undefined; | ||
| } | ||
|
Contributor
There was a problem hiding this comment. The PR description says "so nothing is recomputed", but extracting this into a public method actually adds a computation: Worth memoizing either |
||
|
|
||
| const allBackAliasMembers = this.query.allBackAliasMembers(); | ||
|
|
||
| let matchedTimeDimension: BaseTimeDimension | undefined; | ||
|
|
||
| if (!this.hasCumulativeMeasures()) { | ||
| matchedTimeDimension = this.query.timeDimensions.find(td => { | ||
| if (!td.dateRange) { | ||
| return false; | ||
|
|
@@ -349,24 +353,77 @@ export class PreAggregations { | |
| }); | ||
| } | ||
|
|
||
| const filters = this.query.filters?.filter((td): td is BaseFilter => { | ||
| // TODO support all date operators | ||
| if (td.isDateOperator() && 'camelizeOperator' in td && td.camelizeOperator === 'inDateRange') { | ||
| if (td.dimension === foundPreAggregation.references.timeDimensions[0].dimension) { | ||
| return true; | ||
| } | ||
|
|
||
| // Handling for views | ||
| return td.dimension === allBackAliasMembers[foundPreAggregation.references.timeDimensions[0].dimension]; | ||
| } | ||
|
|
||
| return false; | ||
| }); | ||
|
|
||
| return matchedTimeDimension?.boundaryDateRangeFormatted() || | ||
| filters?.[0]?.formattedDateRange() || // TODO intersect all date ranges | ||
| undefined; | ||
| } | ||
|
|
||
| /** | ||
| * Mirrors the merge done in preAggregationDescriptionsForUsageInfos(), otherwise a forward | ||
| * shifted usage would be bounded tighter than the partitions it unions with and lose | ||
| * source rows. | ||
| */ | ||
| public lambdaSourceDateRange( | ||
| lambdaPreAggregation: PreAggregationForQuery, | ||
| rollupLambda: PreAggregationForQuery | ||
| ): [string, string] | undefined { | ||
| const matchedDateRange = this.matchedTimeDimensionDateRangeFor(lambdaPreAggregation); | ||
|
|
||
| if (!matchedDateRange) { | ||
| return undefined; | ||
| } | ||
|
|
||
| const usageInfos = (this.preAggregationUsageInfos || []).filter( | ||
| usageInfo => usageInfo.cubeName === rollupLambda.cube && | ||
| usageInfo.preAggregationName === rollupLambda.preAggregationName | ||
| ); | ||
|
|
||
| if (usageInfos.length === 0) { | ||
| return matchedDateRange; | ||
| } | ||
|
|
||
| let merged: [string, string] | undefined; | ||
|
|
||
| for (const usageInfo of usageInfos) { | ||
| const usageDateRange = PreAggregations.mergeUsageDateRanges(usageInfo.usages); | ||
| if (!usageDateRange) { | ||
| // An unknown usage range may need anything, so bound nothing. | ||
| return undefined; | ||
| } | ||
| merged = merged | ||
| ? [ | ||
| usageDateRange[0] < merged[0] ? usageDateRange[0] : merged[0], | ||
| usageDateRange[1] > merged[1] ? usageDateRange[1] : merged[1], | ||
| ] | ||
| : usageDateRange; | ||
| } | ||
|
|
||
| return merged; | ||
| } | ||
|
|
||
| private preAggregationDescriptionFor(cube: string, foundPreAggregation: PreAggregationForQuery): FullPreAggregationDescription { | ||
| const { preAggregationName, preAggregation, references } = foundPreAggregation; | ||
|
|
||
| const tableName = this.preAggregationTableName(cube, preAggregationName, preAggregation); | ||
| const invalidateKeyQueries = this.query.preAggregationInvalidateKeyQueries(cube, preAggregation, preAggregationName); | ||
| const queryForSqlEvaluation = this.query.preAggregationQueryForSqlEvaluation(cube, preAggregation); | ||
| // Atm this is only defined in KsqlQuery but without it partitions are recreated on every refresh | ||
| const partitionInvalidateKeyQueries = queryForSqlEvaluation.partitionInvalidateKeyQueries?.(cube, preAggregation); | ||
|
|
||
| const uniqueKeyColumnsDefault = () => null; | ||
| const uniqueKeyColumns = ({ | ||
| rollup: () => queryForSqlEvaluation.preAggregationUniqueKeyColumns(cube, preAggregation), | ||
|
|
@@ -403,11 +460,7 @@ export class PreAggregations { | |
| preAggregationStartEndQueries: | ||
| (preAggregation.partitionGranularity || references.timeDimensions[0]?.granularity) && | ||
| this.refreshRangeQuery(cube).preAggregationStartEndQueries(cube, preAggregation), | ||
| matchedTimeDimensionDateRange: this.matchedTimeDimensionDateRangeFor(foundPreAggregation), | ||
| indexesSql: Object.keys(preAggregation.indexes || {}) | ||
| .map( | ||
| index => { | ||
|
|
||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The skip decision and the SQL bound can disagree in the multi-usage case, and the disagreement is in the unsafe direction.
lambdaSourceDateRange()bounds the SQL by the union of every usage range for the rollupLambda, but this check readsthis.preAggregation.matchedTimeDimensionDateRange, which comes from one description. With several usage infos,preAggregationDescriptionsForUsageInfos()produces one description per usage (each with its own merged range) and thenpreAggregationsDescription()collapses them withR.uniqBy(desc => desc.tableName)(schema-compiler/src/adapter/PreAggregations.ts:145) — the first one wins.So if usage A ends inside the built range and usage B extends past it, the surviving description carries A's range,
lambdaSourceDataCovered()returnstrue, and the fresh tail that B's SQL bound was explicitly widened to fetch is never downloaded. Before this change the download always ran.Given the PR already notes that multi-usage descriptions don't add the lambda table to their per-usage unions, this may be masked today — but the two computations should be fed from the same value rather than relying on that. Consider having the schema compiler stamp the effective source range onto
LambdaQuery(next tomaxSourceRowLimit) and comparing against that here, so the skip can never be wider than the bound actually rendered into the SQL.