Skip to content

Rearchitect the warehouse executor: named backends, one route resolver, structural ingest routing #231

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Makisuo merged 2 commits into from
Jul 19, 2026
Merged

Rearchitect the warehouse executor: named backends, one route resolver, structural ingest routing #231

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
230d269
refactor(query-engine): rearchitect warehouse executor around named b…
Makisuo Jul 18, 2026
c6c72c3
w
Makisuo Jul 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 5 additions & 4 deletions .agents/skills/maple-telemetry-conventions/rules/service-map-attribution.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down Expand Up @@ -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

Expand All @@ -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 |
Expand Down
51 changes: 34 additions & 17 deletions .agents/skills/maple-telemetry-conventions/rules/span-attributes.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.

Expand Down
90 changes: 35 additions & 55 deletions apps/api/src/lib/WarehouseQueryService.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 })
Expand All @@ -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))
Expand All @@ -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))
})
Expand Down Expand Up @@ -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))
})
Expand Down Expand Up @@ -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",
Expand Down Expand Up @@ -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",
}
Expand Down Expand Up @@ -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",
Expand All @@ -722,63 +715,48 @@ 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<string> = []
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()

return Effect.gen(function* () {
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", () => {
Expand All @@ -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 })
},
}))

Expand All @@ -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))
})
Expand Down
Loading
Loading