diff --git a/apps/api/src/routes/v1/session-replay.schema.test.ts b/apps/api/src/routes/v1/session-replay.schema.test.ts index e39c85356..4b2498aef 100644 --- a/apps/api/src/routes/v1/session-replay.schema.test.ts +++ b/apps/api/src/routes/v1/session-replay.schema.test.ts @@ -20,6 +20,12 @@ const baseRow = { durationMs: null, status: "ended", userId: null, + // `""` is the never-identified state (see SessionReplayListItem), which is + // what an anonymous `userId: null` row carries. + userName: "", + userEmail: "", + groupId: "", + groupName: "", visitorId: "b0f2b0c6-8a9f-4a1f-9a0e-2b4f9e2f1a77", utmSource: "", entryPath: "/", diff --git a/apps/api/src/services/dashboards/ServiceMapRollupService.ts b/apps/api/src/services/dashboards/ServiceMapRollupService.ts index 1c3f8172b..ecfb480d9 100644 --- a/apps/api/src/services/dashboards/ServiceMapRollupService.ts +++ b/apps/api/src/services/dashboards/ServiceMapRollupService.ts @@ -1,5 +1,5 @@ import { RoleName, UserId as UserIdSchema, type OrgId } from "@maple/domain/http" -import { orgIngestKeys } from "@maple/db" +import { orgClickHouseSettings, orgIngestKeys } from "@maple/db" import * as CH from "@maple/query-engine/ch" import { Clock, Cause, Context, Effect, Layer, Schema } from "effect" import type { TenantContext } from "@/services/auth/AuthService" @@ -15,12 +15,24 @@ const HOUR_MS = 3_600_000 /** * How many completed hours back the rollup re-checks on every run. Bounds the * per-run cost to a constant and lets a few missed cron ticks catch up. Hours - * already present in `service_map_edges_hourly` are skipped, so re-checking is - * cheap; an hour with genuinely zero cross-service calls is re-attempted each - * run until it ages out of this window — also cheap (an empty join). + * already present in `service_map_edges_hourly` are skipped; an hour with + * genuinely zero cross-service calls is re-attempted each run until it ages out + * of this window. + * + * That re-attempt is NOT free — the join is empty only *after* it has scanned + * the hour's spans. Keep the org set gated (see `resolveActiveOrgs`): when the + * rollup fanned out across every org that ever held an ingest key, this window + * turned ~260 orgs into ~6 executions each per tick, of two queries, one of + * them a raw-`traces` self-join. */ const LOOKBACK_HOURS = 6 +/** + * Discovery window for the active-org scan. Must be a SUPERSET of the per-org + * lookback so no org that produced spans in a candidate hour is skipped. + */ +const ACTIVE_DISCOVERY_HOURS = LOOKBACK_HOURS + 2 + /** Concurrency for per-org rollup processing. */ const ORG_CONCURRENCY = 4 @@ -149,8 +161,26 @@ export class ServiceMapRollupService extends Context.Service< // ingest can fail independently. Recompute resolution rows for sealed // hours in the bounded lookback as an idempotent repair pass // (ReplacingMergeTree deduplicates the same mapping key). + // + // Only for hours that have no resolution rows yet: the repair join + // reads raw `traces`, so re-running it for every sealed hour on every + // tick was the single most expensive thing this service did. + const resolvedRows = yield* warehouse.compiledQuery( + tenant, + CH.serviceMapResolutionsExistingHoursSQL({ + orgId, + startTime: formatWarehouseDateTime(oldestHourMs), + endTime: formatWarehouseDateTime(currentHourMs), + }), + { context: "serviceMapResolutionsExistingHours" }, + ) + const resolved = new Set(resolvedRows.map((row) => Number(row.hourTs))) + yield* Effect.forEach( - candidates.filter((hourMs) => existing.has(Math.floor(hourMs / 1000))), + candidates.filter( + (hourMs) => + existing.has(Math.floor(hourMs / 1000)) && !resolved.has(Math.floor(hourMs / 1000)), + ), (hourMs) => Effect.gen(function* () { const resolutionsRows = yield* warehouse.compiledQuery( @@ -182,6 +212,70 @@ export class ServiceMapRollupService extends Context.Service< } }) + /** + * Which orgs actually produced spans in the discovery window. + * + * One cross-org scan of `traces_aggregates_hourly` replaces per-org + * guessing — the same gating the error-issue and anomaly ticks already + * use (`packages/query-engine/src/ch/queries/activity.ts`). Idle orgs + * have no spans to join, so every hour they are handed costs two + * warehouse scans to prove a join is empty. + * + * BYO-ClickHouse orgs are invisible to this managed-workspace scan and + * are always processed. + * + * Fails OPEN, unlike the anomaly tick: a skipped rollup hour is a + * permanent gap in the service map once it ages past `LOOKBACK_HOURS`, + * whereas a skipped anomaly evaluation is only a late alert. A discovery + * outage costs one hour of the old fan-out; failing closed could cost + * the edges outright. + */ + const resolveActiveOrgs = Effect.fn("ServiceMapRollupService.resolveActiveOrgs")(function* ( + knownOrgs: ReadonlyArray, + ) { + if (knownOrgs.length === 0) return undefined + + const nowMs = yield* Clock.currentTimeMillis + const startTime = formatWarehouseDateTime(nowMs - ACTIVE_DISCOVERY_HOURS * HOUR_MS) + const byoRows = yield* database + .execute((db) => + db.selectDistinct({ orgId: orgClickHouseSettings.orgId }).from(orgClickHouseSettings), + ) + .pipe(Effect.orElseSucceed(() => [] as ReadonlyArray<{ orgId: string }>)) + + return yield* warehouse + .crossOrgQuery( + systemTenant(knownOrgs[0]!), + CH.compile(CH.activeOrgsByTracesQuery(), { startTime }), + { + profile: "discovery", + context: "serviceMapRollupActiveOrgs", + justification: + "enumerate orgs with recent span aggregates so the hourly service-map rollup skips idle orgs", + }, + ) + .pipe( + Effect.map((rows) => { + const active = new Set(byoRows.map((row) => row.orgId)) + for (const row of rows) { + const orgId = String((row as { orgId?: unknown }).orgId ?? "") + if (orgId) active.add(orgId) + } + return active as ReadonlySet + }), + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.as( + Effect.logWarning( + "Service map rollup active-org discovery failed; processing every known org", + ).pipe(Effect.annotateLogs({ error: Cause.pretty(cause) })), + undefined, + ), + ), + ) + }) + const runRollupTick: ServiceMapRollupServiceShape["runRollupTick"] = Effect.fn( "ServiceMapRollupService.runRollupTick", )(function* () { @@ -189,10 +283,20 @@ export class ServiceMapRollupService extends Context.Service< db.selectDistinct({ orgId: orgIngestKeys.orgId }).from(orgIngestKeys), ) + const knownOrgs = orgRows.map((row) => row.orgId as OrgId) + const active = yield* resolveActiveOrgs(knownOrgs) + const targetOrgs = + active === undefined ? knownOrgs : knownOrgs.filter((orgId) => active.has(orgId)) + yield* Effect.annotateCurrentSpan({ + knownOrgs: knownOrgs.length, + targetOrgs: targetOrgs.length, + activeOrgDiscovery: active === undefined ? "failed" : "ok", + }) + const results = yield* Effect.forEach( - orgRows, - (row) => - processOrg(row.orgId as OrgId).pipe( + targetOrgs, + (orgId) => + processOrg(orgId).pipe( // Interrupts (isolate teardown) are NOT failures — re-raise them // so the tick cancels promptly instead of logging phantom // per-org failures. @@ -202,7 +306,7 @@ export class ServiceMapRollupService extends Context.Service< : Effect.as( Effect.logError("Service map rollup failed for org").pipe( Effect.annotateLogs({ - orgId: row.orgId, + orgId, error: Cause.pretty(cause), }), ), @@ -221,7 +325,7 @@ export class ServiceMapRollupService extends Context.Service< ) return { - orgsProcessed: orgRows.length, + orgsProcessed: targetOrgs.length, hoursRolledUp: results.reduce((sum, r) => sum + r.hoursRolledUp, 0), edgesWritten: results.reduce((sum, r) => sum + r.edgesWritten, 0), resolutionsWritten: results.reduce((sum, r) => sum + r.resolutionsWritten, 0), diff --git a/apps/cli/src/server/schema/local-inserts.json b/apps/cli/src/server/schema/local-inserts.json index 54ae65f31..1df89201d 100644 --- a/apps/cli/src/server/schema/local-inserts.json +++ b/apps/cli/src/server/schema/local-inserts.json @@ -1,5 +1,5 @@ { - "projectRevision": "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06", + "projectRevision": "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa", "orgPlaceholder": "__ORG__", "datasources": { "traces": { diff --git a/apps/cli/src/server/schema/local-schema.sql b/apps/cli/src/server/schema/local-schema.sql index e788933c7..d5764ec42 100644 --- a/apps/cli/src/server/schema/local-schema.sql +++ b/apps/cli/src/server/schema/local-schema.sql @@ -1,6 +1,6 @@ -- This file is generated by scripts/generate-clickhouse-schema-sql.ts -- Do not edit manually. Run `bun run clickhouse:schema` to regenerate. --- projectRevision: 6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06 +-- projectRevision: 12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa -- localSchemaVersion: 1 CREATE TABLE IF NOT EXISTS alert_checks ( diff --git a/apps/ingest/src/clickhouse_insert_mappings.rs b/apps/ingest/src/clickhouse_insert_mappings.rs index 00e4963b1..00020bb5b 100644 --- a/apps/ingest/src/clickhouse_insert_mappings.rs +++ b/apps/ingest/src/clickhouse_insert_mappings.rs @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-clickhouse-insert-mappings.ts // Do not edit manually. -pub const PROJECT_REVISION: &str = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06"; +pub const PROJECT_REVISION: &str = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa"; // Gate for BYO-ClickHouse ingest readiness — the migration version, NOT the // Tinybird-coupled PROJECT_REVISION. Compared against // org_clickhouse_settings.schema_version. See @maple/domain/clickhouse diff --git a/packages/domain/src/generated/clickhouse-schema.ts b/packages/domain/src/generated/clickhouse-schema.ts index 75ecf5b70..32885dd85 100644 --- a/packages/domain/src/generated/clickhouse-schema.ts +++ b/packages/domain/src/generated/clickhouse-schema.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-clickhouse-schema.ts // Do not edit manually. -export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const +export const projectRevision = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa" as const export const latestSnapshotStatements: ReadonlyArray = [ "CREATE TABLE IF NOT EXISTS alert_checks (\n OrgId LowCardinality(String),\n RuleId String,\n GroupKey String,\n Timestamp DateTime64(3),\n Status LowCardinality(String),\n SignalType LowCardinality(String),\n Comparator LowCardinality(String),\n Threshold Float64,\n ObservedValue Nullable(Float64),\n SampleCount UInt32,\n WindowMinutes UInt16,\n WindowStart DateTime64(3),\n WindowEnd DateTime64(3),\n ConsecutiveBreaches UInt16,\n ConsecutiveHealthy UInt16,\n IncidentId Nullable(String),\n IncidentTransition LowCardinality(String),\n EvaluationDurationMs UInt32,\n ErrorMessage Nullable(String),\n ErrorCategory LowCardinality(String)\n)\nENGINE = MergeTree\nPARTITION BY toDate(Timestamp)\nORDER BY (OrgId, RuleId, GroupKey, Timestamp)\nTTL toDate(Timestamp) + INTERVAL 365 DAY", diff --git a/packages/domain/src/generated/tinybird-project-manifest.ts b/packages/domain/src/generated/tinybird-project-manifest.ts index 35c953b8a..0c02f6147 100644 --- a/packages/domain/src/generated/tinybird-project-manifest.ts +++ b/packages/domain/src/generated/tinybird-project-manifest.ts @@ -1,7 +1,7 @@ // This file is generated by scripts/generate-tinybird-project-manifest.ts // Do not edit manually. -export const projectRevision = "6d960544a383391d479e6ec3174fc498e9e7201bdf1a07c7f11a778b2accce06" as const +export const projectRevision = "12a7685236a4ebe4a8a40900e74c0239ab853afec09659d78a9b2f92811561fa" as const export const datasources = [ { @@ -72,7 +72,7 @@ export const datasources = [ { name: "service_address_resolutions_hourly", content: - 'DESCRIPTION >\n Resolved (sourceService, parent.server.address) → resolved targetService facts emitted by the ServiceMapRollupService rollup. Used to anti-join internal-service overlap out of the external-edges query.\n\nSCHEMA >\n OrgId LowCardinality(String),\n Hour DateTime,\n SourceService LowCardinality(String),\n ParentServerAddress String,\n ResolvedTargetService LowCardinality(String),\n DeploymentEnv LowCardinality(String)\n\nENGINE "ReplacingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, ParentServerAddress, ResolvedTargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"', + 'DESCRIPTION >\n Resolved (sourceService, parent.server.address) → resolved targetService facts emitted by the ServiceMapRollupService rollup. Used to anti-join internal-service overlap out of the external-edges query.\n\nSCHEMA >\n OrgId LowCardinality(String) `json:$.OrgId`,\n Hour DateTime `json:$.Hour`,\n SourceService LowCardinality(String) `json:$.SourceService`,\n ParentServerAddress String `json:$.ParentServerAddress`,\n ResolvedTargetService LowCardinality(String) `json:$.ResolvedTargetService`,\n DeploymentEnv LowCardinality(String) `json:$.DeploymentEnv`\n\nENGINE "ReplacingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, ParentServerAddress, ResolvedTargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"', }, { name: "service_external_edges_hourly", @@ -97,7 +97,7 @@ export const datasources = [ { name: "service_map_edges_hourly", content: - 'DESCRIPTION >\n Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).\n\nSCHEMA >\n OrgId LowCardinality(String),\n Hour DateTime,\n SourceService LowCardinality(String),\n TargetService String,\n DeploymentEnv LowCardinality(String),\n CallCount SimpleAggregateFunction(sum, UInt64),\n ErrorCount SimpleAggregateFunction(sum, UInt64),\n DurationSumMs SimpleAggregateFunction(sum, Float64),\n MaxDurationMs SimpleAggregateFunction(max, Float64),\n SampledSpanCount SimpleAggregateFunction(sum, UInt64),\n UnsampledSpanCount SimpleAggregateFunction(sum, UInt64),\n SampleRateSum SimpleAggregateFunction(sum, Float64)\n\nENGINE "AggregatingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, TargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"', + 'DESCRIPTION >\n Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).\n\nSCHEMA >\n OrgId LowCardinality(String) `json:$.OrgId`,\n Hour DateTime `json:$.Hour`,\n SourceService LowCardinality(String) `json:$.SourceService`,\n TargetService String `json:$.TargetService`,\n DeploymentEnv LowCardinality(String) `json:$.DeploymentEnv`,\n CallCount SimpleAggregateFunction(sum, UInt64) `json:$.CallCount`,\n ErrorCount SimpleAggregateFunction(sum, UInt64) `json:$.ErrorCount`,\n DurationSumMs SimpleAggregateFunction(sum, Float64) `json:$.DurationSumMs`,\n MaxDurationMs SimpleAggregateFunction(max, Float64) `json:$.MaxDurationMs`,\n SampledSpanCount SimpleAggregateFunction(sum, UInt64) `json:$.SampledSpanCount`,\n UnsampledSpanCount SimpleAggregateFunction(sum, UInt64) `json:$.UnsampledSpanCount`,\n SampleRateSum SimpleAggregateFunction(sum, Float64) `json:$.SampleRateSum`\n\nENGINE "AggregatingMergeTree"\nENGINE_PARTITION_KEY "toDate(Hour)"\nENGINE_SORTING_KEY "OrgId, Hour, DeploymentEnv, SourceService, TargetService"\nENGINE_TTL "toDate(Hour) + INTERVAL 365 DAY"', }, { name: "service_map_spans", diff --git a/packages/domain/src/tinybird/datasources.ts b/packages/domain/src/tinybird/datasources.ts index da22e7eef..a935162a3 100644 --- a/packages/domain/src/tinybird/datasources.ts +++ b/packages/domain/src/tinybird/datasources.ts @@ -413,7 +413,12 @@ export type ServiceMapChildrenRow = InferRow export const serviceMapEdgesHourly = defineDatasource("service_map_edges_hourly", { description: "Pre-aggregated hourly service-to-service edges for the service map. Uses AggregatingMergeTree for incremental aggregation. Populated by the scheduled ServiceMapRollupService rollup (one write per completed hour).", - jsonPaths: false, + // jsonPaths enabled: this is ingested directly via POST /v0/events from + // ServiceMapRollupService, not by a materialized view. Declaring + // `jsonPaths: false` made Tinybird reject every write with "Data Source + // needs to have JSONPaths defined", so the table stopped filling — and + // because an hour is sealed only once its edge rows land, the rollup then + // re-ran all six lookback hours for every org on every tick, forever. schema: { OrgId: t.string().lowCardinality(), Hour: t.dateTime(), @@ -609,7 +614,8 @@ export type ServiceExternalEdgesHourlyRow = InferRow= '2026-01-01 10:30:00' + AND Hour < '2026-01-03 14:15:00' + GROUP BY hourTs + FORMAT JSON + -- builder:service-map:serviceDependenciesForServiceQuery:default [32f3ee6f] SELECT sourceService AS sourceService, @@ -4410,7 +4420,7 @@ SELECT LIMIT 5000 FORMAT JSON --- pipe:span_search:default:baseline [d4e4a298] +-- pipe:span_search:default:baseline [bfce1ab3] SELECT TraceId AS traceId, SpanId AS spanId, @@ -4427,11 +4437,20 @@ SELECT AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' AND StatusCode != 'Error' + AND Timestamp >= (SELECT min(ts) FROM (SELECT + Timestamp AS ts + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND StatusCode != 'Error' + ORDER BY ts DESC + LIMIT 20)) ORDER BY timestamp DESC LIMIT 20 FORMAT JSON --- pipe:span_search:default:bloom [d4e4a298] +-- pipe:span_search:default:bloom [bfce1ab3] SELECT TraceId AS traceId, SpanId AS spanId, @@ -4448,11 +4467,20 @@ SELECT AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' AND StatusCode != 'Error' + AND Timestamp >= (SELECT min(ts) FROM (SELECT + Timestamp AS ts + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND StatusCode != 'Error' + ORDER BY ts DESC + LIMIT 20)) ORDER BY timestamp DESC LIMIT 20 FORMAT JSON --- pipe:span_search:default:text [d4e4a298] +-- pipe:span_search:default:text [bfce1ab3] SELECT TraceId AS traceId, SpanId AS spanId, @@ -4469,6 +4497,15 @@ SELECT AND Timestamp >= '2026-01-01 10:30:00' AND Timestamp <= '2026-01-03 14:15:00' AND StatusCode != 'Error' + AND Timestamp >= (SELECT min(ts) FROM (SELECT + Timestamp AS ts + FROM traces + WHERE OrgId = 'org_sql_catalog' + AND Timestamp >= '2026-01-01 10:30:00' + AND Timestamp <= '2026-01-03 14:15:00' + AND StatusCode != 'Error' + ORDER BY ts DESC + LIMIT 20)) ORDER BY timestamp DESC LIMIT 20 FORMAT JSON diff --git a/packages/query-engine/src/ch/builder-fixtures.ts b/packages/query-engine/src/ch/builder-fixtures.ts index 8ea2284b9..938ea23ea 100644 --- a/packages/query-engine/src/ch/builder-fixtures.ts +++ b/packages/query-engine/src/ch/builder-fixtures.ts @@ -473,6 +473,12 @@ export const builderFixtures: ReadonlyArray = [ label: "default", compile: () => CH.serviceMapEdgesExistingHoursSQL(window), }, + { + module: "service-map-rollup", + name: "serviceMapResolutionsExistingHoursSQL", + label: "default", + compile: () => CH.serviceMapResolutionsExistingHoursSQL(window), + }, { // Org-wide: hourly MV branch UNION ALL two partial-hour live joins. module: "service-map", diff --git a/packages/query-engine/src/ch/index.ts b/packages/query-engine/src/ch/index.ts index aa6408b59..47f95fee4 100644 --- a/packages/query-engine/src/ch/index.ts +++ b/packages/query-engine/src/ch/index.ts @@ -293,6 +293,7 @@ export { export { serviceMapEdgesRollupSQL, serviceMapEdgesExistingHoursSQL, + serviceMapResolutionsExistingHoursSQL, serviceMapResolutionsRollupSQL, type ServiceMapEdgesRollupParams, type ServiceMapEdgesHourlyOutput, diff --git a/packages/query-engine/src/ch/queries/service-map-rollup.ts b/packages/query-engine/src/ch/queries/service-map-rollup.ts index 8317c7092..9689fc5b4 100644 --- a/packages/query-engine/src/ch/queries/service-map-rollup.ts +++ b/packages/query-engine/src/ch/queries/service-map-rollup.ts @@ -20,7 +20,7 @@ import { compileCH } from "@maple-dev/clickhouse-builder" import * as CH from "@maple-dev/clickhouse-builder/expr" import { param } from "@maple-dev/clickhouse-builder" import { from, fromQuery } from "@maple-dev/clickhouse-builder" -import { ServiceMapEdgesHourly, Traces } from "../tables" +import { ServiceAddressResolutionsHourly, ServiceMapEdgesHourly, Traces } from "../tables" import { serviceMapEdgeJoinQuery } from "./service-map" import { CHNumber } from "../schema" @@ -108,6 +108,43 @@ export function serviceMapEdgesExistingHoursSQL(params: { ) } +/** + * SQL listing the distinct hours already present in + * `service_address_resolutions_hourly` for an org within `[startTime, endTime)`. + * + * The companion resolutions write can fail independently of the edges write, so + * the rollup runs a repair pass over sealed hours. Without this probe that pass + * was unconditional: it re-ran the resolutions join — a raw-`traces` self-join, + * the most expensive query in the tick — for every sealed hour on every tick, + * forever. Asking which hours already resolved costs one cheap sorted-prefix + * read and skips nearly all of them. + */ +export function serviceMapResolutionsExistingHoursSQL(params: { + orgId: string + startTime: string + endTime: string +}): CompiledQuery { + const query = from(ServiceAddressResolutionsHourly) + .select(($) => ({ hourTs: CH.toUnixTimestamp($.Hour) })) + .where(($) => [ + $.OrgId.eq(param.string("orgId")), + $.Hour.gte(param.dateTime("startTime")), + $.Hour.lt(param.dateTime("endTime")), + ]) + .groupBy("hourTs") + .format("JSON") + + return compileCH( + query, + { + orgId: params.orgId, + startTime: params.startTime, + endTime: params.endTime, + }, + { rowSchema: ServiceMapEdgesExistingHourSchema }, + ) +} + /** * SQL that computes the service-to-service edges for one completed hour * `[hourStart, hourEnd)`. Output rows are ready to `ingest` into diff --git a/packages/query-engine/src/ch/queries/traces.test.ts b/packages/query-engine/src/ch/queries/traces.test.ts index 6d3a8f6b9..71caf37c2 100644 --- a/packages/query-engine/src/ch/queries/traces.test.ts +++ b/packages/query-engine/src/ch/queries/traces.test.ts @@ -453,4 +453,25 @@ describe("spanSearchQuery", () => { expect(sql).not.toContain("FROM trace_detail_spans") expect(sql).toContain("SpanName = 'GET /users'") }) + + it("two-stages the raw-traces path so the attribute Maps are read after a cutoff", () => { + const q = spanSearchQuery({ spanName: "GET /users", limit: 20, offset: 5 }) + const { sql } = compileCH(q, baseParams) + + // `traces` is sorted (OrgId, ServiceName, SpanName, toDateTime(Timestamp)), + // so `ORDER BY Timestamp DESC` cannot read in order — single-stage meant + // materializing SpanAttributes and ResourceAttributes for every matching + // row in range before LIMIT discarded all but N. + expect(sql).toContain("Timestamp >= (SELECT min(ts) FROM (") + // The cutoff must cover every row the outer query can examine. + expect(sql).toContain("LIMIT 25") + expect(sql).toContain("SpanAttributes AS spanAttributes") + }) + + it("does not add a cutoff on the trace-detail path", () => { + const q = spanSearchQuery({ traceId: "trace_123", limit: 50 }) + const { sql } = compileCH(q, baseParams) + + expect(sql).not.toContain("SELECT min(ts)") + }) }) diff --git a/packages/query-engine/src/ch/queries/traces.ts b/packages/query-engine/src/ch/queries/traces.ts index 942908dc2..c97e7ef8a 100644 --- a/packages/query-engine/src/ch/queries/traces.ts +++ b/packages/query-engine/src/ch/queries/traces.ts @@ -947,6 +947,7 @@ function spanSearchFrom( opts: SpanSearchOpts, limit: number, offset: number, + cutoff?: CH.Expr, ) { const q = from(source) .select(($) => ({ @@ -964,6 +965,7 @@ function spanSearchFrom( .where(($) => [ ...tracesBaseWhereConditions($, opts), CH.when(opts.traceId, (v: string) => $.TraceId.eq(v)), + cutoff === undefined ? undefined : $.Timestamp.gte(cutoff), ]) .orderBy(["timestamp", "desc"]) .limit(limit) @@ -976,11 +978,32 @@ export function spanSearchQuery(opts: SpanSearchOpts) { const limit = opts.limit ?? 20 const offset = opts.offset ?? 0 + // With a trace id, `trace_detail_spans` is keyed `(OrgId, TraceId, SpanId)`: + // the filter is a sort-key prefix and the row set is one trace, so the whole + // Maps are cheap to read directly. if (opts.traceId) { return spanSearchFrom(TraceDetailSpans, opts, limit, offset) } - return spanSearchFrom(Traces, opts, limit, offset) + // Without one, this reads raw `traces`, whose sort key + // `(OrgId, ServiceName, SpanName, toDateTime(Timestamp))` cannot serve + // `ORDER BY Timestamp DESC` — so a single-stage query materialized both + // attribute Maps for every matching row in range before `LIMIT` discarded + // all but N (9–13 GB reads in production). Two-stage it exactly like + // `tracesListQuery`: stage 1 reads only `Timestamp` to find the cutoff, + // stage 2 reads the Maps only at/after it. + const cutoffInner = from(Traces) + .select(($) => ({ ts: $.Timestamp })) + .where(($) => [ + ...tracesBaseWhereConditions($, opts), + CH.when(opts.traceId, (v: string) => $.TraceId.eq(v)), + ]) + .orderBy(["ts", "desc"]) + .limit(limit + offset) + const cutoffSql = compileCH(cutoffInner, {}, { skipFormat: true }).sql + const cutoff = CH.rawExpr(`(SELECT min(ts) FROM (${cutoffSql}))`) + + return spanSearchFrom(Traces, opts, limit, offset, cutoff) } // --------------------------------------------------------------------------- diff --git a/packages/query-engine/src/execution/backend.ts b/packages/query-engine/src/execution/backend.ts index e1dfd61f4..5c3ec669a 100644 --- a/packages/query-engine/src/execution/backend.ts +++ b/packages/query-engine/src/execution/backend.ts @@ -112,6 +112,16 @@ export interface WarehouseBackendDialect { * SQL-catalog e2e sweep enforces. */ readonly unquote64BitIntegers: boolean + /** + * True when this backend runs the schema *we* deploy, so its columns and + * skip indices are known at compile time and capability probing is both + * unnecessary and — on Tinybird — impossible (`system.*` answers `403` + * through the SDK and takes ~2.2s through the gateway). BYO ClickHouse is + * the user's own cluster and must still be inspected live; `chdb` applies + * the same generated schema but is cheap to probe and can lag a migration, + * so it keeps the live answer too. + */ + readonly managedSchema: boolean } /** Single source of truth for per-backend behavior. */ @@ -125,6 +135,7 @@ export const BackendDialect: Record { }), ) - it.effect("does not select text indexes when the Tinybird gateway rejects setting overrides", () => + it.effect("answers Tinybird gateway capabilities from the deployed schema, not a probe", () => Effect.gen(function* () { const sqls: string[] = [] const executor = makeWarehouseExecutor({ createClient: () => ({ sql: async (sql) => { sqls.push(sql) - if (sql.includes("SELECT version()")) return { data: [{ version: "26.2.1" }] } - if (sql.includes("system.data_skipping_indices")) { - return { - data: [ - { - table: "logs", - name: "idx_lower_body_text", - type: "text", - expression: "lower(Body)", - }, - ], - } - } - if (sql.includes("system.settings")) { - return { data: [{ name: "enable_full_text_index", value: "0" }] } - } + // A probe here would be a bug: Tinybird answers `403` for + // `system.columns` / `system.data_skipping_indices`, so any + // live inspection collapses to the conservative plan. return { data: [] } }, insert: async () => {}, @@ -627,16 +614,20 @@ describe("makeWarehouseExecutor capability-aware compilation", () => { unsafeCompiledQuery<{ readonly c: number }>({ reason: "test-fixture", note: "Synthetic SQL asserting executor/compile behaviour, not a product query.", - sql: `SELECT count() AS c FROM logs WHERE OrgId = 'org_test' AND '${logBodySearchMode(capabilities)}' = 'scan' FORMAT JSON`, + sql: `SELECT count() AS c FROM logs WHERE OrgId = 'org_test' AND '${logBodySearchMode(capabilities)}' = 'tokenbf' FORMAT JSON`, tenantScope: "org", }), { context: "tinybird-gateway-capabilities" }, ) + // The tokenbf prefilter is selected because `logs.idx_lower_body` is in + // the deployed schema — the whole point of resolving statically. const executed = sqls.find((sql) => sql.includes("FROM logs WHERE")) assert.isDefined(executed) - assert.include(executed!, "'scan' = 'scan'") + assert.include(executed!, "'tokenbf' = 'tokenbf'") + // Managed backends reject inline setting overrides, so no text plan. assert.notInclude(executed!, "enable_full_text_index") + assert.isEmpty(sqls.filter((sql) => sql.includes("system."))) }), ) diff --git a/packages/query-engine/src/execution/executor.ts b/packages/query-engine/src/execution/executor.ts index 7295fd908..f9e2d4850 100644 --- a/packages/query-engine/src/execution/executor.ts +++ b/packages/query-engine/src/execution/executor.ts @@ -28,6 +28,7 @@ import { truncateSql, } from "./fingerprint" import { BackendDialect, warehouseTargetAttributes } from "./backend" +import { managedWarehouseCapabilities } from "./managed-capabilities" import { findIngestPinnedTable } from "./datasource-routing" import type { ExecutionTenant, @@ -44,7 +45,6 @@ import { deriveWarehouseCapabilities, logBodySearchMode, type WarehouseCapabilities, - type WarehouseSettingMetadataRow, WarehouseColumnMetadataSchema, WarehouseIndexMetadataSchema, WarehouseSettingMetadataSchema, @@ -52,7 +52,12 @@ import { } from "../capabilities" const CLIENT_CACHE_TTL_MS = 30_000 -const CAPABILITIES_CACHE_TTL_MS = 5 * 60_000 +/** + * Only BYO ClickHouse is probed now, and its answer changes solely when the + * user migrates their cluster. The cache is isolate-local (and there are two + * per isolate), so the TTL — not the map — is what bounds probe volume. + */ +const CAPABILITIES_CACHE_TTL_MS = 60 * 60_000 const CAPABILITIES_INSPECTION_TIMEOUT = Duration.seconds(2) const WarehouseCapabilityMetadataTarget = Schema.Literals(["version", "indexes", "columns", "settings"]) type WarehouseCapabilityMetadataTarget = Schema.Schema.Type @@ -178,46 +183,68 @@ export const makeWarehouseExecutor = (deps: WarehouseExecutorDeps): WarehouseQue }), ) + // Every probe degrades to an empty result rather than failing the whole + // inspection. A backend that denies one `system.*` table (Tinybird answers + // `403` for `system.columns` and `system.data_skipping_indices` but serves + // `system.settings`) should lose only the features that depend on that + // table: collapsing to `baselineWarehouseCapabilities()` silently disables + // the bloom and tokenbf prefilters on a backend that does have them. + const degradeToEmpty = ( + effect: Effect.Effect, WarehouseCapabilityProbeError>, + ): Effect.Effect> => + effect.pipe( + Effect.catchTag("@maple/query-engine/execution/WarehouseCapabilityProbeError", (error) => + logProbeFailure(error).pipe(Effect.as>([])), + ), + ) + const inspection = Effect.all( [ - queryRows("version", "SELECT version() AS version").pipe( - Effect.flatMap((rows) => - Schema.decodeUnknownEffect(WarehouseVersionMetadataSchema)(rows), + degradeToEmpty( + queryRows("version", "SELECT version() AS version").pipe( + Effect.flatMap((rows) => + Schema.decodeUnknownEffect(WarehouseVersionMetadataSchema)(rows), + ), + Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("version", cause))), ), - Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("version", cause))), ), - queryRows( - "indexes", - `SELECT table, name, type, expr AS expression + degradeToEmpty( + queryRows( + "indexes", + `SELECT table, name, type, expr AS expression FROM system.data_skipping_indices WHERE database = currentDatabase() AND table IN ('logs', 'traces')`, - ).pipe( - Effect.flatMap((rows) => Schema.decodeUnknownEffect(WarehouseIndexMetadataSchema)(rows)), - Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("indexes", cause))), + ).pipe( + Effect.flatMap((rows) => + Schema.decodeUnknownEffect(WarehouseIndexMetadataSchema)(rows), + ), + Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("indexes", cause))), + ), ), - queryRows( - "columns", - `SELECT table, name + degradeToEmpty( + queryRows( + "columns", + `SELECT table, name FROM system.columns WHERE database = currentDatabase() AND table IN ('logs', 'traces')`, - ).pipe( - Effect.flatMap((rows) => Schema.decodeUnknownEffect(WarehouseColumnMetadataSchema)(rows)), - Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("columns", cause))), + ).pipe( + Effect.flatMap((rows) => + Schema.decodeUnknownEffect(WarehouseColumnMetadataSchema)(rows), + ), + Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("columns", cause))), + ), ), - queryRows( - "settings", - `SELECT name, value + degradeToEmpty( + queryRows( + "settings", + `SELECT name, value FROM system.settings WHERE name = 'enable_full_text_index'`, - ).pipe( - Effect.flatMap((rows) => - Schema.decodeUnknownEffect(WarehouseSettingMetadataSchema)(rows), - ), - Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("settings", cause))), - Effect.catchTag("@maple/query-engine/execution/WarehouseCapabilityProbeError", (error) => - logProbeFailure(error).pipe( - Effect.as>([]), + ).pipe( + Effect.flatMap((rows) => + Schema.decodeUnknownEffect(WarehouseSettingMetadataSchema)(rows), ), + Effect.catchTag("SchemaError", (cause) => Effect.fail(probeError("settings", cause))), ), ), ], @@ -234,16 +261,12 @@ WHERE name = 'enable_full_text_index'`, ), ) - const timed: Effect.Effect< - WarehouseCapabilities, - WarehouseCapabilityProbeError | Cause.TimeoutError - > = inspection.pipe(Effect.timeout(CAPABILITIES_INSPECTION_TIMEOUT)) - const probeRecovered: Effect.Effect = timed.pipe( - Effect.catchTag("@maple/query-engine/execution/WarehouseCapabilityProbeError", (error) => - logProbeFailure(error).pipe(Effect.as(baselineWarehouseCapabilities())), - ), + // Individual probe failures are already absorbed by `degradeToEmpty`, so + // only a whole-inspection timeout can still reach the conservative plan. + const timed: Effect.Effect = inspection.pipe( + Effect.timeout(CAPABILITIES_INSPECTION_TIMEOUT), ) - return probeRecovered.pipe( + return timed.pipe( Effect.catchTag("TimeoutError", (error) => Effect.logWarning("Warehouse capability inspection fell back to conservative plan").pipe( Effect.annotateLogs({ target: "inspection", error: error.message }), @@ -259,6 +282,23 @@ WHERE name = 'enable_full_text_index'`, ) { const purpose: RoutePurpose = options?.route === "ingest" ? "ingest" : "read" const resolved = yield* deps.resolveRoute(tenant, purpose, "capabilities") + + // Backends running the schema we deploy answer from the generated + // snapshot: no client, no round-trip, no cache entry, and no way to fall + // back to a conservative plan because a `system.*` probe was denied. + if (BackendDialect[resolved.config.kind].managedSchema) { + const capabilities = managedWarehouseCapabilities() + yield* Effect.annotateCurrentSpan({ + "maple.query.capabilities.cache": "static", + "maple.query.capabilities.metadata_available": capabilities.metadataAvailable, + "warehouse.backend": resolved.config.kind, + "warehouse.route": purpose, + "warehouse.config_source": resolved.source, + orgId: tenant.orgId, + }) + return capabilities + } + const nowMs = yield* Clock.currentTimeMillis const configKey = sqlClientCacheKey(resolved.config) const cache = yield* Ref.get(capabilitiesCache) @@ -951,9 +991,15 @@ WHERE name = 'enable_full_text_index'`, }) return { - query: (tenant, payload, options) => unbounded(query(tenant, payload, withoutResponseLimits(options))), + query: (tenant, payload, options) => + unbounded(query(tenant, payload, withoutResponseLimits(options))), crossOrgQuery: (tenant, compiled, options) => - unbounded(crossOrgQuery(tenant, compiled, { ...withoutResponseLimits(options), justification: options.justification })), + unbounded( + crossOrgQuery(tenant, compiled, { + ...withoutResponseLimits(options), + justification: options.justification, + }), + ), rawSqlQuery: (tenant, sql, options) => unbounded(rawSqlQuery(tenant, sql, options)), compiledQuery, compiledQueryBounded, diff --git a/packages/query-engine/src/execution/managed-capabilities.test.ts b/packages/query-engine/src/execution/managed-capabilities.test.ts new file mode 100644 index 000000000..d7774f939 --- /dev/null +++ b/packages/query-engine/src/execution/managed-capabilities.test.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from "vitest" +import { attributeIndexMode, logBodySearchMode } from "../capabilities" +import { managedWarehouseCapabilities } from "./managed-capabilities" + +describe("managed warehouse capabilities", () => { + /** + * The regression this guards: production capability inspection failed on + * every request (Tinybird answers `403` for `system.columns` and + * `system.data_skipping_indices`; its gateway takes ~2.2s, over the 2s + * budget), so every log and trace search ran with the prefilters off even + * though the indices are deployed. If this test says "none"/"scan" again, + * the managed schema is being read wrong. + */ + it("enables the bloom and tokenbf prefilters that the deployed schema carries", () => { + const capabilities = managedWarehouseCapabilities() + + expect(capabilities.metadataAvailable).toBe(true) + expect(attributeIndexMode(capabilities, "logs")).toBe("bloom") + expect(attributeIndexMode(capabilities, "traces")).toBe("bloom") + expect(logBodySearchMode(capabilities)).toBe("tokenbf") + }) + + it("parses the generated snapshot rather than a hand-maintained list", () => { + const capabilities = managedWarehouseCapabilities() + + // Index rows carry the parsed type and expression, not just a name. + expect(capabilities.indexes.get("traces.idx_span_attr_keys")).toEqual({ + table: "traces", + name: "idx_span_attr_keys", + type: "bloom_filter(0.01)", + expression: "mapKeys(SpanAttributes)", + }) + expect(capabilities.indexes.get("logs.idx_lower_body")?.type).toBe("tokenbf_v1(32768, 3, 0)") + + // Columns come from the same parse, and index lines must not leak in. + expect(capabilities.columns.has("logs.Body")).toBe(true) + expect(capabilities.columns.has("traces.SpanAttributes")).toBe(true) + expect([...capabilities.columns].some((key) => key.includes("INDEX"))).toBe(false) + }) + + it("keeps full-text search unavailable — managed backends reject setting overrides", () => { + expect(managedWarehouseCapabilities().fullTextSearchSetting).toBe("unavailable") + }) +}) diff --git a/packages/query-engine/src/execution/managed-capabilities.ts b/packages/query-engine/src/execution/managed-capabilities.ts new file mode 100644 index 000000000..a32537781 --- /dev/null +++ b/packages/query-engine/src/execution/managed-capabilities.ts @@ -0,0 +1,110 @@ +import { latestSnapshotStatements } from "@maple/domain/generated/clickhouse-schema" +import { + deriveWarehouseCapabilities, + type WarehouseCapabilities, + type WarehouseColumnMetadataRow, + type WarehouseIndexMetadataRow, +} from "../capabilities" + +/** + * Capabilities for the warehouses whose schema *we* deploy. + * + * Probing `system.columns` / `system.data_skipping_indices` at runtime is the + * wrong question to ask a managed backend: Tinybird's `/v0/sql` answers `403` + * for both (it does not expose `system.*` to workspace tokens), and its + * ClickHouse-compatible gateway answers the `system.columns` scan in ~2.2s — + * over the inspection budget. Every inspection therefore fell back to + * `baselineWarehouseCapabilities()`, which pins `attributeIndexMode` to `"none"` + * and `logBodySearchMode` to `"scan"` — so the bloom and tokenbf indices that + * are in the deployed schema were never used by a single production query. + * + * For a schema we own, the indices are a compile-time fact. These are read out + * of the generated snapshot rather than hand-listed so that adding an index to + * `packages/domain/src/tinybird/datasources.ts` enables the matching feature + * with no change here. + */ + +/** The tables `deriveWarehouseCapabilities` reasons about. */ +const CAPABILITY_TABLES = ["logs", "traces"] as const + +/** + * `INDEX TYPE [GRANULARITY ]`. Both `` and + * `` contain parentheses and commas (`mapKeys(SpanAttributes)`, + * `tokenbf_v1(32768, 3, 0)`), so the keywords — not punctuation — delimit them. + */ +const INDEX_LINE = /^INDEX\s+(\S+)\s+(.+?)\s+TYPE\s+(.+?)(?:\s+GRANULARITY\s+\d+)?$/ + +interface TableMetadata { + readonly columns: ReadonlyArray + readonly indexes: ReadonlyArray +} + +/** + * Parse the column and index list out of one generated `CREATE TABLE`. The + * statements are machine-generated with exactly one column or index per line, + * so this stays a line scan rather than a DDL parser. + */ +const parseTableDdl = (table: string, ddl: string): TableMetadata => { + const lines = ddl.split("\n") + const end = lines.findIndex((line, index) => index > 0 && line.startsWith(")")) + const body = lines.slice(1, end === -1 ? lines.length : end) + + const columns: Array = [] + const indexes: Array = [] + + for (const line of body) { + const entry = line.trim().replace(/,$/, "") + if (entry === "") continue + const index = INDEX_LINE.exec(entry) + if (index) { + indexes.push({ table, name: index[1]!, type: index[3]!, expression: index[2]! }) + continue + } + const name = entry.split(/\s+/)[0] + if (name !== undefined) columns.push({ table, name }) + } + + return { columns, indexes } +} + +const readManagedSchemaMetadata = (): TableMetadata => { + const columns: Array = [] + const indexes: Array = [] + + for (const table of CAPABILITY_TABLES) { + const ddl = latestSnapshotStatements.find((statement) => + statement.startsWith(`CREATE TABLE IF NOT EXISTS ${table} (`), + ) + if (ddl === undefined) continue + const parsed = parseTableDdl(table, ddl) + columns.push(...parsed.columns) + indexes.push(...parsed.indexes) + } + + return { columns, indexes } +} + +let cached: WarehouseCapabilities | undefined + +/** + * Capabilities derived from the deployed managed schema. Computed once per + * isolate — the generated snapshot cannot change at runtime. + * + * `settings` is deliberately empty: both managed backends set + * `stripTinybirdRestrictedSettings`, so `enable_full_text_index` can never be + * overridden inline and `fullTextSearchSetting` resolves to `"unavailable"` + * either way. The managed schema carries no `text` indices, so no text feature + * depends on it. + */ +export const managedWarehouseCapabilities = (): WarehouseCapabilities => { + if (cached === undefined) { + const { columns, indexes } = readManagedSchemaMetadata() + cached = deriveWarehouseCapabilities({ + indexes, + columns, + settings: [], + allowSettingOverrides: false, + }) + } + return cached +}