diff --git a/docs-mintlify/docs.json b/docs-mintlify/docs.json index 6353ba0c4f484..3e44f5d77961e 100644 --- a/docs-mintlify/docs.json +++ b/docs-mintlify/docs.json @@ -437,6 +437,8 @@ "embedding/iframe/analytics-chat", "embedding/iframe/creator-mode", "embedding/iframe/customization", + "embedding/iframe/localization", + "embedding/iframe/events", { "group": "Authentication", "pages": [ diff --git a/docs-mintlify/embedding/iframe/customization.mdx b/docs-mintlify/embedding/iframe/customization.mdx index 91ef6550e3bee..3bcece8cf2cfd 100644 --- a/docs-mintlify/embedding/iframe/customization.mdx +++ b/docs-mintlify/embedding/iframe/customization.mdx @@ -12,6 +12,10 @@ You can customize the appearance of iframe-embedded Cube content in two ways: Customization options will expand over time. This page documents what is supported today. + + To set the language of embedded surfaces — account-wide, per embed, or at runtime — see [Localization](/embedding/iframe/localization). + + To brand the whole embedded experience — accent color, background, text, logo, and fonts — at the account level, configure the [app theme](/admin/customization/app-theme). It applies to all embedded surfaces automatically (and, optionally, to the entire Cube Cloud console). diff --git a/docs-mintlify/embedding/iframe/dashboards.mdx b/docs-mintlify/embedding/iframe/dashboards.mdx index 46a2abb0ce2fd..e8430459fda47 100644 --- a/docs-mintlify/embedding/iframe/dashboards.mdx +++ b/docs-mintlify/embedding/iframe/dashboards.mdx @@ -83,6 +83,18 @@ CSV is generated client-side from the data already loaded into the widget, so no additional query is issued. The parameter is opt-in — omit it (the default) to keep the download action hidden. +## Set the language + +Embedded dashboards render their UI in the account's default language, which you can +override per embed by adding the `?locale=` query parameter: + +```text +https://your-tenant.cubecloud.dev/embed/dashboard/YOUR_DASHBOARD_PUBLIC_ID?session=YOUR_SESSION_ID&locale=es-MX +``` + +See [Localization](/embedding/iframe/localization) for the list of supported languages +and the other ways to set the language. + ## Customize appearance You can style an embedded dashboard — background, padding, widget borders, titles, and fonts — from the **Styling** panel in the Dashboard Builder. See [Dashboards → Styling](/docs/explore-analyze/dashboards/styling) for the full list of options. diff --git a/docs-mintlify/embedding/iframe/events.mdx b/docs-mintlify/embedding/iframe/events.mdx new file mode 100644 index 0000000000000..1aa5aa0fc3b04 --- /dev/null +++ b/docs-mintlify/embedding/iframe/events.mdx @@ -0,0 +1,468 @@ +--- +title: Events & actions +description: Listen to user-interaction events from an embedded Cube iframe and send actions back into it, over the browser postMessage API. +--- + + + +Available on [Premium and above plans](https://cube.dev/pricing). + + + +Embedded Cube surfaces communicate with your host page over the browser +[`postMessage`](https://developer.mozilla.org/docs/Web/API/Window/postMessage) +API, in both directions: + +- **Events** (`cube:event:*`) travel **embed → host**. Subscribe to them to learn + how a viewer interacts with the embed — when it loads, what they view, + download, drill into, search for with AI, and any errors they hit. Feed them + into your own analytics / event bus. +- **Actions** (`cube:action:*`) travel **host → embed**. Send them to drive the + embed from your app — switch the color scheme, set a filter, navigate, or + refresh the data. + +No SDK is required — it's plain `window.postMessage` and a `message` listener. + +## Message envelope + +Every message — in either direction — is a single object with the same shape: + +```ts +{ + source: "cube-embed", // discriminator — always this string + direction: "event" | "action", // "event" = embed→host, "action" = host→embed + type: string, // e.g. "cube:event:download" / "cube:action:set-filter" + payload: object, // shape depends on `type` (see catalogs below) + timestamp: number, // epoch milliseconds + surface?: "dashboard" | "app" | "chat" // always on events; never on actions +} +``` + +| Field | Description | +| --- | --- | +| `source` | Always `"cube-embed"`. Check this first to tell Cube messages apart from other `postMessage` traffic on the page (browser extensions, other libraries, your own app). | +| `direction` | `"event"` for messages emitted by the embed, `"action"` for messages you send into it. | +| `type` | The event or action name (see the catalogs below). | +| `payload` | Event/action-specific data. | +| `timestamp` | When the message was created, in epoch milliseconds. | +| `surface` | Which embedded surface the message relates to. **Always present on events**; never sent on actions — so event listeners never need to handle a missing `surface`. | + +## Listening to events + +Attach a single `message` listener to `window`. Always validate `event.origin` +against your tenant's origin and check `data.source === "cube-embed"` before +trusting a message. + +```js +const CUBE_ORIGIN = "https://your-tenant.cubecloud.dev"; + +window.addEventListener("message", (event) => { + // 1. Only trust messages from your Cube tenant. + if (event.origin !== CUBE_ORIGIN) return; + + const data = event.data; + + // 2. Only handle Cube embed events. + if (!data || data.source !== "cube-embed" || data.direction !== "event") return; + + // 3. Dispatch on the event type. + switch (data.type) { + case "cube:event:ready": + console.log("Embed ready", data.payload.embedTenant, data.payload.deploymentId); + break; + case "cube:event:download": + myAnalytics.track("embed_download", data.payload); + break; + case "cube:event:ai-query": + myAnalytics.track("embed_ai_query", { query: data.payload.query }); + break; + case "cube:event:error": + console.error("Embed error", data.payload.message); + break; + default: + // ready, view, navigate, dashboard-loaded, drilldown, … + myAnalytics.track(data.type, { surface: data.surface, ...data.payload }); + } +}); +``` + +### Event catalog + +| Event | Fires when | Surfaces | +| --- | --- | --- | +| [`cube:event:ready`](#cube-event-ready) | The embed has authenticated and mounted (the handshake) | all | +| [`cube:event:view`](#cube-event-view) | A surface is viewed, on load and on each in-embed navigation | all | +| [`cube:event:navigate`](#cube-event-navigate) | The viewer navigates within the embed | all | +| [`cube:event:dashboard-loaded`](#cube-event-dashboard-loaded) | All widgets on a dashboard have rendered | dashboard | +| [`cube:event:download`](#cube-event-download) | The viewer exports data or an image | dashboard | +| [`cube:event:drilldown`](#cube-event-drilldown) | The viewer drills into a measure | dashboard | +| [`cube:event:ai-query`](#cube-event-ai-query) | The viewer runs an AI / natural-language query | all | +| [`cube:event:error`](#cube-event-error) | The embed surfaces an error | all | + + + Every event payload is also delivered with the envelope's `surface` field, so + you can always tell which surface (`dashboard`, `app`, or `chat`) it came from + — including AI queries, which report `app` when run inside the embedded app and + `chat` on the standalone chat surface. + + +#### `cube:event:ready` + +Emitted once per session, as soon as the embed authenticates and mounts. The +handshake — the first event you receive, and the moment to record an +"embed opened". + +| Field | Type | Description | +| --- | --- | --- | +| `embedTenant` | `string \| null` | The embed tenant the iframe resolved to, when known. | +| `deploymentId` | `number \| null` | The deployment the embed is bound to, when known. | +| `mode` | `"signed" \| "private"` | How the viewer was authenticated. | +| `surface` | `"dashboard" \| "app" \| "chat"` | The surface that mounted. | +| `publicId` | `string` _(optional)_ | The dashboard's public id, for the dashboard surface. | + +```json +{ + "embedTenant": "acme", + "deploymentId": 42, + "mode": "signed", + "surface": "dashboard", + "publicId": "a1b2c3d4" +} +``` + +#### `cube:event:view` + +Emitted when a surface is viewed — on the initial load and again whenever the +viewer navigates within the embed. + +| Field | Type | Description | +| --- | --- | --- | +| `surface` | `"dashboard" \| "app" \| "chat"` | The surface viewed. | +| `path` | `string` | The in-embed route path that was viewed. | +| `publicId` | `string` _(optional)_ | The dashboard's public id, when applicable. | +| `title` | `string` _(optional)_ | Human-readable title of the surface, when available. | + +```json +{ + "surface": "app", + "path": "/embed/d/42/app/workbook/130" +} +``` + +#### `cube:event:navigate` + +Emitted when the viewer navigates within the embed (a route change). Use it to +mirror the embed's location in your own router or analytics. + +| Field | Type | Description | +| --- | --- | --- | +| `path` | `string` | The new path. | +| `previousPath` | `string` _(optional)_ | The path navigated away from. | + +```json +{ + "path": "/embed/d/42/app/workbook/130", + "previousPath": "/embed/d/42/app" +} +``` + +#### `cube:event:dashboard-loaded` + +Emitted when a dashboard has finished rendering all of its widgets — the "fully +painted" signal (distinct from `ready`, which fires at mount, before data loads). + +| Field | Type | Description | +| --- | --- | --- | +| `publicId` | `string` _(optional)_ | The dashboard's public id. | +| `widgetCount` | `number` _(optional)_ | Number of widgets on the dashboard. | +| `loadDurationMs` | `number` _(optional)_ | Milliseconds from mount to all widgets loaded, when measurable. | + +```json +{ + "publicId": "a1b2c3d4", + "widgetCount": 6 +} +``` + +#### `cube:event:download` + +Emitted when a viewer exports something — a widget's data as CSV, or (in future) +a dashboard image. Reports _that_ an export happened and its shape — never the +exported rows themselves. + +| Field | Type | Description | +| --- | --- | --- | +| `format` | `"csv" \| "xlsx" \| "png" \| "pdf"` | The file format produced. | +| `target` | `"widget" \| "dashboard"` | Whether a single widget or the whole dashboard was exported. | +| `widgetId` | `string` _(optional)_ | Id of the source widget, when `target` is `"widget"`. | +| `title` | `string` _(optional)_ | Title of the exported widget / dashboard. | +| `rowCount` | `number` _(optional)_ | Rows exported, for data exports (`csv` / `xlsx`). | + +```json +{ + "format": "csv", + "target": "widget", + "widgetId": "37", + "title": "Revenue by month", + "rowCount": 128 +} +``` + + + The CSV download action on a dashboard widget only appears when the embed URL + includes `allowExport=true` (see [Dashboards → Allow CSV + export](/embedding/iframe/dashboards#allow-csv-export)). The event fires when a + viewer uses it. + + +#### `cube:event:drilldown` + +Emitted when a viewer drills into a measure (clicks a chart mark or table cell to +see its detail rows). + +| Field | Type | Description | +| --- | --- | --- | +| `member` | `string` | The fully-qualified measure that was drilled into. | +| `value` | `unknown` _(optional)_ | The clicked value, when the click carried one. | +| `widgetId` | `string` _(optional)_ | Id of the originating widget. | + +```json +{ + "member": "orders.count", + "value": "completed", + "widgetId": "37" +} +``` + +#### `cube:event:ai-query` + +Emitted around an AI / natural-language query — capturing _what_ the viewer asked +and the lifecycle stage. Fires wherever AI chat is used: the standalone chat +surface, the dashboard agent, and the embedded app. + +| Field | Type | Description | +| --- | --- | --- | +| `query` | `string` | The natural-language query the viewer submitted. | +| `status` | `"submitted" \| "completed" \| "error"` | Lifecycle stage of the query. | +| `chatId` | `string` _(optional)_ | The chat/session id, when applicable. | +| `agentId` | `string` _(optional)_ | The agent that answered, when applicable. | + +```json +{ + "query": "top 10 customers by revenue this quarter", + "status": "submitted", + "agentId": "1" +} +``` + +#### `cube:event:error` + +Emitted when the embed surfaces an error (a render error, a query failure, an +auth/session problem). `fatal` distinguishes an error that took the whole surface +down from a recoverable one. + +| Field | Type | Description | +| --- | --- | --- | +| `message` | `string` | Human-readable message. | +| `name` | `string` _(optional)_ | Error name/class, e.g. `"TypeError"`. | +| `context` | `string` _(optional)_ | Where it originated, e.g. `"embed-render"`. | +| `fatal` | `boolean` _(optional)_ | `true` when the error took down the whole surface. | + +```json +{ + "message": "Failed to load data", + "context": "embed-render", + "fatal": true +} +``` + +## Sending actions + +Send actions into the embed by posting a message to the iframe's +`contentWindow`. Always target your tenant's origin (not `"*"`) so the message +can't leak to another document if the iframe navigates away. + +```js +const iframe = document.querySelector("iframe#cube"); +const CUBE_ORIGIN = "https://your-tenant.cubecloud.dev"; + +function sendAction(type, payload = {}) { + iframe.contentWindow.postMessage( + { + source: "cube-embed", + direction: "action", + type, + payload, + timestamp: Date.now(), + }, + CUBE_ORIGIN + ); +} + +// Examples +sendAction("cube:action:set-color-scheme", { scheme: "dark" }); +sendAction("cube:action:set-filter", { + filterUrlParameter: 'f_orders.status={"value":"completed"}', +}); +sendAction("cube:action:refresh"); +``` + +### Action catalog + +| Action | Effect | Payload | +| --- | --- | --- | +| [`cube:action:set-color-scheme`](#cube-action-set-color-scheme) | Switch light / dark / auto | `{ scheme }` | +| [`cube:action:set-theme`](#cube-action-set-theme) | Apply a brand theme (colors, fonts) | `embedTheme` object | +| [`cube:action:set-locale`](#cube-action-set-locale) | Switch the UI language | `{ locale }` | +| [`cube:action:set-filter`](#cube-action-set-filter) | Push a filter into a dashboard | `{ filterUrlParameter }` | +| [`cube:action:navigate`](#cube-action-navigate) | Navigate the embed to a path | `{ path }` | +| [`cube:action:refresh`](#cube-action-refresh) | Re-run the embed's queries | _none_ | + +#### `cube:action:set-color-scheme` + +Switch the embed's color scheme at runtime. + +| Field | Type | Description | +| --- | --- | --- | +| `scheme` | `"light" \| "dark" \| "auto"` | `"auto"` follows the viewer's OS preference. | + +```js +sendAction("cube:action:set-color-scheme", { scheme: "dark" }); +``` + +#### `cube:action:set-theme` + +Apply a brand theme (colors, fonts) to the embed at runtime. The payload is an +`embedTheme` object — the same shape the [Generate Session API](/reference/embed-apis/generate-session) +accepts. Common fields are `primaryColor`, `borderRadius`, and `font`; see +[App customization](/embedding/iframe/customization#app-customization) for the +full list. + +```js +sendAction("cube:action:set-theme", { + primaryColor: "#7c5cff", + borderRadius: 8, +}); +``` + +#### `cube:action:set-locale` + +Switch the embed's UI language. Accepts a full code (`es-ES`), a short code +(`es`), or a regional variant. See [Localization](/embedding/iframe/localization) +for the list of supported languages and the other ways to set the language. + +| Field | Type | Description | +| --- | --- | --- | +| `locale` | `string` | The locale to switch to. | + +```js +sendAction("cube:action:set-locale", { locale: "es" }); +``` + +#### `cube:action:set-filter` + +Push a filter into a dashboard. The `filterUrlParameter` is the same +`f_.=` form used to +[pre-set filters via URL](/embedding/iframe/dashboards#pre-set-dashboard-filters-via-url), +so you can capture a viewer's filters and restore them later. + +| Field | Type | Description | +| --- | --- | --- | +| `filterUrlParameter` | `string` | Filter(s) in URL-query form, e.g. `f_orders.status={"value":"completed"}`. | + +```js +sendAction("cube:action:set-filter", { + filterUrlParameter: 'f_orders.status={"value":"completed"}', +}); +``` + +#### `cube:action:navigate` + +Navigate the embed to an in-embed path. + +| Field | Type | Description | +| --- | --- | --- | +| `path` | `string` | The in-embed path to navigate to. | + +```js +sendAction("cube:action:navigate", { path: "/embed/d/42/app/workbook/130" }); +``` + +#### `cube:action:refresh` + +Re-run the embed's queries and refresh its data. No payload. + +```js +sendAction("cube:action:refresh"); +``` + +## Surfaces + +Events come from one of three customer-facing surfaces, reported in the envelope's +`surface` field: + +- `dashboard` — a [published dashboard](/embedding/iframe/dashboards). +- `app` — the [Creator-mode app](/embedding/iframe/creator-mode) (workbooks, + folders, in-app dashboards). AI chat run _inside_ the app reports `app`. +- `chat` — the standalone [analytics chat](/embedding/iframe/analytics-chat). + +## Security + +- **Always validate `event.origin`** against your tenant's origin in your + `message` listener, and check `data.source === "cube-embed"`. Never act on a + message that fails either check. +- **Target your tenant's origin when sending actions** (`iframe.contentWindow.postMessage(msg, CUBE_ORIGIN)`), + not `"*"`, so an action can't be delivered to an unexpected document. + +## Complete example + +A minimal host page that loads a signed dashboard embed, logs every event, and +exposes buttons to drive it. Generate the `session` on your backend with the +[Generate Session API](/reference/embed-apis/generate-session) — see +[Signed embedding](/embedding/iframe/auth/signed) for the full flow. + +```html + + + + + + + + + + + +``` diff --git a/docs-mintlify/embedding/iframe/localization.mdx b/docs-mintlify/embedding/iframe/localization.mdx new file mode 100644 index 0000000000000..d1d86fe2f5d6d --- /dev/null +++ b/docs-mintlify/embedding/iframe/localization.mdx @@ -0,0 +1,90 @@ +--- +title: Localization +description: Set the language for embedded Cube surfaces — account-wide, per embed via URL, or at runtime. +--- + +Embedded Cube surfaces — dashboards, [Analytics Chat](/embedding/iframe/analytics-chat), +and the full app in [Creator Mode](/embedding/iframe/creator-mode) — can render their +UI in any of the supported languages. You can set a default language for the whole +account, override it per embed with a URL parameter, or switch it at runtime from the +host page. + + + Available on [Premium and Enterprise plans](https://cube.dev/pricing). + + +## Supported languages + +| Language | Locale code | +| -------------------------- | ----------- | +| English (US) | `en-US` | +| Deutsch (Deutschland) | `de-DE` | +| Español (España) | `es-ES` | +| Español (Latinoamérica) | `es-MX` | +| Français (France) | `fr-FR` | +| Italiano (Italia) | `it-IT` | +| 日本語 (日本) | `ja-JP` | +| Norsk bokmål (Norge) | `nb-NO` | +| Português (Brasil) | `pt-BR` | +| Português (Portugal) | `pt-PT` | +| Svenska (Sverige) | `sv-SE` | +| Tiếng Việt (Việt Nam) | `vi-VN` | + +English (`en-US`) is the default and the fallback when no language is configured. + +A locale value can be a full code (`es-ES`), a short language code (`es`), or a regional +variant that isn't shipped (`es-AR`). Short codes and unsupported regional variants +resolve to the first supported locale for that language — for example, both `es` and +`es-AR` resolve to `es-ES`. A value that doesn't match any supported language is ignored. + +## How the language is resolved + +The language of an embedded surface is resolved from the following sources, highest +priority first: + +1. **Runtime override** — a [`cube:action:set-locale`](/embedding/iframe/events#cube-action-set-locale) + message sent from the host page (see [At runtime](#at-runtime)). +2. **URL parameter** — the `?locale=` query parameter on the embed URL (see + [Per embed via URL](#per-embed-via-url)). +3. **Account default** — the language configured in **Embed → Settings** (see + [Account-wide default](#account-wide-default)). +4. **Fallback** — `en-US`. + +## Account-wide default + +Set a default language for all embedded surfaces from the Cube Cloud console: + +1. Go to **Embed → Settings**. +2. In the **Language** card, pick a language from the dropdown. + +The selected language applies to every embedded surface across the account, unless a +specific embed overrides it with a `?locale=` URL parameter or a runtime +`cube:action:set-locale` message. + +Clearing the setting removes the stored default, and embeds fall back to `en-US` (or to +whatever a per-embed override specifies). + +## Per embed via URL + +Override the account default for an individual embed by adding the `?locale=` query +parameter to the embed URL: + +```text +https://your-tenant.cubecloud.dev/embed/dashboard/YOUR_DASHBOARD_PUBLIC_ID?session=YOUR_SESSION_ID&locale=es-MX +``` + +The parameter is read once when the embed loads and pinned for the session, so in-app +navigation won't drop it. + +## At runtime + +Switch the language after the embed has loaded by sending a +[`cube:action:set-locale`](/embedding/iframe/events#cube-action-set-locale) message from +the host page. This takes precedence over both the URL parameter and the account default: + +```js +sendAction("cube:action:set-locale", { locale: "es" }); +``` + +See [Events and actions](/embedding/iframe/events) for the full host ↔ embed messaging +contract. diff --git a/docs/content/product/configuration/reference/environment-variables.mdx b/docs/content/product/configuration/reference/environment-variables.mdx index cab568d98042d..17e645b46ad98 100644 --- a/docs/content/product/configuration/reference/environment-variables.mdx +++ b/docs/content/product/configuration/reference/environment-variables.mdx @@ -1463,6 +1463,15 @@ Define which metrics collector format. | --------------------- | ---------------------- | --------------------- | | `statsd`, `dogstatsd` | `statsd` | `statsd` | +## `CUBESTORE_METRICS_BIND_ADDRESS` + +Define the IP address to bind the metrics collector server to. Binding to +the default loopback interface will prevent sending of metrics to other devices. + +| Possible Values | Default in Development | Default in Production | +| ------------------ | ---------------------- | --------------------- | +| A valid IP address | `127.0.0.1` | `127.0.0.1` | + ## `CUBESTORE_METRICS_ADDRESS` Required IP address to send metrics. diff --git a/packages/cubejs-backend-native/test/bridge/object-bridges-coverage.test.ts b/packages/cubejs-backend-native/test/bridge/object-bridges-coverage.test.ts index a49039c536502..9a3717ad3a8b1 100644 --- a/packages/cubejs-backend-native/test/bridge/object-bridges-coverage.test.ts +++ b/packages/cubejs-backend-native/test/bridge/object-bridges-coverage.test.ts @@ -54,6 +54,7 @@ const BRIDGES: BridgeSpec[] = [ 'order', 'pre_aggregation_id', 'pre_aggregation_query', + 'pre_aggregations_match_only', 'row_limit', 'security_context', 'segments', diff --git a/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js b/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js index 077744e7898fe..fed818e029d57 100644 --- a/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js +++ b/packages/cubejs-schema-compiler/src/adapter/BaseQuery.js @@ -1009,6 +1009,11 @@ export class BaseQuery { ungrouped: this.options.ungrouped, exportAnnotatedSql: false, preAggregationQuery: this.options.preAggregationQuery, + // We only consume the pre-aggregation match result here (sql/params are discarded + // below), so tell the native planner to skip building the outer query SQL. Otherwise + // a rolling-window measure without a date range would throw while rendering its time + // series, even though matching itself doesn't need it. + preAggregationsMatchOnly: true, preAggregationId: this.options.preAggregationId || null, securityContext: this.contextSymbols.securityContext, cubestoreSupportMultistage: this.options.cubestoreSupportMultistage ?? getEnv('cubeStoreRollingWindowJoin'), diff --git a/packages/cubejs-schema-compiler/test/unit/pre-aggregations.test.ts b/packages/cubejs-schema-compiler/test/unit/pre-aggregations.test.ts index 9e39b56c84f6b..eb60b07785c2b 100644 --- a/packages/cubejs-schema-compiler/test/unit/pre-aggregations.test.ts +++ b/packages/cubejs-schema-compiler/test/unit/pre-aggregations.test.ts @@ -3,6 +3,7 @@ import path from 'path'; import { prepareJsCompiler, prepareYamlCompiler } from './PrepareCompiler'; import { createECommerceSchema, createSchemaYaml } from './utils'; import { PostgresQuery, queryClass, QueryFactory } from '../../src'; +import { RedshiftQuery } from '../../src/adapter/RedshiftQuery'; describe('pre-aggregations', () => { it('rollupJoin scheduledRefresh', async () => { @@ -979,4 +980,82 @@ describe('pre-aggregations', () => { expect(originalSqlDesc.loadSql[0]).toMatch(/SELECT \* FROM public\.orders/); }); }); + + // Regression for the pre-aggregation refresh/metadata path: it builds a query from a + // rolling pre-agg's references (rolling measure + a time dimension with granularity, but + // NO date range) just to find the matching pre-aggregation. Matching must succeed without + // building the outer query's rolling-window time series — which can't render without a date + // range on dialects that lack generated time series (e.g. Redshift). Pure matching, no DB. + describe('rolling pre-aggregation matching without a date range', () => { + const { compiler, joinGraph, cubeEvaluator } = prepareJsCompiler(` + cube(\`visitors\`, { + sql: \`SELECT * FROM visitors\`, + + measures: { + count: { + type: \`count\`, + }, + rollingCount: { + type: \`count\`, + rollingWindow: { + trailing: \`unbounded\`, + }, + }, + }, + + dimensions: { + id: { + sql: \`id\`, + type: \`number\`, + primaryKey: true, + }, + source: { + sql: \`source\`, + type: \`string\`, + }, + createdAt: { + sql: \`created_at\`, + type: \`time\`, + }, + }, + + preAggregations: { + partitionedRolling: { + type: \`rollup\`, + measures: [CUBE.rollingCount], + dimensions: [CUBE.source], + timeDimension: CUBE.createdAt, + granularity: \`hour\`, + partitionGranularity: \`month\`, + }, + }, + }); + `); + + beforeAll(async () => { + await compiler.compile(); + }); + + [PostgresQuery, RedshiftQuery].forEach((QueryClass) => { + it(`matches the rolling pre-aggregation (${QueryClass.name})`, async () => { + const query = new QueryClass({ joinGraph, cubeEvaluator, compiler }, { + measures: ['visitors.rollingCount'], + dimensions: ['visitors.source'], + timeDimensions: [{ + dimension: 'visitors.createdAt', + granularity: 'day', + // no dateRange — the shape the refresh/metadata path builds + }], + timezone: 'UTC', + preAggregationsSchema: '', + }); + + // Must not throw while determining the matching pre-aggregation. + const preAggregationsDescription: any = query.preAggregations?.preAggregationsDescription(); + const ids = (preAggregationsDescription || []).map((d: any) => d.preAggregationId); + expect(ids).toContain('visitors.partitionedRolling'); + expect(query.preAggregations?.preAggregationForQuery?.canUsePreAggregation).toEqual(true); + }); + }); + }); }); diff --git a/packages/cubejs-server-core/src/core/CompilerApi.ts b/packages/cubejs-server-core/src/core/CompilerApi.ts index 8ce288843a5e0..e1f62fea4dfed 100644 --- a/packages/cubejs-server-core/src/core/CompilerApi.ts +++ b/packages/cubejs-server-core/src/core/CompilerApi.ts @@ -58,6 +58,11 @@ export interface GetSqlOptions { includeDebugInfo?: boolean; exportAnnotatedSql?: boolean; requestId?: string; + // Build only the pre-aggregation descriptions, skipping the outer query SQL. + // Used by the refresh/metadata path, which consumes `preAggregations` but not `sql`. + // Avoids building a rolling-window time series that would require a date range the + // refresh path doesn't provide. + preAggregationsOnly?: boolean; } export interface SqlResult { @@ -347,13 +352,13 @@ export class CompilerApi { } public async getSql(query: NormalizedQuery, options: GetSqlOptions = {}): Promise { - const { includeDebugInfo, exportAnnotatedSql } = options; + const { includeDebugInfo, exportAnnotatedSql, preAggregationsOnly } = options; const { sqlGenerator, compilers } = await this.getSqlGenerator(query); const getSqlFn = () => compilers.compiler.withQuery(sqlGenerator, () => ({ external: sqlGenerator.externalPreAggregationQuery(), - sql: sqlGenerator.buildSqlAndParams(exportAnnotatedSql), - lambdaQueries: sqlGenerator.buildLambdaQuery(), + sql: preAggregationsOnly ? null : sqlGenerator.buildSqlAndParams(exportAnnotatedSql), + lambdaQueries: preAggregationsOnly ? [] : sqlGenerator.buildLambdaQuery(), timeDimensionAlias: sqlGenerator.timeDimensions[0]?.unescapedAliasName(), timeDimensionField: sqlGenerator.timeDimensions[0]?.dimension, order: sqlGenerator.order, diff --git a/packages/cubejs-server-core/src/core/RefreshScheduler.ts b/packages/cubejs-server-core/src/core/RefreshScheduler.ts index e94001b9a6bb4..a334a673ec183 100644 --- a/packages/cubejs-server-core/src/core/RefreshScheduler.ts +++ b/packages/cubejs-server-core/src/core/RefreshScheduler.ts @@ -146,7 +146,7 @@ export class RefreshScheduler { queryingOptions: ScheduledRefreshQueryingOptions ): Promise { const baseQuery = await this.baseQueryForPreAggregation(compilerApi, preAggregation, queryingOptions); - const baseQuerySql = await compilerApi.getSql(baseQuery); + const baseQuerySql = await compilerApi.getSql(baseQuery, { preAggregationsOnly: true }); const preAggregationDescriptionList = baseQuerySql.preAggregations; const preAggregationDescription = preAggregationDescriptionList.find(p => p.preAggregationId === preAggregation.id); const orchestratorApi = await this.serverCore.getOrchestratorApi(context); diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_query_options.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_query_options.rs index 5208860ba436d..30f34f1bf751b 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_query_options.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/cube_bridge/base_query_options.rs @@ -206,6 +206,8 @@ pub struct BaseQueryOptionsStatic { pub export_annotated_sql: bool, #[serde(rename = "preAggregationQuery")] pub pre_aggregation_query: Option, + #[serde(rename = "preAggregationsMatchOnly")] + pub pre_aggregations_match_only: Option, #[serde(rename = "useOriginalSqlPreAggregationsInPreAggregation")] pub use_original_sql_pre_aggregations_in_pre_aggregation: Option, #[serde(rename = "totalQuery")] diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/compiled_pre_aggregation.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/compiled_pre_aggregation.rs index db0d707d468a4..60dd590dd8382 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/compiled_pre_aggregation.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/compiled_pre_aggregation.rs @@ -20,7 +20,21 @@ pub struct PreAggregationJoin { #[derive(Clone, Debug)] pub struct PreAggregationUnion { - pub items: Vec>, + pub items: Vec, +} + +/// A single member rollup of a `rollupLambda` union, paired with the +/// member symbols of *that* rollup. The lambda exposes the first member +/// rollup's symbols, but each branch stores its columns under its own +/// cube aliases (e.g. `requests_stream__tenant_id` vs `requests__tenant_id`), +/// so the physical builder needs each branch's own symbols to read the +/// right column while projecting the lambda's unified alias. +#[derive(Clone, Debug)] +pub struct PreAggregationUnionItem { + pub table: Rc, + pub measures: Vec>, + pub dimensions: Vec>, + pub time_dimensions: Vec>, } #[derive(Clone, Debug)] diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/optimizer.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/optimizer.rs index 239de72bcb5c6..0d6ee57f17ef6 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/optimizer.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/optimizer.rs @@ -414,11 +414,14 @@ impl PreAggregationOptimizer { let items = union .items .iter() - .map(|t| { - Rc::new(PreAggregationTable { + .map(|item| PreAggregationUnionItem { + table: Rc::new(PreAggregationTable { usage_index: Some(usage_index), - ..t.as_ref().clone() - }) + ..item.table.as_ref().clone() + }), + measures: item.measures.clone(), + dimensions: item.dimensions.clone(), + time_dimensions: item.time_dimensions.clone(), }) .collect(); Rc::new(PreAggregationSource::Union(PreAggregationUnion { items })) diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/pre_aggregations_compiler.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/pre_aggregations_compiler.rs index 413fe93149967..ed31b21fc2b07 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/pre_aggregations_compiler.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/optimizers/pre_aggregation/pre_aggregations_compiler.rs @@ -7,6 +7,7 @@ use crate::logical_plan::PreAggregationJoin; use crate::logical_plan::PreAggregationJoinItem; use crate::logical_plan::PreAggregationTable; use crate::logical_plan::PreAggregationUnion; +use crate::logical_plan::PreAggregationUnionItem; use crate::planner::join_hints::JoinHints; use crate::planner::multi_fact_join_groups::{MeasuresJoinHints, MultiFactJoinGroups}; use crate::planner::planners::JoinPlanner; @@ -302,13 +303,22 @@ impl PreAggregationsCompiler { for (i, rollup) in pre_aggrs_for_lambda.clone().iter().enumerate() { match rollup.source.as_ref() { PreAggregationSource::Single(table) => { - sources.push(Rc::new(table.clone())); + // Carry this branch's own symbols: each member rollup stores its + // columns under its own cube aliases, while the lambda exposes the + // first rollup's symbols. The physical builder maps lambda members + // to each branch's stored column through these. + sources.push(PreAggregationUnionItem { + table: Rc::new(table.clone()), + measures: rollup.measures.clone(), + dimensions: rollup.dimensions.clone(), + time_dimensions: rollup.time_dimensions.clone(), + }); } _ => { return Err(CubeError::user(format!("Rollup lambda can't be nested"))); } } - if i > 1 { + if i >= 1 { Self::match_symbols(&rollup.measures, &pre_aggrs_for_lambda[0].measures)?; Self::match_symbols(&rollup.dimensions, &pre_aggrs_for_lambda[0].dimensions)?; Self::match_time_dimensions( diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/pre_aggregation.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/pre_aggregation.rs index e3f0a0d4893dc..386812d766b2c 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/pre_aggregation.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/logical_plan/pre_aggregation.rs @@ -178,7 +178,10 @@ impl PrettyPrint for PreAggregation { result.println("Union:", &state); let state = state.new_level(); for item in union.items.iter() { - result.println(&format!("-{}.{}", item.cube_name, item.name), &state); + result.println( + &format!("-{}.{}", item.table.cube_name, item.table.name), + &state, + ); } } } diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/sql_nodes/factory.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/sql_nodes/factory.rs index 6743740d88177..302aec8a5efc3 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/sql_nodes/factory.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan/sql_nodes/factory.rs @@ -255,6 +255,21 @@ impl SqlNodesFactory { ); RenderReferencesSqlNode::new(root_node, self.render_references.clone()) } + + /// When an ungrouped query reads from a pre-aggregation, a measure must + /// resolve to its stored rollup column. Unlike the grouped case, the column + /// already holds the aggregated value and is returned as-is, so it's a plain + /// column reference with no `sum()` wrap. Wrapping the ungrouped node keeps + /// the reference outermost so the measure is intercepted before it would + /// otherwise re-render its base-table SQL. + fn wrap_ungrouped_pre_aggregation_measure(&self, node: Rc) -> Rc { + if !self.pre_aggregation_measures_references.is_empty() { + RenderReferencesSqlNode::new(node, self.pre_aggregation_measures_references.clone()) + } else { + node + } + } + fn add_ungrouped_measure_reference_if_needed( &self, default: Rc, @@ -288,9 +303,11 @@ impl SqlNodesFactory { fn final_measure_node_processor(&self, input: Rc) -> Rc { if self.ungrouped_measure { - UngroupedMeasureSqlNode::new(input) + self.wrap_ungrouped_pre_aggregation_measure(UngroupedMeasureSqlNode::new(input)) } else if self.ungrouped { - UngroupedQueryFinalMeasureSqlNode::new(input) + self.wrap_ungrouped_pre_aggregation_measure(UngroupedQueryFinalMeasureSqlNode::new( + input, + )) } else { let final_processor: Rc = FinalMeasureSqlNode::new(input.clone(), self.count_approx_as_state); diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/pre_aggregation.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/pre_aggregation.rs index 7826a3169bea9..6effa2e9ac0e0 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/pre_aggregation.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/pre_aggregation.rs @@ -11,6 +11,7 @@ use crate::physical_plan::{ }; use crate::physical_plan_builder::PhysicalPlanBuilder; use crate::planner::sql_templates::PlanSqlTemplates; +use crate::planner::MemberSymbol; use crate::planner::SqlJoinCondition; use cubenativeutils::CubeError; use std::rc::Rc; @@ -73,68 +74,110 @@ impl PreAggregationProcessor<'_> { Ok(from) } + /// Resolve the column a union branch stores a lambda member under. The + /// lambda exposes the first member rollup's symbols, but each branch keeps + /// its own cube aliases (e.g. `requests_stream__tenant_id` vs the lambda's + /// `requests__tenant_id`). + fn find_branch_member( + lambda_member: &Rc, + branch_members: &[Rc], + branch_cube_name: &str, + ) -> Result, CubeError> { + if let Some(member) = branch_members + .iter() + .find(|m| m.full_name() == lambda_member.full_name()) + { + return Ok(member.clone()); + } + let short_name = lambda_member.name(); + if let Some(member) = branch_members + .iter() + .find(|m| m.name() == short_name && m.cube_name() == branch_cube_name) + { + return Ok(member.clone()); + } + Err(CubeError::internal(format!( + "Lambda pre-aggregation member '{}' has no match in union branch '{}'", + lambda_member.full_name(), + branch_cube_name + ))) + } + fn make_pre_aggregation_union_source( &self, pre_aggregation: &PreAggregation, union: &PreAggregationUnion, ) -> Result, CubeError> { if union.items.len() == 1 { - let table_source = self.make_pre_aggregation_table_source(&union.items[0])?; + let table_source = self.make_pre_aggregation_table_source(&union.items[0].table)?; return Ok(From::new(FromSource::Single(table_source))); } let query_tools = self.builder.query_tools(); let mut union_sources = Vec::new(); for item in union.items.iter() { - let table_source = self.make_pre_aggregation_table_source(&item)?; + let branch_cube_name = &item.table.cube_name; + let table_source = self.make_pre_aggregation_table_source(&item.table)?; let from = From::new(FromSource::Single(table_source)); let mut select_builder = SelectBuilder::new(from); for dim in pre_aggregation.dimensions().iter() { - let name_in_table = - PlanSqlTemplates::member_alias_name(&item.cube_alias, &dim.name(), &None); - let alias = dim.alias(); + // Read this branch's stored column for the lambda dimension and + // project it under the lambda's unified alias. + let branch_dim = Self::find_branch_member(dim, &item.dimensions, branch_cube_name)?; select_builder.add_projection_reference_member( &dim, - QualifiedColumnName::new(None, name_in_table), - Some(alias), + QualifiedColumnName::new(None, branch_dim.alias()), + Some(dim.alias()), ); } + // Match time dimensions on their base member so the granularity + // suffix is applied consistently on both the read and output sides. + let branch_time_bases = item + .time_dimensions + .iter() + .map(|td| { + if let Ok(t) = td.as_time_dimension() { + t.base_symbol().clone() + } else { + td.clone() + } + }) + .collect::>(); for dim in pre_aggregation.time_dimensions().iter() { - let (alias, granularity) = if let Ok(td) = dim.as_time_dimension() { - (td.base_symbol().alias(), td.granularity().clone()) + let (lambda_base, granularity) = if let Ok(td) = dim.as_time_dimension() { + (td.base_symbol().clone(), td.granularity().clone()) } else { - (dim.alias(), None) + (dim.clone(), None) }; - let name_in_table = PlanSqlTemplates::member_alias_name( - &item.cube_alias, - &dim.name(), - &granularity, - ); + let branch_base = + Self::find_branch_member(&lambda_base, &branch_time_bases, branch_cube_name)?; - let suffix = if let Some(granularity) = granularity { - format!("_{}", granularity.clone()) + let read_suffix = if let Some(granularity) = &granularity { + format!("_{}", granularity) + } else { + String::new() + }; + let name_in_table = format!("{}{}", branch_base.alias(), read_suffix); + + let out_suffix = if let Some(granularity) = &granularity { + format!("_{}", granularity) } else { "_day".to_string() }; - let alias = format!("{}{}", alias, suffix); + let alias = format!("{}{}", lambda_base.alias(), out_suffix); select_builder.add_projection_reference_member( &dim, - QualifiedColumnName::new(None, name_in_table.clone()), + QualifiedColumnName::new(None, name_in_table), Some(alias), ); } for meas in pre_aggregation.measures().iter() { - let name_in_table = PlanSqlTemplates::member_alias_name( - &item.cube_alias, - &meas.name(), - &meas.alias_suffix(), - ); - let alias = meas.alias(); + let branch_meas = Self::find_branch_member(meas, &item.measures, branch_cube_name)?; select_builder.add_projection_reference_member( &meas, - QualifiedColumnName::new(None, name_in_table.clone()), - Some(alias), + QualifiedColumnName::new(None, branch_meas.alias()), + Some(meas.alias()), ); } let context = SqlNodesFactory::new(); diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/query.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/query.rs index 7b1fa4c930c66..a388859ac3cb1 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/query.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/physical_plan_builder/processors/query.rs @@ -195,9 +195,29 @@ impl<'a> LogicalNodeProcessor<'a, Query> for QueryProcessor<'a> { context_factory.set_ungrouped(true); } + // When reading from a pre-aggregation, drop ORDER BY keys on measures that + // are not part of the selection. CubeStore cannot ORDER BY an aggregate of a + // rollup column that isn't projected. + let order_by = if is_pre_aggregation { + logical_plan + .modifers() + .order_by + .iter() + .filter(|o| { + !(o.member_symbol().is_measure() + && logical_plan + .schema() + .find_member_positions(&o.name()) + .is_empty()) + }) + .cloned() + .collect() + } else { + logical_plan.modifers().order_by.clone() + }; select_builder.set_order_by( self.builder - .make_order_by(logical_plan.schema(), &logical_plan.modifers().order_by)?, + .make_order_by(logical_plan.schema(), &order_by)?, ); let res = Rc::new(select_builder.build(query_tools.clone(), context_factory)); diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties.rs index 37a7d696aea1f..70ba0941b5b73 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties.rs @@ -167,6 +167,13 @@ pub struct QueryProperties { ungrouped: bool, #[builder(default)] pre_aggregation_query: bool, + /// Pre-aggregation matching only: run the pre-aggregation optimizer to determine + /// which pre-aggregation a query would use, but skip building the outer query's + /// physical SQL. Used by the refresh/metadata path, which needs the match (and the + /// pre-agg's own load SQL) but not the outer query — that outer query may include a + /// rolling-window time series which requires a date range the refresh path doesn't have. + #[builder(default)] + pre_aggregations_match_only: bool, /// When building a rollup pre-aggregation, source it from the cube's /// `originalSql` pre-aggregation table instead of the raw cube SQL. #[builder(default)] @@ -358,6 +365,10 @@ impl QueryProperties { self.pre_aggregation_query } + pub fn is_pre_aggregations_match_only(&self) -> bool { + self.pre_aggregations_match_only + } + pub fn use_original_sql_pre_aggregations_in_pre_aggregation(&self) -> bool { self.use_original_sql_pre_aggregations_in_pre_aggregation } @@ -1115,6 +1126,7 @@ impl PartialEq for QueryProperties { ungrouped, ignore_cumulative, pre_aggregation_query, + pre_aggregations_match_only, use_original_sql_pre_aggregations_in_pre_aggregation, total_query, allow_multi_stage, @@ -1141,6 +1153,7 @@ impl PartialEq for QueryProperties { && *ungrouped == other.ungrouped && *ignore_cumulative == other.ignore_cumulative && *pre_aggregation_query == other.pre_aggregation_query + && *pre_aggregations_match_only == other.pre_aggregations_match_only && *use_original_sql_pre_aggregations_in_pre_aggregation == other.use_original_sql_pre_aggregations_in_pre_aggregation && *total_query == other.total_query diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties_compiler.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties_compiler.rs index 3f6dc85a092d0..c184016389ac0 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties_compiler.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/query_properties_compiler.rs @@ -84,6 +84,10 @@ impl QueryPropertiesCompiler { .and_then(|v| v.parse::().ok()); let ungrouped = options.static_data().ungrouped.unwrap_or(false); let pre_aggregation_query = options.static_data().pre_aggregation_query.unwrap_or(false); + let pre_aggregations_match_only = options + .static_data() + .pre_aggregations_match_only + .unwrap_or(false); let use_original_sql_pre_aggregations_in_pre_aggregation = options .static_data() .use_original_sql_pre_aggregations_in_pre_aggregation @@ -113,6 +117,7 @@ impl QueryPropertiesCompiler { .offset(offset) .ungrouped(ungrouped) .pre_aggregation_query(pre_aggregation_query) + .pre_aggregations_match_only(pre_aggregations_match_only) .use_original_sql_pre_aggregations_in_pre_aggregation( use_original_sql_pre_aggregations_in_pre_aggregation, ) diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/top_level_planner.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/top_level_planner.rs index 427eb8c2b46ad..9ed93b0b05099 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/top_level_planner.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/planner/top_level_planner.rs @@ -43,6 +43,15 @@ impl TopLevelPlanner { let (optimized_plan, usages) = self.try_pre_aggregations(logical_plan.clone())?; + // Match-only mode (refresh/metadata path): the caller only needs the matched + // pre-aggregation(s), not the outer query SQL. Skip the physical build, which for a + // rolling-window measure would render a time series that requires a date range the + // refresh path doesn't provide (and which non-generated-time-series dialects can't + // build without one). The pre-agg's own load SQL is built separately on the JS side. + if self.request.is_pre_aggregations_match_only() { + return Ok((String::new(), usages)); + } + let is_external = if !usages.is_empty() { usages.iter().all(|usage| usage.pre_aggregation.external()) } else { diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/cube_bridge/base_query_options.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/cube_bridge/base_query_options.rs index 3b091aadea29a..4b98a3ce07d75 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/cube_bridge/base_query_options.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/cube_bridge/base_query_options.rs @@ -62,6 +62,8 @@ pub struct MockBaseQueryOptions { #[builder(default)] pre_aggregation_query: Option, #[builder(default)] + pre_aggregations_match_only: Option, + #[builder(default)] use_original_sql_pre_aggregations_in_pre_aggregation: Option, #[builder(default)] total_query: Option, @@ -92,6 +94,7 @@ impl_static_data!( ungrouped, export_annotated_sql, pre_aggregation_query, + pre_aggregations_match_only, use_original_sql_pre_aggregations_in_pre_aggregation, total_query, cubestore_support_multistage, diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/test_utils/test_context.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/test_utils/test_context.rs index 2cf973cc7e9b3..26e5afe1b8c60 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/test_utils/test_context.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/test_fixtures/test_utils/test_context.rs @@ -673,61 +673,108 @@ impl TestContext { for key in &order { let usages = &grouped[key]; let pre_agg = &usages[0].pre_aggregation; - let tables = Self::collect_pre_agg_source_tables(pre_agg.source()); - let mut union_measures: Vec = Vec::new(); - let mut seen: std::collections::HashSet = std::collections::HashSet::new(); - for u in usages { - for m in u.pre_aggregation.measures() { - let n = m.full_name(); - if seen.insert(n.clone()) { - union_measures.push(n); + + match pre_agg.source().as_ref() { + // A rollupLambda unions member rollups that may live in different + // cubes. Production refreshes each member rollup independently, so + // every stored table carries ITS OWN cube aliases (e.g. + // `visitor_checkins2__visitor_id`). + PreAggregationSource::Union(union) => { + for item in &union.items { + let yaml = Self::build_pre_agg_query_yaml_from_members( + &item.measures, + &item.dimensions, + &item.time_dimensions, + ); + let inlined_sql = self.build_pre_agg_table_sql(&yaml); + let name = item + .table + .alias + .clone() + .unwrap_or_else(|| item.table.name.clone()); + let table_name = PlanSqlTemplates::alias_name(&format!( + "{}.{}", + item.table.cube_name, name + )); + Self::create_pg_pre_agg_table(client, &table_name, &inlined_sql).await; + } + } + _ => { + let tables = Self::collect_pre_agg_source_tables(pre_agg.source()); + // Dedup usages by (cube, name): the optimizer creates a separate + // PreAggregationUsage per subquery with only the measures consumed + // in that subquery; the physical pre-agg table must expose the union. + let mut union_measures: Vec = Vec::new(); + let mut seen: std::collections::HashSet = + std::collections::HashSet::new(); + for u in usages { + for m in u.pre_aggregation.measures() { + let n = m.full_name(); + if seen.insert(n.clone()) { + union_measures.push(n); + } + } + } + let yaml = Self::build_pre_agg_query_yaml(pre_agg, &union_measures); + let inlined_sql = self.build_pre_agg_table_sql(&yaml); + + for table in &tables { + let name = table.alias.clone().unwrap_or_else(|| table.name.clone()); + let table_name = + PlanSqlTemplates::alias_name(&format!("{}.{}", table.cube_name, name)); + Self::create_pg_pre_agg_table(client, &table_name, &inlined_sql).await; } } } - let yaml = Self::build_pre_agg_query_yaml(pre_agg, &union_measures); - - let pa_ctx = - Self::new_with_options(self.schema.clone(), Tz::UTC, None, None, false, false) - .expect("Failed to create pre-agg context"); - - let (raw_sql, _) = pa_ctx - .build_sql_with_used_pre_aggregations(&yaml) - .unwrap_or_else(|e| { - panic!( - "Failed to build pre-agg SQL.\nQuery YAML:\n{}\nError: {}", - yaml, e - ) - }); + } + } - let templates = pa_ctx - .query_tools - .plan_sql_templates(false) - .expect("Failed to get SQL templates"); - let (sql, params) = pa_ctx - .query_tools - .build_sql_and_params(&raw_sql, true, &templates) - .expect("Failed to build pre-agg SQL and params"); - let inlined_sql = Self::inline_params(&sql, ¶ms); + /// Builds SQL that materializes a pre-aggregation table from a + /// `pre_aggregation_query: true` YAML (the rollup-defining SELECT), with + /// parameters inlined so it can run as `CREATE TABLE ... AS (...)`. + #[cfg(feature = "integration-postgres")] + fn build_pre_agg_table_sql(&self, yaml: &str) -> String { + let pa_ctx = Self::new_with_options(self.schema.clone(), Tz::UTC, None, None, false, false) + .expect("Failed to create pre-agg context"); + + let (raw_sql, _) = pa_ctx + .build_sql_with_used_pre_aggregations(yaml) + .unwrap_or_else(|e| { + panic!( + "Failed to build pre-agg SQL.\nQuery YAML:\n{}\nError: {}", + yaml, e + ) + }); - for table in &tables { - let name = table.alias.clone().unwrap_or_else(|| table.name.clone()); - let table_name = - PlanSqlTemplates::alias_name(&format!("{}.{}", table.cube_name, name)); + let templates = pa_ctx + .query_tools + .plan_sql_templates(false) + .expect("Failed to get SQL templates"); + let (sql, params) = pa_ctx + .query_tools + .build_sql_and_params(&raw_sql, true, &templates) + .expect("Failed to build pre-agg SQL and params"); + Self::inline_params(&sql, ¶ms) + } - client - .batch_execute(&format!( - "DROP TABLE IF EXISTS \"{table_name}\";\n\ - CREATE TABLE \"{table_name}\" AS ({inlined_sql})" - )) - .await - .unwrap_or_else(|e| { - panic!( - "Failed to create pre-agg table {}: {}\nSQL: {}", - table_name, e, inlined_sql - ) - }); - } - } + #[cfg(feature = "integration-postgres")] + async fn create_pg_pre_agg_table( + client: &tokio_postgres::Client, + table_name: &str, + inlined_sql: &str, + ) { + client + .batch_execute(&format!( + "DROP TABLE IF EXISTS \"{table_name}\";\n\ + CREATE TABLE \"{table_name}\" AS ({inlined_sql})" + )) + .await + .unwrap_or_else(|e| { + panic!( + "Failed to create pre-agg table {}: {}\nSQL: {}", + table_name, e, inlined_sql + ) + }); } #[cfg(feature = "integration-postgres")] @@ -741,9 +788,11 @@ impl TestContext { } tables } - PreAggregationSource::Union(union) => { - union.items.iter().map(|t| t.as_ref().clone()).collect() - } + PreAggregationSource::Union(union) => union + .items + .iter() + .map(|item| item.table.as_ref().clone()) + .collect(), } } @@ -799,6 +848,52 @@ impl TestContext { yaml } + /// Same as `build_pre_agg_query_yaml` but driven by explicit member symbols + /// instead of a `PreAggregation`. Used to materialize each rollupLambda union + /// member from its own cube's members. + #[cfg(feature = "integration-postgres")] + fn build_pre_agg_query_yaml_from_members( + measures: &[Rc], + dimensions: &[Rc], + time_dimensions: &[Rc], + ) -> String { + let mut yaml = String::new(); + + if !measures.is_empty() { + yaml.push_str("measures:\n"); + for m in measures { + yaml.push_str(&format!(" - {}\n", m.full_name())); + } + } + + if !dimensions.is_empty() { + yaml.push_str("dimensions:\n"); + for d in dimensions { + yaml.push_str(&format!(" - {}\n", d.full_name())); + } + } + + if !time_dimensions.is_empty() { + yaml.push_str("time_dimensions:\n"); + for td in time_dimensions { + if let Ok(td_sym) = td.as_time_dimension() { + yaml.push_str(&format!( + " - dimension: {}\n", + td_sym.base_symbol().full_name() + )); + if let Some(gran) = td_sym.granularity() { + yaml.push_str(&format!(" granularity: {}\n", gran)); + } + } else { + yaml.push_str(&format!(" - dimension: {}\n", td.full_name())); + } + } + } + + yaml.push_str("pre_aggregation_query: true\n"); + yaml + } + #[cfg(not(feature = "integration-postgres"))] pub async fn try_execute_pg_from_options( &self, diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__order_by_only_measure_dropped_from_pre_agg_cubestore_result.snap b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__order_by_only_measure_dropped_from_pre_agg_cubestore_result.snap new file mode 100644 index 0000000000000..bbb22d4614a66 --- /dev/null +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__order_by_only_measure_dropped_from_pre_agg_cubestore_result.snap @@ -0,0 +1,13 @@ +--- +source: cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs +expression: result +--- +orders__country | orders__created_at_day | orders__count +----------------+--------------------------+-------------- +New York | 2025-01-10T00:00:00.000Z | 2 +New York | 2025-01-11T00:00:00.000Z | 1 +Boston | 2025-01-31T00:00:00.000Z | 1 +Boston | 2025-02-01T00:00:00.000Z | 2 +Chicago | 2025-02-15T00:00:00.000Z | 1 +Chicago | 2025-03-01T00:00:00.000Z | 2 +New York | 2025-03-02T00:00:00.000Z | 1 diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__rollup_lambda_cross_cube_union_cubestore_result.snap b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__rollup_lambda_cross_cube_union_cubestore_result.snap new file mode 100644 index 0000000000000..af68495ce8ca0 --- /dev/null +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__rollup_lambda_cross_cube_union_cubestore_result.snap @@ -0,0 +1,14 @@ +--- +source: cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs +expression: result +--- +visitor_checkins__visitor_id | visitor_checkins__created_at_day | visitor_checkins__count +-----------------------------+----------------------------------+------------------------ +1 | 2025-01-10T00:00:00.000Z | 4 +1 | 2025-01-11T00:00:00.000Z | 2 +4 | 2025-01-31T00:00:00.000Z | 2 +4 | 2025-02-01T00:00:00.000Z | 2 +5 | 2025-02-01T00:00:00.000Z | 2 +6 | 2025-02-15T00:00:00.000Z | 2 +7 | 2025-02-16T00:00:00.000Z | 2 +8 | 2025-03-01T00:00:00.000Z | 2 diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__ungrouped_pre_agg_measure_reads_rollup_column_cubestore_result.snap b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__ungrouped_pre_agg_measure_reads_rollup_column_cubestore_result.snap new file mode 100644 index 0000000000000..559dcf5521533 --- /dev/null +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/snapshots/cubesqlplanner__tests__integration__pre_aggregations__sql_generation__ungrouped_pre_agg_measure_reads_rollup_column_cubestore_result.snap @@ -0,0 +1,13 @@ +--- +source: cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs +expression: result +--- +orders__status | orders__city | orders__total_amount +---------------+--------------+--------------------- +cancelled | Chicago | 25 +completed | Boston | 375 +completed | Chicago | 400 +completed | New York | 150 +pending | Boston | 150 +pending | Chicago | 175 +pending | New York | 260 diff --git a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs index adfefec5b010d..a75cd162fcf87 100644 --- a/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs +++ b/rust/cube/cubesqlplanner/cubesqlplanner/src/tests/integration/pre_aggregations/sql_generation.rs @@ -15,9 +15,9 @@ use indoc::indoc; use std::rc::Rc; #[tokio::test(flavor = "multi_thread")] -async fn test_basic_pre_agg_sql() { +async fn test_basic_pre_agg_sql() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml"); - let test_context = TestContext::new(schema).unwrap(); + let test_context = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -28,9 +28,7 @@ async fn test_basic_pre_agg_sql() { - id: visitors.source "}; - let (_sql, pre_aggrs) = test_context - .build_sql_with_used_pre_aggregations(query_yaml) - .expect("Should generate SQL without pre-aggregations"); + let (_sql, pre_aggrs) = test_context.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1, "Should use one pre-aggregation"); assert_eq!(pre_aggrs[0].name(), "daily_rollup"); @@ -41,13 +39,15 @@ async fn test_basic_pre_agg_sql() { { insta::assert_snapshot!("basic_pre_agg_sql_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_full_match_main_rollup() { +async fn test_full_match_main_rollup() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -61,9 +61,7 @@ async fn test_full_match_main_rollup() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main_rollup"); @@ -74,13 +72,15 @@ async fn test_full_match_main_rollup() { { insta::assert_snapshot!("full_match_main_rollup_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_partial_match_main_rollup() { +async fn test_partial_match_main_rollup() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -91,9 +91,7 @@ async fn test_partial_match_main_rollup() { - id: orders.status "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main_rollup"); @@ -104,13 +102,15 @@ async fn test_partial_match_main_rollup() { { insta::assert_snapshot!("partial_match_main_rollup_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_full_match_non_additive_measure() { +async fn test_full_match_non_additive_measure() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -123,9 +123,7 @@ async fn test_full_match_non_additive_measure() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main_rollup"); @@ -136,31 +134,33 @@ async fn test_full_match_non_additive_measure() { { insta::assert_snapshot!("full_match_non_additive_measure_cubestore_result", result); } + + Ok(()) } #[test] -fn test_no_match_non_additive_measure_partial() { +fn test_no_match_non_additive_measure_partial() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.avg_amount dimensions: - orders.status - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_daily_rollup_full_match() { +async fn test_daily_rollup_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["daily_countries_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -175,9 +175,7 @@ async fn test_daily_rollup_full_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "daily_countries_rollup"); @@ -188,13 +186,15 @@ async fn test_daily_rollup_full_match() { { insta::assert_snapshot!("daily_rollup_full_match_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_daily_rollup_coarser_granularity() { +async fn test_daily_rollup_coarser_granularity() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["daily_countries_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -209,9 +209,7 @@ async fn test_daily_rollup_coarser_granularity() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "daily_countries_rollup"); @@ -222,16 +220,17 @@ async fn test_daily_rollup_coarser_granularity() { { insta::assert_snapshot!("daily_rollup_coarser_granularity_cubestore_result", result); } + + Ok(()) } #[test] -fn test_daily_rollup_finer_granularity_no_match() { +fn test_daily_rollup_finer_granularity_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["daily_countries_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.count dimensions: @@ -239,17 +238,18 @@ fn test_daily_rollup_finer_granularity_no_match() { time_dimensions: - dimension: orders.created_at granularity: hour - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_daily_rollup_non_additive_full_match() { +async fn test_daily_rollup_non_additive_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["daily_countries_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -264,9 +264,7 @@ async fn test_daily_rollup_non_additive_full_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "daily_countries_rollup"); @@ -280,16 +278,17 @@ async fn test_daily_rollup_non_additive_full_match() { result ); } + + Ok(()) } #[test] -fn test_daily_rollup_non_additive_coarser_granularity_no_match() { +fn test_daily_rollup_non_additive_coarser_granularity_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["daily_countries_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.avg_amount dimensions: @@ -297,19 +296,20 @@ fn test_daily_rollup_non_additive_coarser_granularity_no_match() { time_dimensions: - dimension: orders.created_at granularity: month - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } // --- multi_level_measure across different pre-aggregations --- #[tokio::test(flavor = "multi_thread")] -async fn test_multi_level_all_base_measures_full_match() { +async fn test_multi_level_all_base_measures_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["all_base_measures_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -322,9 +322,7 @@ async fn test_multi_level_all_base_measures_full_match() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "all_base_measures_rollup"); @@ -338,13 +336,15 @@ async fn test_multi_level_all_base_measures_full_match() { result ); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_multi_level_all_base_measures_partial_match() { +async fn test_multi_level_all_base_measures_partial_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["all_base_measures_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -355,9 +355,7 @@ async fn test_multi_level_all_base_measures_partial_match() { - id: orders.status "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "all_base_measures_rollup"); @@ -371,31 +369,33 @@ async fn test_multi_level_all_base_measures_partial_match() { result ); } + + Ok(()) } #[test] -fn test_multi_level_calculated_measure_no_match() { +fn test_multi_level_calculated_measure_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["calculated_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.multi_level_measure dimensions: - orders.status - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_multi_level_calculated_measure_full_match() { +async fn test_multi_level_calculated_measure_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["calculated_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -408,9 +408,7 @@ async fn test_multi_level_calculated_measure_full_match() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "calculated_measure_rollup"); @@ -424,13 +422,15 @@ async fn test_multi_level_calculated_measure_full_match() { result ); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_multi_level_mixed_measure_full_match() { +async fn test_multi_level_mixed_measure_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["mixed_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -443,9 +443,7 @@ async fn test_multi_level_mixed_measure_full_match() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "mixed_measure_rollup"); @@ -459,31 +457,33 @@ async fn test_multi_level_mixed_measure_full_match() { result ); } + + Ok(()) } #[test] -fn test_multi_level_mixed_measure_partial_no_match() { +fn test_multi_level_mixed_measure_partial_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["mixed_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.multi_level_measure dimensions: - orders.status - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_base_and_calculated_measure_full_match() { +async fn test_base_and_calculated_measure_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["base_and_calculated_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -496,9 +496,7 @@ async fn test_base_and_calculated_measure_full_match() { - id: orders.city "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "base_and_calculated_measure_rollup"); @@ -512,13 +510,15 @@ async fn test_base_and_calculated_measure_full_match() { result ); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_base_and_calculated_measure_parital_match() { +async fn test_base_and_calculated_measure_parital_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["base_and_calculated_measure_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -529,9 +529,7 @@ async fn test_base_and_calculated_measure_parital_match() { - id: orders.status "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "base_and_calculated_measure_rollup"); @@ -545,6 +543,8 @@ async fn test_base_and_calculated_measure_parital_match() { result ); } + + Ok(()) } // --- Segment matching tests --- @@ -555,10 +555,10 @@ async fn test_base_and_calculated_measure_parital_match() { // references no members (empty dependencies), so it's a constant filter on top // of the rollup and must not disqualify pre-aggregation matching. #[tokio::test(flavor = "multi_thread")] -async fn test_constant_member_expression_segment_keeps_pre_aggregation() { +async fn test_constant_member_expression_segment_keeps_pre_aggregation() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -568,7 +568,7 @@ async fn test_constant_member_expression_segment_keeps_pre_aggregation() { "}; let access_denied_segment = { - let sql: Rc = Rc::new(MockMemberSql::new("1 = 0").unwrap()); + let sql: Rc = Rc::new(MockMemberSql::new("1 = 0")?); let expr = MockMemberExpressionDefinition::builder() .expression_name(Some("rlsAccessDenied".to_string())) .cube_name(Some("orders".to_string())) @@ -577,9 +577,10 @@ async fn test_constant_member_expression_segment_keeps_pre_aggregation() { OptionsMember::MemberExpression(Rc::new(expr)) }; - let (sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations_with_segments(query_yaml, vec![access_denied_segment]) - .unwrap(); + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations_with_segments( + query_yaml, + vec![access_denied_segment], + )?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main_rollup"); @@ -587,13 +588,15 @@ async fn test_constant_member_expression_segment_keeps_pre_aggregation() { sql.contains("1 = 0"), "expected the constant access-denied segment in SQL, got:\n{sql}" ); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_segment_full_match() { +async fn test_segment_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["segment_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -610,9 +613,7 @@ async fn test_segment_full_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "segment_rollup"); @@ -623,13 +624,15 @@ async fn test_segment_full_match() { { insta::assert_snapshot!("segment_full_match_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_segment_partial_match_unused_segment() { +async fn test_segment_partial_match_unused_segment() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["segment_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -644,9 +647,7 @@ async fn test_segment_partial_match_unused_segment() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "segment_rollup"); @@ -660,16 +661,17 @@ async fn test_segment_partial_match_unused_segment() { result ); } + + Ok(()) } #[test] -fn test_segment_no_match_missing_in_pre_agg() { +fn test_segment_no_match_missing_in_pre_agg() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["main_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.count dimensions: @@ -677,19 +679,20 @@ fn test_segment_no_match_missing_in_pre_agg() { - orders.city segments: - orders.high_priority - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } // --- Custom granularity pre-aggregation tests --- #[tokio::test(flavor = "multi_thread")] -async fn test_custom_granularity_full_match() { +async fn test_custom_granularity_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["custom_half_year_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -704,9 +707,7 @@ async fn test_custom_granularity_full_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "custom_half_year_rollup"); @@ -717,13 +718,15 @@ async fn test_custom_granularity_full_match() { { insta::assert_snapshot!("custom_granularity_full_match_cubestore_result", result); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_standard_pre_agg_coarser_custom_query() { +async fn test_standard_pre_agg_coarser_custom_query() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["daily_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -738,9 +741,7 @@ async fn test_standard_pre_agg_coarser_custom_query() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "daily_rollup"); @@ -754,16 +755,17 @@ async fn test_standard_pre_agg_coarser_custom_query() { result ); } + + Ok(()) } #[test] -fn test_custom_pre_agg_finer_query_no_match() { +fn test_custom_pre_agg_finer_query_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["custom_half_year_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.count dimensions: @@ -771,20 +773,20 @@ fn test_custom_pre_agg_finer_query_no_match() { time_dimensions: - dimension: orders.created_at granularity: day - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[test] -fn test_custom_pre_agg_finer_standard_query_no_match() { +fn test_custom_pre_agg_finer_standard_query_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["custom_half_year_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.count dimensions: @@ -792,17 +794,18 @@ fn test_custom_pre_agg_finer_standard_query_no_match() { time_dimensions: - dimension: orders.created_at granularity: month - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_custom_granularity_non_additive_full_match() { +async fn test_custom_granularity_non_additive_full_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["custom_half_year_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -817,9 +820,7 @@ async fn test_custom_granularity_non_additive_full_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "custom_half_year_rollup"); @@ -833,16 +834,17 @@ async fn test_custom_granularity_non_additive_full_match() { result ); } + + Ok(()) } #[test] -fn test_custom_granularity_non_additive_coarser_no_match() { +fn test_custom_granularity_non_additive_coarser_no_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["daily_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - orders.avg_amount dimensions: @@ -850,17 +852,18 @@ fn test_custom_granularity_non_additive_coarser_no_match() { time_dimensions: - dimension: orders.created_at granularity: half_year - "}) - .unwrap(); + "})?; assert!(pre_aggrs.is_empty()); + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_custom_granularity_non_strict_self_match() { +async fn test_custom_granularity_non_strict_self_match() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/custom_granularity_test.yaml") .only_pre_aggregations(&["custom_half_year_non_strict"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -872,9 +875,7 @@ async fn test_custom_granularity_non_strict_self_match() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "custom_half_year_non_strict"); @@ -888,13 +889,15 @@ async fn test_custom_granularity_non_strict_self_match() { result ); } + + Ok(()) } #[tokio::test(flavor = "multi_thread")] -async fn test_segment_with_coarser_granularity() { +async fn test_segment_with_coarser_granularity() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["segment_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -911,9 +914,7 @@ async fn test_segment_with_coarser_granularity() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "segment_rollup"); @@ -924,14 +925,17 @@ async fn test_segment_with_coarser_granularity() { { insta::assert_snapshot!("segment_with_coarser_granularity_cubestore_result", result); } + + Ok(()) } // --- Multi-stage count_distinct sum by quarter with pre-aggregation --- #[tokio::test(flavor = "multi_thread")] -async fn test_multi_stage_count_distinct_sum_by_quarter_with_pre_aggregation() { +async fn test_multi_stage_count_distinct_sum_by_quarter_with_pre_aggregation( +) -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/multi_stage_sum_by_quarter_test.yaml"); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -939,9 +943,7 @@ async fn test_multi_stage_count_distinct_sum_by_quarter_with_pre_aggregation() { cubestoreSupportMultistage: true "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main"); @@ -955,14 +957,16 @@ async fn test_multi_stage_count_distinct_sum_by_quarter_with_pre_aggregation() { result ); } + + Ok(()) } // --- Multi-stage with separate pre-aggregations --- #[tokio::test(flavor = "multi_thread")] -async fn test_multi_stage_separate_pre_aggregations() { +async fn test_multi_stage_separate_pre_aggregations() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/multi_stage_separate_pre_aggs_test.yaml"); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -971,9 +975,7 @@ async fn test_multi_stage_separate_pre_aggregations() { cubestoreSupportMultistage: true "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 2, "Expected 2 pre-aggregation usages"); @@ -998,14 +1000,16 @@ async fn test_multi_stage_separate_pre_aggregations() { { insta::assert_snapshot!("multi_stage_separate_pre_aggs_cubestore_result", result); } + + Ok(()) } // --- Multi-stage with separate pre-aggregations and time shift --- #[tokio::test(flavor = "multi_thread")] -async fn test_multi_stage_separate_pre_aggs_with_time_shift() { +async fn test_multi_stage_separate_pre_aggs_with_time_shift() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/multi_stage_pre_agg_time_shift_test.yaml"); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -1022,9 +1026,7 @@ async fn test_multi_stage_separate_pre_aggs_with_time_shift() { - id: orders.created_at "}; - let (_sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 2, "Expected 2 pre-aggregation usages"); @@ -1067,24 +1069,24 @@ async fn test_multi_stage_separate_pre_aggs_with_time_shift() { result ); } + + Ok(()) } // --- rollupJoin with calculated measures through view --- #[test] -fn test_rollup_join_calculated_measures_through_view() { +fn test_rollup_join_calculated_measures_through_view() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/rollup_join_calculated_measures.yaml"); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; - let (sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(indoc! {" + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(indoc! {" measures: - my_view.facts_avg_cost time_dimensions: - dimension: my_view.facts_day granularity: day - "}) - .unwrap(); + "})?; let pre_agg_names: Vec<_> = pre_aggrs .iter() @@ -1113,6 +1115,8 @@ fn test_rollup_join_calculated_measures_through_view() { "SQL should reference li_rollup, got:\n{}", sql ); + + Ok(()) } // A rolling-window count_distinct_approx whose pre-aggregation stores the @@ -1121,11 +1125,10 @@ fn test_rollup_join_calculated_measures_through_view() { // across the window and finalize to a cardinality. This pins the state // branch — the read must NOT collapse the state to a cardinality too early. #[test] -fn test_count_distinct_approx_rolling_pre_agg_keeps_state() { +fn test_count_distinct_approx_rolling_pre_agg_keeps_state() -> Result<(), CubeError> { let ctx = TestContext::new(MockSchema::from_yaml_file( "common/integration_rolling_window.yaml", - )) - .unwrap(); + ))?; let query = indoc! {r#" measures: @@ -1141,7 +1144,7 @@ fn test_count_distinct_approx_rolling_pre_agg_keeps_state() { cubestoreSupportMultistage: true "#}; - let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query).unwrap(); + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "approx_rolling"); @@ -1158,6 +1161,8 @@ fn test_count_distinct_approx_rolling_pre_agg_keeps_state() { "Rolling window should finalize merged states to a cardinality, got:\n{}", sql ); + + Ok(()) } // --- HLL count_distinct_approx through a pre-aggregation --- @@ -1175,10 +1180,10 @@ fn test_count_distinct_approx_rolling_pre_agg_keeps_state() { // count_distinct_approx -> round(hll_cardinality(hll_add_agg(hll_hash_any(x)))) #[test] -fn test_count_distinct_approx_pre_agg_read_merges_state() { +fn test_count_distinct_approx_pre_agg_read_merges_state() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["approx_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -1188,9 +1193,7 @@ fn test_count_distinct_approx_pre_agg_read_merges_state() { - orders.city "}; - let (sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "approx_rollup"); @@ -1208,13 +1211,15 @@ fn test_count_distinct_approx_pre_agg_read_merges_state() { "Read query should not re-init HLL from the state column, got:\n{}", sql ); + + Ok(()) } #[test] -fn test_count_distinct_approx_pre_agg_build_emits_state() { +fn test_count_distinct_approx_pre_agg_build_emits_state() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") .only_pre_aggregations(&["approx_rollup"]); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; // pre_aggregation_query: true renders the rollup build (load) SQL. let query_yaml = indoc! {" @@ -1226,7 +1231,7 @@ fn test_count_distinct_approx_pre_agg_build_emits_state() { pre_aggregation_query: true "}; - let sql = ctx.build_sql(query_yaml).unwrap(); + let sql = ctx.build_sql(query_yaml)?; // The build must serialize the HLL state (hll_init) without merging // or taking its cardinality — that happens only on read. @@ -1245,6 +1250,8 @@ fn test_count_distinct_approx_pre_agg_build_emits_state() { "Build SQL should not merge states, got:\n{}", sql ); + + Ok(()) } // A multi-stage measure that sums a count_distinct_approx must read the @@ -1252,9 +1259,9 @@ fn test_count_distinct_approx_pre_agg_build_emits_state() { // pre-aggregation — the outer `sum` aggregates counts, not raw HLL states. // This pins that the pre-agg read does not leak a bare merged state here. #[test] -fn test_count_distinct_approx_multistage_pre_agg_reads_cardinality() { +fn test_count_distinct_approx_multistage_pre_agg_reads_cardinality() -> Result<(), CubeError> { let schema = MockSchema::from_yaml_file("common/multi_stage_sum_by_quarter_test.yaml"); - let ctx = TestContext::new(schema).unwrap(); + let ctx = TestContext::new(schema)?; let query_yaml = indoc! {" measures: @@ -1262,9 +1269,7 @@ fn test_count_distinct_approx_multistage_pre_agg_reads_cardinality() { cubestoreSupportMultistage: true "}; - let (sql, pre_aggrs) = ctx - .build_sql_with_used_pre_aggregations(query_yaml) - .unwrap(); + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; assert_eq!(pre_aggrs.len(), 1); assert_eq!(pre_aggrs[0].name(), "main_approx"); @@ -1276,6 +1281,8 @@ fn test_count_distinct_approx_multistage_pre_agg_reads_cardinality() { "Multi-stage leaf should finalize HLL to cardinality, got:\n{}", sql ); + + Ok(()) } // A cube `foo` whose `originalSql` pre-aggregation (`main`) materializes its base @@ -1368,3 +1375,148 @@ fn test_rollup_build_without_use_original_sql_pre_aggregations_in_pre_aggregatio ); Ok(()) } + +// A measure referenced only in ORDER BY (not in the selected measures) is dropped +// from the ORDER BY when reading a pre-aggregation. CubeStore cannot ORDER BY an +// aggregate of a rollup column that isn't projected, and the legacy planner +// likewise ignores such keys, so the remaining keys (here the time dimension and +// the selected dimension) drive the order. Mirrors the driver-test "partitioned +// pre-agg" queries that order by `created_at asc, desc, asc`. +#[tokio::test(flavor = "multi_thread")] +async fn test_order_by_only_measure_dropped_from_pre_agg() -> Result<(), CubeError> { + let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") + .only_pre_aggregations(&["daily_countries_rollup"]); + let ctx = TestContext::new(schema)?; + + let query_yaml = indoc! {" + measures: + - orders.count + dimensions: + - orders.country + time_dimensions: + - dimension: orders.created_at + granularity: day + order: + - id: orders.created_at + desc: false + - id: orders.total_amount + desc: true + - id: orders.country + desc: false + "}; + + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; + + assert_eq!(pre_aggrs.len(), 1); + assert_eq!(pre_aggrs[0].name(), "daily_countries_rollup"); + // total_amount is neither selected nor projected by the rollup read, so it must + // not appear in ORDER BY — neither as the base column nor the rollup column. + assert!( + !sql.contains("\"orders\".amount"), + "ORDER BY must not reference the base-table column, got:\n{sql}" + ); + assert!( + !sql.contains("orders__total_amount"), + "order-by-only measure must be dropped, not reference the rollup column, got:\n{sql}" + ); + + if let Some(result) = ctx + .try_execute(query_yaml, "pre_aggregation_matching_tables.sql") + .await + { + insta::assert_snapshot!( + "order_by_only_measure_dropped_from_pre_agg_cubestore_result", + result + ); + } + + Ok(()) +} + +// An ungrouped query reading from a rollup must reference the stored measure +// column as-is (no `sum()`), not re-render the measure's base-table SQL. +#[tokio::test(flavor = "multi_thread")] +async fn test_ungrouped_pre_agg_measure_reads_rollup_column() -> Result<(), CubeError> { + let schema = MockSchema::from_yaml_file("common/pre_aggregation_matching_test.yaml") + .only_pre_aggregations(&["main_rollup"]); + let ctx = TestContext::new(schema)?; + + let query_yaml = indoc! {" + measures: + - orders.total_amount + dimensions: + - orders.status + - orders.city + ungrouped: true + order: + - id: orders.status + - id: orders.city + "}; + + let (sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; + + assert_eq!(pre_aggrs.len(), 1); + assert_eq!(pre_aggrs[0].name(), "main_rollup"); + assert!( + sql.contains("\"orders__total_amount\" \"orders__total_amount\""), + "ungrouped measure should project the rollup column, got:\n{sql}" + ); + assert!( + !sql.contains("\"orders\".amount"), + "ungrouped measure must not reference the base-table column, got:\n{sql}" + ); + + if let Some(result) = ctx + .try_execute(query_yaml, "pre_aggregation_matching_tables.sql") + .await + { + insta::assert_snapshot!( + "ungrouped_pre_agg_measure_reads_rollup_column_cubestore_result", + result + ); + } + + Ok(()) +} + +// `lambda_union` UNIONs `visitor_checkins.for_lambda` with +// `visitor_checkins2.for_lambda`. The lambda exposes the first member +// rollup's symbols, so the dimension's unified output alias is +// `visitor_checkins__visitor_id`, but the second branch stores it as +// `visitor_checkins2__visitor_id`. +#[tokio::test(flavor = "multi_thread")] +async fn test_rollup_lambda_cross_cube_union_aliases() -> Result<(), CubeError> { + let schema = MockSchema::from_yaml_file("common/pre_aggregations_test.yaml"); + let ctx = TestContext::new(schema)?; + + // Force the lambda; otherwise the matcher would pick the plain `for_lambda` + // rollup that the lambda is built from (it is declared first). The total + // `order:` pins row order so the CubeStore result snapshot is stable. + let query_yaml = indoc! {" + measures: + - visitor_checkins.count + dimensions: + - visitor_checkins.visitor_id + time_dimensions: + - dimension: visitor_checkins.created_at + granularity: day + order: + - id: visitor_checkins.visitor_id + - id: visitor_checkins.created_at + pre_aggregation_id: visitor_checkins.lambda_union + "}; + + let (_sql, pre_aggrs) = ctx.build_sql_with_used_pre_aggregations(query_yaml)?; + + assert_eq!(pre_aggrs.len(), 1); + assert_eq!(pre_aggrs[0].name(), "lambda_union"); + + if let Some(result) = ctx + .try_execute(query_yaml, "pre_aggregation_tables.sql") + .await + { + insta::assert_snapshot!("rollup_lambda_cross_cube_union_cubestore_result", result); + } + + Ok(()) +} diff --git a/rust/cubestore/cubestore/src/bin/cubestored.rs b/rust/cubestore/cubestore/src/bin/cubestored.rs index 8da198a504ef9..8752cf0bb1523 100644 --- a/rust/cubestore/cubestore/src/bin/cubestored.rs +++ b/rust/cubestore/cubestore/src/bin/cubestored.rs @@ -31,13 +31,15 @@ fn main() { ), Err(_) => metrics::Compatibility::StatsD, }; + let metrics_bind_address = + std::env::var("CUBESTORE_METRICS_BIND_ADDRESS").unwrap_or("127.0.0.1".to_string()); let metrics_addr = std::env::var("CUBESTORE_METRICS_ADDRESS").unwrap_or("127.0.0.1".to_string()); let metrics_port = std::env::var("CUBESTORE_METRICS_PORT").unwrap_or("8125".to_string()); let metrics_server_address = format!("{}:{}", metrics_addr, metrics_port); init_metrics( - "127.0.0.1:0", + format!("{}:0", metrics_bind_address), metrics_server_address, metrics_format, vec![], diff --git a/rust/cubestore/cubestore/src/metastore/mod.rs b/rust/cubestore/cubestore/src/metastore/mod.rs index 5d6fadd650638..98d179410598b 100644 --- a/rust/cubestore/cubestore/src/metastore/mod.rs +++ b/rust/cubestore/cubestore/src/metastore/mod.rs @@ -42,7 +42,8 @@ use crate::metastore::multi_index::{ }; use crate::metastore::partition::PartitionIndexKey; use crate::metastore::replay_handle::{ - ReplayHandle, ReplayHandleIndexKey, ReplayHandleRocksIndex, ReplayHandleRocksTable, SeqPointer, + validate_seq_pointers_by_location, ReplayHandle, ReplayHandleIndexKey, ReplayHandleRocksIndex, + ReplayHandleRocksTable, SeqPointer, }; use crate::metastore::source::{ Source, SourceCredentials, SourceIndexKey, SourceRocksIndex, SourceRocksTable, @@ -4616,6 +4617,8 @@ impl MetaStore for RocksMetaStore { self.write_operation( "create_replay_handle_from_seq_pointers", move |db_ref, batch_pipe| { + let table = TableRocksTable::new(db_ref.clone()).get_row_or_not_found(table_id)?; + validate_seq_pointers_by_location(&table, &seq_pointers)?; let handle = ReplayHandle::new_from_seq_pointers(table_id, seq_pointers); Ok(ReplayHandleRocksTable::new(db_ref.clone()).insert(handle, batch_pipe)?) }, @@ -4691,7 +4694,7 @@ impl MetaStore for RocksMetaStore { return Err(CubeError::internal("Can't merge empty replay handles list".to_string())); } let table = ReplayHandleRocksTable::new(db_ref.clone()); - let chunks_table = ChunkRocksTable::new(db_ref); + let chunks_table = ChunkRocksTable::new(db_ref.clone()); let mut replay_handles: Vec> = Vec::new(); for id in old_ids.into_iter() { let replay_handle = table.get_row_or_not_found(id)?; @@ -4723,7 +4726,11 @@ impl MetaStore for RocksMetaStore { replay_handles.push(replay_handle); } let new_handle = if let Some(_) = new_seq_pointer { - let new_replay_handle = ReplayHandle::new_from_seq_pointers(replay_handles[0].get_row().table_id(), new_seq_pointer); + let table_id = replay_handles[0].get_row().table_id(); + let tables_table = TableRocksTable::new(db_ref.clone()); + let tables_row = tables_table.get_row_or_not_found(table_id)?; + validate_seq_pointers_by_location(&tables_row, &new_seq_pointer)?; + let new_replay_handle = ReplayHandle::new_from_seq_pointers(table_id, new_seq_pointer); Some(table.insert(new_replay_handle, batch_pipe)?) } else { @@ -7860,6 +7867,81 @@ mod tests { Ok(()) } + + #[tokio::test] + async fn replay_handle_location_length_guard_test() -> Result<(), CubeError> { + let config = Config::test("replay_handle_location_length_guard_test"); + let store_path = env::current_dir()?.join("rh-guard-local"); + let remote_store_path = env::current_dir()?.join("rh-guard-remote"); + let _ = fs::remove_dir_all(store_path.clone()); + let _ = fs::remove_dir_all(remote_store_path.clone()); + let remote_fs = LocalDirRemoteFs::new(Some(remote_store_path.clone()), store_path.clone()); + + let meta_store = RocksMetaStore::new( + store_path.join("metastore").as_path(), + BaseRocksStoreFs::new_for_metastore(remote_fs.clone(), config.config_obj()), + config.config_obj(), + )?; + + meta_store.create_schema("foo".to_string(), false).await?; + let mut columns = Vec::new(); + columns.push(Column::new("col1".to_string(), ColumnType::Int, 0)); + + let locations = vec![ + "stream://k/T/0".to_string(), + "stream://k/T/1".to_string(), + "stream://k/T/2".to_string(), + ]; + let table = meta_store + .create_table( + "foo".to_string(), + "boo".to_string(), + columns.clone(), + Some(locations), + None, + vec![], + true, + None, + None, + None, + None, + None, + None, + None, + None, + None, + false, + None, + ) + .await?; + + let mismatching = Some(vec![Some(SeqPointer::new(Some(0), Some(1))); 6]); + assert!(meta_store + .create_replay_handle_from_seq_pointers(table.get_id(), mismatching) + .await + .is_err()); + assert!(meta_store + .get_replay_handles_by_table(table.get_id()) + .await? + .is_empty()); + + let matching = Some(vec![Some(SeqPointer::new(Some(0), Some(1))); 3]); + meta_store + .create_replay_handle_from_seq_pointers(table.get_id(), matching) + .await?; + assert_eq!( + meta_store + .get_replay_handles_by_table(table.get_id()) + .await? + .len(), + 1 + ); + + let _ = fs::remove_dir_all(store_path.clone()); + let _ = fs::remove_dir_all(remote_store_path.clone()); + + Ok(()) + } } impl RocksMetaStore { diff --git a/rust/cubestore/cubestore/src/metastore/replay_handle.rs b/rust/cubestore/cubestore/src/metastore/replay_handle.rs index d3d07dfbd6004..3649f00ef8923 100644 --- a/rust/cubestore/cubestore/src/metastore/replay_handle.rs +++ b/rust/cubestore/cubestore/src/metastore/replay_handle.rs @@ -203,7 +203,7 @@ pub fn seq_pointer_for_location<'a>( )) })?; if locations.len() != seq_pointers_by_location.len() { - return Err(CubeError::internal(format!( + return Err(CubeError::corrupt_data(format!( "Location array size mismatch during accessing seq pointers: {:?} and {:?}", table.get_row().locations(), seq_pointers_by_location @@ -213,6 +213,25 @@ pub fn seq_pointer_for_location<'a>( Ok(&seq_pointers_by_location[pos]) } +pub fn validate_seq_pointers_by_location( + table: &IdRow, + seq_pointers_by_location: &Option>>, +) -> Result<(), CubeError> { + if let Some(seq_pointers) = seq_pointers_by_location { + let locations_len = table.get_row().locations().map(|l| l.len()).unwrap_or(0); + if locations_len != seq_pointers.len() { + return Err(CubeError::internal(format!( + "Refusing to persist replay handle for table {}: {} locations but {} seq pointers: {:?}", + table.get_id(), + locations_len, + seq_pointers.len(), + seq_pointers + ))); + } + } + Ok(()) +} + pub fn location_position(table: &IdRow
, location: &str) -> Result { let locations = table.get_row().locations().ok_or_else(|| { CubeError::internal(format!( @@ -365,3 +384,56 @@ impl RocksSecondaryIndex for ReplayHandleRoc *self as IndexId } } + +#[cfg(test)] +mod tests { + use super::*; + + fn table_with_locations(count: usize) -> IdRow
{ + let locations = (0..count).map(|i| format!("loc-{}", i)).collect::>(); + IdRow::new( + 1, + Table::new( + "t".to_string(), + 1, + Vec::new(), + Some(locations), + None, + true, + None, + None, + None, + None, + None, + None, + Vec::new(), + None, + None, + None, + ), + ) + } + + fn pointers(count: usize) -> Option>> { + Some(vec![Some(SeqPointer::new(Some(0), Some(1))); count]) + } + + #[test] + fn validate_matching_length_ok() { + let table = table_with_locations(3); + assert!(validate_seq_pointers_by_location(&table, &pointers(3)).is_ok()); + } + + #[test] + fn validate_mismatching_length_err() { + let table = table_with_locations(3); + assert!(validate_seq_pointers_by_location(&table, &pointers(6)).is_err()); + assert!(validate_seq_pointers_by_location(&table, &pointers(2)).is_err()); + } + + #[test] + fn validate_none_pointers_ok() { + let table = table_with_locations(3); + assert!(validate_seq_pointers_by_location(&table, &None).is_ok()); + } +} diff --git a/rust/cubestore/cubestore/src/scheduler/mod.rs b/rust/cubestore/cubestore/src/scheduler/mod.rs index 719af5aa45e48..36813d32ee966 100644 --- a/rust/cubestore/cubestore/src/scheduler/mod.rs +++ b/rust/cubestore/cubestore/src/scheduler/mod.rs @@ -6,7 +6,7 @@ use crate::metastore::partition::partition_file_name; use crate::metastore::replay_handle::ReplayHandle; use crate::metastore::replay_handle::{ subtract_from_right_seq_pointer_by_location, subtract_if_covers_seq_pointer_by_location, - union_seq_pointer_by_location, SeqPointerForLocation, + union_seq_pointer_by_location, SeqPointer, SeqPointerForLocation, }; use crate::metastore::table::Table; use crate::metastore::{ @@ -512,67 +512,85 @@ impl SchedulerImpl { .into_iter() .chunk_by(|(h, _)| h.get_row().table_id()) { - let mut seq_pointer_by_location = None; - let mut ids = Vec::new(); let handles = handles.collect::>(); - for (handle, _) in handles - .iter() - .filter(|(handle, no_active_chunks)| !is_newest_handle(handle) && *no_active_chunks) - { - union_seq_pointer_by_location( - &mut seq_pointer_by_location, - handle.get_row().seq_pointers_by_location(), - )?; - ids.push(handle.get_id()); - } let empty_vec = Vec::new(); let failed = table_to_failed.get(&table_id).unwrap_or(&empty_vec); - for (failed_handle, no_active_chunks) in failed.iter() { - let mut failed_seq_pointers = - failed_handle.get_row().seq_pointers_by_location().clone(); - let mut replay_after_failed_union = None; - let replay_after_failed = handles - .iter() - .filter(|(h, _)| { - h.get_id() > failed_handle.get_id() - && !h.get_row().has_failed_to_persist_chunks() - }) - .collect::>(); - for (replay, _) in replay_after_failed.iter() { - union_seq_pointer_by_location( - &mut replay_after_failed_union, - replay.get_row().seq_pointers_by_location(), - )?; - } - subtract_if_covers_seq_pointer_by_location( - &mut failed_seq_pointers, - &replay_after_failed_union, - )?; - let empty_seq_pointers = failed_seq_pointers - .map(|p| { - p.iter() - .all(|p| p.as_ref().map(|p| p.is_empty()).unwrap_or(true)) - }) - .unwrap_or(true); - if empty_seq_pointers && *no_active_chunks { - ids.push(failed_handle.get_id()); - } else if !empty_seq_pointers { - subtract_from_right_seq_pointer_by_location( - &mut seq_pointer_by_location, - failed_handle.get_row().seq_pointers_by_location(), - )?; + // Isolate per-table merge: a corrupt handle (e.g. seq pointer / location + // length mismatch) must not abort merging of the remaining tables. + let table_merge = + (|| -> Result<(Vec, Option>>), CubeError> { + let mut seq_pointer_by_location = None; + let mut ids = Vec::new(); + for (handle, _) in handles.iter().filter(|(handle, no_active_chunks)| { + !is_newest_handle(handle) && *no_active_chunks + }) { + union_seq_pointer_by_location( + &mut seq_pointer_by_location, + handle.get_row().seq_pointers_by_location(), + )?; + ids.push(handle.get_id()); + } + + for (failed_handle, no_active_chunks) in failed.iter() { + let mut failed_seq_pointers = + failed_handle.get_row().seq_pointers_by_location().clone(); + let mut replay_after_failed_union = None; + let replay_after_failed = handles + .iter() + .filter(|(h, _)| { + h.get_id() > failed_handle.get_id() + && !h.get_row().has_failed_to_persist_chunks() + }) + .collect::>(); + for (replay, _) in replay_after_failed.iter() { + union_seq_pointer_by_location( + &mut replay_after_failed_union, + replay.get_row().seq_pointers_by_location(), + )?; + } + subtract_if_covers_seq_pointer_by_location( + &mut failed_seq_pointers, + &replay_after_failed_union, + )?; + let empty_seq_pointers = failed_seq_pointers + .map(|p| { + p.iter() + .all(|p| p.as_ref().map(|p| p.is_empty()).unwrap_or(true)) + }) + .unwrap_or(true); + if empty_seq_pointers && *no_active_chunks { + ids.push(failed_handle.get_id()); + } else if !empty_seq_pointers { + subtract_from_right_seq_pointer_by_location( + &mut seq_pointer_by_location, + failed_handle.get_row().seq_pointers_by_location(), + )?; + } + } + + Ok((ids, seq_pointer_by_location)) + })(); + + match table_merge { + Ok(merge) => to_merge.push(merge), + Err(e) => { + error!("Skipping replay handle merge for table {}: {}", table_id, e); + continue; } } - - to_merge.push((ids, seq_pointer_by_location)); } for (ids, seq_pointer_by_location) in to_merge.into_iter() { if !ids.is_empty() { - self.meta_store + if let Err(e) = self + .meta_store .replace_replay_handles(ids, seq_pointer_by_location) - .await?; + .await + { + error!("Skipping replay handle merge: {}", e); + continue; + } } }