Skip to content

Query engine v2 #20

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 11 commits into from
Apr 6, 2026
Merged

Query engine v2 #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
d3f0634
init
Makisuo Apr 6, 2026
01144ab
fix
Makisuo Apr 6, 2026
62a536d
more stuff
Makisuo Apr 6, 2026
acdc056
fix subqueries
Makisuo Apr 6, 2026
a2390c3
make more typesafe
Makisuo Apr 6, 2026
8b12a8a
s
Makisuo Apr 6, 2026
ba3f2cb
ad dtests
Makisuo Apr 6, 2026
8959433
fix
Makisuo Apr 6, 2026
2f3d3f1
add type tests
Makisuo Apr 6, 2026
4a07f1e
fix stuff
Makisuo Apr 6, 2026
7588297
g
Makisuo Apr 6, 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
8 changes: 4 additions & 4 deletions apps/api/src/services/PipeQueryDispatcher.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -168,17 +168,17 @@ export function compilePipeQuery(
}), { orgId, startTime, endTime })),
),
Match.when("errors_summary", () =>
eraseType(CH.errorsSummarySQL({
eraseType(CH.compile(CH.errorsSummaryQuery({
rootOnly: bool("root_only"), services: str("services")?.split(",").filter(Boolean),
deploymentEnvs: str("deployment_envs")?.split(",").filter(Boolean),
errorTypes: str("error_types")?.split(",").filter(Boolean),
}, { orgId, startTime, endTime })),
}), { orgId, startTime, endTime })),
),
Match.when("error_detail_traces", () =>
eraseType(CH.errorDetailTracesSQL({
eraseType(CH.compile(CH.errorDetailTracesQuery({
errorType: String(params.error_type), rootOnly: bool("root_only"),
services: str("services")?.split(",").filter(Boolean), limit: int("limit", 10),
}, { orgId, startTime, endTime })),
}), { orgId, startTime, endTime })),
),
// ----- Metrics -----
Match.when("list_metrics", () =>
Expand Down
201 changes: 90 additions & 111 deletions apps/api/src/services/QueryEngineService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ const truncateSql = (s: string, maxLen = 1000) =>
const executeCHQuery = <Output extends Record<string, any>, Params extends Record<string, any>>(
tinybird: Pick<TinybirdServiceShape, "sqlQuery">,
tenant: TenantContext,
query: CH.CHQuery<any, Output, Params>,
query: CH.CHQuery<any, Output>,
params: Params,
context: string,
): Effect.Effect<ReadonlyArray<Output>, QueryEngineExecutionError> => {
Expand Down Expand Up @@ -498,6 +498,61 @@ function resolveAttributeScope(
return scope === "resource" ? "resource" : "span"
}

type AttrFilterArray = Array<{
key: string
value?: string
mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains"
}>

function extractTracesOpts(filters: Record<string, unknown> | undefined) {
return {
serviceName: filters?.serviceName as string | undefined,
spanName: filters?.spanName as string | undefined,
rootOnly: filters?.rootSpansOnly as boolean | undefined,
errorsOnly: filters?.errorsOnly as boolean | undefined,
environments: filters?.environments as string[] | undefined,
commitShas: filters?.commitShas as string[] | undefined,
minDurationMs: filters?.minDurationMs as number | undefined,
maxDurationMs: filters?.maxDurationMs as number | undefined,
matchModes: filters?.matchModes as {
serviceName?: "contains"
spanName?: "contains"
deploymentEnv?: "contains"
} | undefined,
attributeFilters: filters?.attributeFilters as AttrFilterArray | undefined,
resourceAttributeFilters: filters?.resourceAttributeFilters as AttrFilterArray | undefined,
groupByAttributeKeys: filters?.groupByAttributeKeys as string[] | undefined,
}
}

function shapeMetricsGroupRows<T extends { bucket: string | Date; serviceName: string; attributeValue: string }>(
rows: ReadonlyArray<T>,
valueExtractor: (row: T) => number,
groupBy: readonly string[] | undefined,
groupByAttributeKey: string | undefined,
fillOptions: BucketFillOptions | undefined,
): Array<TimeseriesPoint> {
if (groupBy?.includes("none") || !groupBy?.length) {
return groupTimeSeriesRows(
rows.map((row) => ({ bucket: row.bucket, groupName: "all" as const, value: valueExtractor(row) })),
(r) => r.value,
fillOptions,
)
}
if (groupByAttributeKey) {
return groupTimeSeriesRows(
rows.map((row) => ({ bucket: row.bucket, groupName: row.attributeValue || "(empty)", value: valueExtractor(row) })),
(r) => r.value,
fillOptions,
)
}
return groupTimeSeriesRows(
rows.map((row) => ({ bucket: row.bucket, groupName: row.serviceName, value: valueExtractor(row) })),
(r) => r.value,
fillOptions,
)
}

const isScalarAlertQuery = (
query: QuerySpec,
): query is Extract<QuerySpec, { kind: "timeseries"; source: "traces" | "logs" | "metrics" }> => {
Expand Down Expand Up @@ -550,22 +605,15 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
: undefined

if (request.query.source === "traces" && request.query.kind === "timeseries") {
const opts = extractTracesOpts(request.query.filters as Record<string, unknown>)
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesTimeseriesQuery({
...opts,
metric: request.query.metric,
needsSampling: false,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
groupBy: request.query.groupBy as string[] | undefined,
groupByAttributeKeys: request.query.filters?.groupByAttributeKeys as string[] | undefined,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime, bucketSeconds: bucketSeconds! },
Expand Down Expand Up @@ -614,13 +662,13 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
const isRateOrIncrease = request.query.metric === "rate" || request.query.metric === "increase"

if (isRateOrIncrease) {
const compiled = CH.metricsTimeseriesRateSQL(
{
const compiled = CH.compile(
CH.metricsTimeseriesRateQuery({
serviceName: request.query.filters.serviceName,
groupByAttributeKey,
attributeKey: attributeFilter?.key,
attributeValue: attributeFilter?.value,
},
}),
{
orgId: tenant.orgId,
metricName: request.query.filters.metricName,
Expand All @@ -645,35 +693,13 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>

const rateValueField = request.query.metric === "rate" ? "rateValue" : "increaseValue"

const data = (request.query.groupBy?.includes("none") || !request.query.groupBy?.length)
? groupTimeSeriesRows(
rateResult.map((row) => ({
bucket: row.bucket,
groupName: "all" as const,
value: Number(row[rateValueField]),
})),
(row) => row.value,
fillOptions,
)
: groupByAttributeKey
? groupTimeSeriesRows(
rateResult.map((row) => ({
bucket: row.bucket,
groupName: row.attributeValue || "(empty)",
value: Number(row[rateValueField]),
})),
(row) => row.value,
fillOptions,
)
: groupTimeSeriesRows(
rateResult.map((row) => ({
bucket: row.bucket,
groupName: row.serviceName,
value: Number(row[rateValueField]),
})),
(row) => row.value,
fillOptions,
)
const data = shapeMetricsGroupRows(
rateResult,
(row) => Number(row[rateValueField]),
request.query.groupBy,
groupByAttributeKey,
fillOptions,
)

return new QueryEngineExecuteResponse({
result: {
Expand Down Expand Up @@ -719,25 +745,13 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
(row) => row.value,
fillOptions,
)
: groupByAttributeKey
? groupTimeSeriesRows(
result.map((row) => ({
bucket: row.bucket,
groupName: row.attributeValue || "(empty)",
value: Number(row[valueField]),
})),
(row) => row.value,
fillOptions,
)
: groupTimeSeriesRows(
result.map((row) => ({
bucket: row.bucket,
groupName: row.serviceName,
value: Number(row[valueField]),
})),
(row) => row.value,
fillOptions,
)
: shapeMetricsGroupRows(
result,
(row) => Number(row[valueField]),
request.query.groupBy,
groupByAttributeKey,
fillOptions,
)

return new QueryEngineExecuteResponse({
result: {
Expand All @@ -749,25 +763,19 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
}

if (request.query.source === "traces" && request.query.kind === "breakdown") {
const opts = extractTracesOpts(request.query.filters as Record<string, unknown>)
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesBreakdownQuery({
...opts,
metric: request.query.metric,
groupBy: request.query.groupBy,
groupByAttributeKey:
request.query.groupBy === "attribute"
? (request.query.filters?.groupByAttributeKeys as string[] | undefined)?.[0]
? opts.groupByAttributeKeys?.[0]
: undefined,
limit: request.query.limit,
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as Array<{ key: string; value?: string; mode: "equals" | "exists" }> | undefined,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
Expand Down Expand Up @@ -850,38 +858,22 @@ export const makeQueryEngineExecute = (tinybird: QueryEngineTinybird) =>
}

if (request.query.source === "traces" && request.query.kind === "list") {
type AttrFilterArray = Array<{ key: string; value?: string; mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains" }>
const filters = request.query.filters
const opts = extractTracesOpts(request.query.filters as Record<string, unknown>)

// Graceful limit clamping: cap at 200, auto-reduce to 50 when no indexed filters
const hasIndexedFilter = !!(filters?.serviceName || filters?.spanName || filters?.errorsOnly || filters?.rootSpansOnly)
const hasIndexedFilter = !!(opts.serviceName || opts.spanName || opts.errorsOnly || opts.rootOnly)
const maxLimit = hasIndexedFilter ? 200 : 50
const clampedLimit = Math.min(request.query.limit ?? 25, maxLimit)

const sharedOpts = {
limit: clampedLimit,
offset: request.query.offset,
serviceName: filters?.serviceName,
spanName: filters?.spanName,
rootOnly: filters?.rootSpansOnly,
errorsOnly: filters?.errorsOnly,
environments: filters?.environments as string[] | undefined,
commitShas: filters?.commitShas as string[] | undefined,
minDurationMs: filters?.minDurationMs,
maxDurationMs: filters?.maxDurationMs,
matchModes: filters?.matchModes as { serviceName?: "contains"; spanName?: "contains"; deploymentEnv?: "contains" } | undefined,
attributeFilters: filters?.attributeFilters as AttrFilterArray | undefined,
resourceAttributeFilters: filters?.resourceAttributeFilters as AttrFilterArray | undefined,
columns: (request.query as { columns?: readonly string[] }).columns as string[] | undefined,
}

// Always use tracesListQuery — it handles MV usage when rootOnly=true
// and returns a consistent field shape for dashboard widgets.
// (tracesRootListQuery has a different output shape for the built-in trace list page)
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.tracesListQuery(sharedOpts),
CH.tracesListQuery({
...opts,
limit: clampedLimit,
offset: request.query.offset,
columns: (request.query as { columns?: readonly string[] }).columns as string[] | undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
"Failed to execute traces list query",
)
Expand Down Expand Up @@ -972,19 +964,12 @@ export const makeQueryEngineEvaluate = (tinybird: QueryEngineTinybird) =>
let observations: ReadonlyArray<AlertObservation>

if (request.query.source === "traces") {
type AttrFilterArray = Array<{ key: string; value?: string; mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains" }>
const opts = extractTracesOpts(request.query.filters as Record<string, unknown>)
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.alertTracesAggregateQuery({
serviceName: request.query.filters?.serviceName,
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as AttrFilterArray | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as AttrFilterArray | undefined,
...opts,
apdexThresholdMs: request.query.metric === "apdex" ? request.query.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
Expand Down Expand Up @@ -1097,19 +1082,13 @@ export const makeQueryEngineEvaluateGrouped = (tinybird: QueryEngineTinybird) =>

if (groupBy === "service") {
if (request.query.source === "traces") {
type AttrFilterArray = Array<{ key: string; value?: string; mode: "equals" | "exists" | "gt" | "gte" | "lt" | "lte" | "contains" }>
const tracesQuery = request.query as Extract<QuerySpec, { source: "traces"; metric: string }>
const opts = extractTracesOpts(request.query.filters as Record<string, unknown>)
const rows = yield* executeCHQuery(
tinybird,
tenant,
CH.alertTracesAggregateByServiceQuery({
spanName: request.query.filters?.spanName,
rootOnly: request.query.filters?.rootSpansOnly,
errorsOnly: request.query.filters?.errorsOnly,
environments: request.query.filters?.environments as string[] | undefined,
commitShas: request.query.filters?.commitShas as string[] | undefined,
attributeFilters: request.query.filters?.attributeFilters as AttrFilterArray | undefined,
resourceAttributeFilters: request.query.filters?.resourceAttributeFilters as AttrFilterArray | undefined,
...opts,
apdexThresholdMs: tracesQuery.metric === "apdex" ? tracesQuery.apdexThresholdMs : undefined,
}),
{ orgId: tenant.orgId, startTime: request.startTime, endTime: request.endTime },
Expand Down
10 changes: 5 additions & 5 deletions apps/api/src/services/TinybirdService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,7 +49,7 @@ export class TinybirdService extends ServiceMap.Service<TinybirdService, Tinybir
const env = yield* Env
const orgTinybirdSettings = yield* OrgTinybirdSettingsService

const toTinybirdQueryError = (pipe: TinybirdQueryRequest["pipe"], error: unknown) =>
const toTinybirdQueryError = (pipe: string, error: unknown) =>
new TinybirdQueryError({
message: error instanceof Error ? error.message : "Tinybird query failed",
pipe,
Expand All @@ -75,7 +75,7 @@ export class TinybirdService extends ServiceMap.Service<TinybirdService, Tinybir

const override = yield* orgTinybirdSettings
.resolveRuntimeConfig(tenant.orgId)
.pipe(Effect.mapError((error) => toTinybirdQueryError(pipe as TinybirdQueryRequest["pipe"], error)))
.pipe(Effect.mapError((error) => toTinybirdQueryError(pipe, error)))

if (Option.isSome(override)) {
yield* Effect.annotateCurrentSpan("clientSource", "org_override")
Expand All @@ -101,7 +101,7 @@ export class TinybirdService extends ServiceMap.Service<TinybirdService, Tinybir
const client = yield* resolveClient(tenant, pipe)
const result = yield* Effect.tryPromise({
try: () => client.sql(sql),
catch: (error) => toTinybirdQueryError(pipe as TinybirdQueryRequest["pipe"], error),
catch: (error) => toTinybirdQueryError(pipe, error),
}).pipe(
Effect.tapError((error) =>
Effect.logError("TinybirdService.executeSql failed", { pipe, error: String(error) }),
Expand Down Expand Up @@ -147,10 +147,10 @@ export class TinybirdService extends ServiceMap.Service<TinybirdService, Tinybir
sql: string,
) {
if (!tenant.orgId || tenant.orgId.trim() === "") {
return yield* new TinybirdQueryError({ pipe: "list_traces", message: "org_id must not be empty (sqlQuery)" })
return yield* new TinybirdQueryError({ pipe: "sqlQuery", message: "org_id must not be empty (sqlQuery)" })
}
if (!sql.includes("OrgId")) {
return yield* new TinybirdQueryError({ pipe: "list_traces", message: "SQL query must contain OrgId filter (sqlQuery)" })
return yield* new TinybirdQueryError({ pipe: "sqlQuery", message: "SQL query must contain OrgId filter (sqlQuery)" })
}
return yield* executeSql(tenant, sql, "sqlQuery")
})
Expand Down
4 changes: 2 additions & 2 deletions apps/web/src/api/tinybird/logs.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,8 +68,8 @@ function transformLog(raw: ListLogsOutput): Log {
severityNumber: Number(raw.severityNumber),
serviceName: raw.serviceName,
body: raw.body,
traceId: toTraceId(raw.traceId),
spanId: toSpanId(raw.spanId),
traceId: raw.traceId ? toTraceId(raw.traceId) : ("" as TraceId),
spanId: raw.spanId ? toSpanId(raw.spanId) : ("" as SpanId),
logAttributes: parseAttributes(raw.logAttributes),
resourceAttributes: parseAttributes(raw.resourceAttributes),
}
Expand Down
1 change: 1 addition & 0 deletions bun.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading
Loading