diff --git a/services/cubejs/src/routes/loadExport.js b/services/cubejs/src/routes/loadExport.js index ce6daa59..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; diff --git a/services/cubejs/test/loadExport.test.js b/services/cubejs/test/loadExport.test.js index 6450814b..320ca728 100644 --- a/services/cubejs/test/loadExport.test.js +++ b/services/cubejs/test/loadExport.test.js @@ -358,6 +358,49 @@ describe("maybeHandleLoadExport", () => { ]); }); + it("compiles without Cube Store pre-aggregations so the export can stream", async () => { + const req = createRequest({ + format: "arrow", + query: { + dimensions: ["Orders.city"], + measures: ["Orders.count"], + }, + }); + const res = new MockResponse(); + const compiledQueries = []; + + const cubejs = createMockCube({ + dbType: "clickhouse", + normalizedQuery: req.body.query, + compiledQueries, + sqlQuery: { + sql: ["SELECT city, count(*) FROM orders GROUP BY 1", []], + 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 }], + }); + + await maybeHandleLoadExport(req, res, () => { + throw new Error("next should not be called — the export streams"); + }, cubejs); + + // 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 () => { const req = createRequest({ format: "csv", @@ -470,6 +513,7 @@ function createMockCube(options) { streamRows = [], nativePreAggs = null, driver = null, + compiledQueries = null, } = options; return { @@ -506,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