[pull] master from cube-js:master

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 |
 

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

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<void> {
+    this.usedCache.delete(tableName);
+
+    await this.queryCache.getCacheDriver().remove(this.tablesUsedRedisKey(tableName));
+  }
+
   public async updateLastTouch(tableName: string): Promise<void> {
     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<void> {
+    this.touchCache.delete(tableName);
+
+    await this.queryCache.getCacheDriver().remove(this.tablesTouchRedisKey(tableName));
+  }
+
   public async updatePreAggBackoff(tableName: string, backoffData: { backoffMultiplier: number, nextTimestamp: Date }): Promise<void> {
     await this.queryCache.getCacheDriver().set(
       this.preAggBackoffRedisKey(tableName),

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;
 

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",
 ] }

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 |
++---+-----+

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() {