From 44c965cf28300c04ddec3f83da341202119202af Mon Sep 17 00:00:00 2001 From: Dmitry Patsura Date: Tue, 21 Jul 2026 16:32:50 +0200 Subject: [PATCH 1/3] fix(query-orchestrator): Drop touch/used keys when pre-aggregation build fails (#11314) --- .../src/orchestrator/PreAggregationLoader.ts | 47 +++++++-- .../src/orchestrator/PreAggregations.ts | 12 +++ .../test/unit/PreAggregations.test.ts | 99 ++++++++++++++++++- 3 files changed, 150 insertions(+), 8 deletions(-) diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregationLoader.ts b/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregationLoader.ts index a478fce68cb16..8b7cd2250783a 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregationLoader.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregationLoader.ts @@ -463,8 +463,12 @@ export class PreAggregationLoader { } public refresh(newVersionEntry: VersionEntry, invalidationKeys: InvalidationKeys, client) { - this.updateLastTouch(this.targetTableName(newVersionEntry)); + const targetTableName = this.targetTableName(newVersionEntry); + this.updateLastTouch(targetTableName); + let refreshStrategy = this.refreshStoreInSourceStrategy; + let dropTableUsedKey = true; + if (this.preAggregation.external) { const readOnly = this.preAggregation.readOnly || @@ -473,17 +477,46 @@ export class PreAggregationLoader { if (readOnly) { refreshStrategy = this.refreshReadOnlyExternalStrategy; + dropTableUsedKey = false; } else { refreshStrategy = this.refreshWriteStrategy; + dropTableUsedKey = false; } } + return cancelCombinator( - saveCancelFn => refreshStrategy.bind(this)( - client, - newVersionEntry, - saveCancelFn, - invalidationKeys - ) + async saveCancelFn => { + try { + return await refreshStrategy.bind(this)( + client, + newVersionEntry, + saveCancelFn, + invalidationKeys + ); + } catch (e) { + // It's required to remove touch keys, because they are unique per run/table, and it causes + // a large number of touch keys in the cache store + try { + await this.preAggregations.removeTableTouched(targetTableName); + } catch (ne: any) { + this.logger('Error on dropping pre-aggregation touch key after failed build', { + error: (ne.stack || ne), preAggregation: this.preAggregation, requestId: this.requestId, + }); + } + + if (dropTableUsedKey) { + try { + await this.preAggregations.removeTableUsed(targetTableName); + } catch (ne: any) { + this.logger('Error on dropping pre-aggregation table used key after failed build', { + error: (ne.stack || ne), preAggregation: this.preAggregation, requestId: this.requestId, + }); + } + } + + throw e; + } + } ); } diff --git a/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts b/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts index 335ba9a9b68c6..df5015a4a957f 100644 --- a/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts +++ b/packages/cubejs-query-orchestrator/src/orchestrator/PreAggregations.ts @@ -363,6 +363,12 @@ export class PreAggregations { .map(k => k.replace(this.tablesUsedRedisKey(''), '')); } + public async removeTableUsed(tableName: string): Promise { + this.usedCache.delete(tableName); + + await this.queryCache.getCacheDriver().remove(this.tablesUsedRedisKey(tableName)); + } + public async updateLastTouch(tableName: string): Promise { if (this.touchCache.has(tableName)) { return; @@ -388,6 +394,12 @@ export class PreAggregations { .map(k => k.replace(this.tablesTouchRedisKey(''), '')); } + public async removeTableTouched(tableName: string): Promise { + this.touchCache.delete(tableName); + + await this.queryCache.getCacheDriver().remove(this.tablesTouchRedisKey(tableName)); + } + public async updatePreAggBackoff(tableName: string, backoffData: { backoffMultiplier: number, nextTimestamp: Date }): Promise { await this.queryCache.getCacheDriver().set( this.preAggBackoffRedisKey(tableName), diff --git a/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts b/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts index 1838a8e083c9e..e63d6a54b8f31 100644 --- a/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts +++ b/packages/cubejs-query-orchestrator/test/unit/PreAggregations.test.ts @@ -7,7 +7,7 @@ import { } from '@cubejs-backend/shared'; import crypto from 'crypto'; -import { PreAggregationPartitionRangeLoader, PreAggregations, QueryCache, LocalCacheDriver, version } from '../../src'; +import { PreAggregationLoadCache, PreAggregationLoader, PreAggregationPartitionRangeLoader, PreAggregations, QueryCache, LocalCacheDriver, version } from '../../src'; class MockDriver { public tables: string[] = []; @@ -183,6 +183,103 @@ describe('PreAggregations', () => { (queryCache.getCacheDriver() as LocalCacheDriver).reset(); }); + describe('touch/used cache key cleanup', () => { + let preAggregations: PreAggregations; + + beforeEach(() => { + preAggregations = new PreAggregations( + 'TEST', + mockDriverFactory as any, + // eslint-disable-next-line @typescript-eslint/no-empty-function + () => {}, + queryCache!, + { + queueOptions: async () => ({ + executionTimeout: 1, + concurrency: 2, + }), + }, + ); + }); + + test('removeTableUsed / removeTableTouched drop keys and evict the in-memory LRU', async () => { + const table = 'stb_pre_aggregations.orders_abc_def_1'; + + await preAggregations.addTableUsed(table); + await preAggregations.updateLastTouch(table); + + expect(await preAggregations.tablesUsed()).toContain(table); + expect(await preAggregations.tablesTouched()).toContain(table); + + await preAggregations.removeTableUsed(table); + await preAggregations.removeTableTouched(table); + + expect(await preAggregations.tablesUsed()).not.toContain(table); + expect(await preAggregations.tablesTouched()).not.toContain(table); + + // Re-adding must succeed. If the LRU guard weren't evicted, the has() + // short-circuit in addTableUsed/updateLastTouch would suppress the write + // and the keys would stay absent. + await preAggregations.addTableUsed(table); + await preAggregations.updateLastTouch(table); + + expect(await preAggregations.tablesUsed()).toContain(table); + expect(await preAggregations.tablesTouched()).toContain(table); + }); + + test('failed pre-aggregation build drops its touch and used keys', async () => { + const preAggregation = { + preAggregationsSchema: 'stb_pre_aggregations', + tableName: 'stb_pre_aggregations.orders_number_and_count', + dataSource: 'default', + external: false, + loadSql: ['CREATE TABLE stb_pre_aggregations.orders_number_and_count AS SELECT 1', []], + invalidateKeyQueries: [], + }; + + const newVersionEntry = { + table_name: 'stb_pre_aggregations.orders_number_and_count', + structure_version: 'aaaa1111', + content_version: 'bbbb2222', + last_updated_at: 1600000000000, + naming_version: 2, + }; + + const targetTableName = PreAggregations.targetTableName(newVersionEntry); + + // Simulate a build that fails inside the datasource. + jest.spyOn(mockDriver!, 'loadPreAggregationIntoTable') + .mockRejectedValue(new Error('build boom')); + + const loadCache = new PreAggregationLoadCache( + mockDriverFactory as any, + queryCache!, + preAggregations, + { dataSource: 'default' }, + ); + + const loader = new PreAggregationLoader( + mockDriverFactory as any, + // eslint-disable-next-line @typescript-eslint/no-empty-function + () => {}, + queryCache!, + preAggregations, + preAggregation, + [], + loadCache, + { requestId: 'failed-build' }, + ); + + await expect(loader.refresh(newVersionEntry as any, [] as any, mockDriver!)) + .rejects.toThrow('build boom'); + + // The failed attempt must leave no touch/used markers behind, otherwise + // repeated failures accumulate keys in cache until TTL (CORE-646). + expect(await preAggregations.tablesTouched()).not.toContain(targetTableName); + expect(await preAggregations.tablesUsed()).not.toContain(targetTableName); + }); + }); + describe('loadAllPreAggregationsIfNeeded', () => { let preAggregations: PreAggregations | null = null; From bef789c452f2a5549675186f2c53769c9d46063d Mon Sep 17 00:00:00 2001 From: Dmitry Patsura Date: Tue, 21 Jul 2026 16:59:32 +0200 Subject: [PATCH 2/3] docs: add v1.6.x to the list of LTS releases --- docs-mintlify/admin/account-billing/distribution.mdx | 1 + 1 file changed, 1 insertion(+) diff --git a/docs-mintlify/admin/account-billing/distribution.mdx b/docs-mintlify/admin/account-billing/distribution.mdx index 762c6c670e4c6..e565e46e2eaba 100644 --- a/docs-mintlify/admin/account-billing/distribution.mdx +++ b/docs-mintlify/admin/account-billing/distribution.mdx @@ -87,6 +87,7 @@ List of LTS releases: | Version | Status | End-of-life date | | --- | --- | --- | +| `v1.6.x` | Active | July 10, 2027 | | `v1.4.x` | Active | October 29, 2026 | | `v1.0.x` | End-of-life | October 29, 2025 | From d216ebc59ef28e35299cd953649ecea7a35d300e Mon Sep 17 00:00:00 2001 From: Alex Qyoun-ae <4062971+MazterQyou@users.noreply.github.com> Date: Tue, 21 Jul 2026 22:25:52 +0400 Subject: [PATCH 3/3] fix(cubesql): Support type coercion in multi-branch `UNION ALL` (#11284) Signed-off-by: Alex Qyoun-ae <4062971+MazterQyou@users.noreply.github.com> --- packages/cubejs-backend-native/Cargo.lock | 12 ++++---- rust/cubesql/Cargo.lock | 12 ++++---- rust/cubesql/cubesql/Cargo.toml | 2 +- ...on__union_all_ctes_with_type_coercion.snap | 11 ++++++++ .../src/compile/test/test_df_execution.rs | 28 +++++++++++++++++++ 5 files changed, 52 insertions(+), 13 deletions(-) create mode 100644 rust/cubesql/cubesql/src/compile/test/snapshots/cubesql__compile__test__test_df_execution__union_all_ctes_with_type_coercion.snap diff --git a/packages/cubejs-backend-native/Cargo.lock b/packages/cubejs-backend-native/Cargo.lock index 496f7d8bedd3f..b868307a87269 100644 --- a/packages/cubejs-backend-native/Cargo.lock +++ b/packages/cubejs-backend-native/Cargo.lock @@ -897,7 +897,7 @@ dependencies = [ [[package]] name = "cube-ext" version = "1.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "arrow 13.0.0", "chrono", @@ -1073,7 +1073,7 @@ checksum = "e8566979429cf69b49a5c740c60791108e86440e8be149bbea4fe54d2c32d6e2" [[package]] name = "datafusion" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow 13.0.0", @@ -1106,7 +1106,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "arrow 13.0.0", "ordered-float 2.10.1", @@ -1117,7 +1117,7 @@ dependencies = [ [[package]] name = "datafusion-data-access" version = "1.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "async-trait", "chrono", @@ -1130,7 +1130,7 @@ dependencies = [ [[package]] name = "datafusion-expr" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow 13.0.0", @@ -1141,7 +1141,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow 13.0.0", diff --git a/rust/cubesql/Cargo.lock b/rust/cubesql/Cargo.lock index 0116c719c5b52..ca57b5f4612fb 100644 --- a/rust/cubesql/Cargo.lock +++ b/rust/cubesql/Cargo.lock @@ -698,7 +698,7 @@ dependencies = [ [[package]] name = "cube-ext" version = "1.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "arrow", "chrono", @@ -822,7 +822,7 @@ dependencies = [ [[package]] name = "datafusion" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow", @@ -855,7 +855,7 @@ dependencies = [ [[package]] name = "datafusion-common" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "arrow", "ordered-float 2.10.0", @@ -866,7 +866,7 @@ dependencies = [ [[package]] name = "datafusion-data-access" version = "1.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "async-trait", "chrono", @@ -879,7 +879,7 @@ dependencies = [ [[package]] name = "datafusion-expr" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow", @@ -890,7 +890,7 @@ dependencies = [ [[package]] name = "datafusion-physical-expr" version = "7.0.0" -source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=400af46a3f2cbc9f893021220af3fb5d6fcabd65#400af46a3f2cbc9f893021220af3fb5d6fcabd65" +source = "git+https://github.com/cube-js/arrow-datafusion.git?rev=b01223b6454b0d72154d15a88aec76efc9da726d#b01223b6454b0d72154d15a88aec76efc9da726d" dependencies = [ "ahash 0.7.8", "arrow", diff --git a/rust/cubesql/cubesql/Cargo.toml b/rust/cubesql/cubesql/Cargo.toml index 900f3713dabd3..e25b02357986c 100644 --- a/rust/cubesql/cubesql/Cargo.toml +++ b/rust/cubesql/cubesql/Cargo.toml @@ -10,7 +10,7 @@ homepage = "https://cube.dev" [dependencies] arc-swap = "1" -datafusion = { git = 'https://github.com/cube-js/arrow-datafusion.git', rev = "400af46a3f2cbc9f893021220af3fb5d6fcabd65", default-features = false, features = [ +datafusion = { git = 'https://github.com/cube-js/arrow-datafusion.git', rev = "b01223b6454b0d72154d15a88aec76efc9da726d", default-features = false, features = [ "regex_expressions", "unicode_expressions", ] } diff --git a/rust/cubesql/cubesql/src/compile/test/snapshots/cubesql__compile__test__test_df_execution__union_all_ctes_with_type_coercion.snap b/rust/cubesql/cubesql/src/compile/test/snapshots/cubesql__compile__test__test_df_execution__union_all_ctes_with_type_coercion.snap new file mode 100644 index 0000000000000..bdf160e45543a --- /dev/null +++ b/rust/cubesql/cubesql/src/compile/test/snapshots/cubesql__compile__test__test_df_execution__union_all_ctes_with_type_coercion.snap @@ -0,0 +1,11 @@ +--- +source: cubesql/src/compile/test/test_df_execution.rs +expression: "execute_query(query.to_string(), DatabaseProtocol::PostgreSQL).await.unwrap()" +--- ++---+-----+ +| l | t | ++---+-----+ +| A | 1 | +| B | 2 | +| C | 3.5 | ++---+-----+ diff --git a/rust/cubesql/cubesql/src/compile/test/test_df_execution.rs b/rust/cubesql/cubesql/src/compile/test/test_df_execution.rs index 2420a09b2fe73..73576a64549b3 100644 --- a/rust/cubesql/cubesql/src/compile/test/test_df_execution.rs +++ b/rust/cubesql/cubesql/src/compile/test/test_df_execution.rs @@ -92,6 +92,34 @@ GROUP BY ); } +#[tokio::test] +async fn union_all_ctes_with_type_coercion() { + init_testing_logger(); + + // language=PostgreSQL + let query = r#" +WITH a AS ( + SELECT 1::bigint AS t LIMIT 1 +), b AS ( + SELECT 2::bigint AS t LIMIT 1 +), c AS ( + SELECT 3.5::float8 AS t LIMIT 1 +) +SELECT 'A' AS l, t FROM a +UNION ALL +SELECT 'B' AS l, t FROM b +UNION ALL +SELECT 'C' AS l, t FROM c +; + "#; + + insta::assert_snapshot!( + execute_query(query.to_string(), DatabaseProtocol::PostgreSQL) + .await + .unwrap() + ); +} + /// See https://www.postgresql.org/docs/current/functions-math.html #[tokio::test] async fn test_round() {