diff --git a/services/cubejs/scripts/patchCubeYamlCompiler.mjs b/services/cubejs/scripts/patchCubeYamlCompiler.mjs index f92a34d4..cf378fce 100644 --- a/services/cubejs/scripts/patchCubeYamlCompiler.mjs +++ b/services/cubejs/scripts/patchCubeYamlCompiler.mjs @@ -45,6 +45,39 @@ const BIGQUERY_PATCHED = ` const rowStream = new HydrationStream_1.Hydrat stream.on('error', (err) => rowStream.destroy(err)); stream.pipe(rowStream);`; +const BIGQUERY_DOWNLOAD_ORIGINAL = ` async loadTablesForDataset(dataset) {`; + +const BIGQUERY_DOWNLOAD_PATCHED = ` // Read-only rollups: column types from BigQuery's result schema, as + // tableColumnTypes gives the staging path. The base driver guesses them + // from the rows: a decimal after whole numbers fails the Cube Store load, + // and an empty partition cannot be typed at all. + async downloadQueryResults(query, values, options) { + const labels = this.buildQueryLabels(options); + const [job] = await this.bigquery.createQueryJob({ + query, + params: values, + parameterMode: 'positional', + useLegacySql: false, + wrapIntegers: true, + ...(labels ? { labels } : {}), + }); + const [rows] = await this.waitForJobResult(job, options, true); + const [, , response] = await job.getQueryResults({ maxResults: 0, autoPaginate: false }); + return { + rows: rows.map(row => (0, HydrationStream_1.transformRow)(row)), + types: response.schema.fields.map((c) => { + if (c.type === 'NUMERIC' || c.type === 'DECIMAL') { + return { name: c.name, type: this.toGenericType(c.type, 38, 9) }; + } + if (c.type === 'BIGNUMERIC' || c.type === 'BIGDECIMAL') { + return { name: c.name, type: this.toGenericType(c.type, 76, 38) }; + } + return { name: c.name, type: this.toGenericType(c.type) }; + }), + }; + } + async loadTablesForDataset(dataset) {`; + // Both stream-download paths wrap the driver's row stream the same way. const LOADER_ORIGINAL = ` tableData.rowStream.pipe(stream); tableData.rowStream = stream;`; @@ -83,6 +116,10 @@ export function patchBigQueryDriverSource(source) { return applyPatch(source, BIGQUERY_ORIGINAL, BIGQUERY_PATCHED, "BigQuery driver"); } +export function patchBigQueryDownloadSource(source) { + return applyPatch(source, BIGQUERY_DOWNLOAD_ORIGINAL, BIGQUERY_DOWNLOAD_PATCHED, "BigQuery read-only download"); +} + export function patchPreAggregationLoaderSource(source) { return applyPatch(source, LOADER_ORIGINAL, LOADER_PATCHED, "pre-aggregation loader", 2); } @@ -118,7 +155,11 @@ export async function patchInstalledCompiler() { await supportedPackageRoot(BIGQUERY_PACKAGE), "dist/src/BigQueryDriver.js", ); - const bigQueryChanged = await patchFile(bigQueryPath, patchBigQueryDriverSource); + const bigQueryChanged = await patchFile(bigQueryPath, (source) => { + const stream = patchBigQueryDriverSource(source); + const download = patchBigQueryDownloadSource(stream.source); + return { source: download.source, changed: stream.changed || download.changed }; + }); const loaderPath = resolve( await supportedPackageRoot(ORCHESTRATOR_PACKAGE), "dist/src/orchestrator/PreAggregationLoader.js", @@ -140,7 +181,7 @@ if ( const result = await patchInstalledCompiler(); console.log( result.changed - ? `Patched Cube ${SUPPORTED_VERSION} YAML metadata handling + ClickHouse string type + stream error forwarding` - : `Cube ${SUPPORTED_VERSION} YAML metadata + ClickHouse string type + stream error forwarding patches already applied`, + ? `Patched Cube ${SUPPORTED_VERSION} YAML metadata handling + ClickHouse string type + stream error forwarding + BigQuery read-only types` + : `Cube ${SUPPORTED_VERSION} YAML metadata + ClickHouse string type + stream error forwarding + BigQuery read-only type patches already applied`, ); } diff --git a/services/cubejs/src/__tests__/cube17Regression.test.js b/services/cubejs/src/__tests__/cube17Regression.test.js index f52632ce..a4ad5e82 100644 --- a/services/cubejs/src/__tests__/cube17Regression.test.js +++ b/services/cubejs/src/__tests__/cube17Regression.test.js @@ -9,6 +9,7 @@ import { prepareCompiler } from "@cubejs-backend/schema-compiler"; import { escapeCSVField } from "../utils/csvSerializer.js"; import { validateFormat } from "../utils/formatValidator.js"; import { + patchBigQueryDownloadSource, patchBigQueryDriverSource, patchClickHouseQuerySource, patchPreAggregationLoaderSource, @@ -190,6 +191,64 @@ after`; await assert.rejects(loaded, /Not found: Table/); }); + it("guards the BigQuery read-only download patch against upstream source drift", () => { + const source = `before + async loadTablesForDataset(dataset) { +after`; + const first = patchBigQueryDownloadSource(source); + assert.equal(first.changed, true); + assert.match(first.source, /async downloadQueryResults\(query, values, options\)/); + assert.deepEqual(patchBigQueryDownloadSource(first.source), { source: first.source, changed: false }); + assert.throws( + () => patchBigQueryDownloadSource("unexpected driver source"), + /expected source anchor was not found exactly once/, + ); + }); + + describe("BigQuery read-only download", () => { + const fields = [ + { name: "club", type: "STRING" }, + { name: "orders", type: "INTEGER" }, + { name: "amount", type: "NUMERIC" }, + { name: "fee", type: "FLOAT" }, + { name: "day", type: "TIMESTAMP" }, + ]; + const fakeBigQuery = (rows) => ({ + createQueryJob: async () => [{ + getMetadata: async () => [{ status: { state: "DONE" }, statistics: {} }], + getQueryResults: async (opts) => (opts?.maxResults === 0 && opts?.autoPaginate === false + ? [[], null, { schema: { fields } }] + : [rows]), + }], + dataset: () => ({ table: () => ({ getMetadata: async () => [{ schema: { fields } }] }) }), + }); + const download = async (rows) => { + const { BigQueryDriver } = require("@cubejs-backend/bigquery-driver"); + const driver = new BigQueryDriver({ projectId: "test", credentials: {} }); + driver.bigquery = fakeBigQuery(rows); + return { driver, result: await driver.downloadQueryResults("SELECT 1", [], {}) }; + }; + + it("types columns from BigQuery's result schema, exactly as the staging path does", async () => { + // whole numbers first: guessing from rows would type amount and fee as integers + const rows = [ + { club: "A", orders: 3, amount: "10", fee: 0, day: "2026-10-01" }, + { club: "B", orders: 4, amount: "12.75", fee: 1.5, day: "2026-10-02" }, + ]; + const { driver, result } = await download(rows); + + assert.deepEqual(result.types, await driver.tableColumnTypes("pre_aggregations_x.t")); + assert.equal(result.rows.length, 2); + }); + + it("types an empty partition instead of failing", async () => { + const { driver, result } = await download([]); + + assert.deepEqual(result.rows, []); + assert.deepEqual(result.types, await driver.tableColumnTypes("pre_aggregations_x.t")); + }); + }); + it("casts composite-key count measures to ClickHouse String (Tesseract)", async () => { const { ClickHouseQuery } = require( "@cubejs-backend/schema-compiler/dist/src/adapter/ClickHouseQuery.js", diff --git a/services/cubejs/src/utils/__tests__/prepareDbParams.test.js b/services/cubejs/src/utils/__tests__/prepareDbParams.test.js new file mode 100644 index 00000000..1a953a76 --- /dev/null +++ b/services/cubejs/src/utils/__tests__/prepareDbParams.test.js @@ -0,0 +1,19 @@ +import assert from "node:assert/strict"; +import { createRequire } from "node:module"; +import { describe, it } from "node:test"; + +import prepareDbParams from "../prepareDbParams.js"; + +const require = createRequire(import.meta.url); + +describe("prepareDbParams", () => { + it("makes BigQuery drivers read-only, so rollups never write to the customer's project", () => { + const { BigQueryDriver } = require("@cubejs-backend/bigquery-driver"); + const params = prepareDbParams( + { projectId: "abler-web", keyFile: JSON.stringify({ client_email: "sa@abler-web.iam" }) }, + "bigquery", + ); + + assert.equal(new BigQueryDriver(params).readOnly(), true); + }); +}); diff --git a/services/cubejs/src/utils/prepareDbParams.js b/services/cubejs/src/utils/prepareDbParams.js index d6e20d0a..49b428e6 100644 --- a/services/cubejs/src/utils/prepareDbParams.js +++ b/services/cubejs/src/utils/prepareDbParams.js @@ -58,6 +58,8 @@ const prepareDbParams = (dbParams, dbType) => { dbConfig = { ...dbConfig, credentials: { ...keyFile }, + // rollups load straight into Cube Store; no staging tables in the customer's project + readOnly: true, }; break; case "mssql":