feat(query-engine): branded identity columns derive branded row schemas; drop redundant declared rowSchemas

apps/api/src/services/dashboards/ServiceMapRollupService.ts

@@ -249,11 +249,7 @@ export class ServiceMapRollupService extends Context.Service<
 			return yield* warehouse
 				.crossOrgQuery(
 					systemTenant(knownOrgs[0]!),
-					CH.compile(
-						CH.activeOrgsByTracesQuery(),
-						{ startTime },
-						{ rowSchema: CH.ActiveOrgsOutputSchema },
-					),
+					CH.compile(CH.activeOrgsByTracesQuery(), { startTime }),
 					{
 						profile: "discovery",
 						context: "serviceMapRollupActiveOrgs",

apps/api/src/services/errors/ErrorIssueReadModelsService.ts

@@ -360,7 +360,6 @@ const make: Effect.Effect<
 					startTime: formatWarehouseDateTime(startMs),
 					endTime: formatWarehouseDateTime(endMs),
 				},
-				{ rowSchema: CH.ErrorIssueSampleTracesOutputSchema },
 			)
 			const samplesEffect = isErrorKind
 				? warehouse.compiledQuery(tenant, samplesCompiled, {

apps/api/src/services/errors/ErrorsService.ts

@@ -302,11 +302,9 @@ const make: Effect.Effect<
 			return byo as ReadonlySet<OrgId>
 		}
 
-		const compiled = CH.compile(
-			CH.activeOrgsByErrorEventsQuery(),
-			{ startTime: formatWarehouseDateTime(nowMs - ERROR_ACTIVE_DISCOVERY_WINDOW_MS) },
-			{ rowSchema: CH.ActiveOrgsOutputSchema },
-		)
+		const compiled = CH.compile(CH.activeOrgsByErrorEventsQuery(), {
+			startTime: formatWarehouseDateTime(nowMs - ERROR_ACTIVE_DISCOVERY_WINDOW_MS),
+		})
 		return yield* warehouse
 			.crossOrgQuery(systemTenant(knownOrgs[0]!), compiled, {
 				// Bound the one cross-org scan (no OrgId predicate ⇒ can't prune the

apps/api/src/services/warehouse/WarehouseQueryService.test.ts

@@ -11,7 +11,7 @@ import {
 	TinybirdOrgTokenConfigError,
 	UserId,
 	WarehouseConfigError,
-	WarehouseQueryError,
+	WarehouseInvalidSqlError,
 	WarehouseResultDecodeError,
 	WarehouseScopeError,
 	WarehouseUpstreamError,
@@ -661,7 +661,10 @@ describe("WarehouseQueryService.ingest writes through the SQL client", () => {
 		}).pipe(Effect.provide(layer))
 	})
 
-	it.effect("maps a failed insert to WarehouseQueryError", () => {
+	// Inserts classify with the read path's default "caller" authorship (the
+	// rows, not Maple's SQL, are what usually earned the rejection), so a
+	// syntax-shaped complaint takes the caller-authored invalid-SQL tag.
+	it.effect("maps a failed insert through the classifier", () => {
 		__testables.setClientFactory(() => ({
 			sql: async () => ({ data: [] }),
 			insert: async () => {
@@ -679,7 +682,7 @@ describe("WarehouseQueryService.ingest writes through the SQL client", () => {
 
 			assert.isTrue(Exit.isFailure(exit))
 			const failure = getError(exit)
-			assert.instanceOf(failure, WarehouseQueryError)
+			assert.instanceOf(failure, WarehouseInvalidSqlError)
 		}).pipe(Effect.provide(layer))
 	})
 })

lib/clickhouse-builder/src/ch/compile.test.ts

@@ -35,6 +35,32 @@ describe("CompiledQuery.decodeRows", () => {
 		}),
 	)
 
+	// `T.custom("String", branded)` is how a caller brands an id column. The
+	// brand must survive derivation — it is the whole reason to declare it — and
+	// the column must still compare against a plain-string param.
+	it.effect("a branded custom column derives a branded row schema", () =>
+		Effect.gen(function* () {
+			const OrgId = Schema.String.check(Schema.isMinLength(1)).pipe(Schema.brand("OrgId"))
+			const table = CH.table(
+				"events",
+				{ OrgId: T.custom("String", OrgId), Count: CH.uint64 },
+				{ tenantColumn: "OrgId" },
+			)
+			const compiled = compileCHUnsafe(
+				CH.from(table)
+					.select(($) => ({ orgId: $.OrgId }))
+					.where(($) => [$.OrgId.eq(CH.param.string("orgId"))]),
+				{ orgId: "org_1" },
+			)
+
+			expect(compiled.rowSchemaSource).toBe("derived")
+			expect(yield* compiled.decodeRows([{ orgId: "org_1" }])).toEqual([{ orgId: "org_1" }])
+			// The brand's checks validate: an empty id is a decode failure.
+			const exit = yield* Effect.exit(compiled.decodeRows([{ orgId: "" }]))
+			expect(Exit.isFailure(exit)).toBe(true)
+		}),
+	)
+
 	it.effect("has no schema when a selected expression has no type to read", () =>
 		Effect.gen(function* () {
 			const table = CH.table("events", { OrgId: CH.string, Count: CH.uint64 })

lib/clickhouse-builder/src/ch/expr.test-d.ts

@@ -166,3 +166,18 @@ expectTypeOf(CH.inList(strExpr, ["a", "b"])).toMatchTypeOf<Condition>()
 
 expectTypeOf(CH.arrayOf(CH.lit("a"), CH.lit("b"))).toMatchTypeOf<Expr<ReadonlyArray<string>>>()
 expectTypeOf(CH.arrayOf(CH.lit(1), CH.lit(2))).toMatchTypeOf<Expr<ReadonlyArray<number>>>()
+
+// Branded column types — comparisons widen to the underlying primitive
+
+type BrandedId = string & { readonly __brand: "BrandedId" }
+declare const brandedRef: Expr<BrandedId>
+declare const plainStringExpr: Expr<string>
+
+// A branded column compares against a plain param/expr and a plain literal…
+expectTypeOf(brandedRef.eq(CH.param.string("orgId"))).toMatchTypeOf<Condition>()
+expectTypeOf(brandedRef.eq(plainStringExpr)).toMatchTypeOf<Condition>()
+expectTypeOf(brandedRef.eq("org_123")).toMatchTypeOf<Condition>()
+expectTypeOf(brandedRef.in_("a", "b")).toMatchTypeOf<Condition>()
+// …and against another ref of its own branded type.
+expectTypeOf(brandedRef.eq(brandedRef)).toMatchTypeOf<Condition>()
+expectTypeOf(CH.inList(brandedRef, ["a", "b"])).toMatchTypeOf<Condition>()

lib/clickhouse-builder/src/ch/expr.ts

@@ -23,6 +23,19 @@ import { encodeColumnLiteral } from "./literal"
  */
 export type Comparable<TSType> = TSType extends DateTime.Utc ? DateTime.Utc | Date | string : TSType
 
+/**
+ * A branded primitive compares as the primitive it brands.
+ *
+ * A column may decode to a branded type (`T.custom("String", OrgId)`), but the
+ * wire value it is compared against is the plain primitive — a param, another
+ * column, a literal. Without widening, `$.OrgId.eq(param.string("orgId"))`
+ * stops compiling the moment the column's schema brands its decoded type,
+ * which would make branding a breaking change instead of an annotation. The
+ * type-level mirror of `literalSchema`: comparisons may accept more than the
+ * column decodes to.
+ */
+export type Widen<TSType> = TSType extends string ? string : TSType extends number ? number : TSType
+
 export interface Expr<TSType> {
 	readonly _brand: "Expr"
 	readonly _phantom?: TSType
@@ -38,22 +51,27 @@ export interface Expr<TSType> {
 	readonly schema?: Schema.Codec<TSType, any>
 	toFragment(): SqlFragment
 
-	// Comparison — returns Condition
-	eq(other: Comparable<TSType> | Expr<TSType>): Condition
-	neq(other: Comparable<TSType> | Expr<TSType>): Condition
-	gt(other: Comparable<TSType> | Expr<TSType>): Condition
-	gte(other: Comparable<TSType> | Expr<TSType>): Condition
-	lt(other: Comparable<TSType> | Expr<TSType>): Condition
-	lte(other: Comparable<TSType> | Expr<TSType>): Condition
+	// Comparison — returns Condition. `Expr<TSType>` is listed alongside the
+	// widened form because `Expr` is invariant: a branded column must accept
+	// both its own refs and plain-primitive exprs (params, other columns).
+	// The widened arms sit in contravariant positions, which TypeScript's
+	// `extends Expr<infer T>` inference would prefer — the reason `InferOutput`
+	// reads the `_phantom` property instead of structurally inferring T.
+	eq(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
+	neq(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
+	gt(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
+	gte(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
+	lt(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
+	lte(other: Comparable<Widen<TSType>> | Expr<TSType> | Expr<Widen<TSType>>): Condition
 
 	// String operations
 	like(this: Expr<string>, pattern: string): Condition
 	notLike(this: Expr<string>, pattern: string): Condition
 	ilike(this: Expr<string>, pattern: string): Condition
 
 	// IN / NOT IN
-	in_(...values: Array<Comparable<TSType>>): Condition
-	notIn(...values: Array<Comparable<TSType>>): Condition
+	in_(...values: Array<Comparable<Widen<TSType>>>): Condition
+	notIn(...values: Array<Comparable<Widen<TSType>>>): Condition
 
 	// Arithmetic — only valid for number expressions
 	div(this: Expr<number>, n: number | Expr<number>): Expr<number>
@@ -348,7 +366,7 @@ export function outerRef<T = string>(name: string): Expr<T> {
 	return makeUntypedExpr<T>(raw(name))
 }
 
-export function inList(expr: Expr<string>, values: readonly string[]): Condition {
+export function inList<T extends string>(expr: Expr<T>, values: readonly string[]): Condition {
 	const escaped = values.map((v) => compile(str(v))).join(", ")
 	return makeCond(raw(`${compile(expr.toFragment())} IN (${escaped})`))
 }

lib/clickhouse-builder/src/ch/query.ts

@@ -46,8 +46,17 @@ export type JoinedColumnAccessor<
 
 type SelectRecord = Record<string, Expr<any>>
 
+/**
+ * Read each selected expression's output type off its `_phantom` property
+ * rather than `S[K] extends Expr<infer T>`. Structural inference prefers the
+ * contravariant candidates in the comparison methods, and those are widened
+ * (`Widen<TSType>`) so branded columns accept plain params — inferring through
+ * them resolved a branded column's output to the bare primitive. The indexed
+ * read is exact; `Exclude` only strips the `undefined` that `_phantom`'s
+ * optionality adds, so a `Nullable(...)` column's `| null` survives.
+ */
 export type InferOutput<S extends SelectRecord> = {
-	readonly [K in keyof S]: S[K] extends Expr<infer T> ? T : never
+	readonly [K in keyof S]: S[K] extends Expr<any> ? Exclude<S[K]["_phantom"], undefined> : never
 }
 
 type OrderBySpec<Output> = [keyof Output & string, "asc" | "desc"]

lib/clickhouse-builder/src/ch/types.ts

@@ -158,6 +158,17 @@ export type CHDateTime = CHType<"DateTime", DateTime.Utc, string>
 export type CHDateTime64 = CHType<"DateTime64", DateTime.Utc, string>
 export type CHBool = CHType<"Bool", boolean, boolean | number>
 
+/**
+ * A `String` column whatever its decoded type — plain, branded, or narrowed.
+ *
+ * The constraint form of `CHString`, for "this table must carry this String
+ * column" (a tenant column, a join key) where how the value decodes is the
+ * table's own business. Spelling such a constraint as `CHString` would reject
+ * a branded column: `CHType` is invariant in its decoded type, so
+ * `custom("String", OrgId)` is not a `CHType<"String", string>`.
+ */
+export type CHStringLike = CHType<"String", any, any>
+
 /**
  * The same columns left as the strings ClickHouse sends.
  *

lib/clickhouse-builder/src/types.ts

@@ -17,6 +17,7 @@ export {
 	type CHMap,
 	type CHNullable,
 	type CHString,
+	type CHStringLike,
 	type CHType,
 	type CHUInt8,
 	type CHUInt16,

packages/query-engine/src/ch/builder-fixtures.ts

@@ -534,6 +534,9 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 				...window,
 				fingerprintHash: FINGERPRINT,
 			}),
+		// TraceId/SpanId decode through their branded schemas (minLength 1), which
+		// the synthetic row's "" would fail.
+		sampleValues: { traceId: "0af7651916cd43dd8448eb211c80319c", spanId: "b7ad6b7169203331" },
 	},
 	{
 		module: "errors",
@@ -748,6 +751,7 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 				}).format("JSON"),
 				{ orgId: ORG_ID, hourStart: START_TIME, hourEnd: END_TIME },
 			),
+		sampleValues: { OrgId: ORG_ID },
 	},
 	{
 		// The service-scoped variant pushes the filter into the parent subquery.
@@ -764,6 +768,7 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 				}).format("JSON"),
 				{ orgId: ORG_ID, hourStart: START_TIME, hourEnd: END_TIME },
 			),
+		sampleValues: { OrgId: ORG_ID },
 	},
 	{
 		module: "service-map-rollup",
@@ -849,17 +854,20 @@ export const builderFixtures: ReadonlyArray<BuilderFixture> = [
 		name: "activeOrgsByErrorEventsQuery",
 		label: "default",
 		compile: () => CH.compileUnsafe(CH.activeOrgsByErrorEventsQuery(), { startTime: START_TIME }),
+		sampleValues: { orgId: ORG_ID },
 	},
 	{
 		module: "activity",
 		name: "activeOrgsByTracesQuery",
 		label: "default",
 		compile: () => CH.compileUnsafe(CH.activeOrgsByTracesQuery(), { startTime: START_TIME }),
+		sampleValues: { orgId: ORG_ID },
 	},
 	{
 		module: "activity",
 		name: "activeOrgsByLogsQuery",
 		label: "default",
 		compile: () => CH.compileUnsafe(CH.activeOrgsByLogsQuery(), { startTime: START_TIME }),
+		sampleValues: { orgId: ORG_ID },
 	},
 ]

packages/query-engine/src/ch/index.ts

@@ -259,7 +259,6 @@ export {
 	errorFingerprintsQuery,
 	errorIssueTimeseriesQuery,
 	errorIssueSampleTracesQuery,
-	ErrorIssueSampleTracesOutputSchema,
 	errorIssueEnvironmentsQuery,
 	errorIssueVersionsSinceQuery,
 	ErrorIssueVersionsSinceOutputSchema,
@@ -319,7 +318,6 @@ export {
 	activeOrgsByErrorEventsQuery,
 	activeOrgsByTracesQuery,
 	activeOrgsByLogsQuery,
-	ActiveOrgsOutputSchema,
 	type ActiveOrgsOutput,
 } from "./queries/activity"
 

packages/query-engine/src/ch/pipe-dispatch.ts

@@ -377,7 +377,6 @@ export function compilePipeQuery(
 							commitShas: str("commit_shas")?.split(",").filter(Boolean),
 						}),
 						{ orgId, startTime, endTime },
-						{ rowSchema: serviceOverviewRowSchema },
 					),
 				),
 			),

packages/query-engine/src/ch/queries/activity.ts

@@ -18,12 +18,14 @@
 // active org is missed for the tick.
 
 import { from, param } from "@maple-dev/clickhouse-builder"
-import { OrgId } from "@maple/domain"
-import { Schema } from "effect"
+import type { OrgId } from "@maple/domain"
 import { ErrorEventsByTime, LogsAggregatesHourly, TracesAggregatesHourly } from "../tables"
 
-export const ActiveOrgsOutputSchema = Schema.Struct({ orgId: OrgId })
-export type ActiveOrgsOutput = Schema.Schema.Type<typeof ActiveOrgsOutputSchema>
+/** The `OrgId` brand comes off the tables' branded `OrgId` column — the
+ *  derived row schema carries it, so no declared schema is needed. */
+export interface ActiveOrgsOutput {
+	readonly orgId: OrgId
+}
 
 /** Orgs with any error events since `startTime` (gates the error-issue detector). */
 export function activeOrgsByErrorEventsQuery() {

packages/query-engine/src/ch/queries/errors.ts

@@ -11,7 +11,7 @@ import { from, fromQuery, type CHQuery, type ColumnAccessor } from "@maple-dev/c
 import type { ColumnDefs } from "@maple-dev/clickhouse-builder/types"
 import * as T from "@maple-dev/clickhouse-builder/types"
 import { unionAll, type CHUnionQuery } from "@maple-dev/clickhouse-builder"
-import { SpanId, TraceId } from "@maple/domain"
+import type { SpanId, TraceId } from "@maple/domain"
 import { Schema } from "effect"
 import {
 	ErrorEvents,
@@ -1131,15 +1131,16 @@ export function errorIssueTimeseriesQuery() {
 
 // Error Issue sample traces — most recent occurrences for one issue
 
-export const ErrorIssueSampleTracesOutputSchema = Schema.Struct({
-	traceId: TraceId,
-	spanId: SpanId,
-	serviceName: Schema.String,
-	timestamp: Schema.String,
-	exceptionMessage: Schema.String,
-	durationMicros: CHNumber,
-})
-export type ErrorIssueSampleTracesOutput = Schema.Schema.Type<typeof ErrorIssueSampleTracesOutputSchema>
+/** `TraceId`/`SpanId` brands come off `ErrorEvents`' branded columns — the
+ *  derived row schema carries them, so no declared schema is needed. */
+export interface ErrorIssueSampleTracesOutput {
+	readonly traceId: TraceId
+	readonly spanId: SpanId
+	readonly serviceName: string
+	readonly timestamp: string
+	readonly exceptionMessage: string
+	readonly durationMicros: number
+}
 
 export function errorIssueSampleTracesQuery(opts: { limit?: number }) {
 	return from(ErrorEvents)

packages/query-engine/src/ch/queries/service-map-rollup.ts

@@ -18,6 +18,7 @@ import { compile } 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 { OrgId } from "@maple/domain"
 import { ServiceAddressResolutionsHourly, ServiceMapEdgesHourly, Traces } from "../tables"
 import { deploymentEnvExpr } from "@maple/domain/tinybird/semconv-renames"
 import { serviceMapEdgeJoinQuery } from "./service-map"
@@ -27,7 +28,7 @@ import type { QueryBuilderError } from "@maple-dev/clickhouse-builder"
 /** One pre-aggregated service-to-service edge bucket — mirrors the columns of
  * the `service_map_edges_hourly` ClickHouse table. */
 export interface ServiceMapEdgesHourlyOutput {
-	readonly OrgId: string
+	readonly OrgId: OrgId
 	readonly Hour: string
 	readonly SourceService: string
 	readonly TargetService: string
@@ -42,7 +43,9 @@ export interface ServiceMapEdgesHourlyOutput {
 }
 
 const ServiceMapEdgesHourlyOutputSchema: CompiledQueryRowSchema<ServiceMapEdgesHourlyOutput> = Schema.Struct({
-	OrgId: Schema.String,
+	// The tables' OrgId column is branded, so the derived output is too — a
+	// declared schema may only narrow, so it has to say the brand as well.
+	OrgId,
 	Hour: Schema.String,
 	SourceService: Schema.String,
 	TargetService: Schema.String,
@@ -173,7 +176,7 @@ export function serviceMapEdgesRollupSQL(
 /** One resolved address-to-service mapping bucket — mirrors the columns of
  * `service_address_resolutions_hourly`. */
 export interface ServiceAddressResolutionsHourlyOutput {
-	readonly OrgId: string
+	readonly OrgId: OrgId
 	readonly Hour: string
 	readonly SourceService: string
 	readonly ParentServerAddress: string