From 47f9bfdaed62c033989eb85efd7aeb857eea5a7a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Helgi=20P=C3=A1ll=20Helgason?= Date: Thu, 15 Oct 2026 08:47:11 +0000 Subject: [PATCH 1/2] fix(cubejs): buffered /load for csv/arrow queries that run on Cube Store Cube's query queue refuses to stream a query whose compiled plan targets Cube Store ("Streaming queries to Cube Store aren't supported"), so arrow/csv /load requests that hit an external rollup failed with a 500. The plan already flags pre-aggregations to skip the native ClickHouse path, but still sent them to semantic streaming. When the compiled query is external, fall through to the buffered /load pipeline instead. Co-Authored-By: Claude Opus 5.5 --- services/cubejs/src/routes/loadExport.js | 6 +++ services/cubejs/test/loadExport.test.js | 49 ++++++++++++++++++++++++ 2 files changed, 55 insertions(+) diff --git a/services/cubejs/src/routes/loadExport.js b/services/cubejs/src/routes/loadExport.js index ce6daa59..db339c82 100644 --- a/services/cubejs/src/routes/loadExport.js +++ b/services/cubejs/src/routes/loadExport.js @@ -486,6 +486,12 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { if (!plan || plan.handled) return true; if (plan.unsupported) return false; + // Cube's query queue refuses to stream a query that runs against Cube Store + // ("Streaming queries to Cube Store aren't supported"). When the compiled + // plan targets the external store, neither the native ClickHouse path nor + // semantic streaming can serve it, so let the buffered /load pipeline do. + if (plan.streamingQuery?.external) return false; + const abortController = createAbortController(res); // 099 T089 (FR-091): `Dataset Exported` audit. Tenant rides tokenPayload diff --git a/services/cubejs/test/loadExport.test.js b/services/cubejs/test/loadExport.test.js index 6450814b..3e660f4b 100644 --- a/services/cubejs/test/loadExport.test.js +++ b/services/cubejs/test/loadExport.test.js @@ -358,6 +358,55 @@ describe("maybeHandleLoadExport", () => { ]); }); + it("falls through to the buffered path when the plan runs against Cube Store", async () => { + const req = createRequest({ + format: "arrow", + query: { + dimensions: ["Orders.city"], + measures: ["Orders.count"], + }, + }); + const res = new MockResponse(); + let nextCalls = 0; + + const cubejs = createMockCube({ + dbType: "clickhouse", + normalizedQuery: req.body.query, + sqlQuery: { + sql: ["SELECT city, count FROM stb_pre_agg.orders_rollup", []], + // Rollups served from Cube Store compile with external: true; Cube's + // queue cannot stream those. + external: true, + preAggregations: [ + { + tableName: "stb_pre_agg.orders_rollup", + dataSource: "default", + external: true, + }, + ], + aliasNameToMember: { + "Orders.city": "Orders.city", + "Orders.count": "Orders.count", + }, + }, + metaConfig: createMetaConfig({ + measures: [{ name: "Orders.count", type: "count" }], + dimensions: [{ name: "Orders.city", type: "string" }], + }), + streamRows: [{ "Orders.city": "Reykjavik", "Orders.count": 3 }], + driver: createClickHouseDriver({ + arrowChunks: [Buffer.from("should-not-be-used")], + }), + }); + + await maybeHandleLoadExport(req, res, () => { + nextCalls++; + }, cubejs); + + assert.equal(nextCalls, 1); + assert.equal(res.headersSent, false); + }); + it("falls through to the buffered path when semantic streaming is unavailable", async () => { const req = createRequest({ format: "csv", From 18bd831f6a08669333ab4b087c916d3418d99c70 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Helgi=20P=C3=A1ll=20Helgason?= Date: Thu, 15 Oct 2026 09:59:25 +0000 Subject: [PATCH 2/2] fix(cubejs): stream csv/arrow exports from the source db, not Cube Store Replaces the buffered fallback: that path cut Arrow at 100k rows, skipped the Dataset Exported audit and held the whole result in memory. Compile the export with disableExternalPreAggregations, as Cube's own SQL API does for streaming, so the query never targets Cube Store. Co-Authored-By: Claude Opus 5.5 --- services/cubejs/src/routes/loadExport.js | 25 +++++++++------- services/cubejs/test/loadExport.test.js | 38 +++++++++++------------- 2 files changed, 32 insertions(+), 31 deletions(-) diff --git a/services/cubejs/src/routes/loadExport.js b/services/cubejs/src/routes/loadExport.js index db339c82..a394d46e 100644 --- a/services/cubejs/src/routes/loadExport.js +++ b/services/cubejs/src/routes/loadExport.js @@ -323,8 +323,18 @@ async function buildLoadExportPlan(req, res, cubejs, query) { metaConfig = apiGateway.filterVisibleItemsInMeta(context, metaConfig); const annotation = prepareAnnotation(metaConfig, normalizedQuery); + // Compile against the source database, not Cube Store: Cube's queue + // refuses to stream an external query ("Streaming queries to Cube Store + // aren't supported"). Cube's own SQL API streams the same way + // (gateway.js sqlApiLoad). Rollups in the source database still apply. const sqlQuery = ( - await apiGateway.getSqlQueriesInternal(context, normalizedQueries) + await apiGateway.getSqlQueriesInternal( + context, + normalizedQueries.map((q) => ({ + ...q, + disableExternalPreAggregations: true, + })), + ) )[0]; const streamingQuery = { ...sqlQuery, @@ -338,10 +348,9 @@ async function buildLoadExportPlan(req, res, cubejs, query) { }; // The compiled sqlQuery carries a preAggregations array when the query - // plan routes through pre-aggregation tables (typically in CubeStore). - // Neither the native ClickHouse path nor semantic streaming can reach - // CubeStore tables, so flag it so tryHandleLoadExport can bail out early - // and let the standard Cube.js load pipeline handle it. + // plan routes through pre-aggregation tables in the source database. The + // native ClickHouse passthrough cannot resolve those, so flag it and let + // semantic streaming (which loads them) serve the query. const usesPreAggregations = Array.isArray(sqlQuery.preAggregations) && sqlQuery.preAggregations.length > 0; @@ -486,12 +495,6 @@ async function tryHandleLoadExport(req, res, cubejs, query, format) { if (!plan || plan.handled) return true; if (plan.unsupported) return false; - // Cube's query queue refuses to stream a query that runs against Cube Store - // ("Streaming queries to Cube Store aren't supported"). When the compiled - // plan targets the external store, neither the native ClickHouse path nor - // semantic streaming can serve it, so let the buffered /load pipeline do. - if (plan.streamingQuery?.external) return false; - const abortController = createAbortController(res); // 099 T089 (FR-091): `Dataset Exported` audit. Tenant rides tokenPayload diff --git a/services/cubejs/test/loadExport.test.js b/services/cubejs/test/loadExport.test.js index 3e660f4b..320ca728 100644 --- a/services/cubejs/test/loadExport.test.js +++ b/services/cubejs/test/loadExport.test.js @@ -358,7 +358,7 @@ describe("maybeHandleLoadExport", () => { ]); }); - it("falls through to the buffered path when the plan runs against Cube Store", async () => { + it("compiles without Cube Store pre-aggregations so the export can stream", async () => { const req = createRequest({ format: "arrow", query: { @@ -367,23 +367,14 @@ describe("maybeHandleLoadExport", () => { }, }); const res = new MockResponse(); - let nextCalls = 0; + const compiledQueries = []; const cubejs = createMockCube({ dbType: "clickhouse", normalizedQuery: req.body.query, + compiledQueries, sqlQuery: { - sql: ["SELECT city, count FROM stb_pre_agg.orders_rollup", []], - // Rollups served from Cube Store compile with external: true; Cube's - // queue cannot stream those. - external: true, - preAggregations: [ - { - tableName: "stb_pre_agg.orders_rollup", - dataSource: "default", - external: true, - }, - ], + sql: ["SELECT city, count(*) FROM orders GROUP BY 1", []], aliasNameToMember: { "Orders.city": "Orders.city", "Orders.count": "Orders.count", @@ -394,17 +385,20 @@ describe("maybeHandleLoadExport", () => { dimensions: [{ name: "Orders.city", type: "string" }], }), streamRows: [{ "Orders.city": "Reykjavik", "Orders.count": 3 }], - driver: createClickHouseDriver({ - arrowChunks: [Buffer.from("should-not-be-used")], - }), }); await maybeHandleLoadExport(req, res, () => { - nextCalls++; + throw new Error("next should not be called — the export streams"); }, cubejs); - assert.equal(nextCalls, 1); - assert.equal(res.headersSent, false); + // Cube's queue refuses to stream a query compiled against Cube Store + // ("Streaming queries to Cube Store aren't supported"). + assert.equal(compiledQueries.length, 1); + assert.equal(compiledQueries[0].disableExternalPreAggregations, true); + const parsed = tableFromIPC(res.binaryOutput); + assert.deepEqual(parsed.toArray().map((row) => row.toJSON()), [ + { "Orders.city": "Reykjavik", "Orders.count": 3 }, + ]); }); it("falls through to the buffered path when semantic streaming is unavailable", async () => { @@ -519,6 +513,7 @@ function createMockCube(options) { streamRows = [], nativePreAggs = null, driver = null, + compiledQueries = null, } = options; return { @@ -555,7 +550,10 @@ function createMockCube(options) { metaConfig: async () => metaConfig, }), filterVisibleItemsInMeta: (context, cubes) => cubes, - getSqlQueriesInternal: async () => [sqlQuery], + getSqlQueriesInternal: async (context, queries) => { + compiledQueries?.push(...queries); + return [sqlQuery]; + }, getAdapterApi: async () => ({ streamQuery: async () => createRowStream(streamRows), getQueryOrchestrator: nativePreAggs