From 230d269ee7e95eae017d1c9fb17bfd15c8d481a7 Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 19 Jul 2026 01:31:32 +0200 Subject: [PATCH 1/2] refactor(query-engine): rearchitect warehouse executor around named backends and route-as-data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three warehouse paths (managed Tinybird, Tinybird CH-gateway, BYO ClickHouse) were encoded as _tag×provider flag combinations branched on across two layers, with route selection split between the package executor and four host resolvers, ingest pinning enforced by a per-call-site boolean, three drifting option types, duplicate facade spans, a 500-flatten on observability errors, and a hand-copied CLI executor. - Name the backends: WarehouseBackendKind + one BackendDialect behavior table (driver, settings strip, SQL normalize, peer service); delete the _tag/provider dual discriminator. - Collapse the four injected resolvers into one resolveRoute(tenant, purpose, label) returning route-as-data (source, config, cache key); raw-SQL isolation invariants and cache-key strings preserved verbatim. - Make ingest routing structural: .routing("ingest") declared at the query definition rides CompiledQuery; delete pinToIngestConfig; warn when a BYO route reads an ingest-pinned table. - Unify on one SqlQueryOptions (profiles), delete ExecutorQuery* dupes, drop the duplicate WarehouseExecutor.* span wrappers; executeSql adds warehouse.backend / warehouse.route / warehouse.config_source (legacy attrs dual-emitted for one release). - Surface warehouse errors with their canonical statuses on the observability endpoints (429 quota / 400 validation / 503 transient instead of a blanket 500); map quota/validation properly in the v2 alerts error map; classify ingest failures. - CLI local mode reuses the real executor via a chdb dialect + constant route, deleting the ~190-line hand-rolled copy; /local/query returns 400 for failing SQL so the executor doesn't retry it; the raw-SQL escape hatch posts directly to /local/query (single-tenant, no OrgId guard weakening). Co-Authored-By: Claude Fable 5 --- .../api/src/lib/WarehouseQueryService.test.ts | 90 +++---- apps/api/src/lib/WarehouseQueryService.ts | 178 +++++++------- apps/api/src/routes/observability.http.ts | 13 +- apps/api/src/routes/v2/alerts-error-map.ts | 22 +- apps/api/src/services/AlertsService.ts | 6 +- .../src/services/AnomalyDetectionService.ts | 4 +- .../CloudflareAnalyticsService.test.ts | 14 +- .../services/CloudflareAnalyticsService.ts | 2 +- apps/api/src/services/ErrorsService.ts | 1 - apps/cli/package.json | 2 +- apps/cli/src/core/executor.test.ts | 78 ++++++ apps/cli/src/core/executor.ts | 223 ++++-------------- apps/cli/src/core/operations.ts | 32 ++- apps/cli/src/core/remote-executor.ts | 10 +- apps/cli/src/core/warehouse.ts | 10 +- apps/cli/src/server/serve.ts | 5 +- lib/clickhouse-builder/src/ch/compile.ts | 12 +- lib/clickhouse-builder/src/ch/query.ts | 14 ++ packages/domain/src/http/observability.ts | 13 +- .../query-engine/src/ch/queries/activity.ts | 11 +- .../src/ch/queries/alert-checks.ts | 4 + .../query-engine/src/execution/backend.ts | 79 +++++++ .../src/execution/datasource-routing.ts | 14 ++ .../src/execution/executor.test.ts | 151 +++++++++--- .../query-engine/src/execution/executor.ts | 157 ++++++------ packages/query-engine/src/execution/index.ts | 2 + packages/query-engine/src/execution/ports.ts | 97 +++----- .../src/observability/WarehouseExecutor.ts | 30 +-- .../query-engine/src/observability/index.ts | 4 +- .../src/profiles/query-profile.ts | 23 ++ .../query-engine/src/runtime/query-engine.ts | 19 +- 31 files changed, 716 insertions(+), 604 deletions(-) create mode 100644 apps/cli/src/core/executor.test.ts create mode 100644 packages/query-engine/src/execution/backend.ts create mode 100644 packages/query-engine/src/execution/datasource-routing.ts diff --git a/apps/api/src/lib/WarehouseQueryService.test.ts b/apps/api/src/lib/WarehouseQueryService.test.ts index bb0b46135..c90a8954c 100644 --- a/apps/api/src/lib/WarehouseQueryService.test.ts +++ b/apps/api/src/lib/WarehouseQueryService.test.ts @@ -111,9 +111,8 @@ describe("WarehouseQueryService raw-SQL provider routing", () => { yield* WarehouseQueryService.use((service) => service.rawSqlQuery(makeTenant(), "SELECT 1 WHERE OrgId = 'org_test'"), ) - assert.strictEqual(captured?._tag, "tinybird") - if (captured?._tag !== "tinybird") throw new Error("expected Tinybird config") - assert.strictEqual(captured.provider, "tinybird") + assert.strictEqual(captured?.kind, "tinybird") + if (captured?.kind !== "tinybird") throw new Error("expected Tinybird config") assert.notStrictEqual(captured.token, "managed-token") assert.strictEqual(decodeJwtPayload(captured.token).workspace_id, "test-workspace") assert.deepStrictEqual(responseLimits, { maxRows: 1000, maxBytes: 5_000_000 }) @@ -135,9 +134,8 @@ describe("WarehouseQueryService raw-SQL provider routing", () => { yield* WarehouseQueryService.use((service) => service.rawSqlQuery(makeTenant(), "SELECT 1 WHERE OrgId = 'org_test'"), ) - assert.strictEqual(captured?._tag, "clickhouse") - if (captured?._tag !== "clickhouse") throw new Error("expected gateway config") - assert.strictEqual(captured.provider, "tinybird") + assert.strictEqual(captured?.kind, "tinybird-gateway") + if (captured?.kind !== "tinybird-gateway") throw new Error("expected gateway config") assert.notStrictEqual(captured.password, "gateway-admin-token") assert.strictEqual(decodeJwtPayload(captured.password).workspace_id, "test-workspace") }).pipe(Effect.provide(layer)) @@ -159,9 +157,8 @@ describe("WarehouseQueryService raw-SQL provider routing", () => { yield* WarehouseQueryService.use((service) => service.rawSqlQuery(makeTenant(), "SELECT 1 WHERE OrgId = 'org_test'"), ) - assert.strictEqual(captured?._tag, "clickhouse") - if (captured?._tag !== "clickhouse") throw new Error("expected ClickHouse config") - assert.strictEqual(captured.provider, "clickhouse") + assert.strictEqual(captured?.kind, "clickhouse") + if (captured?.kind !== "clickhouse") throw new Error("expected ClickHouse config") assert.strictEqual(captured.password, "original-clickhouse-password") }).pipe(Effect.provide(layer)) }) @@ -223,9 +220,8 @@ describe("WarehouseQueryService raw-SQL provider routing", () => { yield* WarehouseQueryService.use((service) => service.rawSqlQuery(makeTenant(), "SELECT 1 WHERE OrgId = 'org_test'"), ) - assert.strictEqual(captured?._tag, "clickhouse") - if (captured?._tag !== "clickhouse") throw new Error("expected ClickHouse config") - assert.strictEqual(captured.provider, "clickhouse") + assert.strictEqual(captured?.kind, "clickhouse") + if (captured?.kind !== "clickhouse") throw new Error("expected ClickHouse config") assert.strictEqual(captured.password, "byo-password") }).pipe(Effect.provide(layer)) }) @@ -608,8 +604,7 @@ describe("createClickHouseSqlClient.insert is disabled (ClickHouse is read-only) // client's insert must fail loudly so it can never silently 500 against the // read-only query gateway ("Only SELECT or DESCRIBE … Got: InsertQuery"). const chConfig = { - _tag: "clickhouse" as const, - provider: "clickhouse" as const, + kind: "clickhouse" as const, url: "https://ch.example.com", username: "u", password: "p", @@ -644,8 +639,7 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert // Inserts in the cloud only need to work on Tinybird. This pins that path so a // future change can't silently break ingest into the managed pipeline. const tbConfig = { - _tag: "tinybird" as const, - provider: "tinybird" as const, + kind: "tinybird" as const, host: "https://api.tinybird.co", token: "tok_123", } @@ -711,8 +705,7 @@ describe("createTinybirdSdkSqlClient.insert wire framing (the production insert describe("ingest routes writes to the managed pipeline, not a per-org read override", () => { const clickhouseReadOverride = { config: { - _tag: "clickhouse" as const, - provider: "clickhouse" as const, + kind: "clickhouse" as const, url: "https://byo-clickhouse.example.com", username: "u", password: "p", @@ -722,30 +715,34 @@ describe("ingest routes writes to the managed pipeline, not a per-org read overr } const tinybirdManaged = { config: { - _tag: "tinybird" as const, - provider: "tinybird" as const, + kind: "tinybird" as const, host: "https://managed.tinybird.co", token: "tok", }, clientCacheKey: "write:managed", } - it.effect("ingest uses resolveIngestConfig (Tinybird) while reads use resolveConfig (override)", () => { - const used: Array<{ op: "sql" | "insert"; tag: string }> = [] + it.effect("ingest routes with purpose 'ingest' (Tinybird) while reads route to the override", () => { + const used: Array<{ op: "sql" | "insert"; kind: string }> = [] + const purposes: Array = [] const executor = makeWarehouseExecutor({ createClient: (config) => ({ sql: async () => { - used.push({ op: "sql", tag: config._tag }) + used.push({ op: "sql", kind: config.kind }) return { data: [] } }, insert: async () => { - used.push({ op: "insert", tag: config._tag }) + used.push({ op: "insert", kind: config.kind }) }, }), - resolveConfig: () => Effect.succeed(clickhouseReadOverride), - resolveRawSqlConfig: () => - Effect.succeed({ ...clickhouseReadOverride, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => Effect.succeed(tinybirdManaged), + resolveRoute: (_tenant, purpose) => { + purposes.push(purpose) + return Effect.succeed( + purpose === "ingest" + ? { source: "managed" as const, ...tinybirdManaged } + : { source: "org-byo" as const, ...clickhouseReadOverride }, + ) + }, }) const tenant = makeTenant() @@ -753,32 +750,13 @@ describe("ingest routes writes to the managed pipeline, not a per-org read overr yield* executor.sqlQuery(tenant, "SELECT 1 FROM traces WHERE OrgId = 'org_test'") yield* executor.ingest(tenant, "traces", [{ trace_id: "a" }]) + assert.deepStrictEqual(purposes, ["read", "ingest"]) assert.deepStrictEqual(used, [ - { op: "sql", tag: "clickhouse" }, - { op: "insert", tag: "tinybird" }, + { op: "sql", kind: "clickhouse" }, + { op: "insert", kind: "tinybird" }, ]) }) }) - - it.effect("falls back to resolveConfig for ingest when resolveIngestConfig is absent", () => { - const used: Array<{ op: "insert"; tag: string }> = [] - const executor = makeWarehouseExecutor({ - createClient: (config) => ({ - sql: async () => ({ data: [] }), - insert: async () => { - used.push({ op: "insert", tag: config._tag }) - }, - }), - resolveConfig: () => Effect.succeed(tinybirdManaged), - resolveRawSqlConfig: () => Effect.succeed({ ...tinybirdManaged, clientCacheKey: "raw:org_test" }), - }) - const tenant = makeTenant() - - return Effect.gen(function* () { - yield* executor.ingest(tenant, "traces", [{ trace_id: "a" }]) - assert.deepStrictEqual(used, [{ op: "insert", tag: "tinybird" }]) - }) - }) }) describe("ingest pins writes to Tinybird even when CLICKHOUSE_URL makes managed reads ClickHouse", () => { @@ -788,14 +766,14 @@ describe("ingest pins writes to Tinybird even when CLICKHOUSE_URL makes managed // MUST resolve to Tinybird regardless. Routing ingest through the managed // resolver (which prefers ClickHouse) is what kept demo-seed onboarding broken. it.effect("reads resolve to managed ClickHouse, but ingest resolves to Tinybird", () => { - const used: Array<{ op: "sql" | "insert"; tag: string }> = [] + const used: Array<{ op: "sql" | "insert"; kind: string }> = [] __testables.setClientFactory((config) => ({ sql: async () => { - used.push({ op: "sql", tag: config._tag }) + used.push({ op: "sql", kind: config.kind }) return { data: [] } }, insert: async () => { - used.push({ op: "insert", tag: config._tag }) + used.push({ op: "insert", kind: config.kind }) }, })) @@ -814,9 +792,11 @@ describe("ingest pins writes to Tinybird even when CLICKHOUSE_URL makes managed service.ingest(tenant, "traces", [{ trace_id: "a" }]), ) + // CLICKHOUSE_PROVIDER defaults to "tinybird", so a bare CLICKHOUSE_URL is + // the Tinybird CH-gateway. assert.deepStrictEqual(used, [ - { op: "sql", tag: "clickhouse" }, - { op: "insert", tag: "tinybird" }, + { op: "sql", kind: "tinybird-gateway" }, + { op: "insert", kind: "tinybird" }, ]) }).pipe(Effect.provide(layer)) }) diff --git a/apps/api/src/lib/WarehouseQueryService.ts b/apps/api/src/lib/WarehouseQueryService.ts index 94e80bf0c..3b280626e 100644 --- a/apps/api/src/lib/WarehouseQueryService.ts +++ b/apps/api/src/lib/WarehouseQueryService.ts @@ -6,8 +6,10 @@ import { makeWarehouseExecutor, toWarehouseQueryError, WarehouseResponseLimitError, + type ClickHouseProtocolBackendConfig, type ResolvedWarehouseConfig, type SqlQueryOptions, + type TinybirdBackendConfig, type WarehouseExecutorDeps, type WarehouseQueryServiceShape, type WarehouseSqlClient, @@ -33,10 +35,7 @@ import { TinybirdOrgTokenService } from "../services/TinybirdOrgTokenService" // Re-export the executor types so existing import sites stay stable. export type { WarehouseQueryServiceShape, SqlQueryOptions } -type ClickHouseConfig = Extract -type TinybirdConfig = Extract - -const createClickHouseSqlClient = (config: ClickHouseConfig): WarehouseSqlClient => { +const createClickHouseSqlClient = (config: ClickHouseProtocolBackendConfig): WarehouseSqlClient => { const client = createClickHouseClient({ url: config.url, username: config.username, @@ -146,7 +145,7 @@ const boundedResponseFetch = }) } -const createTinybirdSdkSqlClient = (config: TinybirdConfig): WarehouseSqlClient => { +const createTinybirdSdkSqlClient = (config: TinybirdBackendConfig): WarehouseSqlClient => { const makeClient = (fetchAdapter?: typeof fetch) => new Tinybird({ baseUrl: config.host, @@ -210,8 +209,11 @@ const createTinybirdSdkSqlClient = (config: TinybirdConfig): WarehouseSqlClient } } +// Driver selection follows `BackendDialect[kind].driver`: the `tinybird` kind is +// the only one on the SDK; every ClickHouse-protocol kind (gateway, BYO/vanilla +// CH, chdb) uses the official web client. const createClient = (config: ResolvedWarehouseConfig): WarehouseSqlClient => - config._tag === "clickhouse" ? createClickHouseSqlClient(config) : createTinybirdSdkSqlClient(config) + config.kind === "tinybird" ? createTinybirdSdkSqlClient(config) : createClickHouseSqlClient(config) let sqlClientFactory: typeof createClient = createClient @@ -224,18 +226,8 @@ export class WarehouseQueryService extends Context.Service< const orgClickHouseSettings = yield* OrgClickHouseSettingsService const orgTokens = yield* TinybirdOrgTokenService - /** - * Resolve the upstream config for this tenant's queries. - * - * Resolution order: - * 1. Per-org BYO ClickHouse row (`org_clickhouse_settings`) - * 2. Env-level managed ClickHouse (`CLICKHOUSE_URL` set) - * 3. Env-level managed Tinybird (`TINYBIRD_HOST` + `TINYBIRD_TOKEN`) - */ - // The managed (env-level) upstream: ClickHouse when CLICKHOUSE_URL is set, - // otherwise the managed Tinybird pipeline. This is the canonical WRITE - // target — demo-seed, service-map rollups and alert-check inserts all land - // here — and the read-path fallback when an org has no BYO override. + // The managed (env-level) READ upstream: the Tinybird CH-gateway or vanilla + // ClickHouse when CLICKHOUSE_URL is set, otherwise the managed Tinybird SDK. const resolveManagedConfig = Effect.fn("WarehouseQueryService.resolveManagedConfig")(function* () { if (Option.isSome(env.CLICKHOUSE_URL)) { const configuredUrl = env.CLICKHOUSE_URL.value @@ -254,16 +246,17 @@ export class WarehouseQueryService extends Context.Service< }) } yield* Effect.annotateCurrentSpan("db.client", "clickhouse") - const provider: "tinybird" | "clickhouse" = - env.CLICKHOUSE_PROVIDER === "tinybird" ? "tinybird" : "clickhouse" + const kind = + env.CLICKHOUSE_PROVIDER === "tinybird" + ? ("tinybird-gateway" as const) + : ("clickhouse" as const) return { config: { - _tag: "clickhouse" as const, - provider, + kind, url: clickhouseUrl.toString().replace(/\/$/, ""), username: env.CLICKHOUSE_USER, password: Option.match(env.CLICKHOUSE_PASSWORD, { - onNone: () => (provider === "tinybird" ? Redacted.value(env.TINYBIRD_TOKEN) : ""), + onNone: () => (kind === "tinybird-gateway" ? Redacted.value(env.TINYBIRD_TOKEN) : ""), onSome: Redacted.value, }), database: env.CLICKHOUSE_DATABASE, @@ -275,8 +268,7 @@ export class WarehouseQueryService extends Context.Service< yield* Effect.annotateCurrentSpan("db.client", "tinybird-sdk") return { config: { - _tag: "tinybird" as const, - provider: "tinybird" as const, + kind: "tinybird" as const, host: env.TINYBIRD_HOST, token: Redacted.value(env.TINYBIRD_TOKEN), }, @@ -285,51 +277,88 @@ export class WarehouseQueryService extends Context.Service< }) /** - * Read-path config. A per-org BYO ClickHouse row (`org_clickhouse_settings`) - * overrides the managed upstream for that org's queries; otherwise we fall - * back to the managed config. + * The single routing decision: purpose → backend + credentials. + * + * ingest → managed Tinybird Events API, always (source: managed) + * read → org BYO row? that org's ClickHouse (source: org-byo) + * else env: tinybird-gateway|clickhouse|tinybird (source: managed) + * raw → org BYO row? that org's ClickHouse (source: org-byo) + * managed tinybird/gateway? org-scoped JWT (source: org-jwt) + * managed vanilla CH? self-hosted mode only (source: managed) + * + * Ingest notes (why writes NEVER follow the read routing): + * - BILLING: this path bypasses the ingest gateway, where Autumn usage + * metering happens. Today's only `ingest` callers — demo seed, service-map + * rollups, alert checks — are derived/internal/demo data and deliberately + * unmetered. Net-new *customer* telemetry must go through the ingest + * gateway (as the Cloudflare edge-metrics poller does) so it is metered. + * - When CLICKHOUSE_URL is set, the managed READ backend is a read-only + * query gateway that rejects inserts ("Only SELECT or DESCRIBE queries + * are supported. Got: InsertQuery"), and a per-org BYO override is a read + * concern. Tinybird is the only writable warehouse (TINYBIRD_HOST/TOKEN + * are required env). Routing writes anywhere else broke demo-seed + * onboarding. + * + * Raw-SQL isolation invariants (defense-in-depth, preserved verbatim): + * BYO creds are already tenant-isolated; shared Tinybird gets a + * datasource-scoped org JWT (works through both the SDK and the CH + * gateway); a shared vanilla ClickHouse credential has no DB-enforced OrgId + * scope, so raw SQL there is allowed only in single-org self-hosted mode. */ - const resolveConfig: WarehouseExecutorDeps["resolveConfig"] = Effect.fn( - "WarehouseQueryService.resolveSqlConfig", - )(function* (tenant, label) { + const resolveRoute: WarehouseExecutorDeps["resolveRoute"] = Effect.fn( + "WarehouseQueryService.resolveRoute", + )(function* (tenant, purpose, label) { + yield* Effect.annotateCurrentSpan("orgId", tenant.orgId) + yield* Effect.annotateCurrentSpan("warehouse.route", purpose) + + if (purpose === "ingest") { + // Legacy attrs, dual-emitted until dashboards move to `warehouse.*`. + yield* Effect.annotateCurrentSpan("clientSource", "managed") + yield* Effect.annotateCurrentSpan("query.routing", "ingest") + yield* Effect.annotateCurrentSpan("db.client", "tinybird-sdk") + return { + source: "managed" as const, + config: { + kind: "tinybird" as const, + host: env.TINYBIRD_HOST, + token: Redacted.value(env.TINYBIRD_TOKEN), + }, + clientCacheKey: "write:managed", + } + } + + // A per-org BYO ClickHouse row (`org_clickhouse_settings`) overrides the + // managed upstream for that org's reads AND raw SQL (the credentials are + // already tenant-isolated). const override = yield* orgClickHouseSettings .resolveRuntimeConfig(tenant.orgId) .pipe(Effect.mapError((error) => toWarehouseQueryError(label, error))) - if (Option.isSome(override)) { yield* Effect.annotateCurrentSpan("clientSource", "org_override") yield* Effect.annotateCurrentSpan("db.client", "clickhouse") return { + source: "org-byo" as const, config: { - _tag: "clickhouse" as const, - provider: "clickhouse" as const, + kind: "clickhouse" as const, url: override.value.url, username: override.value.user, password: override.value.password, database: override.value.database, }, - clientCacheKey: `read:${tenant.orgId}`, + clientCacheKey: + purpose === "raw" ? `raw:${tenant.orgId}` : `read:${tenant.orgId}`, } } yield* Effect.annotateCurrentSpan("clientSource", "managed") - return yield* resolveManagedConfig() - }) + const managed = yield* resolveManagedConfig() + if (purpose === "read") return { source: "managed" as const, ...managed } - const resolveRawSqlConfig: WarehouseExecutorDeps["resolveRawSqlConfig"] = Effect.fn( - "WarehouseQueryService.resolveRawSqlConfig", - )(function* (tenant, label) { - const resolved = yield* resolveConfig(tenant, label) + // Raw SQL on the shared warehouse needs tenant isolation. Shared Tinybird + // is isolated with a datasource-scoped JWT; the same token works through + // both the SDK and Tinybird's ClickHouse-compatible gateway. const clientCacheKey = `raw:${tenant.orgId}` - - // Per-org BYO ClickHouse credentials already isolate the warehouse. - if (resolved.clientCacheKey !== "read:managed") { - return { ...resolved, clientCacheKey } - } - - // Shared Tinybird is isolated with a datasource-scoped JWT. The same token - // works through both the SDK and Tinybird's ClickHouse-compatible gateway. - if (resolved.config.provider === "tinybird") { + if (managed.config.kind === "tinybird" || managed.config.kind === "tinybird-gateway") { const jwt = yield* orgTokens.getOrgReadToken(tenant.orgId).pipe( Effect.mapError( (error) => @@ -342,10 +371,11 @@ export class WarehouseQueryService extends Context.Service< ) yield* Effect.annotateCurrentSpan("maple.tinybird.token.scope", "org_jwt") return { + source: "org-jwt" as const, config: - resolved.config._tag === "tinybird" - ? { ...resolved.config, token: jwt } - : { ...resolved.config, password: jwt }, + managed.config.kind === "tinybird" + ? { ...managed.config, token: jwt } + : { ...managed.config, password: jwt }, clientCacheKey, } } @@ -359,50 +389,12 @@ export class WarehouseQueryService extends Context.Service< "Raw SQL on managed vanilla ClickHouse is available only in single-org self-hosted mode", }) } - return { ...resolved, clientCacheKey } - }) - - /** - * Write-path config. Inserts (demo seed, service-map rollups, alert checks) - * ALWAYS go to the Tinybird ingest pipeline (Events API) — never ClickHouse. - * - * BILLING: this path bypasses the ingest gateway, which is where Autumn usage - * metering happens. Today's only `ingest` callers — demo seed, service-map - * rollups, alert checks — are all derived/internal/demo data and deliberately - * unmetered. Net-new *customer* telemetry must go through the ingest gateway - * (as the Cloudflare edge-metrics poller does) so it is metered, not direct `ingest`. - * - * This is deliberately NOT `resolveManagedConfig()`: when `CLICKHOUSE_URL` is - * set, the *managed* backend is a READ-ONLY ClickHouse query gateway that - * rejects inserts ("DB::Exception: Only SELECT or DESCRIBE queries are - * supported. Got: InsertQuery"). A per-org BYO ClickHouse override is also a - * read concern. Tinybird is the only writable warehouse, so ingest pins to - * it unconditionally (TINYBIRD_HOST/TOKEN are required env, always present). - * Routing writes anywhere else broke demo-seed onboarding. - */ - const resolveIngestConfig: NonNullable = Effect.fn( - "WarehouseQueryService.resolveIngestConfig", - )(function* (tenant, _label) { - yield* Effect.annotateCurrentSpan("orgId", tenant.orgId) - yield* Effect.annotateCurrentSpan("clientSource", "managed") - yield* Effect.annotateCurrentSpan("query.routing", "ingest") - yield* Effect.annotateCurrentSpan("db.client", "tinybird-sdk") - return { - config: { - _tag: "tinybird" as const, - provider: "tinybird" as const, - host: env.TINYBIRD_HOST, - token: Redacted.value(env.TINYBIRD_TOKEN), - }, - clientCacheKey: "write:managed", - } + return { source: "managed" as const, config: managed.config, clientCacheKey } }) return makeWarehouseExecutor({ createClient: (config) => sqlClientFactory(config), - resolveConfig, - resolveRawSqlConfig, - resolveIngestConfig, + resolveRoute, }) }), }) { diff --git a/apps/api/src/routes/observability.http.ts b/apps/api/src/routes/observability.http.ts index ae4e6f429..165ed7ed7 100644 --- a/apps/api/src/routes/observability.http.ts +++ b/apps/api/src/routes/observability.http.ts @@ -1,6 +1,5 @@ import { HttpApiBuilder } from "effect/unstable/httpapi" import { CurrentTenant, MapleApi, SpanId, TraceId } from "@maple/domain/http" -import { ObservabilityApiError } from "@maple/domain/http/observability" import { Effect, Schema } from "effect" import { listServices, @@ -9,12 +8,12 @@ import { findErrors, diagnoseService, searchLogs, - type WarehouseExecutorError, } from "@maple/query-engine/observability" import { makeWarehouseExecutorFromTenant } from "../lib/WarehouseQueryService" -const mapError = (e: WarehouseExecutorError) => - new ObservabilityApiError({ message: e.message, pipeName: e.pipeName, cause: e }) +// Warehouse errors propagate with their canonical per-tag HTTP statuses (503 +// transient, 429 quota, 400 validation, 502 otherwise) — declared on the +// observability endpoints via `warehouseHttpErrors`. No 500-flatten here. const decodeTraceId = Schema.decodeSync(TraceId) const decodeSpanId = Schema.decodeSync(SpanId) @@ -33,7 +32,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context const services = yield* listServices(payload).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), ) return { services: [...services] } }), @@ -43,7 +41,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context return yield* searchTraces(payload).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), Effect.map((r) => ({ ...r, spans: [...r.spans].map((s) => ({ ...s, attributes: { ...s.attributes } })), @@ -56,7 +53,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context const result = yield* inspectTrace(payload.traceId).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), ) return { traceId: result.traceId, @@ -73,7 +69,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context const errors = yield* findErrors(payload).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), ) return { errors: [...errors] } }), @@ -83,7 +78,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context const result = yield* diagnoseService(payload).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), ) return { serviceName: result.serviceName, @@ -103,7 +97,6 @@ export const HttpObservabilityLive = HttpApiBuilder.group(MapleApi, "observabili const tenant = yield* CurrentTenant.Context const result = yield* searchLogs(payload).pipe( Effect.provide(makeWarehouseExecutorFromTenant(tenant)), - Effect.mapError(mapError), ) return { timeRange: result.timeRange, diff --git a/apps/api/src/routes/v2/alerts-error-map.ts b/apps/api/src/routes/v2/alerts-error-map.ts index fe1bf2980..84fbf4cbb 100644 --- a/apps/api/src/routes/v2/alerts-error-map.ts +++ b/apps/api/src/routes/v2/alerts-error-map.ts @@ -6,12 +6,15 @@ import type { AlertPersistenceError, AlertValidationError, WarehouseError, + WarehouseQuotaExceededError, + WarehouseValidationError, } from "@maple/domain/http" import { conflict, dependencyUnavailable, invalidRequest, permissionError, + rateLimited, resourceNotFound, upstreamError, } from "@maple/domain/http/v2" @@ -20,6 +23,7 @@ import type { V2InvalidRequestError, V2NotFoundError, V2PermissionError, + V2RateLimitError, V2ServiceUnavailableError, V2UpstreamError, } from "@maple/domain/http/v2" @@ -34,12 +38,18 @@ type V2ReachableAlertError = | AlertDeliveryError | WarehouseError +type UpstreamMappedWarehouseError = Exclude< + WarehouseError, + WarehouseValidationError | WarehouseQuotaExceededError +> + type MappedV2AlertError = | (E extends AlertForbiddenError ? V2PermissionError : never) - | (E extends AlertValidationError ? V2InvalidRequestError : never) + | (E extends AlertValidationError | WarehouseValidationError ? V2InvalidRequestError : never) | (E extends AlertNotFoundError ? V2NotFoundError : never) | (E extends AlertDestinationInUseError ? V2ConflictError : never) - | (E extends AlertDeliveryError | WarehouseError ? V2UpstreamError : never) + | (E extends WarehouseQuotaExceededError ? V2RateLimitError : never) + | (E extends AlertDeliveryError | UpstreamMappedWarehouseError ? V2UpstreamError : never) | (E extends AlertPersistenceError ? V2ServiceUnavailableError : never) const normalizeAlertResourceType = (resourceType: string) => @@ -82,8 +92,12 @@ const makeAlertErrorMatcher = (operation: string) => { "@maple/http/errors/WarehouseConfigError": warehouseFailure, "@maple/http/errors/WarehouseClientError": warehouseFailure, "@maple/http/errors/WarehouseSchemaDriftError": warehouseFailure, - "@maple/http/errors/WarehouseQuotaExceededError": warehouseFailure, - "@maple/http/errors/WarehouseValidationError": warehouseFailure, + // A quota breach is the caller exceeding cost limits (429), and a + // validation failure is a malformed request (400) — neither is an + // upstream outage. + "@maple/http/errors/WarehouseQuotaExceededError": () => rateLimited(), + "@maple/http/errors/WarehouseValidationError": (error) => + invalidRequest("parameter_invalid", error.message), }), ) } diff --git a/apps/api/src/services/AlertsService.ts b/apps/api/src/services/AlertsService.ts index 2325ef9f4..503982f65 100644 --- a/apps/api/src/services/AlertsService.ts +++ b/apps/api/src/services/AlertsService.ts @@ -3190,13 +3190,11 @@ export class AlertsService extends Context.Service } @@ -1419,9 +1419,9 @@ describe("CloudflareAnalyticsService", () => { assert.strictEqual(worker.totalRequests, 42) // This org has no BYO-CH settings row (managed/Tinybird), so the gateway wrote its - // metrics to Tinybird — the usage read must pin to the ingest config to find them. + // metrics to Tinybird — the usage read must route to the ingest config to find them. assert.strictEqual(queryStub.calls.length, 1) - assert.strictEqual(queryStub.calls[0]?.options?.pinToIngestConfig, true) + assert.strictEqual(queryStub.calls[0]?.options?.route, "ingest") assert.strictEqual(queryStub.calls[0]?.options?.profile, "aggregation") assert.strictEqual(queryStub.calls[0]?.orgId, ORG) }).pipe(Effect.provide(makeLayer(testDb, captured, {}, {}, queryStub))) @@ -1435,13 +1435,13 @@ describe("CloudflareAnalyticsService", () => { yield* TestClock.setTime(T0) yield* seedConnection() // Connected + schema_version == running version → gateway routes metrics to the - // org's own ClickHouse, so the read must NOT pin to the ingest (Tinybird) config. + // org's own ClickHouse, so the read must NOT route to the ingest (Tinybird) config. yield* seedByoClickHouse() const service = yield* CloudflareAnalyticsService yield* service.getUsage(ORG) assert.strictEqual(queryStub.calls.length, 1) - assert.notStrictEqual(queryStub.calls[0]?.options?.pinToIngestConfig, true) + assert.notStrictEqual(queryStub.calls[0]?.options?.route, "ingest") }).pipe(Effect.provide(makeLayer(testDb, captured, {}, {}, queryStub))) }) @@ -1453,13 +1453,13 @@ describe("CloudflareAnalyticsService", () => { yield* TestClock.setTime(T0) yield* seedConnection() // Stale schema_version → gateway falls back to Tinybird for this org's metrics, - // so the read must pin to the ingest config (its own CH would be empty). + // so the read must route to the ingest config (its own CH would be empty). yield* seedByoClickHouse({ schemaVersion: "stale-schema-version" }) const service = yield* CloudflareAnalyticsService yield* service.getUsage(ORG) assert.strictEqual(queryStub.calls.length, 1) - assert.strictEqual(queryStub.calls[0]?.options?.pinToIngestConfig, true) + assert.strictEqual(queryStub.calls[0]?.options?.route, "ingest") }).pipe(Effect.provide(makeLayer(testDb, captured, {}, {}, queryStub))) }) diff --git a/apps/api/src/services/CloudflareAnalyticsService.ts b/apps/api/src/services/CloudflareAnalyticsService.ts index cc1da9bca..a783ad42d 100644 --- a/apps/api/src/services/CloudflareAnalyticsService.ts +++ b/apps/api/src/services/CloudflareAnalyticsService.ts @@ -2104,7 +2104,7 @@ export class CloudflareAnalyticsService extends Context.Service< .compiledQuery(systemTenant(orgId), compiled, { profile: "aggregation", context: "cloudflareUsage", - pinToIngestConfig: !clickHouseReady, + ...(clickHouseReady ? {} : { route: "ingest" as const }), }) .pipe( Effect.mapError( diff --git a/apps/api/src/services/ErrorsService.ts b/apps/api/src/services/ErrorsService.ts index 6c30b028b..03afa84e7 100644 --- a/apps/api/src/services/ErrorsService.ts +++ b/apps/api/src/services/ErrorsService.ts @@ -473,7 +473,6 @@ const make: Effect.Effect< }) return yield* warehouse .compiledQuery(systemTenant(knownOrgs[0] as OrgId), compiled, { - pinToIngestConfig: true, // Bound the one cross-org scan (no OrgId predicate ⇒ can't prune the // primary key): abort server-side at 5s instead of riding the ~30s // client timeout when the warehouse is slow. diff --git a/apps/cli/package.json b/apps/cli/package.json index a865f4ebb..ed85f0233 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -7,7 +7,7 @@ "type": "module", "scripts": { "start": "bun run src/bin.ts", - "test": "bun test test/*.test.ts src/server/otlp/encode.test.ts", + "test": "bun test test/*.test.ts src/server/otlp/encode.test.ts src/core/executor.test.ts", "typecheck": "tsc --noEmit" }, "dependencies": { diff --git a/apps/cli/src/core/executor.test.ts b/apps/cli/src/core/executor.test.ts new file mode 100644 index 000000000..67026fb57 --- /dev/null +++ b/apps/cli/src/core/executor.test.ts @@ -0,0 +1,78 @@ +import { afterEach, describe, expect, it } from "vitest" +import { Effect } from "effect" +import { unsafeCompiledQuery } from "@maple/query-engine/ch" +import { makeLocalWarehouseExecutorShape } from "./executor" + +// The local executor is the REAL makeWarehouseExecutor wired to the `chdb` +// backend — these tests pin the wiring: SQL normalization for the local +// server, row flow, and the OrgId scoping guard. + +const realFetch = globalThis.fetch + +afterEach(() => { + globalThis.fetch = realFetch +}) + +const stubFetch = (handler: (url: string, init?: RequestInit) => Response) => { + globalThis.fetch = (async (input: string | URL | Request, init?: RequestInit) => + handler(String(input), init)) as unknown as typeof fetch +} + +const stubLocalServer = (rows: ReadonlyArray>) => { + const requests: Array<{ url: string; sql: string }> = [] + stubFetch((url, init) => { + const body = JSON.parse(String(init?.body ?? "{}")) as { sql?: string } + requests.push({ url, sql: body.sql ?? "" }) + return new Response(JSON.stringify(rows), { + status: 200, + headers: { "content-type": "application/json" }, + }) + }) + return requests +} + +describe("makeLocalWarehouseExecutorShape", () => { + it("posts compiled SQL to /local/query with the trailing FORMAT stripped (chdb dialect)", async () => { + const requests = stubLocalServer([{ c: 1 }]) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + const compiled = unsafeCompiledQuery<{ readonly c: number }>({ + sql: "SELECT count() AS c FROM traces WHERE OrgId = 'local'\nFORMAT JSON", + }) + + const rows = await Effect.runPromise(shape.compiledQuery(compiled)) + + expect(rows).toEqual([{ c: 1 }]) + expect(requests).toHaveLength(1) + expect(requests[0]!.url).toBe("http://127.0.0.1:4318/local/query") + // The chdb backend speaks the ClickHouse protocol shape: the executor + // strips the trailing FORMAT and the local server owns the output format. + expect(requests[0]!.sql).not.toContain("FORMAT JSON") + expect(requests[0]!.sql).toContain("OrgId = 'local'") + }) + + it("keeps the executor's OrgId scoping guard for trusted SQL", async () => { + stubLocalServer([]) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + + const exit = await Effect.runPromiseExit(shape.sqlQuery("SELECT 1")) + + expect(exit._tag).toBe("Failure") + }) + + it("classifies a local 400 query failure without retrying it", async () => { + let attempts = 0 + stubFetch(() => { + attempts += 1 + return new Response("query failed: Unknown expression identifier", { status: 400 }) + }) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + + const exit = await Effect.runPromiseExit( + shape.sqlQuery("SELECT nope FROM traces WHERE OrgId = 'local'"), + ) + + expect(exit._tag).toBe("Failure") + // 400 is a non-transient client-side failure — exactly one attempt. + expect(attempts).toBe(1) + }) +}) diff --git a/apps/cli/src/core/executor.ts b/apps/cli/src/core/executor.ts index e3f55eb97..75c896771 100644 --- a/apps/cli/src/core/executor.ts +++ b/apps/cli/src/core/executor.ts @@ -1,191 +1,62 @@ import { Effect, Schema } from "effect" -import { compilePipeQuery } from "@maple/query-engine/ch" -import { type WarehouseExecutorShape, type ExecutorQueryOptions } from "@maple/query-engine/observability" import { OrgId } from "@maple/domain/http" -import { - WarehouseQueryError, - WarehouseSchemaDriftError, - WarehouseValidationError, -} from "@maple/domain/http/warehouse-errors" +import { makeWarehouseExecutor, type WarehouseSqlClient } from "@maple/query-engine/execution" +import type { WarehouseExecutorShape } from "@maple/query-engine/observability" import { executeLocalQuery } from "@maple/query-engine/local" import { debugLog } from "../lib/debug" -// Local mode is single-tenant: the Rust binary writes every row under this +// Local mode is single-tenant: the local binary writes every row under this // OrgId, and every compiled query filters on it. `OrgId` is a non-empty trimmed // branded string, so "local" decodes cleanly (no cast needed). const LOCAL_ORG_ID = Schema.decodeUnknownSync(OrgId)("local") -const toWarehouseError = (pipe: string) => (error: unknown) => - new WarehouseQueryError({ - message: error instanceof Error ? error.message : String(error), - pipeName: pipe, - }) +const LOCAL_TENANT = { orgId: LOCAL_ORG_ID, userId: "local", authMode: "local" } -// Cap `db.query.text` at 16 KB to match apps/api's WarehouseQueryService span. -const MAX_DB_QUERY_TEXT = 16 * 1024 -const truncateSql = (sql: string) => (sql.length > MAX_DB_QUERY_TEXT ? sql.slice(0, MAX_DB_QUERY_TEXT) : sql) +// The chDB driver: POST `/local/query` on the local binary. Runs the SQL, +// timing the round-trip and (under --debug) printing the SQL + elapsed ms to +// stderr. The `finally` logs even on failure so a failing query still shows +// its SQL. +const localChdbClient = (baseUrl: string): WarehouseSqlClient => ({ + sql: async (sql) => { + const started = performance.now() + try { + return { data: await executeLocalQuery>(sql, baseUrl) } + } finally { + debugLog(`local query · ${Math.round(performance.now() - started)}ms`, sql) + } + }, + insert: async () => { + // Local mode ingests via OTLP into the embedded chDB, never through the + // warehouse `ingest` path. + throw new Error("local mode is read-only through the warehouse executor — ingest via OTLP") + }, +}) /** - * A `WarehouseExecutor` shape backed by the local Maple binary's `/local/query` - * endpoint. Both executor methods reduce to raw SQL against the embedded chDB: - * - * - `sqlQuery` posts the SQL directly. - * - `query(pipe, params)` compiles the pipe name to SQL via the shared - * `compilePipeQuery` dispatcher (the same one the cloud uses), then posts it. + * A `WarehouseExecutor` backed by the local Maple binary's `/local/query` + * endpoint — the REAL `makeWarehouseExecutor` from `@maple/query-engine` + * (spans, error classification, OrgId scoping) with a chDB client and a + * constant single-tenant route injected. The `chdb` backend dialect strips the + * trailing `FORMAT` (the local server owns the output format) and skips + * Tinybird's restricted-settings policy. * * This makes every `@maple/query-engine/observability` function — which only - * depend on a `WarehouseExecutor` — work unchanged against local mode. + * depends on a `WarehouseExecutor` — work unchanged against local mode, with + * the same `warehouse.backend="chdb"` span contract as the cloud. */ -export const makeLocalWarehouseExecutorShape = (baseUrl: string): WarehouseExecutorShape => { - // Run SQL against the local server, timing the round-trip and (under --debug) - // printing the SQL + elapsed ms to stderr. The `finally` logs even on failure - // so a failing query still shows its SQL. - const exec = async (sql: string, label: string): Promise> => { - const started = performance.now() - try { - return await executeLocalQuery(sql, baseUrl) - } finally { - debugLog(`${label} · ${Math.round(performance.now() - started)}ms`, sql) - } - } - return { - orgId: LOCAL_ORG_ID, - sqlQuery: >(sql: string, options?: ExecutorQueryOptions) => - Effect.gen(function* () { - const started = performance.now() - const rows = yield* Effect.tryPromise({ - try: () => exec(sql, "sqlQuery"), - catch: toWarehouseError("sqlQuery"), - }) - yield* Effect.annotateCurrentSpan({ - "db.duration_ms": Math.round(performance.now() - started), - "result.rowCount": rows.length, - }) - return rows - }).pipe( - Effect.withSpan("warehouse.sqlQuery", { - kind: "client", - attributes: { - "db.system.name": "clickhouse", - "peer.service": "chdb", - "db.query.text": truncateSql(sql), - "db.query.length": sql.length, - "query.context": "sqlQuery", - ...(options?.profile ? { "query.profile": options.profile } : {}), - }, - }), - ), - compiledQuery: (compiled, options?: ExecutorQueryOptions) => - Effect.gen(function* () { - const started = performance.now() - const rows = yield* Effect.tryPromise({ - try: () => exec>(compiled.sql, "compiledQuery"), - catch: toWarehouseError("compiledQuery"), - }) - yield* Effect.annotateCurrentSpan({ - "db.duration_ms": Math.round(performance.now() - started), - "result.rowCount": rows.length, - }) - return yield* compiled.decodeRows(rows).pipe( - Effect.mapError( - (error) => - new WarehouseSchemaDriftError({ - message: error.message, - pipeName: "compiledQuery", - cause: error, - }), - ), - ) - }).pipe( - Effect.withSpan("warehouse.compiledQuery", { - kind: "client", - attributes: { - "db.system.name": "clickhouse", - "peer.service": "chdb", - "db.query.text": truncateSql(compiled.sql), - "db.query.length": compiled.sql.length, - "query.context": "compiledQuery", - ...(options?.profile ? { "query.profile": options.profile } : {}), - }, - }), - ), - compiledQueryFirst: (compiled, options?: ExecutorQueryOptions) => - Effect.gen(function* () { - const started = performance.now() - const rows = yield* Effect.tryPromise({ - try: () => exec>(compiled.sql, "compiledQueryFirst"), - catch: toWarehouseError("compiledQueryFirst"), - }) - yield* Effect.annotateCurrentSpan({ - "db.duration_ms": Math.round(performance.now() - started), - "result.rowCount": rows.length, - }) - return yield* compiled.decodeFirstRow(rows).pipe( - Effect.mapError( - (error) => - new WarehouseSchemaDriftError({ - message: error.message, - pipeName: "compiledQueryFirst", - cause: error, - }), - ), - ) - }).pipe( - Effect.withSpan("warehouse.compiledQueryFirst", { - kind: "client", - attributes: { - "db.system.name": "clickhouse", - "peer.service": "chdb", - "db.query.text": truncateSql(compiled.sql), - "db.query.length": compiled.sql.length, - "query.context": "compiledQueryFirst", - ...(options?.profile ? { "query.profile": options.profile } : {}), - }, - }), - ), - query: (pipe: string, params: Record, _options?: ExecutorQueryOptions) => - Effect.gen(function* () { - const compiled = compilePipeQuery(pipe, { ...params, org_id: LOCAL_ORG_ID }) - if (!compiled) { - return yield* new WarehouseValidationError({ - message: `Unsupported pipe in local mode: ${pipe}`, - pipeName: pipe, - }) - } - yield* Effect.annotateCurrentSpan({ - "db.query.text": truncateSql(compiled.sql), - "db.query.length": compiled.sql.length, - }) - const started = performance.now() - const rows = yield* Effect.tryPromise({ - try: () => exec>(compiled.sql, pipe), - catch: toWarehouseError(pipe), - }) - yield* Effect.annotateCurrentSpan({ - "db.duration_ms": Math.round(performance.now() - started), - "result.rowCount": rows.length, - }) - const decodedRows = yield* compiled.decodeRows(rows).pipe( - Effect.mapError( - (error) => - new WarehouseSchemaDriftError({ - message: error.message, - pipeName: pipe, - cause: error, - }), - ), - ) - // Type-erased executor boundary — mirrors WarehouseQueryService.asExecutor in apps/api. - return { data: decodedRows as ReadonlyArray } - }).pipe( - Effect.withSpan("warehouse.query", { - kind: "client", - attributes: { - "db.system.name": "clickhouse", - "peer.service": "chdb", - "query.context": pipe, - }, - }), - ), - } -} +export const makeLocalWarehouseExecutorShape = (baseUrl: string): WarehouseExecutorShape => + makeWarehouseExecutor({ + createClient: () => localChdbClient(baseUrl), + resolveRoute: () => + Effect.succeed({ + source: "managed" as const, + config: { + kind: "chdb" as const, + url: baseUrl, + username: "", + password: "", + database: "default", + }, + clientCacheKey: "local", + }), + }).asExecutor(LOCAL_TENANT) diff --git a/apps/cli/src/core/operations.ts b/apps/cli/src/core/operations.ts index 6e1ff608d..9c2dd28e6 100644 --- a/apps/cli/src/core/operations.ts +++ b/apps/cli/src/core/operations.ts @@ -21,6 +21,9 @@ import { findSlowTraces as obsFindSlowTraces, topOperations as obsTopOperations, } from "@maple/query-engine/observability" +import { WarehouseClientError, WarehouseQueryError } from "@maple/domain/http/warehouse-errors" +import { executeLocalQuery } from "@maple/query-engine/local" +import { Mode } from "./mode" import type { Range } from "./time" type AttrSource = "traces" | "metrics" | "services" @@ -166,11 +169,34 @@ export const listMetrics = (p: { range: Range; service?: string; search?: string return result.data }) -/** Raw SQL escape hatch against the local chDB store. */ +/** + * Raw SQL escape hatch against the local chDB store — local mode only. + * Arbitrary user SQL carries no OrgId guarantee, so it deliberately bypasses + * the warehouse executor (whose `sqlQuery` enforces the OrgId scoping guard) + * and posts straight to the single-tenant `/local/query` endpoint. + */ export const rawQuery = (sql: string) => Effect.gen(function* () { - const executor = yield* WarehouseExecutor - return yield* executor.sqlQuery(sql) + const mode = yield* Mode + const resolved = yield* mode.resolve.pipe( + Effect.mapError( + (error) => new WarehouseClientError({ message: error.message, pipeName: "rawQuery" }), + ), + ) + if (resolved._tag !== "local") { + return yield* new WarehouseClientError({ + message: "Raw SQL is only available in local mode", + pipeName: "rawQuery", + }) + } + return yield* Effect.tryPromise({ + try: () => executeLocalQuery>(sql, resolved.baseUrl), + catch: (error) => + new WarehouseQueryError({ + message: error instanceof Error ? error.message : String(error), + pipeName: "rawQuery", + }), + }) }) // Custom traces analytics share the `group_by_*` presence-flag convention the diff --git a/apps/cli/src/core/remote-executor.ts b/apps/cli/src/core/remote-executor.ts index cc966b323..34aa1c9b9 100644 --- a/apps/cli/src/core/remote-executor.ts +++ b/apps/cli/src/core/remote-executor.ts @@ -1,5 +1,5 @@ import { Effect, type Option } from "effect" -import { type WarehouseExecutorShape, type ExecutorQueryOptions } from "@maple/query-engine/observability" +import { type WarehouseExecutorShape, type SqlQueryOptions } from "@maple/query-engine/observability" import { WarehouseClientError, WarehouseQueryError } from "@maple/domain/http/warehouse-errors" import { debugLog } from "../lib/debug" @@ -28,7 +28,7 @@ export const makeRemoteWarehouseExecutorShape = ( const endpoint = `${apiUrl.replace(/\/$/, "")}/api/tinybird/query` return { orgId, - query: (pipe: string, params: Record, _options?: ExecutorQueryOptions) => + query: (pipe: string, params: Record, _options?: SqlQueryOptions) => Effect.tryPromise({ try: async (): Promise<{ data: ReadonlyArray }> => { const started = performance.now() @@ -71,15 +71,15 @@ export const makeRemoteWarehouseExecutorShape = ( }, }), ), - sqlQuery: >(_sql: string, _options?: ExecutorQueryOptions) => + sqlQuery: >(_sql: string, _options?: SqlQueryOptions) => Effect.fail( new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "sqlQuery" }), ) as Effect.Effect, WarehouseClientError>, - compiledQuery: (_compiled: unknown, _options?: ExecutorQueryOptions) => + compiledQuery: (_compiled: unknown, _options?: SqlQueryOptions) => Effect.fail( new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "compiledQuery" }), ) as Effect.Effect, WarehouseClientError>, - compiledQueryFirst: (_compiled: unknown, _options?: ExecutorQueryOptions) => + compiledQueryFirst: (_compiled: unknown, _options?: SqlQueryOptions) => Effect.fail( new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "compiledQueryFirst" }), ) as Effect.Effect, WarehouseClientError>, diff --git a/apps/cli/src/core/warehouse.ts b/apps/cli/src/core/warehouse.ts index 62bd0f2c7..679746134 100644 --- a/apps/cli/src/core/warehouse.ts +++ b/apps/cli/src/core/warehouse.ts @@ -2,7 +2,7 @@ import { Effect, Layer } from "effect" import { WarehouseExecutor, type WarehouseExecutorShape, - type ExecutorQueryOptions, + type SqlQueryOptions, } from "@maple/query-engine/observability" import { WarehouseConfigError } from "@maple/domain/http/warehouse-errors" import type { WarehouseQueryName } from "@maple/domain/warehouse-queries" @@ -42,15 +42,15 @@ export const WarehouseExecutorFromMode = Layer.effect( ) return WarehouseExecutor.of({ orgId: "", - query: (pipe: string, params: Record, options?: ExecutorQueryOptions) => + query: (pipe: string, params: Record, options?: SqlQueryOptions) => getShape.pipe( Effect.flatMap((shape) => shape.query(pipe as WarehouseQueryName, params, options)), ), - sqlQuery: >(sql: string, options?: ExecutorQueryOptions) => + sqlQuery: >(sql: string, options?: SqlQueryOptions) => getShape.pipe(Effect.flatMap((shape) => shape.sqlQuery(sql, options))), - compiledQuery: (compiled, options?: ExecutorQueryOptions) => + compiledQuery: (compiled, options?: SqlQueryOptions) => getShape.pipe(Effect.flatMap((shape) => shape.compiledQuery(compiled, options))), - compiledQueryFirst: (compiled, options?: ExecutorQueryOptions) => + compiledQueryFirst: (compiled, options?: SqlQueryOptions) => getShape.pipe(Effect.flatMap((shape) => shape.compiledQueryFirst(compiled, options))), }) }), diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 78715b08e..7f87e47e8 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -235,8 +235,11 @@ async function handleQuery(db: Chdb, req: Request): Promise { try { out = db.query(forceJsonEachRow(sql)) } catch (error) { + // 400, not 500: a failing statement is a problem with the submitted SQL, + // and a 5xx would make the shared warehouse executor classify it as a + // transient upstream error and retry the identical query. return { - response: text(`query failed: ${(error as Error).message}`, 500), + response: text(`query failed: ${(error as Error).message}`, 400), rowCount: 0, durationMs: Math.round(performance.now() - started), sql, diff --git a/lib/clickhouse-builder/src/ch/compile.ts b/lib/clickhouse-builder/src/ch/compile.ts index 9d72382e0..12ed2344f 100644 --- a/lib/clickhouse-builder/src/ch/compile.ts +++ b/lib/clickhouse-builder/src/ch/compile.ts @@ -46,6 +46,11 @@ export class CompiledQueryDecodeError extends Schema.TaggedErrorClass { readonly sql: string + /** Execution-routing metadata: `"ingest"` marks a query whose datasource only + * exists in the managed ingest pipeline (declared via `.routing("ingest")` at + * the query definition), so executors read it there instead of a per-org + * warehouse override. */ + readonly routing?: "ingest" /** Type-safe cast of raw query results. The cast is sound because the * Output type is derived from the SELECT clause that produced the SQL. */ readonly castRows: (rows: ReadonlyArray>) => ReadonlyArray @@ -67,6 +72,7 @@ export type CompiledQueryRowSchema = Schema.Schema const makeCompiledQuery = ( sql: string, rowSchema?: CompiledQueryRowSchema, + routing?: "ingest", ): CompiledQuery => { const decodeRow = rowSchema ? (Schema.decodeUnknownEffect(rowSchema) as (row: unknown) => Effect.Effect) @@ -92,6 +98,7 @@ const makeCompiledQuery = ( return { sql, + ...(routing === undefined ? {} : { routing }), castRows: (rows) => rows as unknown as ReadonlyArray, decodeRows, decodeFirstRow: (rows) => { @@ -122,7 +129,8 @@ const makeCompiledQuery = ( export const unsafeCompiledQuery = (args: { readonly sql: string readonly rowSchema?: CompiledQueryRowSchema -}): CompiledQuery => makeCompiledQuery(args.sql, args.rowSchema) + readonly routing?: "ingest" +}): CompiledQuery => makeCompiledQuery(args.sql, args.rowSchema, args.routing) export function compileCH< Cols extends ColumnDefs, @@ -231,7 +239,7 @@ export function compileCH< } return { - ...makeCompiledQuery(sql, options?.rowSchema), + ...makeCompiledQuery(sql, options?.rowSchema, state.routingValue), } } diff --git a/lib/clickhouse-builder/src/ch/query.ts b/lib/clickhouse-builder/src/ch/query.ts index 15538ff3c..a83f60f2a 100644 --- a/lib/clickhouse-builder/src/ch/query.ts +++ b/lib/clickhouse-builder/src/ch/query.ts @@ -87,6 +87,8 @@ interface CHQueryState { readonly limitValue?: number readonly offsetValue?: number readonly formatValue?: string + /** Execution-routing metadata carried onto the CompiledQuery (see compile.ts). */ + readonly routingValue?: "ingest" /** Typed FROM subquery. Compiled lazily at compileCH time. */ readonly fromQuery?: CHQuery readonly fromQueryAlias?: string @@ -136,6 +138,14 @@ export interface CHQuery< format(fmt: "JSON" | "JSONEachRow"): CHQuery + /** + * Declare that this query reads a datasource that only exists in the managed + * ingest pipeline (an MV-less `ingest`-written table like `alert_checks`, or + * a deliberately cross-org managed scan). The executor routes it to the + * ingest backend instead of a per-org read override. + */ + routing(route: "ingest"): CHQuery + // --------------------------------------------------------------------------- // Type-safe joins with Table // --------------------------------------------------------------------------- @@ -338,6 +348,10 @@ function makeQuery< return makeQuery({ ...state, formatValue: fmt }) }, + routing(route) { + return makeQuery({ ...state, routingValue: route }) + }, + // ----------------------------------------------------------------------- // Type-safe joins with Table // ----------------------------------------------------------------------- diff --git a/packages/domain/src/http/observability.ts b/packages/domain/src/http/observability.ts index 6056cd52d..b59bde8fd 100644 --- a/packages/domain/src/http/observability.ts +++ b/packages/domain/src/http/observability.ts @@ -2,6 +2,7 @@ import { HttpApiEndpoint, HttpApiGroup } from "effect/unstable/httpapi" import { Schema } from "effect" import { SpanId, TraceId } from "../primitives" import { Authorization } from "./current-tenant" +import { warehouseHttpErrors } from "./warehouse-errors" export { UnauthorizedError } from "./current-tenant" @@ -231,42 +232,42 @@ export class ObservabilityApiGroup extends HttpApiGroup.make("observability") HttpApiEndpoint.post("listServices", "/services", { payload: ListServicesRequest, success: ListServicesResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .add( HttpApiEndpoint.post("searchTraces", "/traces/search", { payload: SearchTracesRequest, success: SearchTracesResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .add( HttpApiEndpoint.post("inspectTrace", "/traces/inspect", { payload: InspectTraceRequest, success: InspectTraceResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .add( HttpApiEndpoint.post("findErrors", "/errors", { payload: FindErrorsRequest, success: FindErrorsResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .add( HttpApiEndpoint.post("diagnoseService", "/diagnose", { payload: DiagnoseServiceRequest, success: DiagnoseServiceResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .add( HttpApiEndpoint.post("searchLogs", "/logs", { payload: SearchLogsRequest, success: SearchLogsResponse, - error: ObservabilityApiError, + error: [ObservabilityApiError, ...warehouseHttpErrors], }), ) .prefix("/api/observability") diff --git a/packages/query-engine/src/ch/queries/activity.ts b/packages/query-engine/src/ch/queries/activity.ts index ea99fae6c..4a86926ab 100644 --- a/packages/query-engine/src/ch/queries/activity.ts +++ b/packages/query-engine/src/ch/queries/activity.ts @@ -9,10 +9,10 @@ // They are deliberately CROSS-ORG: there is no `OrgId.eq(...)` predicate, so a // single cheap scan of the small recent window / hourly MVs enumerates active // orgs at once. `OrgId` still appears in SELECT/GROUP BY, which satisfies the -// WarehouseQueryService `sql.includes("OrgId")` guard. Callers MUST route them -// with `pinToIngestConfig: true` so they hit the managed Tinybird workspace -// (where all managed orgs' data lives); BYO-ClickHouse orgs are invisible here -// and are gated separately by the caller (always processed). +// WarehouseQueryService `sql.includes("OrgId")` guard. `.routing("ingest")` +// routes them to the managed Tinybird workspace (where all managed orgs' data +// lives); BYO-ClickHouse orgs are invisible here and are gated separately by +// the caller (always processed). // // The discovery window must be a SUPERSET of the per-org scan window so no // active org is missed for the tick. @@ -32,6 +32,7 @@ export function activeOrgsByErrorEventsQuery() { .where(($) => [$.Timestamp.gte(param.dateTime("startTime"))]) .groupBy("orgId") .format("JSON") + .routing("ingest") } /** Orgs with any span aggregates since `startTime` (gates the anomaly detector). */ @@ -41,6 +42,7 @@ export function activeOrgsByTracesQuery() { .where(($) => [$.Hour.gte(param.dateTime("startTime"))]) .groupBy("orgId") .format("JSON") + .routing("ingest") } /** Orgs with any log aggregates since `startTime` (gates the anomaly detector). */ @@ -50,4 +52,5 @@ export function activeOrgsByLogsQuery() { .where(($) => [$.Hour.gte(param.dateTime("startTime"))]) .groupBy("orgId") .format("JSON") + .routing("ingest") } diff --git a/packages/query-engine/src/ch/queries/alert-checks.ts b/packages/query-engine/src/ch/queries/alert-checks.ts index 6125306ca..dfebb3fd8 100644 --- a/packages/query-engine/src/ch/queries/alert-checks.ts +++ b/packages/query-engine/src/ch/queries/alert-checks.ts @@ -76,4 +76,8 @@ export function listRuleChecksQuery(opts: ListRuleChecksOpts) { .limit(opts.limit) .offset(opts.offset ?? 0) .format("JSON") + // alert_checks is written via `ingest` (Tinybird-pinned) with no per-org + // MV, so reads must hit the same managed pipeline — otherwise a + // BYO-ClickHouse org reads an empty table from its own ClickHouse. + .routing("ingest") } diff --git a/packages/query-engine/src/execution/backend.ts b/packages/query-engine/src/execution/backend.ts new file mode 100644 index 000000000..aadafff58 --- /dev/null +++ b/packages/query-engine/src/execution/backend.ts @@ -0,0 +1,79 @@ +/** + * The warehouse backends Maple can execute against, named for what they ARE + * instead of being encoded as protocol × policy flag combinations: + * + * - `tinybird` — managed Tinybird via its SDK (`/v0/sql` + Events API) + * - `tinybird-gateway` — managed Tinybird via its ClickHouse-compatible gateway + * (`CLICKHOUSE_URL` + `CLICKHOUSE_PROVIDER=tinybird`) + * - `clickhouse` — vanilla ClickHouse: a per-org BYO cluster or an + * env-level self-hosted read endpoint + * - `chdb` — the embedded chDB engine behind the local `maple` binary + */ +export type WarehouseBackendKind = "tinybird" | "tinybird-gateway" | "clickhouse" | "chdb" + +export interface TinybirdBackendConfig { + readonly kind: "tinybird" + readonly host: string + readonly token: string +} + +export interface ClickHouseProtocolBackendConfig { + readonly kind: Exclude + readonly url: string + readonly username: string + readonly password: string + readonly database: string +} + +/** Resolved upstream connection config for a tenant's queries. */ +export type ResolvedWarehouseConfig = TinybirdBackendConfig | ClickHouseProtocolBackendConfig + +export interface WarehouseBackendDialect { + readonly driver: "tinybird-sdk" | "clickhouse-web" + /** + * Tinybird's managed warehouse rejects some ClickHouse settings (e.g. + * "Usage of setting 'max_block_size' is restricted") — through the SDK AND + * through its ClickHouse-compatible gateway. Vanilla ClickHouse allows them. + */ + readonly stripTinybirdRestrictedSettings: boolean + /** + * The official ClickHouse client rejects a trailing `FORMAT …`/`;` (it sets + * the format itself); Tinybird's `/v0/sql` requires them. + */ + readonly normalizeSqlForClient: boolean + /** + * Value for the `db.system.name` / `peer.service` span attributes. The + * gateway reports `clickhouse` (it is addressed over the ClickHouse + * protocol) — service-map DB nodes key on this value, so it must stay + * stable per deployment. + */ + readonly peerService: string +} + +/** Single source of truth for per-backend behavior. */ +export const BackendDialect: Record = { + tinybird: { + driver: "tinybird-sdk", + stripTinybirdRestrictedSettings: true, + normalizeSqlForClient: false, + peerService: "tinybird", + }, + "tinybird-gateway": { + driver: "clickhouse-web", + stripTinybirdRestrictedSettings: true, + normalizeSqlForClient: true, + peerService: "clickhouse", + }, + clickhouse: { + driver: "clickhouse-web", + stripTinybirdRestrictedSettings: false, + normalizeSqlForClient: true, + peerService: "clickhouse", + }, + chdb: { + driver: "clickhouse-web", + stripTinybirdRestrictedSettings: false, + normalizeSqlForClient: true, + peerService: "chdb", + }, +} diff --git a/packages/query-engine/src/execution/datasource-routing.ts b/packages/query-engine/src/execution/datasource-routing.ts new file mode 100644 index 000000000..9ec49ea2c --- /dev/null +++ b/packages/query-engine/src/execution/datasource-routing.ts @@ -0,0 +1,14 @@ +/** + * Datasources that are written via `ingest` (hard-pinned to the managed + * Tinybird pipeline) and have NO per-org materialization — they simply do not + * exist in a BYO ClickHouse. Reads of these tables must route with purpose + * "ingest" (declare `.routing("ingest")` at the query definition). + * + * The executor uses this list as a safety net: a read that resolves to an + * org-BYO backend while referencing one of these tables would silently return + * empty rows, so it logs a warning instead of failing quietly. + */ +export const INGEST_PINNED_TABLES: ReadonlyArray = ["alert_checks"] + +export const findIngestPinnedTable = (sql: string): string | undefined => + INGEST_PINNED_TABLES.find((table) => sql.includes(table)) diff --git a/packages/query-engine/src/execution/executor.test.ts b/packages/query-engine/src/execution/executor.test.ts index 03d698e71..90b2de59e 100644 --- a/packages/query-engine/src/execution/executor.test.ts +++ b/packages/query-engine/src/execution/executor.test.ts @@ -3,7 +3,7 @@ import { Duration, Effect, Ref } from "effect" import { TestClock } from "effect/testing" import type { OrgId } from "@maple/domain" import { RawSqlValidationError } from "@maple/domain/http" -import { compile, listRuleChecksQuery } from "../ch" +import { compile, listRuleChecksQuery, unsafeCompiledQuery } from "../ch" import { makeWarehouseExecutor } from "./executor" import { WarehouseResponseLimitError } from "./response-limits" import type { @@ -22,62 +22,84 @@ const tenant: ExecutionTenant = { // A per-org BYO read override (the read path) vs the managed Tinybird ingest // pipeline (the write path). alert_checks rows only ever land in the latter. const clickhouseConfig: ResolvedWarehouseConfig = { - _tag: "clickhouse", - provider: "clickhouse", + kind: "clickhouse", url: "https://byo.example.com", username: "default", password: "secret", database: "maple", } const tinybirdConfig: ResolvedWarehouseConfig = { - _tag: "tinybird", - provider: "tinybird", + kind: "tinybird", host: "https://api.tinybird.co", token: "tb_token", } const tinybirdGatewayConfig: ResolvedWarehouseConfig = { ...clickhouseConfig, - provider: "tinybird", + kind: "tinybird-gateway", } +// listRuleChecksQuery declares .routing("ingest") at its definition — +// alert_checks only exists in the managed ingest pipeline. const compiled = compile(listRuleChecksQuery({ limit: 1 }), { orgId: "org_test", ruleId: "rule_test", }) +// A plain query with no routing declaration follows the default read route. +const untaggedCompiled = unsafeCompiledQuery<{ readonly c: number }>({ + sql: "SELECT count() AS c FROM traces WHERE OrgId = 'org_test'\nFORMAT JSON", +}) + // Records the backend each constructed client was wired to, so a test can assert -// which config the executor resolved through. -const makeDeps = (createdTags: Array): WarehouseExecutorDeps => ({ +// which route the executor resolved through. Models a BYO-CH org: reads and raw +// SQL hit the org's ClickHouse, ingest hits the managed Tinybird pipeline. +const makeDeps = (createdKinds: Array): WarehouseExecutorDeps => ({ createClient: (config) => { - createdTags.push(config._tag) + createdKinds.push(config.kind) const client: WarehouseSqlClient = { sql: async () => ({ data: [] }), insert: async () => {}, } return client }, - resolveConfig: () => Effect.succeed({ config: clickhouseConfig, clientCacheKey: "read:org_test" }), - resolveRawSqlConfig: () => Effect.succeed({ config: clickhouseConfig, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "write:managed" }), + resolveRoute: (_tenant, purpose) => + Effect.succeed( + purpose === "ingest" + ? { source: "managed" as const, config: tinybirdConfig, clientCacheKey: "write:managed" } + : { + source: "org-byo" as const, + config: clickhouseConfig, + clientCacheKey: purpose === "raw" ? "raw:org_test" : "read:org_test", + }, + ), }) -describe("makeWarehouseExecutor pinToIngestConfig", () => { - it.effect("reads from the per-org (ClickHouse) config by default", () => +describe("makeWarehouseExecutor ingest routing", () => { + it.effect("reads an untagged compiled query from the per-org (ClickHouse) route", () => Effect.gen(function* () { - const created: Array = [] + const created: Array = [] const executor = makeWarehouseExecutor(makeDeps(created)) - yield* executor.compiledQuery(tenant, compiled, { context: "test" }) + yield* executor.compiledQuery(tenant, untaggedCompiled, { context: "test" }) assert.deepStrictEqual(created, ["clickhouse"]) }), ) - it.effect("reads from the ingest (Tinybird) config when pinToIngestConfig is set", () => + it.effect("routes a .routing('ingest')-tagged compiled query to the ingest (Tinybird) route", () => Effect.gen(function* () { - const created: Array = [] + const created: Array = [] const executor = makeWarehouseExecutor(makeDeps(created)) - yield* executor.compiledQuery(tenant, compiled, { + yield* executor.compiledQuery(tenant, compiled, { context: "test" }) + assert.deepStrictEqual(created, ["tinybird"]) + }), + ) + + it.effect("routes hand-written SQL with the route:'ingest' option", () => + Effect.gen(function* () { + const created: Array = [] + const executor = makeWarehouseExecutor(makeDeps(created)) + yield* executor.sqlQuery(tenant, "SELECT 1 WHERE OrgId = 'org_test'", { context: "test", - pinToIngestConfig: true, + route: "ingest", }) assert.deepStrictEqual(created, ["tinybird"]) }), @@ -86,9 +108,8 @@ describe("makeWarehouseExecutor pinToIngestConfig", () => { // Capture the final SQL the executor hands to the client so a test can assert // whether a Tinybird-restricted setting (max_block_size) survived the strip for -// the resolved backend. Provider is independent from protocol and routing -// source: a Tinybird gateway uses the ClickHouse protocol but still enforces -// Tinybird's restricted-settings policy. +// the resolved backend. The Tinybird gateway (`tinybird-gateway`) uses the +// ClickHouse protocol but still enforces Tinybird's restricted-settings policy. const makeRecordingDeps = ( resolved: { config: ResolvedWarehouseConfig; clientCacheKey: string }, sqls: Array, @@ -100,9 +121,12 @@ const makeRecordingDeps = ( }, insert: async () => {}, }), - resolveConfig: () => Effect.succeed(resolved), - resolveRawSqlConfig: () => Effect.succeed({ ...resolved, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => Effect.succeed(resolved), + resolveRoute: (_tenant, purpose) => + Effect.succeed({ + source: "managed" as const, + config: resolved.config, + clientCacheKey: purpose === "raw" ? "raw:org_test" : resolved.clientCacheKey, + }), }) describe("makeWarehouseExecutor restricted-settings strip", () => { @@ -164,6 +188,46 @@ describe("makeWarehouseExecutor restricted-settings strip", () => { ) }) +// The compiled DSL always ends in `FORMAT JSON`; the official ClickHouse client +// rejects a trailing FORMAT/`;` (it sets the format itself) while Tinybird's +// /v0/sql requires it. Normalization follows the wire protocol (the dialect's +// normalizeSqlForClient) — the Tinybird CH-gateway speaks the ClickHouse protocol. +describe("makeWarehouseExecutor SQL normalization", () => { + it.effect("strips the trailing FORMAT JSON for a ClickHouse-protocol backend", () => + Effect.gen(function* () { + const sqls: Array = [] + const executor = makeWarehouseExecutor( + makeRecordingDeps({ config: clickhouseConfig, clientCacheKey: "read:org_test" }, sqls), + ) + yield* executor.compiledQuery(tenant, compiled, { context: "test" }) + assert.isTrue(compiled.sql.trimEnd().endsWith("FORMAT JSON")) + assert.isFalse(sqls[0]?.includes("FORMAT JSON")) + }), + ) + + it.effect("strips the trailing FORMAT JSON for the Tinybird CH-gateway", () => + Effect.gen(function* () { + const sqls: Array = [] + const executor = makeWarehouseExecutor( + makeRecordingDeps({ config: tinybirdGatewayConfig, clientCacheKey: "read:managed" }, sqls), + ) + yield* executor.compiledQuery(tenant, compiled, { context: "test" }) + assert.isFalse(sqls[0]?.includes("FORMAT JSON")) + }), + ) + + it.effect("keeps the trailing FORMAT JSON for the Tinybird SDK backend", () => + Effect.gen(function* () { + const sqls: Array = [] + const executor = makeWarehouseExecutor( + makeRecordingDeps({ config: tinybirdConfig, clientCacheKey: "read:managed" }, sqls), + ) + yield* executor.compiledQuery(tenant, compiled, { context: "test" }) + assert.isTrue(sqls[0]?.trimEnd().endsWith("FORMAT JSON")) + }), + ) +}) + describe("makeWarehouseExecutor raw response limits", () => { it.effect("maps a driver byte abort directly to RawSqlValidationError", () => Effect.gen(function* () { @@ -206,12 +270,17 @@ describe("makeWarehouseExecutor client cache partitions", () => { }, } }, - resolveConfig: () => - Effect.succeed({ config: tinybirdConfig, clientCacheKey: "read:managed" }), - resolveRawSqlConfig: () => - Effect.succeed({ config: tinybirdConfig, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => - Effect.succeed({ config: tinybirdConfig, clientCacheKey: "write:managed" }), + resolveRoute: (_tenant, purpose) => + Effect.succeed({ + source: "managed" as const, + config: tinybirdConfig, + clientCacheKey: + purpose === "ingest" + ? "write:managed" + : purpose === "raw" + ? "raw:org_test" + : "read:managed", + }), }) yield* executor.sqlQuery(tenant, "SELECT 1 WHERE OrgId = 'org_test'") @@ -238,9 +307,12 @@ const makeHangingDeps = (): WarehouseExecutorDeps => ({ sql: () => new Promise<{ data: never[] }>(() => {}), insert: async () => {}, }), - resolveConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "read:managed" }), - resolveRawSqlConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "write:managed" }), + resolveRoute: () => + Effect.succeed({ + source: "managed" as const, + config: tinybirdConfig, + clientCacheKey: "read:managed", + }), }) // Like makeHangingDeps, but counts how many times the client's `sql` is invoked @@ -254,9 +326,12 @@ const makeCountingHangingDeps = (counter: { count: number }): WarehouseExecutorD }, insert: async () => {}, }), - resolveConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "read:managed" }), - resolveRawSqlConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "raw:org_test" }), - resolveIngestConfig: () => Effect.succeed({ config: tinybirdConfig, clientCacheKey: "write:managed" }), + resolveRoute: () => + Effect.succeed({ + source: "managed" as const, + config: tinybirdConfig, + clientCacheKey: "read:managed", + }), }) describe("makeWarehouseExecutor client timeout", () => { diff --git a/packages/query-engine/src/execution/executor.ts b/packages/query-engine/src/execution/executor.ts index 6c68137c8..a18eaf157 100644 --- a/packages/query-engine/src/execution/executor.ts +++ b/packages/query-engine/src/execution/executor.ts @@ -11,7 +11,7 @@ import { } from "@maple/domain/http" import type { WarehouseQueryName } from "@maple/domain/warehouse-queries" import { compilePipeQuery, type CompiledQuery } from "../ch" -import type { ExecutorQueryOptions, WarehouseExecutorShape } from "../observability" +import type { WarehouseExecutorShape } from "../observability" import { appendSettings, type QueryProfileName, @@ -27,9 +27,12 @@ import { normalizeSqlForClickHouseClient, truncateSql, } from "./fingerprint" +import { BackendDialect } from "./backend" +import { findIngestPinnedTable } from "./datasource-routing" import type { ExecutionTenant, ResolvedWarehouseConfig, + RoutePurpose, SqlQueryOptions, WarehouseExecutorDeps, WarehouseQueryServiceShape, @@ -45,9 +48,9 @@ interface CachedClient { } const sqlClientCacheKey = (config: ResolvedWarehouseConfig): string => - config._tag === "clickhouse" - ? `${config.provider}:clickhouse:${config.url}:${config.username}:${config.password}:${config.database}` - : `tinybird:${config.host}:${config.token}` + config.kind === "tinybird" + ? `tinybird:${config.host}:${config.token}` + : `${config.kind}:${config.url}:${config.username}:${config.password}:${config.database}` // Only retry transient upstream failures (5xx, 408, 429, network blips). Non-transient // errors (auth, config, schema_drift, query) re-fail immediately — there's nothing to @@ -140,32 +143,37 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue // Control-plane datasources (e.g. alert_checks) are written via `ingest`, // which is hard-pinned to the managed Tinybird pipeline. They do NOT exist - // in a per-org BYO ClickHouse, so their reads must pin to the same ingest + // in a per-org BYO ClickHouse, so their reads must route to the same ingest // config to stay symmetric with the write — otherwise a BYO-CH org reads an - // empty table from its own ClickHouse. - const resolveFn = - execution === "raw" - ? deps.resolveRawSqlConfig - : options?.pinToIngestConfig && deps.resolveIngestConfig - ? deps.resolveIngestConfig - : deps.resolveConfig - const resolved = yield* resolveFn(tenant, pipe) - if (options?.pinToIngestConfig) yield* Effect.annotateCurrentSpan("query.routing", "ingest") - const peerService = resolved.config._tag === "clickhouse" ? "clickhouse" : "tinybird" - yield* Effect.annotateCurrentSpan("db.system.name", peerService) - yield* Effect.annotateCurrentSpan("peer.service", peerService) - // Tinybird's managed warehouse restricts some settings (e.g. max_block_size: - // "Usage of setting 'max_block_size' is restricted"). It is reached either via - // the Tinybird SDK (_tag "tinybird") or its ClickHouse-compatible gateway - // (_tag "clickhouse", when CLICKHOUSE_PROVIDER=tinybird); both enforce that - // policy. Vanilla ClickHouse supports those settings regardless of whether its - // credentials come from the environment or an org override. - const settings = - resolved.config.provider === "tinybird" - ? stripTinybirdRestrictedSettings(resolveSettings(options)) - : resolveSettings(options) - const sqlForClient = - resolved.config._tag === "clickhouse" ? normalizeSqlForClickHouseClient(sql) : sql + // empty table from its own ClickHouse. That routing is declared at the + // query definition (`.routing("ingest")` → `options.route`). + const purpose: RoutePurpose = + execution === "raw" ? "raw" : options?.route === "ingest" ? "ingest" : "read" + const resolved = yield* deps.resolveRoute(tenant, purpose, pipe) + // Safety net for the silent-empty-table failure: an ingest-pinned table + // read against an org's own BYO ClickHouse returns no rows, not an error. + if (purpose !== "ingest" && resolved.source === "org-byo") { + const pinnedTable = findIngestPinnedTable(sql) + if (pinnedTable !== undefined) { + yield* Effect.logWarning( + "Query reads an ingest-pinned datasource from a BYO ClickHouse — declare .routing(\"ingest\") at the query definition", + { pipe, table: pinnedTable, orgId: tenant.orgId }, + ) + } + } + // Legacy spelling of `warehouse.route: "ingest"` — dual-emitted until + // dashboards move to the `warehouse.*` attributes. + if (purpose === "ingest") yield* Effect.annotateCurrentSpan("query.routing", "ingest") + const dialect = BackendDialect[resolved.config.kind] + yield* Effect.annotateCurrentSpan("warehouse.backend", resolved.config.kind) + yield* Effect.annotateCurrentSpan("warehouse.route", purpose) + yield* Effect.annotateCurrentSpan("warehouse.config_source", resolved.source) + yield* Effect.annotateCurrentSpan("db.system.name", dialect.peerService) + yield* Effect.annotateCurrentSpan("peer.service", dialect.peerService) + const settings = dialect.stripTinybirdRestrictedSettings + ? stripTinybirdRestrictedSettings(resolveSettings(options)) + : resolveSettings(options) + const sqlForClient = dialect.normalizeSqlForClient ? normalizeSqlForClickHouseClient(sql) : sql const finalSql = appendSettings(sqlForClient, settings) const sqlLength = finalSql.length const sqlTruncated = sqlLength > SQL_TRACE_MAX @@ -247,7 +255,7 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue pipe, context: options?.context, orgId: tenant.orgId, - backend: resolved.config._tag, + backend: resolved.config.kind, durationMs: elapsedMs, retryAttempts: attempts, errorTag: error._tag, @@ -359,12 +367,21 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue return yield* executeSql(tenant, sql, "rawSqlQuery", options, "raw") }) + // A compiled query can carry `.routing("ingest")` from its definition — that + // wins over the (absent) per-call option so the table→routing knowledge + // lives next to the query, not at every call site. + const withCompiledRouting = ( + compiled: CompiledQuery, + options?: SqlQueryOptions, + ): SqlQueryOptions | undefined => + compiled.routing === "ingest" ? { ...options, route: "ingest" } : options + const compiledQuery = Effect.fn("WarehouseQueryService.compiledQuery")(function* ( tenant: ExecutionTenant, compiled: CompiledQuery, options?: SqlQueryOptions, ) { - const rows = yield* sqlQuery(tenant, compiled.sql, options) + const rows = yield* sqlQuery(tenant, compiled.sql, withCompiledRouting(compiled, options)) return yield* compiled.decodeRows(rows).pipe( Effect.mapError( (error) => @@ -382,7 +399,7 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue compiled: CompiledQuery, options?: SqlQueryOptions, ) { - const rows = yield* sqlQuery(tenant, compiled.sql, options) + const rows = yield* sqlQuery(tenant, compiled.sql, withCompiledRouting(compiled, options)) return yield* compiled.decodeFirstRow(rows).pipe( Effect.mapError( (error) => @@ -407,14 +424,16 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue if (rows.length === 0) return const label = `ingest:${datasource}` - // Writes resolve through the ingest-specific resolver when provided: the - // host points it at the managed Tinybird pipeline, never a per-org BYO - // ClickHouse READ override (routing writes through the override 500'd every - // insert and broke demo-seed onboarding). Falls back to the read resolver. - const resolveForIngest = deps.resolveIngestConfig ?? deps.resolveConfig - const resolved = yield* resolveForIngest(tenant, label).pipe( + // Writes resolve with purpose "ingest": the host routes them to the managed + // Tinybird pipeline, never a per-org BYO ClickHouse READ override (routing + // writes through the override 500'd every insert and broke demo-seed + // onboarding). + const resolved = yield* deps.resolveRoute(tenant, "ingest", label).pipe( Effect.mapError((error) => toWarehouseQueryError(label, error)), ) + yield* Effect.annotateCurrentSpan("warehouse.backend", resolved.config.kind) + yield* Effect.annotateCurrentSpan("warehouse.route", "ingest") + yield* Effect.annotateCurrentSpan("warehouse.config_source", resolved.source) // Insert through the same client the read path uses (official // @clickhouse/client-web for ClickHouse, Tinybird Events API for @@ -429,13 +448,15 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue yield* Effect.tryPromise({ try: () => client.insert(datasource, rows), - catch: (error) => toWarehouseQueryError(label, error), + // Classify like the read path so an auth failure or quota breach on + // insert surfaces with its real tag instead of a generic query error. + catch: (error) => mapWarehouseError(label, error), }).pipe( Effect.tapError((error) => Effect.logError("WarehouseQueryService.ingest failed", { datasource, rowCount: rows.length, - backend: resolved.config._tag, + backend: resolved.config.kind, errorTag: error._tag, message: error.message, }), @@ -443,58 +464,26 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue ) }) + // The facade only binds the tenant and defaults `query.context` — the + // canonical `WarehouseQueryService.executeSql` span carries all + // instrumentation, so no extra span layer is added here. const asExecutor = (tenant: ExecutionTenant): WarehouseExecutorShape => ({ orgId: tenant.orgId, - query: ( - pipe: WarehouseQueryName, - params: Record, - options?: ExecutorQueryOptions, - ) => - query(tenant, { pipeName: pipe, params }, { ...options, context: `pipe:${pipe}` }).pipe( + query: (pipe: WarehouseQueryName, params: Record, options?: SqlQueryOptions) => + query(tenant, { pipeName: pipe, params }, { context: `pipe:${pipe}`, ...options }).pipe( Effect.map((response) => ({ data: response.data as unknown as ReadonlyArray })), - Effect.withSpan("WarehouseExecutor.query", { - attributes: { - pipe, - orgId: tenant.orgId, - "query.profile": options?.profile, - "query.context": `pipe:${pipe}`, - }, - }), ), - sqlQuery: (sql: string, options?: ExecutorQueryOptions) => - sqlQuery(tenant, sql, { ...options, context: "warehouseExecutor.sqlQuery" }).pipe( + sqlQuery: (sql: string, options?: SqlQueryOptions) => + sqlQuery(tenant, sql, { context: "warehouseExecutor.sqlQuery", ...options }).pipe( Effect.map((rows) => rows as unknown as ReadonlyArray), - Effect.withSpan("WarehouseExecutor.sqlQuery", { - attributes: { - orgId: tenant.orgId, - "query.profile": options?.profile, - "query.context": "warehouseExecutor.sqlQuery", - }, - }), ), - compiledQuery: (compiled: CompiledQuery, options?: ExecutorQueryOptions) => - compiledQuery(tenant, compiled, { ...options, context: "warehouseExecutor.compiledQuery" }).pipe( - Effect.withSpan("WarehouseExecutor.compiledQuery", { - attributes: { - orgId: tenant.orgId, - "query.profile": options?.profile, - "query.context": "warehouseExecutor.compiledQuery", - }, - }), - ), - compiledQueryFirst: (compiled: CompiledQuery, options?: ExecutorQueryOptions) => + compiledQuery: (compiled: CompiledQuery, options?: SqlQueryOptions) => + compiledQuery(tenant, compiled, { context: "warehouseExecutor.compiledQuery", ...options }), + compiledQueryFirst: (compiled: CompiledQuery, options?: SqlQueryOptions) => compiledQueryFirst(tenant, compiled, { - ...options, context: "warehouseExecutor.compiledQueryFirst", - }).pipe( - Effect.withSpan("WarehouseExecutor.compiledQueryFirst", { - attributes: { - orgId: tenant.orgId, - "query.profile": options?.profile, - "query.context": "warehouseExecutor.compiledQueryFirst", - }, - }), - ), + ...options, + }), }) return { diff --git a/packages/query-engine/src/execution/index.ts b/packages/query-engine/src/execution/index.ts index 54008dea0..1f8e2c7e8 100644 --- a/packages/query-engine/src/execution/index.ts +++ b/packages/query-engine/src/execution/index.ts @@ -1,3 +1,5 @@ +export * from "./backend" +export * from "./datasource-routing" export * from "./ports" export * from "./errors" export * from "./fingerprint" diff --git a/packages/query-engine/src/execution/ports.ts b/packages/query-engine/src/execution/ports.ts index 65dd86051..08097df36 100644 --- a/packages/query-engine/src/execution/ports.ts +++ b/packages/query-engine/src/execution/ports.ts @@ -7,9 +7,10 @@ import type { WarehouseQueryError, WarehouseValidationError, } from "@maple/domain/http" +import type { ResolvedWarehouseConfig } from "./backend" import type { CompiledQuery } from "../ch" import type { WarehouseExecutorShape } from "../observability" -import type { QueryProfileName, WarehouseQuerySettings } from "../profiles" +import type { SqlQueryOptions } from "../profiles" import type { WarehouseSqlError } from "./errors" /** The minimal tenant surface the executor reads (org scope + identity for spans). */ @@ -19,44 +20,9 @@ export interface ExecutionTenant { readonly authMode: string } -export type SqlQueryOptions = { - profile?: QueryProfileName - settings?: WarehouseQuerySettings - /** - * Semantic name for the query (e.g. "errorsByType", "spanHierarchy"). - * Annotated on the executeSql span as `query.context` so traces can be - * filtered and grouped by call site without re-running the SQL. - */ - context?: string - /** - * Read this query from the INGEST config (managed Tinybird) instead of the - * per-org read config. Use for Maple control-plane datasources that are - * written via `ingest` (which is hard-pinned to Tinybird) and therefore do - * NOT exist in a per-org BYO ClickHouse — e.g. `alert_checks`. Keeps the - * read symmetric with the write. Falls back to `resolveConfig` if the host - * did not inject a `resolveIngestConfig`. - */ - pinToIngestConfig?: boolean -} +export type { SqlQueryOptions } from "../profiles" -export type WarehouseProvider = "clickhouse" | "tinybird" - -/** Resolved upstream connection config for a tenant's queries. */ -export type ResolvedWarehouseConfig = - | { - readonly _tag: "clickhouse" - readonly provider: WarehouseProvider - readonly url: string - readonly username: string - readonly password: string - readonly database: string - } - | { - readonly _tag: "tinybird" - readonly provider: "tinybird" - readonly host: string - readonly token: string - } +export type { ResolvedWarehouseConfig } from "./backend" /** Minimal client interface — raw SQL execution plus row inserts. */ export interface WarehouseSqlClient { @@ -73,40 +39,45 @@ export interface WarehouseSqlClient { } /** - * The injected dependencies of the warehouse executor. The host app provides - * the driver construction (`createClient`) and the per-org config resolution - * (`resolveConfig` / `resolveRawSqlConfig`, which read the org-override DB row - * or env and return stable logical cache partitions); the executor itself — - * error mapping, retry, client cache, OrgId scoping, span instrumentation — - * lives in this package. + * What a query is FOR — the executor computes this and the host's `resolveRoute` + * turns it into a concrete backend: + * + * - `read` — trusted, Maple-compiled SQL (the default) + * - `raw` — user-authored SQL; must run on tenant-isolated credentials + * - `ingest` — writes, plus reads of control-plane datasources that only exist + * in the managed write pipeline (e.g. `alert_checks`) */ -export interface ResolvedWarehouseTarget { +export type RoutePurpose = "read" | "raw" | "ingest" + +/** The host's routing decision: which backend, with which credentials, and why. */ +export interface WarehouseRoute { + /** + * Why this config was chosen — annotated on the executeSql span as + * `warehouse.config_source`: + * - `managed` — the env-level shared warehouse + * - `org-byo` — the org's own BYO ClickHouse credentials + * - `org-jwt` — the shared Tinybird warehouse behind an org-scoped JWT + */ + readonly source: "managed" | "org-byo" | "org-jwt" readonly config: ResolvedWarehouseConfig /** Stable logical cache partition; config changes are detected independently. */ readonly clientCacheKey: string } +/** + * The injected dependencies of the warehouse executor. The host app provides + * the driver construction (`createClient`) and the routing decision + * (`resolveRoute`, which reads the org-override DB row or env and returns a + * stable logical cache partition); the executor itself — error mapping, retry, + * client cache, OrgId scoping, span instrumentation — lives in this package. + */ export interface WarehouseExecutorDeps { readonly createClient: (config: ResolvedWarehouseConfig) => WarehouseSqlClient - readonly resolveConfig: ( - tenant: ExecutionTenant, - label: string, - ) => Effect.Effect - /** Resolve the isolated credentials used exclusively for user-authored SQL. */ - readonly resolveRawSqlConfig: ( - tenant: ExecutionTenant, - label: string, - ) => Effect.Effect - /** - * Config resolver for the WRITE path (`ingest`). Inserts must land in the - * managed pipeline (Tinybird in the cloud), NOT a per-org BYO ClickHouse - * read override — that override is a query-side concern. Falls back to - * `resolveConfig` when omitted. - */ - readonly resolveIngestConfig?: ( + readonly resolveRoute: ( tenant: ExecutionTenant, + purpose: RoutePurpose, label: string, - ) => Effect.Effect + ) => Effect.Effect } export interface WarehouseQueryServiceShape { @@ -140,7 +111,7 @@ export interface WarehouseQueryServiceShape { tenant: ExecutionTenant, datasource: string, rows: ReadonlyArray, - ) => Effect.Effect + ) => Effect.Effect /** * Present this service as the package-level `WarehouseExecutor` for a given * tenant — the single managed-warehouse implementation of that interface. diff --git a/packages/query-engine/src/observability/WarehouseExecutor.ts b/packages/query-engine/src/observability/WarehouseExecutor.ts index 2e04606f8..731cbcbea 100644 --- a/packages/query-engine/src/observability/WarehouseExecutor.ts +++ b/packages/query-engine/src/observability/WarehouseExecutor.ts @@ -2,6 +2,7 @@ import { Context, type Effect, type Option } from "effect" import type { WarehouseError } from "@maple/domain/http/warehouse-errors" import type { WarehouseQueryName } from "@maple/domain/warehouse-queries" import type { CompiledQuery } from "../ch" +import type { SqlQueryOptions } from "../profiles" /** * The error channel of every `WarehouseExecutor` method. This is the warehouse @@ -11,27 +12,6 @@ import type { CompiledQuery } from "../ch" */ export type WarehouseExecutorError = WarehouseError -/** - * ClickHouse settings a call site may request. Row/byte caps - * (`max_rows_to_read`, `max_result_rows`, `max_bytes_to_read`) are restricted - * by Tinybird and intentionally absent. `maxBlockSize` is ClickHouse-only — - * the executor strips it when the resolved backend is Tinybird (see - * `WarehouseQuerySettings.maxBlockSize` in `../profiles` for the rationale). - */ -export type ExecutorQuerySettings = { - maxExecutionTime?: number - maxMemoryUsage?: number - maxThreads?: number - maxBlockSize?: number -} - -export type ExecutorQueryProfile = "discovery" | "list" | "aggregation" | "explain" | "unbounded" - -export type ExecutorQueryOptions = { - profile?: ExecutorQueryProfile - settings?: ExecutorQuerySettings -} - export interface WarehouseExecutorShape { /** The org ID for the current tenant — needed for raw SQL queries. */ readonly orgId: string @@ -39,23 +19,23 @@ export interface WarehouseExecutorShape { readonly query: ( pipe: WarehouseQueryName, params: Record, - options?: ExecutorQueryOptions, + options?: SqlQueryOptions, ) => Effect.Effect<{ data: ReadonlyArray }, WarehouseExecutorError> /** Execute raw ClickHouse SQL. The SQL MUST include an OrgId filter. */ readonly sqlQuery: >( sql: string, - options?: ExecutorQueryOptions, + options?: SqlQueryOptions, ) => Effect.Effect, WarehouseExecutorError> readonly compiledQuery: ( compiled: CompiledQuery, - options?: ExecutorQueryOptions, + options?: SqlQueryOptions, ) => Effect.Effect, WarehouseExecutorError> readonly compiledQueryFirst: ( compiled: CompiledQuery, - options?: ExecutorQueryOptions, + options?: SqlQueryOptions, ) => Effect.Effect, WarehouseExecutorError> } diff --git a/packages/query-engine/src/observability/index.ts b/packages/query-engine/src/observability/index.ts index 012050ab6..68979b388 100644 --- a/packages/query-engine/src/observability/index.ts +++ b/packages/query-engine/src/observability/index.ts @@ -2,10 +2,8 @@ export { WarehouseExecutor, type WarehouseExecutorError, type WarehouseExecutorShape, - type ExecutorQueryOptions, - type ExecutorQuerySettings, - type ExecutorQueryProfile, } from "./WarehouseExecutor" +export type { SqlQueryOptions } from "../profiles" export type * from "./types" export { toSpanResult, toLogEntry, toErrorSummary } from "./row-mappers" export { aggregateServiceRows, weightedAvg } from "./aggregation" diff --git a/packages/query-engine/src/profiles/query-profile.ts b/packages/query-engine/src/profiles/query-profile.ts index 28f8ca1c5..127fb16f1 100644 --- a/packages/query-engine/src/profiles/query-profile.ts +++ b/packages/query-engine/src/profiles/query-profile.ts @@ -58,6 +58,29 @@ export type WarehouseQueryOptions = { settings?: WarehouseQuerySettings } +/** + * The single per-call option type shared by every warehouse query surface + * (`WarehouseQueryServiceShape`, the `WarehouseExecutor` facade, the runtime + * `QueryEngineWarehouse` port, and the CLI executors). It lives next to the + * profiles because it is a pure type with no execution-layer dependency. + */ +export type SqlQueryOptions = WarehouseQueryOptions & { + /** + * Semantic name for the query (e.g. "errorsByType", "spanHierarchy"). + * Annotated on the executeSql span as `query.context` so traces can be + * filtered and grouped by call site without re-running the SQL. + */ + context?: string + /** + * Route this query to the INGEST backend (managed Tinybird) instead of the + * per-org read config. Prefer declaring this at the query definition via + * `.routing("ingest")` (carried on `CompiledQuery.routing`); use this option + * only for hand-written SQL or when the pin depends on runtime state (e.g. + * reads of gateway-written data gated on write-readiness). + */ + route?: "ingest" +} + /** * Named cost profiles. Pick one at the call site (not at the query * definition) since the same query can be cheap as a one-off and diff --git a/packages/query-engine/src/runtime/query-engine.ts b/packages/query-engine/src/runtime/query-engine.ts index af9bda734..cd53ef7cf 100644 --- a/packages/query-engine/src/runtime/query-engine.ts +++ b/packages/query-engine/src/runtime/query-engine.ts @@ -29,7 +29,12 @@ import { } from "@maple/domain/http" import type { OrgId } from "@maple/domain" import { Array as Arr, Duration, Effect, Match, Option, Result, Schema } from "effect" -import { LOGS_BODY_SEARCH_SETTINGS, type QueryProfileName, type WarehouseQuerySettings } from "../profiles" +import { + LOGS_BODY_SEARCH_SETTINGS, + type QueryProfileName, + type SqlQueryOptions, + type WarehouseQuerySettings, +} from "../profiles" import { computeBucketSeconds } from "../datetime" import { makeExecuteRawSql } from "./raw-sql" import { decodeEvalSeries, encodeEvalPoints, type BucketGroupObs } from "./evaluate-bucket-codec" @@ -54,11 +59,7 @@ export interface QueryEngineWarehouse { readonly sqlQuery: ( tenant: T, sql: string, - options?: { - readonly profile?: QueryProfileName - readonly context?: string - readonly settings?: WarehouseQuerySettings - }, + options?: SqlQueryOptions, ) => Effect.Effect>, WarehouseError> readonly rawSqlQuery: ( tenant: T, @@ -68,11 +69,7 @@ export interface QueryEngineWarehouse { readonly compiledQuery: ( tenant: T, compiled: CH.CompiledQuery, - options?: { - readonly profile?: QueryProfileName - readonly context?: string - readonly settings?: WarehouseQuerySettings - }, + options?: SqlQueryOptions, ) => Effect.Effect, WarehouseError> } From c6c72c393742c6b6916269bc2a809b9ce41e6a8a Mon Sep 17 00:00:00 2001 From: Makisuo Date: Sun, 19 Jul 2026 13:26:22 +0200 Subject: [PATCH 2/2] w --- .../rules/service-map-attribution.md | 9 +- .../rules/span-attributes.md | 51 +++++--- .../src/routes/v2/alerts-error-map.test.ts | 36 ++++++ apps/api/src/routes/v2/alerts-error-map.ts | 7 +- apps/cli/package.json | 2 +- apps/cli/src/core/executor.test.ts | 80 ++++++------ apps/cli/src/core/executor.ts | 5 +- apps/cli/src/core/operations.test.ts | 54 ++++++++ apps/cli/src/core/operations.ts | 54 ++++++-- apps/cli/src/core/remote-executor.test.ts | 70 ++++++++++ apps/cli/src/core/remote-executor.ts | 73 +++++++---- apps/cli/src/server/serve.ts | 23 ++-- apps/cli/test/server-network.test.ts | 52 +++++++- packages/domain/src/http/observability.ts | 21 ++- .../query-engine/src/execution/backend.ts | 18 ++- .../src/execution/executor.test.ts | 120 +++++++++++++++++- .../query-engine/src/execution/executor.ts | 68 ++++++++-- packages/query-engine/src/execution/ports.ts | 5 +- 18 files changed, 611 insertions(+), 137 deletions(-) create mode 100644 apps/api/src/routes/v2/alerts-error-map.test.ts create mode 100644 apps/cli/src/core/operations.test.ts create mode 100644 apps/cli/src/core/remote-executor.test.ts diff --git a/.agents/skills/maple-telemetry-conventions/rules/service-map-attribution.md b/.agents/skills/maple-telemetry-conventions/rules/service-map-attribution.md index fe8a6d026..4620da065 100644 --- a/.agents/skills/maple-telemetry-conventions/rules/service-map-attribution.md +++ b/.agents/skills/maple-telemetry-conventions/rules/service-map-attribution.md @@ -7,7 +7,7 @@ The service map at `apps/web/src/routes/service-map.tsx` is built entirely from | Rendered element | Source query | Required attribute | Lives on | |---|---|---|---| | Service-to-service edge | `service_map_edges_hourly_mv` | `peer.service` | Span (`Client` or `Producer` kind) | -| Service-to-DB edge | `service_map_db_edges_hourly_mv` | `db.system` (and `peer.service` for the main edge) | Span | +| Service-to-DB edge | `service_map_db_edges_hourly_mv` | `db.system.name` (legacy fallback: `db.system`) and `peer.service` | Span | | Platform badge (Cloudflare / AWS / Railway / k8s) | `service_platforms_hourly_mv` | `cloud.platform`, `cloud.provider`, optional `faas.name`, optional `k8s.*` | Resource | | Runtime icon (node / bun / deno / workerd / rust) | `service_platforms_hourly_mv` | `process.runtime.name` | Resource | | SDK badge | `service_platforms_hourly_mv` | `maple.sdk.type` | Resource | @@ -54,17 +54,17 @@ Mirror the TS detection. Set the same keys with the same values — never invent | Call type | Span kind | Required attributes | Notes | |---|---|---|---| | Outbound HTTP / RPC | `Client` or `Producer` | `peer.service`; optional `server.address`, `http.request.method` | `peer.service` is the **logical** destination, not the host | -| Database call | any | `db.system` (e.g. `clickhouse`, `tinybird`, `postgres`) and `peer.service` (same value) | Both keys; `db.system` powers the DB-edge query, `peer.service` powers the main edge query | +| Database call | `Client` or `Producer` | `db.system.name` (e.g. `clickhouse`, `tinybird`, `postgresql`) and `peer.service` | `db.system.name` powers the DB edge; `peer.service` is the logical destination and may differ (for example chDB uses `clickhouse` + `chdb`) | | Message bus produce | `Producer` | `messaging.system`, `messaging.destination.name` | | ### TypeScript ```typescript yield* Effect.annotateCurrentSpan("peer.service", "tinybird") -yield* Effect.annotateCurrentSpan("db.system", "tinybird") +yield* Effect.annotateCurrentSpan("db.system.name", "tinybird") ``` -Canonical example: [apps/api/src/services/WarehouseQueryService.ts](../../../../apps/api/src/services/WarehouseQueryService.ts) `executeSql` — sets both `db.system` and `peer.service` from the resolved warehouse backend. +Canonical example: [packages/query-engine/src/execution/executor.ts](../../../../packages/query-engine/src/execution/executor.ts) `executeSql` — sets both `db.system.name` and `peer.service` from the resolved warehouse backend. ### Rust @@ -89,6 +89,7 @@ Keep these names consistent across services to avoid edge fragmentation on the m |---|---| | `tinybird` | Tinybird hosted warehouse | | `clickhouse` | Self-managed ClickHouse warehouse | +| `chdb` | Embedded chDB behind the local Maple binary | | `collector` | Maple ingest's downstream OTLP collector | | `cloudflare-logpush` | Cloudflare logpush source | | `clerk` | Clerk auth | diff --git a/.agents/skills/maple-telemetry-conventions/rules/span-attributes.md b/.agents/skills/maple-telemetry-conventions/rules/span-attributes.md index ba75e689b..44af9d531 100644 --- a/.agents/skills/maple-telemetry-conventions/rules/span-attributes.md +++ b/.agents/skills/maple-telemetry-conventions/rules/span-attributes.md @@ -14,26 +14,43 @@ Columns: Every SQL execution against Tinybird or ClickHouse emits these. When you add a new attribute to a query path, prefer extending this block over inventing a parallel set. -Source: `apps/api/src/services/WarehouseQueryService.ts:441-510` +Source: `packages/query-engine/src/execution/executor.ts` (`executeSql`) | Key | Type | Set at | Meaning | |---|---|---|---| -| `orgId` | string | `WarehouseQueryService.ts:448` | Tenant org UUID (camelCase — historical, do not rename) | -| `tenant.userId` | string | `WarehouseQueryService.ts:449` | User ID within the tenant | -| `tenant.authMode` | string | `WarehouseQueryService.ts:450` | `"api_key"` / `"user_login"` / etc. | -| `clientSource` | string | `WarehouseQueryService.ts:373, 387` | `"org_override"` or `"managed"` (which config resolved) | -| `db.client` | string | `WarehouseQueryService.ts:374, 389, 405` | `"clickhouse"` or `"tinybird-sdk"` | -| `db.system.name` | string | `WarehouseQueryService.ts:462` | `"clickhouse"` or `"tinybird"` (legacy spans: `db.system`) | -| `db.query.text` | string | `WarehouseQueryService.ts:471` | Full compiled SQL, truncated to 16 KB (legacy spans: `db.statement`) | -| `db.query.length` | int | `WarehouseQueryService.ts:472` | Pre-truncation byte length (legacy: `db.statement.length`) | -| `db.query.truncated` | bool | `WarehouseQueryService.ts:473` | Whether SQL was capped at 16 KB (legacy: `db.statement.truncated`) | -| `db.query.fingerprint` | string | `WarehouseQueryService.ts:474` | 32-bit FNV-1a hash with literals + numbers normalized (legacy: `db.statement.fingerprint`) | -| `db.duration_ms` | int | `WarehouseQueryService.ts:489, 508` | Execution time in ms (emitted on both success and error tap) | -| `query.pipe` | string | `WarehouseQueryService.ts:475` | Original pipe name passed to `sqlQuery()` | -| `query.context` | string | `WarehouseQueryService.ts:476` | Semantic call-site label (e.g. `"errorsByType"`, `"spanHierarchy"`). Set via `SqlQueryOptions.context`. | -| `query.profile` | string | `WarehouseQueryService.ts:477` | Execution profile (e.g. `"list"`, `"analytics"`). Set via `SqlQueryOptions.profile`. | -| `ch.settings` | string (JSON) | `WarehouseQueryService.ts:478` | JSON-encoded ClickHouse settings applied to the query | -| `result.rowCount` | int | `WarehouseQueryService.ts:507` | Number of rows returned | +| `orgId` | string | `executor.ts` | Tenant org UUID (camelCase — historical, do not rename) | +| `tenant.userId` | string | `executor.ts` | User ID within the tenant | +| `tenant.authMode` | string | `executor.ts` | `"api_key"` / `"user_login"` / etc. | +| `clientSource` | string | `executor.ts` | `"org_override"` or `"managed"` (which config resolved) | +| `db.client` | string | `executor.ts` | `"clickhouse"` or `"tinybird-sdk"` | +| `db.system.name` | string | `executor.ts` | `"clickhouse"` or `"tinybird"` (legacy spans: `db.system`) | +| `db.query.text` | string | `executor.ts` | Full compiled SQL, truncated to 16 KB (legacy spans: `db.statement`) | +| `db.query.length` | int | `executor.ts` | Pre-truncation byte length (legacy: `db.statement.length`) | +| `db.query.truncated` | bool | `executor.ts` | Whether SQL was capped at 16 KB (legacy: `db.statement.truncated`) | +| `db.query.fingerprint` | string | `executor.ts` | 32-bit FNV-1a hash with literals + numbers normalized (legacy: `db.statement.fingerprint`) | +| `db.duration_ms` | int | `executor.ts` | Execution time in ms (emitted on both success and error tap) | +| `db.total_duration_ms` | int | `executor.ts` | Total execution-span duration including config resolution and client setup | +| `db.retry.attempts` | int | `executor.ts` | Retries actually performed (not total attempts) | +| `query.pipe` | string | `executor.ts` | Original pipe name passed to `sqlQuery()` | +| `query.context` | string | `executor.ts` | Semantic call-site label (e.g. `"errorsByType"`, `"spanHierarchy"`). Set via `SqlQueryOptions.context`. | +| `query.profile` | string | `executor.ts` | Execution profile (e.g. `"list"`, `"analytics"`). Set via `SqlQueryOptions.profile`. | +| `ch.settings` | string (JSON) | `executor.ts` | JSON-encoded ClickHouse settings applied to the query | +| `result.rowCount` | int | `executor.ts` | Number of rows returned | + +## `warehouse.*` group + +Routing metadata emitted by the shared executor for API and local-CLI queries. + +| Key | Type | Meaning | +|---|---|---| +| `warehouse.backend` | string | Concrete backend kind: `tinybird`, `tinybird-gateway`, `clickhouse`, or `chdb` | +| `warehouse.route` | string | Route purpose: `read`, `raw`, or `ingest` | +| `warehouse.config_source` | string | Config source: `managed`, `org-byo`, or `org-jwt` | + +The historical `clientSource` and `db.client` attributes remain on the same +canonical span. `clientSource` maps `org-byo` to `org_override`; other sources +map to `managed`. `db.client` records the concrete driver family +(`tinybird-sdk` or `clickhouse`). **Rule:** When adding a new query, always pass a `context` string to `SqlQueryOptions` — it becomes filterable as `query.context` in trace search. Don't invent new keys when one of `query.*` fits. diff --git a/apps/api/src/routes/v2/alerts-error-map.test.ts b/apps/api/src/routes/v2/alerts-error-map.test.ts new file mode 100644 index 000000000..68c7e98bf --- /dev/null +++ b/apps/api/src/routes/v2/alerts-error-map.test.ts @@ -0,0 +1,36 @@ +import { assert, describe, it } from "@effect/vitest" +import { Effect } from "effect" +import { WarehouseQuotaExceededError, WarehouseUpstreamError } from "@maple/domain/http" +import { V2RateLimitError, V2ServiceUnavailableError } from "@maple/domain/http/v2" +import { mapAlertError } from "./alerts-error-map" + +describe("mapAlertError", () => { + it.effect("preserves transient warehouse failures as HTTP 503", () => + Effect.gen(function* () { + const error = yield* Effect.fail( + new WarehouseUpstreamError({ + message: "warehouse temporarily unavailable", + pipeName: "listRuleChecks", + upstreamStatus: 503, + }), + ).pipe(mapAlertError("rule_checks_list"), Effect.flip) + + assert.instanceOf(error, V2ServiceUnavailableError) + assert.strictEqual(error.error.code, "alert_rule_checks_list_unavailable") + }), + ) + + it.effect("keeps warehouse quota failures as HTTP 429", () => + Effect.gen(function* () { + const error = yield* Effect.fail( + new WarehouseQuotaExceededError({ + message: "query quota exceeded", + pipeName: "listRuleChecks", + setting: "max_execution_time", + }), + ).pipe(mapAlertError("rule_checks_list"), Effect.flip) + + assert.instanceOf(error, V2RateLimitError) + }), + ) +}) diff --git a/apps/api/src/routes/v2/alerts-error-map.ts b/apps/api/src/routes/v2/alerts-error-map.ts index 84fbf4cbb..1f6442f73 100644 --- a/apps/api/src/routes/v2/alerts-error-map.ts +++ b/apps/api/src/routes/v2/alerts-error-map.ts @@ -7,6 +7,7 @@ import type { AlertValidationError, WarehouseError, WarehouseQuotaExceededError, + WarehouseUpstreamError, WarehouseValidationError, } from "@maple/domain/http" import { @@ -40,7 +41,7 @@ type V2ReachableAlertError = type UpstreamMappedWarehouseError = Exclude< WarehouseError, - WarehouseValidationError | WarehouseQuotaExceededError + WarehouseValidationError | WarehouseQuotaExceededError | WarehouseUpstreamError > type MappedV2AlertError = @@ -49,6 +50,7 @@ type MappedV2AlertError = | (E extends AlertNotFoundError ? V2NotFoundError : never) | (E extends AlertDestinationInUseError ? V2ConflictError : never) | (E extends WarehouseQuotaExceededError ? V2RateLimitError : never) + | (E extends WarehouseUpstreamError ? V2ServiceUnavailableError : never) | (E extends AlertDeliveryError | UpstreamMappedWarehouseError ? V2UpstreamError : never) | (E extends AlertPersistenceError ? V2ServiceUnavailableError : never) @@ -87,7 +89,8 @@ const makeAlertErrorMatcher = (operation: string) => { "@maple/http/errors/AlertDeliveryError": () => upstreamError(`alert_${operation}_upstream_failed`, "The alert provider request failed."), "@maple/http/errors/WarehouseQueryError": warehouseFailure, - "@maple/http/errors/WarehouseUpstreamError": warehouseFailure, + "@maple/http/errors/WarehouseUpstreamError": () => + dependencyUnavailable(`alert_${operation}_unavailable`), "@maple/http/errors/WarehouseAuthError": warehouseFailure, "@maple/http/errors/WarehouseConfigError": warehouseFailure, "@maple/http/errors/WarehouseClientError": warehouseFailure, diff --git a/apps/cli/package.json b/apps/cli/package.json index ed85f0233..ef9db53e1 100644 --- a/apps/cli/package.json +++ b/apps/cli/package.json @@ -7,7 +7,7 @@ "type": "module", "scripts": { "start": "bun run src/bin.ts", - "test": "bun test test/*.test.ts src/server/otlp/encode.test.ts src/core/executor.test.ts", + "test": "bun test test/*.test.ts src/server/otlp/encode.test.ts src/core/*.test.ts", "typecheck": "tsc --noEmit" }, "dependencies": { diff --git a/apps/cli/src/core/executor.test.ts b/apps/cli/src/core/executor.test.ts index 67026fb57..d223dc42f 100644 --- a/apps/cli/src/core/executor.test.ts +++ b/apps/cli/src/core/executor.test.ts @@ -1,5 +1,5 @@ -import { afterEach, describe, expect, it } from "vitest" -import { Effect } from "effect" +import { afterEach, describe, expect, it } from "@effect/vitest" +import { Effect, Exit } from "effect" import { unsafeCompiledQuery } from "@maple/query-engine/ch" import { makeLocalWarehouseExecutorShape } from "./executor" @@ -32,47 +32,53 @@ const stubLocalServer = (rows: ReadonlyArray>) => { } describe("makeLocalWarehouseExecutorShape", () => { - it("posts compiled SQL to /local/query with the trailing FORMAT stripped (chdb dialect)", async () => { - const requests = stubLocalServer([{ c: 1 }]) - const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") - const compiled = unsafeCompiledQuery<{ readonly c: number }>({ - sql: "SELECT count() AS c FROM traces WHERE OrgId = 'local'\nFORMAT JSON", - }) + it.effect("posts compiled SQL to /local/query with the trailing FORMAT stripped (chdb dialect)", () => + Effect.gen(function* () { + const requests = stubLocalServer([{ c: 1 }]) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + const compiled = unsafeCompiledQuery<{ readonly c: number }>({ + sql: "SELECT count() AS c FROM traces WHERE OrgId = 'local'\nFORMAT JSON", + }) - const rows = await Effect.runPromise(shape.compiledQuery(compiled)) + const rows = yield* shape.compiledQuery(compiled) - expect(rows).toEqual([{ c: 1 }]) - expect(requests).toHaveLength(1) - expect(requests[0]!.url).toBe("http://127.0.0.1:4318/local/query") - // The chdb backend speaks the ClickHouse protocol shape: the executor - // strips the trailing FORMAT and the local server owns the output format. - expect(requests[0]!.sql).not.toContain("FORMAT JSON") - expect(requests[0]!.sql).toContain("OrgId = 'local'") - }) + expect(rows).toEqual([{ c: 1 }]) + expect(requests).toHaveLength(1) + expect(requests[0]!.url).toBe("http://127.0.0.1:4318/local/query") + // The chdb backend speaks the ClickHouse protocol shape: the executor + // strips the trailing FORMAT and the local server owns the output format. + expect(requests[0]!.sql).not.toContain("FORMAT JSON") + expect(requests[0]!.sql).toContain("OrgId = 'local'") + }), + ) - it("keeps the executor's OrgId scoping guard for trusted SQL", async () => { - stubLocalServer([]) - const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + it.effect("keeps the executor's OrgId scoping guard for trusted SQL", () => + Effect.gen(function* () { + stubLocalServer([]) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") - const exit = await Effect.runPromiseExit(shape.sqlQuery("SELECT 1")) + const exit = yield* shape.sqlQuery("SELECT 1").pipe(Effect.exit) - expect(exit._tag).toBe("Failure") - }) + expect(Exit.isFailure(exit)).toBe(true) + }), + ) - it("classifies a local 400 query failure without retrying it", async () => { - let attempts = 0 - stubFetch(() => { - attempts += 1 - return new Response("query failed: Unknown expression identifier", { status: 400 }) - }) - const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") + it.effect("classifies a local 400 query failure without retrying it", () => + Effect.gen(function* () { + let attempts = 0 + stubFetch(() => { + attempts += 1 + return new Response("query failed: Unknown expression identifier", { status: 400 }) + }) + const shape = makeLocalWarehouseExecutorShape("http://127.0.0.1:4318") - const exit = await Effect.runPromiseExit( - shape.sqlQuery("SELECT nope FROM traces WHERE OrgId = 'local'"), - ) + const exit = yield* shape + .sqlQuery("SELECT nope FROM traces WHERE OrgId = 'local'") + .pipe(Effect.exit) - expect(exit._tag).toBe("Failure") - // 400 is a non-transient client-side failure — exactly one attempt. - expect(attempts).toBe(1) - }) + expect(Exit.isFailure(exit)).toBe(true) + // 400 is a non-transient client-side failure — exactly one attempt. + expect(attempts).toBe(1) + }), + ) }) diff --git a/apps/cli/src/core/executor.ts b/apps/cli/src/core/executor.ts index 75c896771..3d744c7c0 100644 --- a/apps/cli/src/core/executor.ts +++ b/apps/cli/src/core/executor.ts @@ -1,5 +1,5 @@ import { Effect, Schema } from "effect" -import { OrgId } from "@maple/domain/http" +import { OrgId, UserId } from "@maple/domain/http" import { makeWarehouseExecutor, type WarehouseSqlClient } from "@maple/query-engine/execution" import type { WarehouseExecutorShape } from "@maple/query-engine/observability" import { executeLocalQuery } from "@maple/query-engine/local" @@ -9,8 +9,9 @@ import { debugLog } from "../lib/debug" // OrgId, and every compiled query filters on it. `OrgId` is a non-empty trimmed // branded string, so "local" decodes cleanly (no cast needed). const LOCAL_ORG_ID = Schema.decodeUnknownSync(OrgId)("local") +const LOCAL_USER_ID = Schema.decodeUnknownSync(UserId)("local") -const LOCAL_TENANT = { orgId: LOCAL_ORG_ID, userId: "local", authMode: "local" } +const LOCAL_TENANT = { orgId: LOCAL_ORG_ID, userId: LOCAL_USER_ID, authMode: "local" } // The chDB driver: POST `/local/query` on the local binary. Runs the SQL, // timing the round-trip and (under --debug) printing the SQL + elapsed ms to diff --git a/apps/cli/src/core/operations.test.ts b/apps/cli/src/core/operations.test.ts new file mode 100644 index 000000000..f4f2df345 --- /dev/null +++ b/apps/cli/src/core/operations.test.ts @@ -0,0 +1,54 @@ +import { afterEach, describe, it } from "@effect/vitest" +import { strict as assert } from "node:assert" +import { Effect, Layer, Tracer } from "effect" +import { Mode } from "./mode" +import { rawQuery } from "./operations" + +const realFetch = globalThis.fetch + +afterEach(() => { + globalThis.fetch = realFetch +}) + +const makeRecordingTracer = () => { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + return { spans, tracer } +} + +describe("rawQuery instrumentation", () => { + it.effect("emits the canonical chDB Client span", () => + Effect.gen(function* () { + globalThis.fetch = (async () => + new Response(JSON.stringify([{ value: 1 }]), { + status: 200, + headers: { "content-type": "application/json" }, + })) as unknown as typeof fetch + const { spans, tracer } = makeRecordingTracer() + const modeLayer = Layer.succeed(Mode, { + resolve: Effect.succeed({ _tag: "local" as const, baseUrl: "http://127.0.0.1:4318" }), + }) + + const rows = yield* rawQuery("SELECT 1").pipe( + Effect.provide(modeLayer), + Effect.withTracer(tracer), + ) + + assert.deepStrictEqual(rows, [{ value: 1 }]) + const span = spans.find((candidate) => candidate.name === "WarehouseExecutor.rawQuery") + assert.ok(span) + assert.strictEqual(span.kind, "client") + assert.strictEqual(span.attributes.get("db.system.name"), "clickhouse") + assert.strictEqual(span.attributes.get("peer.service"), "chdb") + assert.strictEqual(span.attributes.get("query.context"), "cli.rawQuery") + assert.strictEqual(span.attributes.get("result.rowCount"), 1) + assert.strictEqual(typeof span.attributes.get("db.duration_ms"), "number") + }), + ) +}) diff --git a/apps/cli/src/core/operations.ts b/apps/cli/src/core/operations.ts index 9c2dd28e6..d4e936855 100644 --- a/apps/cli/src/core/operations.ts +++ b/apps/cli/src/core/operations.ts @@ -3,7 +3,7 @@ // commands call these. Every operation returns an Effect requiring // `WarehouseExecutor` (provided by the local executor layer). -import { Effect } from "effect" +import { Clock, Effect } from "effect" import type { TracesMetric } from "@maple/query-engine" import { WarehouseExecutor, @@ -21,8 +21,9 @@ import { findSlowTraces as obsFindSlowTraces, topOperations as obsTopOperations, } from "@maple/query-engine/observability" -import { WarehouseClientError, WarehouseQueryError } from "@maple/domain/http/warehouse-errors" +import { WarehouseClientError } from "@maple/domain/http/warehouse-errors" import { executeLocalQuery } from "@maple/query-engine/local" +import { fingerprintSql, mapWarehouseError, SQL_TRACE_MAX, truncateSql } from "@maple/query-engine/execution" import { Mode } from "./mode" import type { Range } from "./time" @@ -175,12 +176,50 @@ export const listMetrics = (p: { range: Range; service?: string; search?: string * the warehouse executor (whose `sqlQuery` enforces the OrgId scoping guard) * and posts straight to the single-tenant `/local/query` endpoint. */ +const executeRawLocalQuery = Effect.fn("WarehouseExecutor.rawQuery", { kind: "client" })(function* ( + sql: string, + baseUrl: string, +) { + const startedAtMs = yield* Clock.currentTimeMillis + yield* Effect.annotateCurrentSpan({ + clientSource: "managed", + "db.client": "clickhouse", + "db.system.name": "clickhouse", + "peer.service": "chdb", + "warehouse.backend": "chdb", + "warehouse.route": "raw", + "warehouse.config_source": "managed", + "db.query.text": truncateSql(sql, SQL_TRACE_MAX), + "db.query.length": sql.length, + "db.query.truncated": sql.length > SQL_TRACE_MAX, + "db.query.fingerprint": fingerprintSql(sql), + "query.pipe": "rawSqlQuery", + "query.context": "cli.rawQuery", + }) + const rows = yield* Effect.tryPromise({ + try: () => executeLocalQuery>(sql, baseUrl), + catch: (error) => mapWarehouseError("rawQuery", error), + }).pipe( + Effect.tapError(() => + Clock.currentTimeMillis.pipe( + Effect.flatMap((completedAtMs) => + Effect.annotateCurrentSpan("db.duration_ms", completedAtMs - startedAtMs), + ), + ), + ), + ) + yield* Effect.annotateCurrentSpan("result.rowCount", rows.length) + yield* Effect.annotateCurrentSpan("db.duration_ms", (yield* Clock.currentTimeMillis) - startedAtMs) + return rows +}) + export const rawQuery = (sql: string) => Effect.gen(function* () { const mode = yield* Mode const resolved = yield* mode.resolve.pipe( Effect.mapError( - (error) => new WarehouseClientError({ message: error.message, pipeName: "rawQuery" }), + (error) => + new WarehouseClientError({ message: error.message, pipeName: "rawQuery", cause: error }), ), ) if (resolved._tag !== "local") { @@ -189,14 +228,7 @@ export const rawQuery = (sql: string) => pipeName: "rawQuery", }) } - return yield* Effect.tryPromise({ - try: () => executeLocalQuery>(sql, resolved.baseUrl), - catch: (error) => - new WarehouseQueryError({ - message: error instanceof Error ? error.message : String(error), - pipeName: "rawQuery", - }), - }) + return yield* executeRawLocalQuery(sql, resolved.baseUrl) }) // Custom traces analytics share the `group_by_*` presence-flag convention the diff --git a/apps/cli/src/core/remote-executor.test.ts b/apps/cli/src/core/remote-executor.test.ts new file mode 100644 index 000000000..7f36dcaad --- /dev/null +++ b/apps/cli/src/core/remote-executor.test.ts @@ -0,0 +1,70 @@ +import { afterEach, describe, it } from "@effect/vitest" +import { strict as assert } from "node:assert" +import { Effect, Tracer } from "effect" +import { WarehouseUpstreamError } from "@maple/domain/http/warehouse-errors" +import { makeRemoteWarehouseExecutorShape } from "./remote-executor" + +const realFetch = globalThis.fetch + +afterEach(() => { + globalThis.fetch = realFetch +}) + +const makeRecordingTracer = () => { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + return { spans, tracer } +} + +describe("remote warehouse executor", () => { + it.effect("preserves canonical warehouse error tags from the API", () => + Effect.gen(function* () { + const upstream = new WarehouseUpstreamError({ + message: "warehouse unavailable", + pipeName: "list_services", + upstreamStatus: 503, + }) + globalThis.fetch = (async () => + new Response(JSON.stringify(upstream), { + status: 503, + headers: { "content-type": "application/json" }, + })) as unknown as typeof fetch + const shape = makeRemoteWarehouseExecutorShape("https://api.maple.dev", "test-token", "org_test") + + const error = yield* Effect.flip(shape.query("get_service_usage", {})) + + assert.ok(error instanceof WarehouseUpstreamError) + assert.strictEqual(error.upstreamStatus, 503) + }), + ) + + it.effect("attributes the HTTP peer without inventing a database system", () => + Effect.gen(function* () { + globalThis.fetch = (async () => + new Response(JSON.stringify({ data: [] }), { + status: 200, + headers: { "content-type": "application/json" }, + })) as unknown as typeof fetch + const { spans, tracer } = makeRecordingTracer() + const shape = makeRemoteWarehouseExecutorShape("https://api.maple.dev", "test-token", "org_test") + + yield* shape + .query("get_service_usage", {}, { context: "remoteServices", profile: "list" }) + .pipe(Effect.withTracer(tracer)) + + const span = spans.find((candidate) => candidate.name === "warehouse.query") + assert.ok(span) + assert.strictEqual(span.kind, "client") + assert.strictEqual(span.attributes.get("peer.service"), "maple-api") + assert.strictEqual(span.attributes.get("db.system.name"), undefined) + assert.strictEqual(span.attributes.get("http.request.method"), "POST") + assert.strictEqual(span.attributes.get("query.context"), "remoteServices") + }), + ) +}) diff --git a/apps/cli/src/core/remote-executor.ts b/apps/cli/src/core/remote-executor.ts index 34aa1c9b9..227c4bc25 100644 --- a/apps/cli/src/core/remote-executor.ts +++ b/apps/cli/src/core/remote-executor.ts @@ -1,11 +1,35 @@ -import { Effect, type Option } from "effect" +import { Effect, Option, Schema, type Option as EffectOption } from "effect" import { type WarehouseExecutorShape, type SqlQueryOptions } from "@maple/query-engine/observability" -import { WarehouseClientError, WarehouseQueryError } from "@maple/domain/http/warehouse-errors" +import { + type WarehouseError, + WarehouseClientError, + WarehouseQueryError, + warehouseHttpErrors, +} from "@maple/domain/http/warehouse-errors" import { debugLog } from "../lib/debug" const RAW_SQL_REMOTE_MESSAGE = "Raw SQL (`maple query`) is only available in local mode. In remote mode, use the typed commands (services, traces, errors, logs, timeseries, …)." +const WarehouseErrorSchema = Schema.Union([...warehouseHttpErrors]) +const WarehouseErrorEnvelope = Schema.Struct({ error: WarehouseErrorSchema }) +const decodeWarehouseErrorJson = Schema.decodeUnknownOption(Schema.fromJsonString(WarehouseErrorSchema)) +const decodeWarehouseErrorEnvelopeJson = Schema.decodeUnknownOption( + Schema.fromJsonString(WarehouseErrorEnvelope), +) +const RemoteQueryResponse = Schema.Struct({ data: Schema.optionalKey(Schema.Array(Schema.Unknown)) }) +const decodeRemoteQueryResponseJson = Schema.decodeUnknownSync(Schema.fromJsonString(RemoteQueryResponse)) + +const decodeRemoteWarehouseError = (text: string): WarehouseError | undefined => { + const direct = decodeWarehouseErrorJson(text) + if (Option.isSome(direct)) return direct.value + const envelope = decodeWarehouseErrorEnvelopeJson(text) + return Option.isSome(envelope) ? envelope.value.error : undefined +} + +const unsupported = (pipeName: string): Effect.Effect => + Effect.fail(new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName })) + /** * A `WarehouseExecutor` shape backed by the remote Maple API's generic * `POST /api/tinybird/query` endpoint — the cloud counterpart to the local @@ -26,9 +50,10 @@ export const makeRemoteWarehouseExecutorShape = ( orgId: string, ): WarehouseExecutorShape => { const endpoint = `${apiUrl.replace(/\/$/, "")}/api/tinybird/query` + const serverAddress = new URL(endpoint).hostname return { orgId, - query: (pipe: string, params: Record, _options?: SqlQueryOptions) => + query: (pipe: string, params: Record, options?: SqlQueryOptions) => Effect.tryPromise({ try: async (): Promise<{ data: ReadonlyArray }> => { const started = performance.now() @@ -41,12 +66,14 @@ export const makeRemoteWarehouseExecutorShape = ( }, body: JSON.stringify({ pipe, params }), }) + const text = await res.text() if (!res.ok) { - const text = await res.text().catch(() => "") + const decoded = decodeRemoteWarehouseError(text) + if (decoded !== undefined) throw decoded throw new Error(`HTTP ${res.status}${text ? `: ${text}` : ""}`) } - const json = (await res.json()) as { data?: ReadonlyArray } - return { data: json.data ?? [] } + const json = decodeRemoteQueryResponseJson(text) + return { data: (json.data ?? []) as ReadonlyArray } } finally { // Server-side SQL isn't returned; log the pipe + params instead. debugLog( @@ -55,33 +82,35 @@ export const makeRemoteWarehouseExecutorShape = ( ) } }, - catch: (error) => - new WarehouseQueryError({ - message: error instanceof Error ? error.message : String(error), - pipeName: pipe, - }), + catch: (error) => { + const decoded = Schema.decodeUnknownOption(WarehouseErrorSchema)(error) + return Option.isSome(decoded) + ? decoded.value + : new WarehouseQueryError({ + message: error instanceof Error ? error.message : String(error), + pipeName: pipe, + cause: error, + }) + }, }).pipe( Effect.tap((result) => Effect.annotateCurrentSpan({ "result.rowCount": result.data.length })), Effect.withSpan("warehouse.query", { kind: "client", attributes: { "peer.service": "maple-api", - "db.system.name": "clickhouse", - "query.context": pipe, + "http.request.method": "POST", + "url.full": endpoint, + "server.address": serverAddress, + "query.context": options?.context ?? pipe, + "query.profile": options?.profile, }, }), ), sqlQuery: >(_sql: string, _options?: SqlQueryOptions) => - Effect.fail( - new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "sqlQuery" }), - ) as Effect.Effect, WarehouseClientError>, + unsupported>("sqlQuery"), compiledQuery: (_compiled: unknown, _options?: SqlQueryOptions) => - Effect.fail( - new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "compiledQuery" }), - ) as Effect.Effect, WarehouseClientError>, + unsupported>("compiledQuery"), compiledQueryFirst: (_compiled: unknown, _options?: SqlQueryOptions) => - Effect.fail( - new WarehouseClientError({ message: RAW_SQL_REMOTE_MESSAGE, pipeName: "compiledQueryFirst" }), - ) as Effect.Effect, WarehouseClientError>, + unsupported>("compiledQueryFirst"), } } diff --git a/apps/cli/src/server/serve.ts b/apps/cli/src/server/serve.ts index 7f87e47e8..5ee6b1f0a 100644 --- a/apps/cli/src/server/serve.ts +++ b/apps/cli/src/server/serve.ts @@ -278,7 +278,7 @@ const truncateSql = (sql: string) => (sql.length > MAX_DB_QUERY_TEXT ? sql.slice * `startServer`). The effect always succeeds with a `Response`. */ type SpanRunner = (effect: Effect.Effect) => Promise -// A rejected (4xx/5xx) ingest/query response, surfaced through the Effect error +// A rejected 5xx ingest/query response, surfaced through the Effect error // channel. `message` carries the handler's descriptive body so the span records // a real `exception.message`; `response` is the original, untouched response we // hand back to the client in `recoverResponse`. (Failing with a bare `Response` @@ -290,19 +290,20 @@ class IngestRejected extends Schema.TaggedErrorClass()("@maple/c message: Schema.String, }) {} -// The Effect tracer derives span status from the effect's outcome — success → -// `Ok`, failure → `Error` (conventions say never set the status string by hand -// in TS). So to mark a 4xx/5xx span `Error`, we fail *inside* the span with an -// `IngestRejected` carrying the reason, then recover the original response with -// `Effect.match` *outside* the span — the span has already closed `Error` by then. -const failIfError = (response: Response): Effect.Effect => +// The Effect tracer derives span status from the effect's outcome. OTel HTTP +// Server semantics treat 4xx as a successful server outcome (the caller sent a +// bad request) and only 5xx as Error. Annotate every rejection, but fail inside +// the Server span only for 5xx; recover the original response outside the span. +const recordServerResponse = (response: Response): Effect.Effect => Effect.gen(function* () { if (response.status >= 400) { // Clone to read the body without consuming the response we return below. const body = (yield* Effect.promise(() => response.clone().text())).trim() const message = body.length > 0 ? body : `HTTP ${response.status}` yield* Effect.annotateCurrentSpan({ "error.type": `HTTP ${response.status}` }) - return yield* new IngestRejected({ response, status: response.status, message }) + if (response.status >= 500) { + return yield* new IngestRejected({ response, status: response.status, message }) + } } return response }) @@ -324,7 +325,7 @@ const ingestSpan = (runSpan: SpanRunner, db: Chdb, signal: Signal, req: Request) "maple.ingest.item_count": accepted, "http.response.status_code": response.status, }) - return yield* failIfError(response) + return yield* recordServerResponse(response) }).pipe( Effect.withSpan(`POST /v1/${signal}`, { kind: "server", @@ -353,7 +354,7 @@ const querySpan = (runSpan: SpanRunner, db: Chdb, req: Request): Promise { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + return { spans, tracer } +} + +describe("local HTTP server span status", () => { + it.effect("records a 4xx response as a successful Server span", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + const response = yield* __testables + .recordServerResponse(new Response("invalid SQL", { status: 400 })) + .pipe(Effect.withSpan("POST /local/query", { kind: "server" }), Effect.withTracer(tracer)) + + strictEqual(response.status, 400) + const span = spans.find((candidate) => candidate.name === "POST /local/query") + ok(span) + strictEqual(span.attributes.get("error.type"), "HTTP 400") + ok(span.status._tag === "Ended" && Exit.isSuccess(span.status.exit)) + }), + ) + + it.effect("records a 5xx response as a failed Server span", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + const exit = yield* __testables + .recordServerResponse(new Response("database unavailable", { status: 503 })) + .pipe( + Effect.withSpan("POST /local/query", { kind: "server" }), + Effect.withTracer(tracer), + Effect.exit, + ) + + ok(Exit.isFailure(exit)) + const span = spans.find((candidate) => candidate.name === "POST /local/query") + ok(span) + ok(span.status._tag === "Ended" && Exit.isFailure(span.status.exit)) + }), + ) +}) + describe("local listener addresses", () => { it("reaches an IPv4 wildcard listener through the loopback probe URL", async () => { const server = Bun.serve({ diff --git a/packages/domain/src/http/observability.ts b/packages/domain/src/http/observability.ts index b59bde8fd..2b7ffe915 100644 --- a/packages/domain/src/http/observability.ts +++ b/packages/domain/src/http/observability.ts @@ -97,7 +97,7 @@ const LogEntry = Schema.Struct({ }) interface SpanNodeResponse { - readonly spanId: string + readonly spanId: Schema.Schema.Type readonly parentSpanId: string readonly spanName: string readonly serviceName: string @@ -109,7 +109,20 @@ interface SpanNodeResponse { readonly children: ReadonlyArray } -const SpanNode: Schema.Codec = Schema.Struct({ +interface SpanNodeEncoded { + readonly spanId: string + readonly parentSpanId: string + readonly spanName: string + readonly serviceName: string + readonly durationMs: number + readonly statusCode: string + readonly statusMessage: string + readonly attributes: Record + readonly resourceAttributes: Record + readonly children: ReadonlyArray +} + +const SpanNode: Schema.Codec = Schema.Struct({ spanId: SpanId, parentSpanId: Schema.String, spanName: Schema.String, @@ -119,7 +132,7 @@ const SpanNode: Schema.Codec = Schema.Struct({ statusMessage: Schema.String, attributes: Schema.Record(Schema.String, Schema.String), resourceAttributes: Schema.Record(Schema.String, Schema.String), - children: Schema.Array(Schema.suspend((): Schema.Codec => SpanNode)), + children: Schema.Array(Schema.suspend((): Schema.Codec => SpanNode)), }) const InspectTraceResponse = Schema.Struct({ @@ -197,7 +210,7 @@ const SearchLogsRequest = Schema.Struct({ service: Schema.optionalKey(Schema.String), severity: Schema.optionalKey(Schema.String), search: Schema.optionalKey(Schema.String), - traceId: Schema.optionalKey(Schema.String), + traceId: Schema.optionalKey(TraceId), limit: Schema.optionalKey(Schema.Number), offset: Schema.optionalKey(Schema.Number), }) diff --git a/packages/query-engine/src/execution/backend.ts b/packages/query-engine/src/execution/backend.ts index aadafff58..26a3f0fb8 100644 --- a/packages/query-engine/src/execution/backend.ts +++ b/packages/query-engine/src/execution/backend.ts @@ -30,6 +30,8 @@ export type ResolvedWarehouseConfig = TinybirdBackendConfig | ClickHouseProtocol export interface WarehouseBackendDialect { readonly driver: "tinybird-sdk" | "clickhouse-web" + /** Legacy driver label emitted as `db.client` on canonical query spans. */ + readonly dbClient: "tinybird-sdk" | "clickhouse" /** * Tinybird's managed warehouse rejects some ClickHouse settings (e.g. * "Usage of setting 'max_block_size' is restricted") — through the SDK AND @@ -42,11 +44,11 @@ export interface WarehouseBackendDialect { */ readonly normalizeSqlForClient: boolean /** - * Value for the `db.system.name` / `peer.service` span attributes. The - * gateway reports `clickhouse` (it is addressed over the ClickHouse - * protocol) — service-map DB nodes key on this value, so it must stay - * stable per deployment. + * OTel database-system identity. chDB implements the ClickHouse interface, + * so its system remains `clickhouse` even though the logical peer is `chdb`. */ + readonly dbSystemName: "tinybird" | "clickhouse" + /** Logical destination used by the service-map `peer.service` edge. */ readonly peerService: string } @@ -54,26 +56,34 @@ export interface WarehouseBackendDialect { export const BackendDialect: Record = { tinybird: { driver: "tinybird-sdk", + dbClient: "tinybird-sdk", stripTinybirdRestrictedSettings: true, normalizeSqlForClient: false, + dbSystemName: "tinybird", peerService: "tinybird", }, "tinybird-gateway": { driver: "clickhouse-web", + dbClient: "clickhouse", stripTinybirdRestrictedSettings: true, normalizeSqlForClient: true, + dbSystemName: "clickhouse", peerService: "clickhouse", }, clickhouse: { driver: "clickhouse-web", + dbClient: "clickhouse", stripTinybirdRestrictedSettings: false, normalizeSqlForClient: true, + dbSystemName: "clickhouse", peerService: "clickhouse", }, chdb: { driver: "clickhouse-web", + dbClient: "clickhouse", stripTinybirdRestrictedSettings: false, normalizeSqlForClient: true, + dbSystemName: "clickhouse", peerService: "chdb", }, } diff --git a/packages/query-engine/src/execution/executor.test.ts b/packages/query-engine/src/execution/executor.test.ts index 90b2de59e..ce44c5184 100644 --- a/packages/query-engine/src/execution/executor.test.ts +++ b/packages/query-engine/src/execution/executor.test.ts @@ -1,7 +1,7 @@ import { assert, describe, it } from "@effect/vitest" -import { Duration, Effect, Ref } from "effect" +import { Duration, Effect, Ref, Schema, Tracer } from "effect" import { TestClock } from "effect/testing" -import type { OrgId } from "@maple/domain" +import { OrgId, UserId } from "@maple/domain" import { RawSqlValidationError } from "@maple/domain/http" import { compile, listRuleChecksQuery, unsafeCompiledQuery } from "../ch" import { makeWarehouseExecutor } from "./executor" @@ -14,11 +14,23 @@ import type { } from "./ports" const tenant: ExecutionTenant = { - orgId: "org_test" as OrgId, - userId: "user_test", + orgId: Schema.decodeUnknownSync(OrgId)("org_test"), + userId: Schema.decodeUnknownSync(UserId)("user_test"), authMode: "system", } +const makeRecordingTracer = () => { + const spans: Array = [] + const tracer = Tracer.make({ + span(options) { + const span = new Tracer.NativeSpan(options) + spans.push(span) + return span + }, + }) + return { spans, tracer } +} + // A per-org BYO read override (the read path) vs the managed Tinybird ingest // pipeline (the write path). alert_checks rows only ever land in the latter. const clickhouseConfig: ResolvedWarehouseConfig = { @@ -37,6 +49,10 @@ const tinybirdGatewayConfig: ResolvedWarehouseConfig = { ...clickhouseConfig, kind: "tinybird-gateway", } +const chdbConfig: ResolvedWarehouseConfig = { + ...clickhouseConfig, + kind: "chdb", +} // listRuleChecksQuery declares .routing("ingest") at its definition — // alert_checks only exists in the managed ingest pipeline. @@ -106,6 +122,102 @@ describe("makeWarehouseExecutor ingest routing", () => { ) }) +describe("makeWarehouseExecutor span instrumentation", () => { + it.effect("emits the canonical SQL attributes on the Client span", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + const executor = makeWarehouseExecutor(makeDeps([])) + + yield* executor + .sqlQuery(tenant, "SELECT 1 WHERE OrgId = 'org_test'", { + profile: "list", + context: "spanContract", + }) + .pipe(Effect.withTracer(tracer)) + + const span = spans.find((candidate) => candidate.name === "WarehouseQueryService.executeSql") + assert.isDefined(span) + assert.strictEqual(span.kind, "client") + assert.strictEqual(span.attributes.get("orgId"), "org_test") + assert.strictEqual(span.attributes.get("tenant.userId"), "user_test") + assert.strictEqual(span.attributes.get("tenant.authMode"), "system") + assert.strictEqual(span.attributes.get("clientSource"), "org_override") + assert.strictEqual(span.attributes.get("db.client"), "clickhouse") + assert.strictEqual(span.attributes.get("db.system.name"), "clickhouse") + assert.strictEqual(span.attributes.get("peer.service"), "clickhouse") + assert.strictEqual(span.attributes.get("query.context"), "spanContract") + assert.strictEqual(span.attributes.get("query.profile"), "list") + assert.strictEqual(span.attributes.get("result.rowCount"), 0) + assert.isNumber(span.attributes.get("db.duration_ms")) + assert.match(span.attributes.get("db.query.fingerprint") as string, /^[0-9a-f]{8}$/) + }), + ) + + it.effect("keeps chDB as the peer while reporting ClickHouse as the DB system", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + const executor = makeWarehouseExecutor( + makeRecordingDeps({ config: chdbConfig, clientCacheKey: "local" }, []), + ) + + yield* executor + .sqlQuery(tenant, "SELECT 1 WHERE OrgId = 'org_test'", { context: "localSpan" }) + .pipe(Effect.withTracer(tracer)) + + const span = spans.find((candidate) => candidate.name === "WarehouseQueryService.executeSql") + assert.isDefined(span) + assert.strictEqual(span.attributes.get("db.system.name"), "clickhouse") + assert.strictEqual(span.attributes.get("peer.service"), "chdb") + }), + ) + + it.effect("wraps warehouse inserts in an attributed Client span", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + const executor = makeWarehouseExecutor(makeDeps([])) + + yield* executor + .ingest(tenant, "alert_checks", [{ OrgId: "org_test" }]) + .pipe(Effect.withTracer(tracer)) + + const span = spans.find((candidate) => candidate.name === "WarehouseQueryService.insert") + assert.isDefined(span) + assert.strictEqual(span.kind, "client") + assert.strictEqual(span.attributes.get("db.system.name"), "tinybird") + assert.strictEqual(span.attributes.get("peer.service"), "tinybird") + assert.strictEqual(span.attributes.get("db.client"), "tinybird-sdk") + assert.strictEqual(span.attributes.get("result.rowCount"), 1) + }), + ) + + it.live("reports retries performed rather than failed attempts on terminal failure", () => + Effect.gen(function* () { + const { spans, tracer } = makeRecordingTracer() + let attempts = 0 + const executor = makeWarehouseExecutor({ + ...makeDeps([]), + createClient: () => ({ + sql: async () => { + attempts += 1 + throw new Error("HTTP status 503 service temporarily unavailable") + }, + insert: async () => {}, + }), + }) + + const exit = yield* executor + .sqlQuery(tenant, "SELECT 1 WHERE OrgId = 'org_test'", { context: "retrySpan" }) + .pipe(Effect.withTracer(tracer), Effect.exit) + + assert.strictEqual(exit._tag, "Failure") + assert.strictEqual(attempts, 3) + const span = spans.find((candidate) => candidate.name === "WarehouseQueryService.executeSql") + assert.isDefined(span) + assert.strictEqual(span.attributes.get("db.retry.attempts"), 2) + }), + ) +}) + // Capture the final SQL the executor hands to the client so a test can assert // whether a Tinybird-restricted setting (max_block_size) survived the strip for // the resolved backend. The Tinybird gateway (`tinybird-gateway`) uses the diff --git a/packages/query-engine/src/execution/executor.ts b/packages/query-engine/src/execution/executor.ts index a18eaf157..70a00d245 100644 --- a/packages/query-engine/src/execution/executor.ts +++ b/packages/query-engine/src/execution/executor.ts @@ -18,7 +18,7 @@ import { resolveSettings, stripTinybirdRestrictedSettings, } from "../profiles" -import { mapWarehouseError, toWarehouseQueryError, type WarehouseSqlError } from "./errors" +import { mapWarehouseError, toWarehouseQueryError } from "./errors" import { WarehouseResponseLimitError } from "./response-limits" import { SQL_LOG_MAX, @@ -156,7 +156,7 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue const pinnedTable = findIngestPinnedTable(sql) if (pinnedTable !== undefined) { yield* Effect.logWarning( - "Query reads an ingest-pinned datasource from a BYO ClickHouse — declare .routing(\"ingest\") at the query definition", + 'Query reads an ingest-pinned datasource from a BYO ClickHouse — declare .routing("ingest") at the query definition', { pipe, table: pinnedTable, orgId: tenant.orgId }, ) } @@ -165,10 +165,13 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue // dashboards move to the `warehouse.*` attributes. if (purpose === "ingest") yield* Effect.annotateCurrentSpan("query.routing", "ingest") const dialect = BackendDialect[resolved.config.kind] + const clientSource = resolved.source === "org-byo" ? "org_override" : "managed" yield* Effect.annotateCurrentSpan("warehouse.backend", resolved.config.kind) yield* Effect.annotateCurrentSpan("warehouse.route", purpose) yield* Effect.annotateCurrentSpan("warehouse.config_source", resolved.source) - yield* Effect.annotateCurrentSpan("db.system.name", dialect.peerService) + yield* Effect.annotateCurrentSpan("clientSource", clientSource) + yield* Effect.annotateCurrentSpan("db.client", dialect.dbClient) + yield* Effect.annotateCurrentSpan("db.system.name", dialect.dbSystemName) yield* Effect.annotateCurrentSpan("peer.service", dialect.peerService) const settings = dialect.stripTinybirdRestrictedSettings ? stripTinybirdRestrictedSettings(resolveSettings(options)) @@ -247,7 +250,10 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue const nowMs = yield* Clock.currentTimeMillis const elapsedMs = nowMs - sqlStartedMs const totalElapsedMs = nowMs - startedAtMs - const attempts = yield* Ref.get(retryAttempts) + const failedTransientAttempts = yield* Ref.get(retryAttempts) + const attempts = isTransientUpstreamError(error) + ? Math.max(0, failedTransientAttempts - 1) + : failedTransientAttempts yield* Effect.annotateCurrentSpan("db.duration_ms", elapsedMs) yield* Effect.annotateCurrentSpan("db.total_duration_ms", totalElapsedMs) yield* Effect.annotateCurrentSpan("db.retry.attempts", attempts) @@ -428,9 +434,9 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue // Tinybird pipeline, never a per-org BYO ClickHouse READ override (routing // writes through the override 500'd every insert and broke demo-seed // onboarding). - const resolved = yield* deps.resolveRoute(tenant, "ingest", label).pipe( - Effect.mapError((error) => toWarehouseQueryError(label, error)), - ) + const resolved = yield* deps.resolveRoute(tenant, "ingest", label) + const dialect = BackendDialect[resolved.config.kind] + const clientSource = resolved.source === "org-byo" ? "org_override" : "managed" yield* Effect.annotateCurrentSpan("warehouse.backend", resolved.config.kind) yield* Effect.annotateCurrentSpan("warehouse.route", "ingest") yield* Effect.annotateCurrentSpan("warehouse.config_source", resolved.source) @@ -445,6 +451,7 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue resolved.config, yield* Clock.currentTimeMillis, ) + const insertStartedAtMs = yield* Clock.currentTimeMillis yield* Effect.tryPromise({ try: () => client.insert(datasource, rows), @@ -452,15 +459,48 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue // insert surfaces with its real tag instead of a generic query error. catch: (error) => mapWarehouseError(label, error), }).pipe( + Effect.tap(() => + Clock.currentTimeMillis.pipe( + Effect.flatMap((completedAtMs) => + Effect.annotateCurrentSpan({ + "db.duration_ms": completedAtMs - insertStartedAtMs, + "result.rowCount": rows.length, + }), + ), + ), + ), Effect.tapError((error) => - Effect.logError("WarehouseQueryService.ingest failed", { - datasource, - rowCount: rows.length, - backend: resolved.config.kind, - errorTag: error._tag, - message: error.message, - }), + Clock.currentTimeMillis.pipe( + Effect.flatMap((completedAtMs) => + Effect.annotateCurrentSpan("db.duration_ms", completedAtMs - insertStartedAtMs), + ), + Effect.andThen( + Effect.logError("WarehouseQueryService.ingest failed", { + datasource, + rowCount: rows.length, + backend: resolved.config.kind, + errorTag: error._tag, + message: error.message, + }), + ), + ), ), + Effect.withSpan("WarehouseQueryService.insert", { + kind: "client", + attributes: { + orgId: tenant.orgId, + "tenant.userId": tenant.userId, + "tenant.authMode": tenant.authMode, + clientSource, + "db.client": dialect.dbClient, + "db.system.name": dialect.dbSystemName, + "peer.service": dialect.peerService, + "warehouse.backend": resolved.config.kind, + "warehouse.route": "ingest", + "warehouse.config_source": resolved.source, + datasource, + }, + }), ) }) diff --git a/packages/query-engine/src/execution/ports.ts b/packages/query-engine/src/execution/ports.ts index 08097df36..4025fd72e 100644 --- a/packages/query-engine/src/execution/ports.ts +++ b/packages/query-engine/src/execution/ports.ts @@ -1,10 +1,9 @@ import type { Effect, Option } from "effect" -import type { OrgId } from "@maple/domain" +import type { OrgId, UserId } from "@maple/domain" import type { RawSqlValidationError, WarehouseQueryRequest, WarehouseQueryResponse, - WarehouseQueryError, WarehouseValidationError, } from "@maple/domain/http" import type { ResolvedWarehouseConfig } from "./backend" @@ -16,7 +15,7 @@ import type { WarehouseSqlError } from "./errors" /** The minimal tenant surface the executor reads (org scope + identity for spans). */ export interface ExecutionTenant { readonly orgId: OrgId - readonly userId: string + readonly userId: UserId readonly authMode: string }