diff --git a/packages/opencode/src/altimate/mcp-server/server.ts b/packages/opencode/src/altimate/mcp-server/server.ts index 0e63beac0d..852df3fb56 100644 --- a/packages/opencode/src/altimate/mcp-server/server.ts +++ b/packages/opencode/src/altimate/mcp-server/server.ts @@ -43,6 +43,81 @@ export async function serverInstructions(): Promise { } } +// altimate_change start — lineage tools answer with structured content as well as text +// Lineage tools also return their metadata as structured content, so a client that draws lineage +// (the Claude Code plugin's lineage pane) reads edges and downstream lists without parsing text. +// Claude Code shows structured content to the model in place of the text, so `report` carries the +// (already truncated) text (dbt_lineage: its header, its mappings travel as data), only the lineage +// fields are passed, and the whole is capped. +const STRUCTURED_FIELDS: Record = { + dbt_lineage: ["model_name", "model_unique_id", "confidence", "confidence_factors", "upstream", "columns", "edges", "error"], + impact_analysis: [ + "model", + "column", + "change_type", + "severity", + "direct_count", + "transitive_count", + "test_count", + "downstream", + "exposures", + "column_reach", + "column_reach_error", + "exposures_error", + "error", + ], +} +export const STRUCTURED_TOOLS: ReadonlySet = new Set(Object.keys(STRUCTURED_FIELDS)) +const STRUCTURED_OUTPUT_SCHEMA = { + type: "object" as const, + properties: { + report: { type: "string", description: "The tool's text report" }, + truncated: { type: "boolean", description: "True when the lineage fields were left out for size; the report is kept, shortened if it is too large on its own" }, + }, + required: ["report"], + additionalProperties: true, +} +// The structured result reaches the model in Claude Code, so it stays within the same order of +// size as a text result: past this, only the report is sent. Claude Code caps an MCP result at +// 25k tokens by default and JSON of ids and edges runs ~3-4 characters a token, so this stays under that. +const STRUCTURED_MAX_BYTES = 60_000 + +export function structuredResult(tool: string, metadata: Record, report: string): Record { + // A tool that builds its own structured result (dbt_lineage from the dbt integration) hands it + // over whole; otherwise the lineage fields of its metadata are passed. + const own = metadata.structuredContent + if (own && typeof own === "object" && !Array.isArray(own)) { + // The text repeats every edge that `columnLineage` already carries: only its header goes with them. + const head = report.split(/\n(?:Column edges \(|Trace of )/)[0]!.trim() + const full = { ...(own as Record), report: head } + return Buffer.byteLength(JSON.stringify(full)) <= STRUCTURED_MAX_BYTES ? full : capReport(report, true) + } + const fields = Object.fromEntries((STRUCTURED_FIELDS[tool] ?? []).filter((key) => key in metadata).map((key) => [key, metadata[key]])) + // dbt_lineage's text prints the same mappings and edges as `columns` and `edges`: keep its header. + const head = tool === "dbt_lineage" && "columns" in fields ? report.split(/\n(?:Column Mappings|Lineage Edges):/)[0]!.trim() : report + const full = { ...fields, report: head } + return Buffer.byteLength(JSON.stringify(full)) <= STRUCTURED_MAX_BYTES ? full : capReport(report, true) +} +/** An error report held to the same cap as a result, so a huge exception message cannot exceed it. */ +export function structuredError(message: string): Record { + return capReport(message, false) +} +// `dropped`: the result's other fields were left out, so it is marked truncated even when the +// report itself fits. +function capReport(message: string, dropped: boolean): Record { + // Measured as sent: JSON escaping can grow the text (a newline is two bytes, a control + // character six), so the cap applies to the serialized result, shortening until it fits. + const size = (value: Record) => Buffer.byteLength(JSON.stringify(value)) + const whole = dropped ? { report: message, truncated: true } : { report: message } + if (size(whole) <= STRUCTURED_MAX_BYTES) return whole + let report = message + while (report.length > 0 && size({ report, truncated: true }) > STRUCTURED_MAX_BYTES) { + report = report.slice(0, Math.floor(report.length * 0.9)) + } + return { report, truncated: true } +} +// altimate_change end + /** Spilled results: a resource_link next to the text, which keeps the preview, row count and path. */ function resultLinks(metadata: Record) { // A batched sql_execute call reports one saved file per spilled query. @@ -97,6 +172,7 @@ export async function createServer(options: ServeOptions = {}) { name, description: descriptions[name] ?? tool.description, inputSchema: tool.jsonSchema as McpTool["inputSchema"], + ...(STRUCTURED_TOOLS.has(name) && { outputSchema: STRUCTURED_OUTPUT_SCHEMA }), })), ...(workspace?.tools ?? [])], })) // Ask the user once per session when the model weighed two readings and the client can ask; any @@ -146,19 +222,19 @@ export async function createServer(options: ServeOptions = {}) { const clarified = request.params.name === "answer_commit" ? await clarifyReading(request.params.arguments, result, extra.signal) : undefined if (clarified) return { content: [{ type: "text" as const, text: clarified }], isError: true } const isError = resultFailed(result) - // A tool that builds a structured result (dbt_lineage) hands it over beside its text. - const structured = result.metadata?.structuredContent return { content: [{ type: "text" as const, text: result.output }, ...resultLinks(result.metadata)], - ...(structured && typeof structured === "object" && !isError - ? { structuredContent: structured as Record } - : {}), isError, + ...(STRUCTURED_TOOLS.has(request.params.name) && { + structuredContent: structuredResult(request.params.name, result.metadata, result.output), + }), } } catch (error) { + const message = error instanceof Error ? error.message : String(error) return { - content: [{ type: "text" as const, text: error instanceof Error ? error.message : String(error) }], + content: [{ type: "text" as const, text: message }], isError: true, + ...(STRUCTURED_TOOLS.has(request.params.name) && { structuredContent: structuredError(message) }), } } } diff --git a/packages/opencode/src/altimate/native/dbt/column-reach.ts b/packages/opencode/src/altimate/native/dbt/column-reach.ts new file mode 100644 index 0000000000..b1de28984b --- /dev/null +++ b/packages/opencode/src/altimate/native/dbt/column-reach.ts @@ -0,0 +1,491 @@ +// altimate_change - new file +// +// Which downstream models read a column, and which exposures (dashboards) sit on them. Each +// downstream model's (or snapshot's) compiled SQL is traced with dbtLineage, in dependency order, so +// a column that is renamed or carried through an intermediate model is still followed. What the +// lineage engine cannot settle is reported as such: models it could not trace (no compiled SQL, or +// SQL it could not parse), models that name the column only outside their select list (where the +// engine reports no edge), and models that read it through a table name several parents share. +import { loadRawManifest, findModel } from "./helpers" +import { dbtLineage } from "./lineage" + +export interface ColumnReachModel { + name: string + unique_id: string + /** Its output columns that come from the traced column. */ + columns: string[] +} + +export interface ColumnReachUnsure { + name: string + unique_id: string + /** Why: no compiled SQL or a parse error, a mention outside the select list, an ambiguous table name, or an unsettled parent. */ + reason: string +} + +export interface ColumnReachExposure { + /** Identifies the exposure across packages that reuse a name. */ + unique_id: string + name: string + /** The exposure's label when it has one, else its name. */ + label: string + type: string +} + +export interface ColumnReach { + /** Models whose output columns certainly come from the column. */ + models: ColumnReachModel[] + /** + * Models that may read it: they name it outside their select list (or a column of that name + * another parent shares), read it through a table name more than one parent shares, or sit + * downstream of a model that could not be traced or read it ambiguously. + */ + uncertain: ColumnReachUnsure[] + /** Downstream models that could not be traced (no compiled SQL, or SQL the engine could not parse). */ + untraced: ColumnReachUnsure[] + /** Exposures on the model or on any model that certainly reads the column. */ + exposures: ColumnReachExposure[] + /** Exposures on models that may read it or could not be traced (and on none of the certain readers). */ + uncertain_exposures: ColumnReachExposure[] + /** True when the walk stopped at the model limit before covering everything downstream. */ + truncated: boolean +} + +const MAX_MODELS = 300 +const TRACED = new Set(["model", "snapshot"]) + +/** "db"."schema"."orders"."amount" → { parts: ["db","schema","orders"], column: "amount" }, lower-cased. */ +function splitSource(source: unknown): { parts: string[]; column: string; exact: string } | null { + const text = + typeof source === "string" ? source : typeof (source as any)?.source === "string" ? (source as any).source : null + if (!text) return null + // Split on dots outside quotes: "a"."net.amount" is table a, column net.amount. + const raw = [...text.matchAll(/"([^"]*)"|`([^`]*)`|\[([^\]]*)\]|([^."`\[\]]+)/g)].map( + (m) => m[1] ?? m[2] ?? m[3] ?? m[4] ?? "", + ) + if (raw.length < 2) return null + // `exact` keeps the column's case, for parents that document two columns differing only by case. + return { + parts: raw.slice(0, -1).map((part) => part.toLowerCase()), + column: raw.at(-1)!.toLowerCase(), + exact: raw.at(-1)!, + } +} + +/** The ways a parent's relation can be written in a child's compiled SQL, most qualified first. */ +function relationKeys(node: any): string[] { + const table = [node?.alias, node?.identifier, node?.name].filter( + (v): v is string => typeof v === "string" && v !== "", + ) + const schema = typeof node?.schema === "string" ? node.schema : null + const database = typeof node?.database === "string" ? node.database : null + const keys = new Set() + for (const name of table) { + if (database && schema) keys.add(`${database}.${schema}.${name}`.toLowerCase()) + if (schema) keys.add(`${schema}.${name}`.toLowerCase()) + keys.add(name.toLowerCase()) + } + return [...keys] +} + +/** + * The parents an edge's source table names, matched at the qualification the edge carries: a + * qualified name that matches no parent names none (it is some other relation), and a bare name + * more than one parent answers to names them all. + */ +export function parentsNamed(parts: string[], parents: string[], lookup: (id: string) => any): string[] { + const key = parts.slice(-3).join(".") + return parents.filter((parent) => relationKeys(lookup(parent)).includes(key)) +} + +/** Whether a node documents more than one column with this name once case is ignored ("revenue" and "Revenue"). */ +function caseCollides(node: any, lower: string): boolean { + const names = new Set(Object.entries(node?.columns ?? {}).map(([key, col]) => String(col?.name ?? key))) + return [...names].filter((name) => name.toLowerCase() === lower).length > 1 +} + +/** Column names the manifest documents for a node, lower-cased. */ +function documentedColumns(node: any): Set { + return new Set(Object.entries(node?.columns ?? {}).map(([key, col]) => String(col?.name ?? key).toLowerCase())) +} + +const escapeRegExp = (text: string) => text.replace(/[.*+?^${}()|[\]\\]/g, "\\$&") + +/** + * SQL with comments blanked and string literals (quoted or dollar-quoted) emptied, in one pass so + * none can swallow another. + */ +function stripSql(sql: string): string { + return sql.replace(/'(?:[^']|'')*'|\$([A-Za-z_]\w*)?\$[\s\S]*?\$\1\$|--[^\r\n]*|\/\*[\s\S]*?\*\//g, (match) => + match.startsWith("'") || match.startsWith("$") ? "''" : " ", + ) +} + +/** + * Whether the SQL selects `*` or `t.*` in any form (DISTINCT ON, TOP n, AS STRUCT, multi-part or + * quoted qualifiers, EXCLUDE/EXCEPT/REPLACE/RENAME modifiers). Every `*` counts except count(*) and + * multiplication, which has an operand on both sides that is not a keyword ending the projection. + */ +export function selectsStar(sql: string): boolean { + const code = stripSql(sql) + const ends = /^(from|exclude|except|replace|rename|ilike|columns)\b/i + for (let at = code.indexOf("*"); at !== -1; at = code.indexOf("*", at + 1)) { + const before = code.slice(0, at).trimEnd() + const after = code.slice(at + 1).trimStart() + const prev = before.at(-1) ?? "" + const next = after[0] ?? "" + // count(*) reads no column; hash(*), object_construct(*), struct(*) read them all. + if (prev === "(" && next === ")" && /\bcount(?:_big|_if)?\s*\($/i.test(before)) continue + if (prev === "." || next === "" || next === "," || next === ")" || ends.test(after)) return true + const operand = /[\w"'`\])]/ + // A sign on the right operand (`revenue * -1`) belongs to it. + const right = after.replace(/^[+-]\s*/, "")[0] ?? "" + if (operand.test(prev) && (operand.test(right) || right === "(" || /\d/.test(right))) { + // An operand before the `*` that is itself a keyword (SELECT, DISTINCT, ALL, STRUCT) is no operand. + if (/\b(select|distinct|all|struct)$/i.test(before)) return true + continue + } + return true + } + return false +} + +/** + * Whether compiled SQL names the column anywhere but as an output alias, comments and string + * literals aside. Used only to mark a model as one that may read the column. + */ +function namesColumn(sql: string, column: string): boolean { + const code = stripSql(sql) + const name = escapeRegExp(column) + const ident = new RegExp(`(^|[^A-Za-z0-9_$])("${name}"|${name})(?![A-Za-z0-9_$])`, "gi") + for (const match of code.matchAll(ident)) { + const before = code.slice(0, match.index! + match[1]!.length) + if (/\bas\s*`?$/i.test(before)) continue + return true + } + return false +} + +/** + * Whether the SQL may have a NATURAL JOIN, whose join columns the lineage engine reports no edge for. + * Read from the raw SQL: the words inside a string, comment or quoted name count too, which only adds + * doubt (a comment mentioning one marks the model uncertain), while masking quotes would let a dialect's + * quoting rules hide a real join. Comments between the keywords are read as gaps. + * Limit: a parent's whole row passed as a value (`to_json(o)`) is not seen; telling a row source from + * a column needs the engine's parse, not a pattern. + */ +// Between the keywords: whitespace, or a comment (`--`, `#` or `//` to the line's end, `/* … */` to its +// first `*/`). Limit: a nested Postgres comment between them is not read as one. +const GAP = String.raw`(?:\s|(?:--|#|\/\/)[^\r\n]*[\r\n]|\/\*(?:[^*]|\*(?!\/))*\*\/)+` +// The join's modifiers in any dialect: LEFT/RIGHT/FULL/INNER/CROSS, OUTER, Spark's SEMI/ANTI, Snowflake's DIRECTED. +const NATURAL = new RegExp( + String.raw`^natural${GAP}(?:(?:left|right|full|inner|cross|outer|semi|anti|directed)${GAP}){0,4}join\b`, + "i", +) +// How far past a `natural` the rest of the join is looked for, and how many are looked at: past that +// many the SQL may well have one, which only adds doubt. An unclosed `/*` makes each check read to the +// window's end, so the window is what keeps crafted input cheap. +// Limit: more than 10,000 characters of comments between the keywords hide the join. +const NATURAL_WINDOW = 10_000 +const MAX_NATURALS = 100 +export function joinsNaturally(sql: string): boolean { + let seen = 0 + for (const match of sql.matchAll(/(? MAX_NATURALS || NATURAL.test(sql.slice(match.index!, match.index! + NATURAL_WINDOW))) return true + } + return false +} + +// `IDENTIFIER (` with comments allowed around its parts; when its whole argument is one string +// literal, that string (group 1) and the closing parenthesis (group 2). +const IDENTIFIER_CALL = new RegExp( + String.raw`^identifier(?:${GAP})?\((?:${GAP})?(?:'((?:[^'\\]|''|\\[\s\S])*)'(?:${GAP})?(\))?)?`, + "i", +) + +/** + * Whether the SQL may name one of `columns` through `IDENTIFIER(…)` (Snowflake, Databricks): the + * lineage engine reads the argument as a string, so it reports no edge for the column. Only an + * argument that is one plain string literal naming none of them is cleared; anything else (a bind or + * session variable, a concatenation, a function, an escape) counts, which only adds doubt. Bounded like + * joinsNaturally: each `identifier` is read in a window, and past 100 of them the SQL counts. + */ +export function namesThroughIdentifier(sql: string, columns: string[]): boolean { + const wanted = new Set(columns.map((one) => one.toLowerCase())) + let seen = 0 + for (const match of sql.matchAll(/(? MAX_NATURALS) return true + const call = IDENTIFIER_CALL.exec(sql.slice(match.index!, match.index! + NATURAL_WINDOW)) + // `identifier` not called is a name, not the function. + if (!call) continue + if (call[1] === undefined || !call[2]) return true + const literal = call[1].replace(/''/g, "'") + if (!/^[\w$."]*$/.test(literal)) return true + // Any mention of the column in the string counts (qualified, quoted, with dots in a quoted name); + // a doubled quote inside a quoted name (`"a""b"`) is one quote. + const text = literal.replace(/""/g, '"').toLowerCase() + if ([...wanted].some((column) => text.includes(column))) return true + } + return false +} + +function childMapOf(manifest: any): Record { + if (manifest.child_map) return manifest.child_map + const children: Record = {} + for (const group of [manifest.nodes, manifest.exposures]) { + for (const [id, node] of Object.entries(group ?? {})) { + for (const parent of node?.depends_on?.nodes ?? []) (children[parent] ??= []).push(id) + } + } + return children +} + +export function columnReach(manifestPath: string, model: string, column: string, dialect?: string): ColumnReach { + const manifest = loadRawManifest(manifestPath) + if (!manifest) throw new Error(`Manifest not found at ${manifestPath}`) + const nodes: Record = manifest.nodes ?? {} + const lookup = (id: string) => nodes[id] ?? manifest.sources?.[id] + const target = findModel(nodes, model) + if (!target) throw new Error(`Model '${model}' not found in manifest`) + const targetId: string = target.unique_id ?? Object.keys(nodes).find((id) => nodes[id] === target) + const children = childMapOf(manifest) + + // Everything downstream, then a topological order over it, so each node is traced after all its + // traced parents. + const downstream = new Set() + const queue = [targetId] + while (queue.length) { + for (const child of children[queue.shift()!] ?? []) { + if (downstream.has(child) || !TRACED.has(nodes[child]?.resource_type)) continue + downstream.add(child) + queue.push(child) + } + } + const pending = new Map() + for (const id of downstream) { + const parents: string[] = nodes[id]?.depends_on?.nodes ?? [] + pending.set(id, parents.filter((p) => downstream.has(p)).length) + } + const ready = [...downstream].filter((id) => pending.get(id) === 0) + const order: string[] = [] + while (ready.length) { + const id = ready.shift()! + order.push(id) + for (const child of children[id] ?? []) { + if (!pending.has(child)) continue + const left = pending.get(child)! - 1 + pending.set(child, left) + if (left === 0) ready.push(child) + } + } + + const column0 = column.toLowerCase() + // Columns certainly carried from the traced column, per node. + const tracked = new Map>([[targetId, new Set([column0])]]) + // The same columns with their case kept: consulted only where a parent's names differ by case alone. + // The target's own spelling comes from the manifest: the caller's when documented as typed, else + // the one spelling matching ignoring case, else the caller's. Two documented spellings that only + // case tells apart, neither typed as given, leave the column unsettled: no reader of either is certain. + const spellings = Object.entries(target.columns ?? {}).map(([key, col]) => String(col?.name ?? key)) + const matches = spellings.filter((name) => name.toLowerCase() === column0) + const exact = spellings.includes(column) + const unsettled = !exact && matches.length > 1 + let spelled: string[] = [column] + if (unsettled) spelled = [] + else if (!exact && matches.length > 0) spelled = matches + const trackedExact = new Map>([[targetId, new Set(spelled)]]) + const carries = (parent: string, source: { column: string; exact: string }) => + !!tracked.get(parent)?.has(source.column) && + (!caseCollides(lookup(parent), source.column) || !!trackedExact.get(parent)?.has(source.exact)) + // Nodes whose outputs are unknown (untraced, or read through an ambiguous name): their children + // may read the column too. A node that only filters on it was traced and carries nothing on. + const unsure = new Set(unsettled ? [targetId] : []) + const reached: ColumnReachModel[] = [] + const uncertain: ColumnReachUnsure[] = [] + const untraced: ColumnReachUnsure[] = [] + let truncated = false + let traced = 0 + const entry = (id: string, reason: string) => ({ name: nodes[id]?.name ?? id, unique_id: id, reason }) + + for (const id of order) { + const allParents: string[] = nodes[id]?.depends_on?.nodes ?? [] + const parents = allParents.filter((p) => tracked.has(p)) + const fromUnsure = allParents.some((p) => unsure.has(p)) + if (parents.length === 0 && !fromUnsure) continue + if (traced++ >= MAX_MODELS) { + truncated = true + break + } + + const result = dbtLineage({ manifest_path: manifestPath, model: id, dialect }) + const lineage = result.raw_lineage as any + // dbtLineage rates a model low when the engine threw or answered SQL it cannot parse (`errors`, no edges). + if (result.confidence === "low") { + untraced.push(entry(id, result.confidence_factors[0] ?? "the lineage engine could not trace it")) + unsure.add(id) + continue + } + const edges: { source: unknown; target: string; lineage_type?: string }[] = lineage?.column_lineage?.length + ? lineage.column_lineage + : Object.entries(lineage?.column_dict ?? {}).flatMap(([out, sources]) => + (Array.isArray(sources) ? sources : [sources]).map((source) => ({ + source, + target: out, + lineage_type: "direct", + })), + ) + // Errors beside some edges: the engine traced only part of the SQL. + const partial = Array.isArray(lineage?.errors) && lineage.errors.length > 0 + + // Two parents with the same table name share one entry in the engine's schema, so its edges + // cannot say which one is read; a `select *` from a parent whose schema does not list the + // column cannot be expanded to it. Both are possible reads with unknown outputs. + const sqlText: string = nodes[id]?.compiled_code || nodes[id]?.compiled_sql || "" + const bareName = (p: string) => { + const node = lookup(p) + return String(node?.alias || node?.identifier || node?.name || p).toLowerCase() + } + const collides = parents.some((p) => allParents.some((other) => other !== p && bareName(other) === bareName(p))) + const wildcard = + selectsStar(sqlText) && + parents.some((p) => [...tracked.get(p)!].some((col) => !documentedColumns(lookup(p)).has(col))) + if (collides || wildcard) { + unsure.add(id) + uncertain.push( + entry( + id, + collides + ? "reads a parent whose table name another parent shares" + : "selects * from a parent whose schema.yml does not list the column", + ), + ) + continue + } + + const carried = new Set() + const carriedExact = new Set() + let reads = false + let ambiguous = false + for (const edge of edges) { + const source = splitSource(edge.source) + if (!source) continue + const candidates = parentsNamed(source.parts, allParents, lookup) + if (!candidates.some((p) => carries(p, source))) continue + // A bare table name that more than one parent answers to cannot say which one is read. + if (candidates.length > 1) { + ambiguous = true + continue + } + reads = true + if (edge.lineage_type !== "indirect" && typeof edge.target === "string") { + carried.add(edge.target.toLowerCase()) + carriedExact.add(edge.target) + } + } + // The engine has no schema for the tables inside an inlined ephemeral parent, so it reports no + // edge through it: unknown, not unread, even when another parent's edges are traced. + const viaEphemeral = parents.some((p) => lookup(p)?.config?.materialized === "ephemeral") + const natural = joinsNaturally(sqlText) + // The traced columns this model's parents carry. + const trackedColumns = parents.flatMap((parent) => [...tracked.get(parent)!]) + // The column named as a string the engine cannot trace: its outputs are unknown, like a NATURAL JOIN's. + const viaIdentifier = namesThroughIdentifier(sqlText, trackedColumns) + const opaque = natural || viaIdentifier + if (reads) { + tracked.set(id, carried) + trackedExact.set(id, carriedExact) + // Outputs fed by an unsettled or ephemeral parent, or read through an ambiguous name, stay unsettled. + // A NATURAL JOIN beside the column may join on it, or carry it on in another output. + if (ambiguous || fromUnsure || viaEphemeral || partial || opaque) unsure.add(id) + reached.push({ name: nodes[id]?.name ?? id, unique_id: id, columns: [...carried].sort() }) + continue + } + + if (partial) { + unsure.add(id) + uncertain.push(entry(id, "the lineage engine traced only part of its SQL")) + continue + } + if (viaEphemeral) { + unsure.add(id) + uncertain.push(entry(id, "reads an ephemeral parent, inlined as a CTE the lineage engine cannot trace through")) + continue + } + const named = trackedColumns.filter((col) => namesColumn(sqlText, col)) + // No edge at all while the SQL names the column (a PIVOT, say) is the engine not seeing through + // it, so its outputs are unknown. SQL that never names it (`select 1`, `count(*)`) reads nothing and + // falls through. Below an unsure parent it is already reported as uncertain. + if (edges.length === 0 && !fromUnsure && named.length > 0) { + uncertain.push(entry(id, "names it, but the lineage engine found no column edges in its SQL")) + unsure.add(id) + continue + } + // A NATURAL JOIN may join on the column with no edge the engine can report. + if (!fromUnsure && !ambiguous && named.length === 0 && opaque) { + unsure.add(id) + uncertain.push( + entry( + id, + viaIdentifier + ? "names it through IDENTIFIER(), which the lineage engine cannot trace" + : "uses a NATURAL JOIN, whose join columns the lineage engine cannot trace", + ), + ) + continue + } + const shared = named.some( + (col) => allParents.filter((other) => documentedColumns(lookup(other)).has(col)).length > 1, + ) + if (ambiguous || named.length > 0 || fromUnsure) { + if (ambiguous || fromUnsure || opaque) unsure.add(id) + uncertain.push( + entry( + id, + ambiguous + ? "reads it through a table name more than one parent shares" + : named.length > 0 + ? shared + ? "names it outside its select list, or names another parent's column of the same name" + : "names it only outside its select list (WHERE, JOIN, GROUP BY or HAVING)" + : "is downstream of a model that could not be traced or reads it ambiguously", + ), + ) + } + } + + // An unsettled spelling leaves the target's own dashboards uncertain too: which column they read is unknown. + const reachedIds = new Set([...(unsettled ? [] : [targetId]), ...reached.map((m) => m.unique_id)]) + const certain = exposuresOn(manifestPath, reachedIds) + const unsureIds = [...uncertain, ...untraced].map((m) => m.unique_id) + const maybe = exposuresOn(manifestPath, unsettled ? [targetId, ...unsureIds] : unsureIds).filter( + (e) => !certain.some((c) => c.unique_id === e.unique_id), + ) + return { models: reached, uncertain, untraced, exposures: certain, uncertain_exposures: maybe, truncated } +} + +/** Exposures that depend on any of the given unique_ids, read from the raw manifest. */ +export function exposuresOn(manifestPath: string, ids: Iterable): ColumnReachExposure[] { + const manifest = loadRawManifest(manifestPath) + const wanted = new Set(ids) + // A metric reads its semantic model, which reads a model: whatever depends on one depends on the others. + const layers = [ + ...Object.entries(manifest?.semantic_models ?? {}), + ...Object.entries(manifest?.metrics ?? {}), + ] + for (let grew = true; grew; ) { + grew = false + for (const [id, node] of layers) { + if (wanted.has(id) || !(node?.depends_on?.nodes ?? []).some((dep: string) => wanted.has(dep))) continue + wanted.add(id) + grew = true + } + } + return Object.entries(manifest?.exposures ?? {}) + .filter(([, node]) => (node?.depends_on?.nodes ?? []).some((dep: string) => wanted.has(dep))) + .map(([id, node]) => ({ + unique_id: id, + name: node?.name ?? id, + label: node?.label || node?.name || id, + type: node?.type ?? "exposure", + })) +} diff --git a/packages/opencode/src/altimate/native/dbt/graph.ts b/packages/opencode/src/altimate/native/dbt/graph.ts new file mode 100644 index 0000000000..3edff2baf9 --- /dev/null +++ b/packages/opencode/src/altimate/native/dbt/graph.ts @@ -0,0 +1,120 @@ +// altimate_change - new file +// +// The part of a dbt manifest's graph a lineage viewer draws for one model: everything upstream, +// everything downstream (with the tests counted as ids), and the links between them. Clients that +// cannot read a large manifest themselves (the Claude Code plugin reads files up to 4 MiB) get it +// from the CLI (`altimate tool run dbt_lineage --json` with include_graph); MCP carries no graph. + +export interface GraphNode { + id: string + name: string + /** An exposure's label; null for other nodes. */ + label: string | null + type: string + path: string | null + materialized: string | null + /** database.schema.table, lower-cased. */ + relation: string | null + columns: string[] +} + +export interface LineageGraph { + nodes: Record + parents: Record + children: Record + /** True when downstream stopped at the node limit. */ + truncated: boolean + /** True when upstream stopped at the node limit. */ + upTruncated: boolean +} + +const GROUPS = ["nodes", "sources", "exposures", "metrics", "semantic_models", "saved_queries", "unit_tests"] as const +/** Limits on each side; the target and test nodes come on top. A source-to-mart chain of a few hundred models fits. */ +const MAX_DOWNSTREAM = 500 +const MAX_UPSTREAM = 500 + +function lookup(manifest: any, id: string): any { + for (const group of GROUPS) if (manifest[group]?.[id]) return manifest[group][id] + return undefined +} + +function relationOf(node: any): string | null { + // A manifest that carries relation_name with null (ephemeral models, analyses, tests) has no relation. + if (node && "relation_name" in node && !node.relation_name) return null + if (typeof node?.relation_name === "string" && node.relation_name) + return node.relation_name.replace(/[`"\[\]]/g, "").toLowerCase() + const parts = [node?.database, node?.schema, node?.alias ?? node?.identifier ?? node?.name].filter( + (part): part is string => typeof part === "string" && part !== "", + ) + return parts.length === 3 ? parts.join(".").toLowerCase() : null +} + +function nodeOf(id: string, node: any): GraphNode { + const type: string = node?.resource_type ?? id.split(".")[0] ?? "unknown" + return { + id, + name: node?.name ?? id.split(".").pop() ?? id, + label: type === "exposure" ? node?.label || node?.name || id : null, + type, + path: node?.original_file_path ?? null, + materialized: node?.config?.materialized ?? null, + relation: relationOf(node), + columns: Object.entries(node?.columns ?? {}).map(([key, col]) => String(col?.name ?? key)), + } +} + +function mapsOf(manifest: any): { parents: Record; children: Record } { + if (manifest.parent_map && manifest.child_map) return { parents: manifest.parent_map, children: manifest.child_map } + const parents: Record = {} + const children: Record = {} + for (const group of GROUPS) { + for (const [id, node] of Object.entries(manifest[group] ?? {})) { + parents[id] = node?.depends_on?.nodes ?? [] + for (const parent of parents[id]) (children[parent] ??= []).push(id) + } + } + return { parents, children } +} + +export function lineageGraph(manifest: any, id: string): LineageGraph { + const { parents, children } = mapsOf(manifest) + const isTest = (one: string) => + ["test", "unit_test"].includes(lookup(manifest, one)?.resource_type ?? one.split(".")[0]) + const keep = new Set([id]) + // Breadth first, so a cut keeps the nearest nodes. + const walk = (links: Record, limit: number, seen = new Set([id])) => { + const queue = [id] + let count = 0 + while (queue.length) { + for (const next of links[queue.shift()!] ?? []) { + if (seen.has(next) || isTest(next)) continue + if (count >= limit) return true + seen.add(next) + keep.add(next) + count++ + queue.push(next) + } + } + return false + } + const upTruncated = walk(parents, MAX_UPSTREAM) + // Tests are counted for the target and what is downstream of it only; upstream ones would only add weight. + const counted = new Set([id]) + const truncated = walk(children, MAX_DOWNSTREAM, counted) + const graph: LineageGraph = { nodes: {}, parents: {}, children: {}, truncated, upTruncated } + for (const one of keep) { + graph.nodes[one] = nodeOf(one, lookup(manifest, one)) + graph.parents[one] = (parents[one] ?? []).filter((other) => keep.has(other)) + graph.children[one] = (children[one] ?? []).filter((other) => { + if (keep.has(other)) return true + if (!isTest(other) || !counted.has(one)) return false + if (!graph.nodes[other]) { + graph.nodes[other] = { ...nodeOf(other, lookup(manifest, other)), type: "test" } + graph.parents[other] = (parents[other] ?? []).filter((p) => keep.has(p)) + graph.children[other] = [] + } + return true + }) + } + return graph +} diff --git a/packages/opencode/src/altimate/native/dbt/lineage.ts b/packages/opencode/src/altimate/native/dbt/lineage.ts index 49b4cf9a96..eac0cff688 100644 --- a/packages/opencode/src/altimate/native/dbt/lineage.ts +++ b/packages/opencode/src/altimate/native/dbt/lineage.ts @@ -6,7 +6,14 @@ import * as core from "@altimateai/altimate-core" import type { DbtLineageParams, DbtLineageResult } from "../types" -import { loadRawManifest, findModel, getUniqueId, detectDialect, buildSchemaContext } from "./helpers" +import { loadRawManifest, findModel, detectDialect, buildSchemaContext } from "./helpers" +import { lineageGraph } from "./graph" + +function unparsed(lineage: Record): boolean { + const errors = Array.isArray(lineage.errors) ? lineage.errors : [] + const edges = Array.isArray(lineage.column_lineage) ? lineage.column_lineage : [] + return errors.length > 0 && edges.length === 0 +} /** * Compute column-level lineage for a dbt model. @@ -35,16 +42,48 @@ export function dbtLineage(params: DbtLineageParams): DbtLineageResult { const nodes = manifest.nodes || {} const sources = manifest.sources || {} - // Find target model by name or unique_id - const modelNode = findModel(nodes, params.model) + // A model by name or unique_id; a snapshot by unique_id; either by its project-relative file path. + // `./models/x.sql` and `models\\x.sql` are the manifest's `models/x.sql`. + const filePath = (p: unknown) => + String(p ?? "") + .replace(/\\/g, "/") + .replace(/^(?:\.\/)+/, "") + // An empty request (`./`) is no model's path. + const wanted = filePath(params.model) + const byPath = () => + wanted + ? Object.values(nodes).find( + (node) => ["model", "snapshot"].includes(node?.resource_type) && filePath(node.original_file_path) === wanted, + ) + : undefined + const modelNode = + findModel(nodes, params.model) ?? + (nodes[params.model]?.resource_type === "snapshot" ? nodes[params.model] : null) ?? + byPath() ?? + Object.values(nodes).find((node) => node?.resource_type === "snapshot" && node.name === params.model) ?? + null if (!modelNode) { return emptyResult([`Model '${params.model}' not found in manifest`]) } + // The id of the node that was selected, not of whatever else in the manifest the request string spells. + const uniqueId = + (typeof modelNode.unique_id === "string" && nodes[modelNode.unique_id] === modelNode + ? modelNode.unique_id + : undefined) ?? + Object.keys(nodes).find((id) => nodes[id] === modelNode) ?? + params.model + const graph = params.include_graph && uniqueId ? { graph: lineageGraph(manifest, uniqueId) } : {} // Extract compiled SQL const sql = modelNode.compiled_code || modelNode.compiled_sql || "" if (!sql) { - return emptyResult(["No compiled SQL found — run `dbt compile` first"]) + return { + ...emptyResult(["No compiled SQL found — run `dbt compile` first"]), + model_name: modelNode.name || params.model, + model_unique_id: uniqueId, + upstream: modelNode.depends_on?.nodes ?? [], + ...graph, + } } // Detect dialect @@ -66,10 +105,17 @@ export function dbtLineage(params: DbtLineageParams): DbtLineageResult { return { model_name: modelNode.name || params.model, - model_unique_id: getUniqueId(nodes, params.model), + model_unique_id: uniqueId, compiled_sql: sql, raw_lineage: rawLineage, - confidence: rawLineage.error ? "low" : "high", - confidence_factors: rawLineage.error ? [String(rawLineage.error)] : [], + // The engine reports SQL it cannot parse in `errors` with no edges, rather than throwing. + confidence: rawLineage.error || unparsed(rawLineage) ? "low" : "high", + confidence_factors: rawLineage.error + ? [String(rawLineage.error)] + : unparsed(rawLineage) + ? [String((rawLineage.errors as unknown[])[0])] + : [], + upstream: upstreamIds, + ...graph, } } diff --git a/packages/opencode/src/altimate/native/types.ts b/packages/opencode/src/altimate/native/types.ts index 431d1f41a5..5de10c591f 100644 --- a/packages/opencode/src/altimate/native/types.ts +++ b/packages/opencode/src/altimate/native/types.ts @@ -1076,6 +1076,8 @@ export interface DbtLineageParams { manifest_path: string model: string dialect?: string + /** Also return the model's lineage graph (everything upstream and downstream, up to 500 nodes each side). */ + include_graph?: boolean } export interface DbtLineageResult { @@ -1085,6 +1087,10 @@ export interface DbtLineageResult { raw_lineage: Record confidence: string confidence_factors: string[] + /** unique_ids the model depends on (`depends_on.nodes`). */ + upstream?: string[] + /** With include_graph: the graph a lineage viewer draws for the model. */ + graph?: import("./dbt/graph").LineageGraph } // --- dbt Profile Discovery --- diff --git a/packages/opencode/src/altimate/tools/dbt-lineage.ts b/packages/opencode/src/altimate/tools/dbt-lineage.ts index 8cdcba9596..7b27dd2a7e 100644 --- a/packages/opencode/src/altimate/tools/dbt-lineage.ts +++ b/packages/opencode/src/altimate/tools/dbt-lineage.ts @@ -1,8 +1,11 @@ import z from "zod" +import fs from "fs" import path from "path" import { Tool } from "../../tool/tool" import { Dispatcher } from "../native" import type { DbtLineageResult } from "../native/types" +import { loadRawManifest } from "../native/dbt/helpers" +import { lineageGraph } from "../native/dbt/graph" /** The structured result of dbt-integration's `getModelLineage` (version 1), relayed as is. */ export type ModelLineage = { @@ -36,27 +39,44 @@ export const DbtLineageTool = Tool.define("dbt_lineage", { description: "Column-level lineage for a dbt model: its upstream models, seeds and sources, its downstream models, their columns, and the column edges between them across every hop. Pass `column` to trace one column. Uses the project's dbt integration (compiles on demand, no `dbt compile` needed) and returns the result as structured content; without it, reads the compiled SQL in the manifest.", parameters: z.object({ - model: z.string().describe("Model name or unique_id (e.g. 'my_model' or 'model.project.my_model')"), + model: z + .string() + .describe("Model name, unique_id or project-relative SQL path (e.g. 'my_model', 'model.project.my_model', 'models/my_model.sql'); a snapshot by name, unique_id or path"), column: z.string().optional().describe("Trace only this column of the model, upstream and downstream"), manifest_path: z .string() .optional() .describe( - "manifest.json for the manifest-only fallback, read only when the dbt integration isn't available (default: the dbt project's target/manifest.json)", + "manifest.json for the manifest-only fallback, read only when the dbt integration isn't available, and for the graph (default: the dbt project's target/manifest.json)", ), dialect: z.string().optional().describe("SQL dialect override for the manifest-only fallback"), + include_graph: z + .boolean() + .optional() + .describe("Also return the model's graph from the manifest (everything upstream and downstream, up to 500 nodes each side; the exposures and tests are those of the model and its downstream, upstream tests are not included; graph.upTruncated / graph.truncated are set when a side is cut) in the metadata; for lineage viewers. Over MCP the graph is left out of the structured result: read it with `altimate tool run dbt_lineage --json`"), }), async execute(args, ctx) { - const viaAdapter = await fromAdapter(args.model, args.column, ctx.abort) + let viaAdapter = await fromAdapter(args.model, args.column, ctx.abort) + // The adapter answers for the project it found; a manifest named outside that project's folder + // wins, so its graph and its lineage are never mixed with this one's. + if (viaAdapter.result && args.manifest_path) { + const outside = await manifestOutsideAdapterProject(args.manifest_path) + if (outside) viaAdapter = { reason: outside } + } if (viaAdapter.result) { const result = viaAdapter.result const what = result.focus.column ? `${result.focus.name}.${result.focus.column}` : result.focus.name + // A lineage viewer also wants the manifest's graph: exposures and tests sit there, not here. + const graph = args.include_graph ? manifestGraph(args.manifest_path ?? (await defaultManifestPath()), result.focus.uniqueId) : undefined return { title: `dbt Lineage: ${what} [${result.confidence}]`, metadata: { model_name: result.focus.name, + model_unique_id: result.focus.uniqueId, confidence: result.confidence, structuredContent: result, + ...(graph?.graph && { graph: graph.graph }), + ...(graph?.error && { graph_error: graph.error }), }, output: formatModelLineage(result), } @@ -69,16 +89,26 @@ export const DbtLineageTool = Tool.define("dbt_lineage", { manifest_path: manifestPath, model: args.model, dialect: args.dialect, + include_graph: args.include_graph, }) if (notUsed) result.confidence_factors.push(notUsed) if (args.column) { result.confidence_factors.push(`tracing ${args.column} needs the dbt integration; this is the whole model's lineage`) } + const lineage = result.raw_lineage as { column_dict?: Record; column_lineage?: unknown[] } return { title: `dbt Lineage: ${result.model_name} [${result.confidence}]`, + // The same lineage as the text, as data: callers (the MCP server's structured result, the + // Claude Code plugin's lineage pane) read it here instead of parsing the text. metadata: { model_name: result.model_name, + model_unique_id: result.model_unique_id ?? null, confidence: result.confidence, + confidence_factors: result.confidence_factors, + upstream: result.upstream ?? [], + columns: lineage?.column_dict ?? {}, + edges: Array.isArray(lineage?.column_lineage) ? lineage.column_lineage : [], + ...(result.graph && { graph: result.graph }), }, output: formatDbtLineage(result), } @@ -93,6 +123,80 @@ export const DbtLineageTool = Tool.define("dbt_lineage", { }, }) +/** + * Why a named manifest is not the adapter's project's, or undefined when it is: it must lie in that + * project's folder (links resolved) with no other dbt_project.yml between (a nested project or an + * installed package is another project). A manifest not written yet (an uncompiled project) is placed + * by its path, its nearest existing folder's links resolved, so the project's own default path keeps + * the adapter's answer; one with no dbt project above it at all (a default path resolved against + * another folder) names no other project either. A link to a manifest not written yet is placed where + * it points. + */ +async function manifestOutsideAdapterProject(manifestPath: string): Promise { + const { getDbtProjectRoot } = await import("../native/connections/register") + const root = await getDbtProjectRoot() + if (!root) return "the dbt adapter has no project root" + // The path with its links resolved, the parts not written yet appended; a link to a missing file is + // followed to where it points. `..` is collapsed first, as the manifest reader (loadRawManifest's + // realpathSync) does, so this places the file that reader opens. Undefined past 64 links (a loop); + // climbing to the nearest existing folder is not counted. + const real = (p: string) => { + const rest: string[] = [] + let at = path.resolve(p) + for (let links = 0; links < 64; ) { + try { + return path.join(fs.realpathSync(at), ...rest) + } catch {} + let link: string | undefined + try { + if (fs.lstatSync(at).isSymbolicLink()) link = fs.readlinkSync(at) + } catch {} + if (link !== undefined) { + links++ + // From the link's real folder, as the system resolves it: `..` climbs from there. + at = path.resolve(fs.realpathSync(path.dirname(at)), link) + continue + } + if (path.dirname(at) === at) return undefined + rest.unshift(path.basename(at)) + at = path.dirname(at) + } + return undefined + } + const projectDir = real(root) + const manifest = real(manifestPath) + if (!projectDir || !manifest) return `the manifest ${manifestPath} cannot be placed` + const inside = path.relative(projectDir, manifest) + // Outside the project: the relative path climbs out (`..` as a whole segment, so `..data/` is inside). + if (!inside || inside === ".." || inside.startsWith(`..${path.sep}`) || path.isAbsolute(inside)) { + if (!fs.existsSync(manifest) && !projectAbove(manifest)) return undefined + return `the manifest ${manifestPath} is not in the dbt project at ${root}` + } + // The manifest is inside, so walking up from it reaches the project folder itself. + for (let dir = path.dirname(manifest); dir !== projectDir && dir !== path.dirname(dir); dir = path.dirname(dir)) { + if (fs.existsSync(path.join(dir, "dbt_project.yml"))) return `the manifest ${manifestPath} belongs to the dbt project at ${dir}` + } + return undefined +} + +/** Whether a dbt_project.yml sits in any folder above a path. */ +function projectAbove(file: string): boolean { + for (let dir = path.dirname(file); ; dir = path.dirname(dir)) { + if (fs.existsSync(path.join(dir, "dbt_project.yml"))) return true + if (path.dirname(dir) === dir) return false + } +} + +/** The model's graph from the compiled manifest, or why there is none. */ +function manifestGraph(manifestPath: string, uniqueId: string): { graph?: ReturnType; error?: string } { + try { + const manifest = loadRawManifest(manifestPath) + return manifest ? { graph: lineageGraph(manifest, uniqueId) } : { error: `no manifest at ${manifestPath}` } + } catch (e) { + return { error: e instanceof Error ? e.message : String(e) } + } +} + /** * The lineage from the project's dbt adapter, which the tool starts in read-only mode too: it * compiles and reads column metadata, and runs no user SQL. `reason` says why it wasn't used. diff --git a/packages/opencode/src/altimate/tools/impact-analysis.ts b/packages/opencode/src/altimate/tools/impact-analysis.ts index 205811f81e..5a5e3de299 100644 --- a/packages/opencode/src/altimate/tools/impact-analysis.ts +++ b/packages/opencode/src/altimate/tools/impact-analysis.ts @@ -6,6 +6,7 @@ import z from "zod" import { Tool } from "../../tool/tool" import { Dispatcher } from "../native" import type { Telemetry } from "../telemetry" +import { columnReach, exposuresOn, type ColumnReach, type ColumnReachExposure } from "../native/dbt/column-reach" export const ImpactAnalysisTool = Tool.define("impact_analysis", { description: [ @@ -20,14 +21,14 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { '- impact_analysis({ manifest_path: "target/manifest.json", model: "dim_customers", change_type: "retype" })', ].join("\n"), parameters: z.object({ - model: z.string().describe("dbt model name to analyze impact for (e.g., 'stg_orders', 'dim_customers')"), + model: z.string().describe("dbt model name or unique_id to analyze impact for (e.g., 'stg_orders', 'model.shop.dim_customers')"), column: z .string() .optional() .describe("Specific column to trace impact for. If omitted, analyzes model-level impact."), change_type: z.enum(["remove", "rename", "retype", "add", "modify"]).describe("Type of change being considered"), manifest_path: z.string().optional().default("target/manifest.json").describe("Path to dbt manifest.json file"), - dialect: z.string().optional().default("snowflake").describe("SQL dialect for lineage analysis"), + dialect: z.string().optional().describe("SQL dialect for column lineage (auto-detected from the manifest if omitted)"), }), async execute(args, ctx) { try { @@ -43,9 +44,9 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { } // Step 2: Find the target model and its downstream dependents - const targetModel = manifest.models.find( - (m: { name: string }) => m.name === args.model || m.name.endsWith(`.${args.model}`), - ) + const targetModel = + manifest.models.find((m: { unique_id?: string }) => m.unique_id === args.model) ?? + manifest.models.find((m: { name: string }) => m.name === args.model || m.name.endsWith(`.${args.model}`)) if (!targetModel) { const available = manifest.models @@ -60,58 +61,56 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { } // Step 3: Build the dependency graph and find all downstream models - const modelsByName = new Map() - for (const m of manifest.models) { - modelsByName.set(m.name, m) - } - // Find all models that depend on the target (direct + transitive) - const downstream = findDownstream(args.model, manifest.models) + const downstream = findDownstream(targetModel.name, manifest.models, targetModel.unique_id) const direct = downstream.filter((d) => d.depth === 1) const transitive = downstream.filter((d) => d.depth > 1) // Step 4: Count only tests that reference the target model or its downstream models - const affectedModelIds = new Set([ - targetModel.unique_id, - ...downstream.map((d) => modelsByName.get(d.name)?.unique_id).filter(Boolean), - ]) + const affectedModelIds = new Set( + [targetModel.unique_id, ...downstream.map((d) => d.unique_id)].filter( + (id): id is string => Boolean(id), + ), + ) const affectedTests = (manifest.tests ?? []).filter((t) => t.depends_on?.some((dep) => affectedModelIds.has(dep)), ) const affectedTestCount = affectedTests.length - // Step 5: If column specified, attempt column-level lineage - let columnImpact: string[] = [] + // Step 5: Exposures (dashboards) on the model or anything downstream, and, for a column, the + // downstream models whose SQL reads it, traced model by model through the compiled manifest. + let exposures: ColumnReachExposure[] = [] + let reach: ColumnReach | undefined + let reachError: string | undefined + let exposuresError: string | undefined + try { + exposures = exposuresOn(args.manifest_path, affectedModelIds) + } catch (error) { + // A manifest the parser read but the raw loader cannot: say so rather than imply no dashboards. + exposuresError = error instanceof Error ? error.message : String(error) + } if (args.column) { try { - const lineageResult = await Dispatcher.call("lineage.check", { - sql: `SELECT * FROM ${args.model}`, // Use model reference for lineage tracing - dialect: args.dialect, - }) - if (lineageResult.data?.column_dict) { - // Find which downstream columns reference our target column - for (const [outCol, sources] of Object.entries(lineageResult.data.column_dict)) { - const srcArray = Array.isArray(sources) ? sources : [sources] - if (srcArray.some((s: any) => JSON.stringify(s).includes(args.column!))) { - columnImpact.push(outCol) - } - } - } - } catch { - // Column lineage is best-effort — continue without it + reach = columnReach(args.manifest_path, targetModel.unique_id ?? args.model, args.column, args.dialect) + } catch (error) { + // The model-level impact still stands; the reason is reported beside it. + reachError = error instanceof Error ? error.message : String(error) } } // Step 6: Format the impact report const output = formatImpactReport({ - model: args.model, + model: targetModel.name, column: args.column, changeType: args.change_type, direct, transitive, affectedTestCount, - columnImpact, totalModels: manifest.model_count, + exposures, + exposuresError, + columnReach: reach, + columnReachError: reachError, }) const totalAffected = downstream.length @@ -138,8 +137,16 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { direct_count: direct.length, transitive_count: transitive.length, test_count: affectedTestCount, - column_impact: columnImpact.length, + column_reader_count: reach ? reach.models.length : 0, has_schema: false, + model: targetModel.name, + column: args.column ?? null, + change_type: args.change_type, + downstream: downstream.map((d) => ({ name: d.name, ...(d.unique_id && { unique_id: d.unique_id }), depth: d.depth, materialized: d.materialized ?? null })), + exposures, + ...(reach && { column_reach: reach }), + ...(reachError && { column_reach_error: reachError }), + ...(exposuresError && { exposures_error: exposuresError }), ...(findings.length > 0 && { findings }), }, output, @@ -157,6 +164,8 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { export interface DownstreamModel { name: string + /** The node's unique_id when the manifest gives one (versions of a model share a name). */ + unique_id?: string depth: number materialized?: string path: string[] @@ -164,30 +173,41 @@ export interface DownstreamModel { export function findDownstream( targetName: string, - models: Array<{ name: string; depends_on: string[]; materialized?: string }>, + models: Array<{ name: string; unique_id?: string; depends_on: string[]; materialized?: string }>, + targetId?: string, ): DownstreamModel[] { const results: DownstreamModel[] = [] const visited = new Set() - function walk(name: string, depth: number, path: string[]) { - for (const model of models) { - if (visited.has(model.name)) continue - const deps = model.depends_on.map((d) => d.split(".").pop()) - if (deps.includes(name)) { - visited.add(model.name) - const newPath = [...path, model.name] + // Breadth-first so a model's depth is its shortest distance from the target. A parent is matched + // by unique_id when known (a versioned model's id ends in its version, not its name; another + // package may have a model of the same name), else by the last segment of each dependency id. + let level: { name: string; id: string | undefined; path: string[] }[] = [ + { name: targetName, id: targetId, path: [targetName] }, + ] + for (let depth = 1; level.length > 0; depth++) { + const next: typeof level = [] + for (const parent of level) { + for (const model of models) { + const key = model.unique_id ?? model.name + if (visited.has(key)) continue + // An exact id decides alone: another package's model of the same name is not this one. + const reads = parent.id ? model.depends_on.includes(parent.id) : model.depends_on.some((d) => d.split(".").pop() === parent.name) + if (!reads) continue + visited.add(key) + const path = [...parent.path, model.name] results.push({ name: model.name, + ...(model.unique_id && { unique_id: model.unique_id }), depth, materialized: model.materialized, - path: newPath, + path, }) - walk(model.name, depth + 1, newPath) + next.push({ name: model.name, id: model.unique_id, path }) } } + level = next } - - walk(targetName, 1, [targetName]) return results } @@ -198,8 +218,15 @@ export function formatImpactReport(data: { direct: DownstreamModel[] transitive: DownstreamModel[] affectedTestCount: number - columnImpact: string[] totalModels: number + /** Exposures (dashboards) on the model or anything downstream of it. */ + exposures?: ColumnReachExposure[] + /** Why exposures could not be read, when reading them failed. */ + exposuresError?: string + /** Downstream models that read the column, when one was given and traced. */ + columnReach?: ColumnReach + /** Why the column could not be traced, when tracing failed. */ + columnReachError?: string }): string { const lines: string[] = [] @@ -248,13 +275,49 @@ export function formatImpactReport(data: { lines.push("") } - // Column impact - if (data.column && data.columnImpact.length > 0) { - lines.push(`Affected Output Columns (${data.columnImpact.length})`) + // Column reach: the downstream models whose SQL reads the column + if (data.column && data.columnReach) { + const reach = data.columnReach + const dashboards = reach.exposures.length + // Stopped early: every count is a lower bound, since the models past the stop were never traced. + const atLeast = reach.truncated ? "at least " : "" + const plural = (n: number, word: string) => `${atLeast}${n} ${word}${n === 1 ? "" : "s"}` + const readers = `${plural(reach.models.length, "model")} and ${plural(dashboards, "exposure")}` + const maybe = reach.uncertain.length ? `; ${atLeast}${reach.uncertain.length} more may read it` : "" + const lost = reach.untraced.length ? `; ${atLeast}${reach.untraced.length} could not be traced` : "" + const stopped = reach.truncated ? " (stopped early)" : "" + lines.push(`Column Reach: ${data.model}.${data.column} is read by ${readers}${maybe}${lost}${stopped}`) lines.push("".padEnd(40, "-")) - for (const col of data.columnImpact) { - lines.push(` ${col}`) - } + for (const m of reach.models) lines.push(` ${m.name}${m.columns.length ? ` → ${m.columns.join(", ")}` : ""}`) + for (const e of reach.exposures) lines.push(` ◆ ${e.label} [${e.type}]`) + for (const m of reach.uncertain) lines.push(` ? ${m.name}: ${m.reason}`) + for (const e of reach.uncertain_exposures) lines.push(` ? ◆ ${e.label} [${e.type}] on a model that may read it`) + for (const m of reach.untraced) lines.push(` ! ${m.name} not traced: ${m.reason}`) + lines.push("") + } + + if (data.column && data.columnReachError) { + lines.push(`Column Reach: not traced (${data.columnReachError}); the model-level impact above stands`) + lines.push("") + } + + // Exposures on anything affected, less those the column reach block already listed + const listed = new Set( + [...(data.columnReach?.exposures ?? []), ...(data.columnReach?.uncertain_exposures ?? [])].map((e) => e.unique_id), + ) + const others = (data.exposures ?? []).filter((e) => !listed.has(e.unique_id)) + if (others.length > 0) { + // When column reach stopped early, some of these may sit on models it never reached. + const title = data.columnReach ? "Other exposures" : "Exposures" + const stoppedNote = ": column reach stopped early, so some of these may not have been checked" + const unchecked = data.columnReach?.truncated ? stoppedNote : "" + lines.push(`${title} (${others.length})${unchecked}`) + lines.push("".padEnd(40, "-")) + for (const e of others) lines.push(` ${e.label} [${e.type}]`) + lines.push("") + } + if (data.exposuresError) { + lines.push(`Exposures: not read (${data.exposuresError.replace(/\r?\n/g, " ")})`) lines.push("") } diff --git a/packages/opencode/test/altimate/column-reach.test.ts b/packages/opencode/test/altimate/column-reach.test.ts new file mode 100644 index 0000000000..9bc1fc6d13 --- /dev/null +++ b/packages/opencode/test/altimate/column-reach.test.ts @@ -0,0 +1,1002 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test" +import * as core from "@altimateai/altimate-core" +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { + columnReach, + exposuresOn, + parentsNamed, + joinsNaturally, + namesThroughIdentifier, + selectsStar, +} from "../../src/altimate/native/dbt/column-reach" +import { lineageGraph } from "../../src/altimate/native/dbt/graph" +import { structuredError, structuredResult } from "../../src/altimate/mcp-server/server" +import { dbtLineage } from "../../src/altimate/native/dbt/lineage" + +const dirs: string[] = [] +afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) +}) + +const column = (name: string) => ({ name, data_type: "double" }) +const model = (name: string, deps: string[], sql: string, columns: string[], extra: Record = {}) => ({ + resource_type: "model", + unique_id: `model.shop.${name}`, + name, + alias: name, + schema: "main", + database: "shop", + original_file_path: `models/${name}.sql`, + depends_on: { nodes: deps }, + compiled_code: sql, + columns: Object.fromEntries(columns.map((c) => [c, column(c)])), + ...extra, +}) + +// fct_orders.revenue is carried by fct_daily, renamed to ltv by mart_ltv (and read through it by +// rpt_ltv), filtered on by rpt_big_orders, and never read by rpt_counts. +function manifest() { + const nodes: Record = { + "model.shop.fct_orders": model("fct_orders", [], "select order_id, revenue, gross_revenue from raw.orders", [ + "order_id", + "revenue", + "gross_revenue", + ]), + "model.shop.fct_daily": model( + "fct_daily", + ["model.shop.fct_orders"], + "select sum(revenue) as revenue from fct_orders", + ["revenue"], + ), + "model.shop.mart_ltv": model( + "mart_ltv", + ["model.shop.fct_orders"], + "select order_id, revenue as ltv from fct_orders", + ["order_id", "ltv"], + ), + "model.shop.rpt_ltv": model("rpt_ltv", ["model.shop.mart_ltv"], "select max(ltv) as top_ltv from mart_ltv", [ + "top_ltv", + ]), + "model.shop.rpt_big_orders": model( + "rpt_big_orders", + ["model.shop.fct_orders"], + "select order_id from fct_orders where revenue > 100", + ["order_id"], + ), + "model.shop.rpt_counts": model( + "rpt_counts", + ["model.shop.fct_orders"], + "select count(order_id) as orders from fct_orders", + ["orders"], + ), + } + return { + metadata: { adapter_type: "duckdb" }, + nodes, + sources: {}, + exposures: { + "exposure.shop.exec_revenue": { + name: "exec_revenue", + label: "Exec Revenue", + type: "dashboard", + depends_on: { nodes: ["model.shop.fct_daily"] }, + }, + "exposure.shop.ops": { name: "ops", type: "application", depends_on: { nodes: ["model.shop.rpt_counts"] } }, + }, + } +} + +function manifestFile(value: Record = manifest()) { + const dir = mkdtempSync(join(tmpdir(), "column-reach-")) + dirs.push(dir) + const file = join(dir, "manifest.json") + writeFileSync(file, JSON.stringify(value)) + return file +} + +describe("columnReach", () => { + test("follows a column through renames to the models and dashboards that read it", () => { + const reach = columnReach(manifestFile(), "fct_orders", "REVENUE") + const byName = Object.fromEntries(reach.models.map((m) => [m.name, m.columns])) + expect(byName).toEqual({ fct_daily: ["revenue"], mart_ltv: ["ltv"], rpt_ltv: ["top_ltv"] }) + // The engine reports no edge for a filter: the model is a possible reader, not a certain one. + expect(reach.uncertain).toEqual([ + { + name: "rpt_big_orders", + unique_id: "model.shop.rpt_big_orders", + reason: "names it only outside its select list (WHERE, JOIN, GROUP BY or HAVING)", + }, + ]) + expect(reach.untraced).toEqual([]) + expect(reach.exposures).toEqual([ + { unique_id: "exposure.shop.exec_revenue", name: "exec_revenue", label: "Exec Revenue", type: "dashboard" }, + ]) + expect(reach.truncated).toBe(false) + }) + + test("an output alias or a comment naming the column is not a read", () => { + const value = manifest() + for (const sql of [ + "select count(order_id) as revenue /* revenue */ from fct_orders -- revenue", + "select count(order_id) as `revenue` from fct_orders", + ]) { + value.nodes["model.shop.rpt_counts"].compiled_code = sql + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toEqual(["rpt_big_orders"]) + } + }) + + test("a downstream model without compiled SQL is reported untraced, and what reads it only may read the column", () => { + const value = manifest() + value.nodes["model.shop.mart_ltv"].compiled_code = "" + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name).sort()).toEqual(["fct_daily"]) + expect(reach.untraced).toEqual([ + { name: "mart_ltv", unique_id: "model.shop.mart_ltv", reason: "No compiled SQL found — run `dbt compile` first" }, + ]) + expect(reach.uncertain.map((m) => [m.name, m.reason])).toContainEqual([ + "rpt_ltv", + "is downstream of a model that could not be traced or reads it ambiguously", + ]) + }) + + test("a model that only filters on the column passes nothing on to its children", () => { + const value = manifest() + value.nodes["model.shop.rpt_after_filter"] = model( + "rpt_after_filter", + ["model.shop.rpt_big_orders"], + "select count(order_id) as n from rpt_big_orders", + ["n"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toEqual(["rpt_big_orders"]) + }) + + test("a join key another parent also documents is a possible read, not a dropped one", () => { + const value = manifest() + value.nodes["model.shop.dim_orders"] = model("dim_orders", [], "select order_id, revenue from raw.dim", [ + "order_id", + "revenue", + ]) + value.nodes["model.shop.rpt_joined"] = model( + "rpt_joined", + ["model.shop.fct_orders", "model.shop.dim_orders"], + "select d.order_id from fct_orders f join dim_orders d on f.order_id = d.order_id where f.revenue > 0", + ["order_id"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.find((m) => m.name === "rpt_joined")?.reason).toBe( + "names it outside its select list, or names another parent's column of the same name", + ) + }) + + test("a source and a model with the same table name: a qualified source column is not the model's", () => { + const value: any = manifest() + value.sources = { + "source.shop.raw.fct_orders": { + resource_type: "source", + name: "fct_orders", + identifier: "fct_orders", + schema: "raw", + database: "shop", + columns: { revenue: column("revenue") }, + }, + } + value.nodes["model.shop.rpt_src"] = model( + "rpt_src", + ["model.shop.fct_orders", "source.shop.raw.fct_orders"], + "select s.revenue as src_revenue from raw.fct_orders s join main.fct_orders m on s.revenue = m.order_id", + ["src_revenue"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_src") + }) + + test("a table name two parents share is ambiguous; a schema-qualified one is not", () => { + const base = manifest() + const other = model("orders_copy", [], "select order_id, revenue from raw.copy", ["order_id", "revenue"], { + alias: "fct_orders", + schema: "archive", + }) + const sql = (from: string) => `select b.revenue as kept from ${from}` + const child = (from: string) => + model("rpt_join", ["model.shop.fct_orders", "model.shop.orders_copy"], sql(from), ["kept"]) + const bare = { + ...base, + nodes: { + ...base.nodes, + "model.shop.orders_copy": other, + "model.shop.rpt_join": child("fct_orders a join fct_orders b on a.order_id = b.order_id"), + }, + } + const ambiguous = columnReach(manifestFile(bare), "fct_orders", "revenue") + expect(ambiguous.models.map((m) => m.name)).not.toContain("rpt_join") + expect(ambiguous.uncertain.find((m) => m.name === "rpt_join")?.reason).toBe( + "reads a parent whose table name another parent shares", + ) + + // Only archive.fct_orders (not traced) is read: no reader. + const qualified = { + ...base, + nodes: { + ...base.nodes, + "model.shop.orders_copy": other, + "model.shop.rpt_join": child("main.fct_orders a join archive.fct_orders b on a.order_id = b.order_id"), + }, + } + const reach = columnReach(manifestFile(qualified), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_join") + }) + + test("a snapshot is traced and carries the column on", () => { + const value = manifest() + value.nodes["snapshot.shop.snap_orders"] = { + ...model("snap_orders", ["model.shop.fct_orders"], "select order_id, revenue from fct_orders", [ + "order_id", + "revenue", + ]), + resource_type: "snapshot", + unique_id: "snapshot.shop.snap_orders", + } + value.nodes["model.shop.rpt_snap"] = model( + "rpt_snap", + ["snapshot.shop.snap_orders"], + "select sum(revenue) as snap_revenue from snap_orders", + ["snap_revenue"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + const byName = Object.fromEntries(reach.models.map((m) => [m.name, m.columns])) + expect(byName.snap_orders).toEqual(["revenue"]) + expect(byName.rpt_snap).toEqual(["snap_revenue"]) + }) + + test("a column no downstream model reads reaches nothing", () => { + const reach = columnReach(manifestFile(), "fct_orders", "gross_revenue") + expect(reach.models).toEqual([]) + expect(reach.exposures).toEqual([]) + }) + + test("an unknown model or a missing manifest throws", () => { + expect(() => columnReach(manifestFile(), "nope", "revenue")).toThrow(/not found/) + expect(() => columnReach(join(tmpdir(), "no-such-manifest.json"), "fct_orders", "revenue")).toThrow( + /Manifest not found/, + ) + }) +}) + +describe("exposuresOn through metrics", () => { + test("an exposure on a metric reaches the model under its semantic model", () => { + const value: any = manifest() + value.semantic_models = { "semantic_model.shop.orders": { depends_on: { nodes: ["model.shop.rpt_counts"] } } } + value.metrics = { + "metric.shop.order_count": { depends_on: { nodes: ["semantic_model.shop.orders"] } }, + "metric.shop.double_count": { depends_on: { nodes: ["metric.shop.order_count"] } }, + } + value.exposures["exposure.shop.kpi"] = { + name: "kpi", + type: "dashboard", + depends_on: { nodes: ["metric.shop.double_count"] }, + } + expect( + exposuresOn(manifestFile(value), ["model.shop.rpt_counts"]) + .map((e) => e.unique_id) + .sort(), + ).toEqual(["exposure.shop.kpi", "exposure.shop.ops"]) + }) +}) + +describe("exposuresOn", () => { + test("lists exposures on any of the given nodes, by label when there is one", () => { + const file = manifestFile() + expect(exposuresOn(file, ["model.shop.rpt_counts"])).toEqual([ + { unique_id: "exposure.shop.ops", name: "ops", label: "ops", type: "application" }, + ]) + expect(exposuresOn(file, ["model.shop.fct_orders"])).toEqual([]) + }) +}) + +describe("parentsNamed", () => { + test("a qualified name matches the parent at that qualification only; a bare name matches every parent that answers to it", () => { + const nodes: Record = { + a: { alias: "orders", schema: "main", database: "shop" }, + b: { alias: "orders", schema: "archive", database: "shop" }, + } + const lookup = (id: string) => nodes[id] + expect(parentsNamed(["shop", "main", "orders"], ["a", "b"], lookup)).toEqual(["a"]) + expect(parentsNamed(["archive", "orders"], ["a", "b"], lookup)).toEqual(["b"]) + expect(parentsNamed(["orders"], ["a", "b"], lookup)).toEqual(["a", "b"]) + expect(parentsNamed(["raw", "orders"], ["a", "b"], lookup)).toEqual([]) + }) +}) + +describe("columnReach where the engine's schema falls short", () => { + test("select * from a parent whose schema.yml does not list the column is a possible read, and its children may read it", () => { + const value = manifest() + value.nodes["model.shop.fct_orders"].columns = { order_id: column("order_id") } + value.nodes["model.shop.rpt_star"] = model( + "rpt_star", + ["model.shop.fct_orders"], + 'select * from "shop"."main"."fct_orders"', + [], + ) + value.nodes["model.shop.rpt_after_star"] = model( + "rpt_after_star", + ["model.shop.rpt_star"], + "select sum(revenue) as total from rpt_star", + ["total"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.find((m) => m.name === "rpt_star")?.reason).toBe( + "selects * from a parent whose schema.yml does not list the column", + ) + expect(reach.uncertain.map((m) => m.name)).toContain("rpt_after_star") + }) + + test("a dashboard downstream of a wildcard reader in any dialect form is reported, not dropped", () => { + for (const sql of [ + "select distinct on (order_id) * from fct_orders", + "select top 10 * from fct_orders", + 'select "shop"."main"."fct_orders".* from "shop"."main"."fct_orders"', + ]) { + const value: any = manifest() + value.nodes["model.shop.fct_orders"].columns = { order_id: column("order_id") } + value.nodes["model.shop.rpt_star"] = model("rpt_star", ["model.shop.fct_orders"], sql, []) + value.exposures["exposure.shop.star_board"] = { + name: "star_board", + label: "Star Board", + type: "dashboard", + depends_on: { nodes: ["model.shop.rpt_star"] }, + } + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toContain("rpt_star") + expect(reach.uncertain_exposures.map((e) => e.label)).toContain("Star Board") + } + }) + + test("a column name with a dot, quoted, is matched whole", () => { + const value = manifest() + value.nodes["model.shop.fct_orders"].columns = { "net.amount": column("net.amount") } + value.nodes["model.shop.rpt_net"] = model( + "rpt_net", + ["model.shop.fct_orders"], + 'select "net.amount" as net from fct_orders', + ["net"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "net.amount") + expect(reach.models.find((m) => m.name === "rpt_net")?.columns).toEqual(["net"]) + }) +}) + +describe("selectsStar", () => { + test("a CR-terminated line comment does not hide the SQL after it", () => { + expect(selectsStar("select 1 -- note\rfrom t where x = 1 and y = (select *\rfrom u)")).toBe(true) + }) + + test("finds * and t.* in a select list, not multiplication or count(*)", () => { + for (const sql of [ + "select * from a", + "select*from a", + "select a.*, b.x from a", + "select x, t.* from t", + "select distinct * from a", + "select /* c */ * from a", + "select all * from a", + "select * exclude (order_id) from a", + "select * except (order_id) from a", + "select a . * from a", + "select distinct on (customer_id) * from a", + "select analytics.dbt_x.orders.* from analytics.dbt_x.orders", + '"db"."sch"."orders".* ', + "select top 10 * from a", + "select as struct * from a", + 'select "my col".* from t', + "select * replace (1 as x) from a", + ]) + expect(selectsStar(sql)).toBe(true) + for (const sql of [ + "select revenue * 100 as cents from a", + "select count(*) as n from a", + "select count_big(*) as n from a", + "select count_if(*) as n from a", + "select a, b*c as d from t", + "select '*' as s from a", + "select (a + b) * c as d from t", + "select a * (b + c) as d from t", + "select x from a -- select *", + "select revenue * -1 as x from t", + "select a*-1 as x from t", + "select a * + b as x from t", + "select a * - (b + c) as x from t", + "select $$ * $$ as s from a", + "select $q$ select * $q$ as s from a", + ]) + expect(selectsStar(sql)).toBe(false) + // A function over the whole row reads every column; only count(*) reads none. + for (const sql of [ + "select id, hash(*) as h from a", + "select object_construct(*) as o from a", + "select to_json(struct(*)) as j from a", + "select columns(*) from a", + ]) + expect(selectsStar(sql)).toBe(true) + }) +}) + +describe("columnReach through an ephemeral parent", () => { + test("a model reading an ephemeral parent is a possible read, never a silent zero", () => { + const value = manifest() + value.nodes["model.shop.eph"] = { + ...model("eph", ["model.shop.fct_orders"], "select order_id, revenue from fct_orders", ["order_id", "revenue"]), + config: { materialized: "ephemeral" }, + } + value.nodes["model.shop.rpt_eph"] = model( + "rpt_eph", + ["model.shop.eph"], + "with __dbt__cte__eph as (select order_id, revenue from fct_orders) select sum(revenue) as total from __dbt__cte__eph", + ["total"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).toContain("eph") + expect(reach.uncertain.find((m) => m.name === "rpt_eph")?.reason).toBe( + "reads an ephemeral parent, inlined as a CTE the lineage engine cannot trace through", + ) + }) +}) + +describe("columnReach with an ephemeral and a regular parent", () => { + test("a traced edge from the regular parent does not hide what flows through the ephemeral one", () => { + const value = manifest() + value.nodes["model.shop.eph"] = { + ...model("eph", ["model.shop.fct_orders"], "select order_id, revenue as renamed from fct_orders", [ + "order_id", + "renamed", + ]), + config: { materialized: "ephemeral" }, + } + value.nodes["model.shop.mix"] = model( + "mix", + ["model.shop.fct_daily", "model.shop.eph"], + "with __dbt__cte__eph as (select order_id, revenue as renamed from fct_orders) select d.revenue as direct, e.renamed as hidden from fct_daily d cross join __dbt__cte__eph e", + ["direct", "hidden"], + ) + value.nodes["model.shop.leaf"] = model("leaf", ["model.shop.mix"], "select max(hidden) as h from mix", ["h"]) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).toContain("mix") + expect(reach.uncertain.map((m) => m.name)).toContain("leaf") + }) +}) + +describe("columnReach with columns that differ only by case", () => { + test('"Revenue" is not revenue where the parent documents both', () => { + const value: any = manifest() + value.nodes["model.shop.fct_orders"].columns = { revenue: column("revenue"), Revenue: column("Revenue") } + value.nodes["model.shop.rpt_case"] = model( + "rpt_case", + ["model.shop.fct_orders"], + 'select "Revenue" as other from fct_orders', + ["other"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_case") + expect(reach.models.map((m) => m.name)).toContain("fct_daily") + }) +}) + +describe("columnReach case handling", () => { + test("a spelling matching one documented column ignoring case reaches its readers", () => { + const reach = columnReach(manifestFile(), "fct_orders", "REVENUE") + expect(reach.models.map((m) => m.name)).toContain("fct_daily") + }) + + test("a spelling matching two columns that differ only by case leaves both columns' readers uncertain, never certain", () => { + const value: any = manifest() + value.nodes["model.shop.fct_orders"].columns = { revenue: column("revenue"), Revenue: column("Revenue") } + value.nodes["model.shop.rpt_case"] = model( + "rpt_case", + ["model.shop.fct_orders"], + 'select "Revenue" as other from fct_orders', + ["other"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "REVENUE") + expect(reach.models.map((m) => m.name)).not.toContain("fct_daily") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_case") + expect(reach.uncertain.map((m) => m.name)).toEqual(expect.arrayContaining(["fct_daily", "rpt_case"])) + }) + + test("with the spelling unsettled, a dashboard on the model itself is uncertain too", () => { + const value: any = manifest() + value.nodes["model.shop.fct_orders"].columns = { revenue: column("revenue"), Revenue: column("Revenue") } + value.exposures["exposure.shop.orders_board"] = { + name: "orders_board", + type: "dashboard", + depends_on: { nodes: ["model.shop.fct_orders"] }, + } + const reach = columnReach(manifestFile(value), "fct_orders", "REVENUE") + expect(reach.exposures.map((e) => e.unique_id)).not.toContain("exposure.shop.orders_board") + expect(reach.uncertain_exposures.map((e) => e.unique_id)).toContain("exposure.shop.orders_board") + }) +}) + +describe("columnReach with an empty column_lineage", () => { + test("falls back to column_dict edges", () => { + const spy = spyOn(core, "columnLineage").mockImplementation((() => ({ + column_lineage: [], + column_dict: { revenue: ['"shop"."main"."fct_orders"."revenue"'] }, + })) as any) + try { + const reach = columnReach(manifestFile(), "fct_orders", "revenue") + expect(reach.models.find((m) => m.name === "fct_daily")?.columns).toEqual(["revenue"]) + } finally { + spy.mockRestore() + } + }) +}) + +describe("columnReach on a model the engine cannot fully trace", () => { + const answer = (byModel: (sql: string) => unknown) => + spyOn(core, "columnLineage").mockImplementation(((sql: string) => byModel(sql)) as any) + const direct = (sql: string) => ({ + column_lineage: [{ source: '"shop"."main"."fct_orders"."revenue"', target: "revenue", lineage_type: "direct" }], + column_dict: {}, + }) + + test("no edge at all: SQL that names the column (a PIVOT) is uncertain and so is its child; SQL that never names it reads nothing", () => { + const pivot = "select * from fct_orders pivot(sum(revenue) for status in ('paid' as paid_total))" + const chain: any = manifest() + chain.nodes["model.shop.fct_daily"].compiled_code = pivot + chain.nodes["model.shop.rpt_ltv"] = model( + "rpt_ltv", + ["model.shop.fct_daily"], + "select max(paid_total) as top from fct_daily", + ["top"], + ) + const spy = answer((sql) => (sql === pivot ? { column_lineage: [], column_dict: {} } : direct(sql))) + try { + const reach = columnReach(manifestFile(chain), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toEqual(expect.arrayContaining(["fct_daily", "rpt_ltv"])) + expect(reach.uncertain.find((m) => m.name === "fct_daily")!.reason).toContain("no column edges") + } finally { + spy.mockRestore() + } + const constant: any = manifest() + constant.nodes["model.shop.rpt_ltv"] = model("rpt_ltv", ["model.shop.mart_ltv"], "select 1 as one", ["one"]) + const again = answer((sql) => (sql === "select 1 as one" ? { column_lineage: [], column_dict: {} } : direct(sql))) + try { + const reach = columnReach(manifestFile(constant), "fct_orders", "revenue") + expect([...reach.untraced, ...reach.uncertain].map((m) => m.name)).not.toContain("rpt_ltv") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_ltv") + } finally { + again.mockRestore() + } + }) + + test("errors beside edges: a reader stays reached but unsure, a non-reader is uncertain", () => { + const value: any = manifest() + value.nodes["model.shop.fct_daily"].compiled_code = "select 3" + value.nodes["model.shop.rpt_counts"].compiled_code = "select 4" + // Its SQL never names the column: only its partly traced parent makes it a possible reader. + value.nodes["model.shop.rpt_ltv"] = model( + "rpt_ltv", + ["model.shop.fct_daily"], + "select max(total) as top from fct_daily", + ["top"], + ) + const spy = answer((sql) => + sql === "select 3" + ? { ...direct(sql), errors: ["partial"] } + : sql === "select 4" + ? { + column_lineage: [ + { source: '"shop"."main"."fct_orders"."order_id"', target: "orders", lineage_type: "direct" }, + ], + column_dict: {}, + errors: ["partial"], + } + : direct(sql), + ) + try { + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).toContain("fct_daily") + // fct_daily is reached but only partly traced: its child may read the column, so it is uncertain, never a confident non-reader. + expect(reach.uncertain.map((m) => m.name)).toContain("rpt_ltv") + expect(reach.models.map((m) => m.name)).not.toContain("rpt_ltv") + // Its own child is checked though fct_daily carried the column on. + expect(reach.uncertain.find((m) => m.name === "rpt_counts")?.reason).toBe( + "the lineage engine traced only part of its SQL", + ) + expect(reach.uncertain_exposures.map((e) => e.unique_id)).toContain("exposure.shop.ops") + expect(reach.exposures.map((e) => e.unique_id)).not.toContain("exposure.shop.ops") + } finally { + spy.mockRestore() + } + }) +}) + +describe("columnReach on SQL the engine cannot parse", () => { + test("is untraced, and a dashboard on it is reported apart", () => { + const value = manifest() + value.nodes["model.shop.fct_daily"].compiled_code = "select sum(revenue) as revenue frm fct_orders whre" + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.untraced.map((m) => m.name)).toEqual(["fct_daily"]) + expect(reach.untraced[0]!.reason).toContain("Parse error") + expect(reach.exposures).toEqual([]) + expect(reach.uncertain_exposures).toEqual([ + { unique_id: "exposure.shop.exec_revenue", name: "exec_revenue", label: "Exec Revenue", type: "dashboard" }, + ]) + }) +}) + +describe("lineageGraph", () => { + test("a twelve-hop chain comes back whole to its source; upstream past 500 nodes is cut and marked", () => { + const chain: Record = {} + for (let i = 1; i <= 12; i++) + chain[`model.shop.m${i}`] = model( + `m${i}`, + [i === 1 ? "source.shop.raw.orders" : `model.shop.m${i - 1}`], + "select 1", + [], + ) + chain["test.shop.m3_unique"] = { + resource_type: "test", + name: "m3_unique", + depends_on: { nodes: ["model.shop.m3"] }, + } + const graph = lineageGraph( + { + nodes: chain, + sources: { "source.shop.raw.orders": { resource_type: "source", name: "orders", source_name: "raw" } }, + }, + "model.shop.m12", + ) + // An upstream model's tests are left out: only the target's and downstream tests are counted. + expect(graph.nodes["test.shop.m3_unique"]).toBeUndefined() + expect(graph.nodes["source.shop.raw.orders"]).toBeDefined() + expect(Object.keys(graph.nodes)).toHaveLength(13) + expect(graph.upTruncated).toBe(false) + + const wide: Record = { + "model.shop.top": model( + "top", + Array.from({ length: 600 }, (_, i) => `model.shop.p${i}`), + "select 1", + [], + ), + } + for (let i = 0; i < 600; i++) wide[`model.shop.p${i}`] = model(`p${i}`, [], "select 1", []) + const cut = lineageGraph({ nodes: wide }, "model.shop.top") + expect(cut.upTruncated).toBe(true) + expect(Object.keys(cut.nodes)).toHaveLength(501) + }) + + test("downstream past 500 nodes is cut and marked; downstream tests are still counted", () => { + const wide: Record = { "model.shop.top": model("top", [], "select 1", []) } + for (let i = 0; i < 600; i++) wide[`model.shop.c${i}`] = model(`c${i}`, ["model.shop.top"], "select 1", []) + wide["test.shop.top_unique"] = { + resource_type: "test", + name: "top_unique", + depends_on: { nodes: ["model.shop.top"] }, + } + wide["test.shop.c0_unique"] = { resource_type: "test", name: "c0_unique", depends_on: { nodes: ["model.shop.c0"] } } + const cut = lineageGraph({ nodes: wide }, "model.shop.top") + expect(cut.truncated).toBe(true) + expect(cut.upTruncated).toBe(false) + const kept = Object.values(cut.nodes) + expect(kept.filter((n) => n.type !== "test")).toHaveLength(501) + expect( + kept + .filter((n) => n.type === "test") + .map((n) => n.id) + .sort(), + ).toEqual(["test.shop.c0_unique", "test.shop.top_unique"]) + }) + + test("everything up and downstream, tests as ids, from dbt's child_map when present", () => { + const value = { + ...manifest(), + nodes: { + ...manifest().nodes, + "test.shop.t1": { resource_type: "test", name: "t1", depends_on: { nodes: ["model.shop.fct_daily"] } }, + }, + } + const graph = lineageGraph(value, "model.shop.mart_ltv") + expect(graph.parents["model.shop.mart_ltv"]).toEqual(["model.shop.fct_orders"]) + expect(graph.children["model.shop.mart_ltv"]).toEqual(["model.shop.rpt_ltv"]) + expect(graph.nodes["model.shop.mart_ltv"]).toEqual({ + id: "model.shop.mart_ltv", + name: "mart_ltv", + label: null, + type: "model", + path: "models/mart_ltv.sql", + materialized: null, + relation: "shop.main.mart_ltv", + columns: ["order_id", "ltv"], + }) + expect(graph.truncated).toBe(false) + const fromFct = lineageGraph( + { + ...value, + child_map: { "model.shop.fct_orders": ["model.shop.fct_daily", "test.shop.t1"], "model.shop.fct_daily": [] }, + parent_map: { "model.shop.fct_orders": [], "test.shop.t1": ["model.shop.fct_orders"] }, + }, + "model.shop.fct_orders", + ) + expect(Object.keys(fromFct.nodes).sort()).toEqual(["model.shop.fct_daily", "model.shop.fct_orders", "test.shop.t1"]) + expect(fromFct.nodes["test.shop.t1"]!.type).toBe("test") + expect(fromFct.parents["test.shop.t1"]).toEqual(["model.shop.fct_orders"]) + expect(fromFct.children["test.shop.t1"]).toEqual([]) + }) + + test("a node whose manifest relation_name is null has no relation", () => { + const base = manifest() + const value = { + ...base, + nodes: { ...base.nodes, "model.shop.mart_ltv": { ...base.nodes["model.shop.mart_ltv"], relation_name: null } }, + } + expect(lineageGraph(value, "model.shop.mart_ltv").nodes["model.shop.mart_ltv"]!.relation).toBeNull() + }) +}) + +describe("structuredError", () => { + test("passes a short message and caps a huge one", () => { + expect(structuredError("boom")).toEqual({ report: "boom" }) + const out = structuredError("x".repeat(300_000)) + expect(out.truncated).toBe(true) + expect(Buffer.byteLength(out.report as string)).toBeLessThanOrEqual(60_000) + }) + test("a tool's own structured result sends the report's header, not its edge lists", () => { + const text = + "Model: x (model.a.x) · table · confidence high\n\nUpstream (0):\n none\n\nColumn edges (1):\n x.a ← y.a\n\nNotes: n" + expect(structuredResult("dbt_lineage", { structuredContent: { columnLineage: [1] } }, text)).toEqual({ + columnLineage: [1], + report: "Model: x (model.a.x) · table · confidence high\n\nUpstream (0):\n none", + }) + expect( + (structuredResult("dbt_lineage", { structuredContent: {} }, "Model: x\n\nTrace of x.a (1 edges):\n e") as any) + .report, + ).toBe("Model: x") + }) + test("exposures_error reaches the structured fields", () => { + expect(structuredResult("impact_analysis", { exposures_error: "bad json" }, "R")).toEqual({ + exposures_error: "bad json", + report: "R", + }) + }) +}) + +describe("structuredResult over the cap", () => { + test("a report too big on its own is shortened too", () => { + const out = structuredResult("dbt_lineage", { structuredContent: { focus: {} } }, "x".repeat(300_000)) + expect(out.truncated).toBe(true) + expect(Buffer.byteLength(JSON.stringify(out))).toBeLessThanOrEqual(60_000) + expect("x".repeat(300_000).startsWith(out.report as string)).toBe(true) + }) +}) + +describe("structuredError escaping", () => { + test("the cap holds for the serialized result, whatever JSON escaping adds", () => { + for (const text of ["\n".repeat(100_000), "\u0000".repeat(300_000), "é".repeat(80_000)]) { + const out = structuredError(text) + expect(Buffer.byteLength(JSON.stringify(out))).toBeLessThanOrEqual(60_000) + expect(out.truncated).toBe(true) + } + }) +}) + +describe("structuredResult", () => { + test("passes the lineage fields with the report, and only the report when they are too big", () => { + expect( + structuredResult("impact_analysis", { severity: "LOW", findings: [{ category: "x" }], downstream: [] }, "Impact"), + ).toEqual({ + severity: "LOW", + downstream: [], + report: "Impact", + }) + const huge = { edges: Array.from({ length: 5000 }, (_, i) => ({ source: `"t"."c${i}"`, target: `c${i}` })) } + expect(structuredResult("dbt_lineage", huge, "Model: x")).toEqual({ report: "Model: x", truncated: true }) + // dbt_lineage's text repeats the columns and edges: only its header goes with them. + expect( + structuredResult( + "dbt_lineage", + { columns: {}, edges: [], graph: { nodes: {} } }, + 'Model: x\nID: y\n\nColumn Mappings:\n a ← ["t.a"]', + ), + ).toEqual({ + columns: {}, + edges: [], + report: "Model: x\nID: y", + }) + // With no column_dict the text has no Column Mappings section: the edges are cut at their own header. + expect( + structuredResult("dbt_lineage", { columns: {}, edges: [] }, "Model: x\nID: y\n\nLineage Edges:\n {}"), + ).toEqual({ + columns: {}, + edges: [], + report: "Model: x\nID: y", + }) + }) +}) + +describe("dbtLineage", () => { + test("reports the model's upstream unique_ids, and the graph when asked", () => { + const file = manifestFile() + const result = dbtLineage({ manifest_path: file, model: "mart_ltv" }) + expect(result.upstream).toEqual(["model.shop.fct_orders"]) + expect(result.graph).toBeUndefined() + expect( + Object.keys(dbtLineage({ manifest_path: file, model: "mart_ltv", include_graph: true }).graph!.nodes), + ).toContain("model.shop.rpt_ltv") + // By project-relative path too. + expect(dbtLineage({ manifest_path: file, model: "models/mart_ltv.sql" }).model_unique_id).toBe( + "model.shop.mart_ltv", + ) + // `./` and Windows separators spell the same path. + expect(dbtLineage({ manifest_path: file, model: "./models/mart_ltv.sql" }).model_unique_id).toBe( + "model.shop.mart_ltv", + ) + expect(dbtLineage({ manifest_path: file, model: "models\\mart_ltv.sql" }).model_unique_id).toBe( + "model.shop.mart_ltv", + ) + // An empty path is no model's, not the first one's without a file path. + const pathless = JSON.parse(readFileSync(file, "utf8")) + delete pathless.nodes["model.shop.fct_orders"].original_file_path + expect(dbtLineage({ manifest_path: manifestFile(pathless), model: "./" }).confidence_factors).toContain( + "Model './' not found in manifest", + ) + const withSnapshot: any = manifest() + withSnapshot.nodes["snapshot.shop.snap"] = { + ...model("snap", ["model.shop.fct_orders"], "select order_id from fct_orders", ["order_id"]), + resource_type: "snapshot", + unique_id: "snapshot.shop.snap", + } + expect(dbtLineage({ manifest_path: manifestFile(withSnapshot), model: "snap" }).model_unique_id).toBe( + "snapshot.shop.snap", + ) + // A snapshot named like another node's unique_id keeps its own id. + withSnapshot.nodes["test.shop.t1"] = { resource_type: "test", name: "t1", depends_on: { nodes: [] } } + withSnapshot.nodes["snapshot.shop.odd"] = { + ...withSnapshot.nodes["snapshot.shop.snap"], + unique_id: "snapshot.shop.odd", + name: "test.shop.t1", + } + expect(dbtLineage({ manifest_path: manifestFile(withSnapshot), model: "test.shop.t1" }).model_unique_id).toBe( + "snapshot.shop.odd", + ) + }) +}) + +describe("columnReach with a NATURAL JOIN", () => { + test("a NATURAL JOIN is a possible read, never a confident non-reader", () => { + const value = manifest() + value.nodes["model.shop.rpt_counts"].compiled_code = "select a.order_id from fct_orders a natural join other b" + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name).sort()).toEqual(["rpt_big_orders", "rpt_counts"]) + expect(reach.uncertain_exposures.map((e) => e.name)).toEqual(["ops"]) + }) + + test("a model that reads the column through a NATURAL JOIN passes the doubt on to its children", () => { + const value = manifest() + value.nodes["model.shop.fct_daily"].compiled_code = + "select sum(revenue) as revenue from fct_orders natural left join other" + value.nodes["model.shop.rpt_after"] = model( + "rpt_after", + ["model.shop.fct_daily"], + "select count(*) as n from fct_daily", + ["n"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.map((m) => m.name)).toContain("fct_daily") + expect(reach.uncertain.map((m) => m.name)).toContain("rpt_after") + }) + + test("every NATURAL join form counts, whatever the quoting around it; aggregates under their own names do not", () => { + for (const sql of [ + "select sum(revenue) as revenue, cast(d as date) as d from fct_orders", + "select natural_key, naturally from fct_orders", + "select 'natural ' as a, 'natural' as b from fct_orders natural_join", + ]) + expect(joinsNaturally(sql)).toBe(false) + for (const sql of [ + "select a.id from fct_orders a natural left semi join other b", + "select * from a natural cross join b", + "select * from a natural full outer join b", + "select a.id from a natural inner directed join b", + "select a.id from a natural left outer directed join b", + "select array[(select max(a.id) from fct_orders a natural join other b)] as ids", + 'select "a\\"b" as c from a natural join b where t = "z"', + `select "it's" as c from a natural join b where y = 'q'`, + "select 'a\\'b' as c from a natural join b where t = 'z'", + 'select "C:\\" as p from a natural join b where t = "z"', + "select * from a natural /* inner */ join b", + "select a.id from a natural -- why\n join b", + "select a.id from a natural left -- x\nouter join b", + "select a.id from a natural # mysql\njoin b", + "select a.id from a natural // snowflake\njoin b", + "select a.id from a natural -- old mac\rjoin b", + "select a.id from a natural /* " + "x".repeat(5000) + " */ join b", + // Many `natural` words in a string before the real join: each is checked in a bounded window. + "select $$" + "natural /* ".repeat(4000) + "$$ as s from a natural join b", + "select '/*' as s from a natural /* c */ join b", + "-- see /* note\nselect * from a natural /* c */ join b", + // The words in a string or comment count too: that only adds doubt. + "select 'natural join' as s from fct_orders", + ]) + expect(joinsNaturally(sql)).toBe(true) + }) + + test("a NATURAL JOIN model that names the column outside its select list also passes the doubt on", () => { + const value = manifest() + value.nodes["model.shop.rpt_big_orders"].compiled_code = + "select a.order_id from fct_orders a natural join other b where a.revenue > 100" + value.nodes["model.shop.rpt_after"] = model( + "rpt_after", + ["model.shop.rpt_big_orders"], + "select count(*) as n from rpt_big_orders", + ["n"], + ) + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toContain("rpt_after") + }) +}) + +describe("columnReach through Snowflake IDENTIFIER()", () => { + test("a column named as IDENTIFIER('…'), plain or escaped, is a possible read, and its child and dashboard are not dropped", () => { + for (const call of ["identifier(/* source column */ 'revenue')", String.raw`identifier('\x72evenue')`]) { + const value = manifest() + value.nodes["model.shop.rpt_counts"].compiled_code = `select ${call} as copied from fct_orders` + value.nodes["model.shop.rpt_copy"] = model( + "rpt_copy", + ["model.shop.rpt_counts"], + "select copied from rpt_counts", + ["copied"], + ) + const exposures: Record = value.exposures + exposures["exposure.shop.copy_board"] = { + name: "copy_board", + type: "dashboard", + depends_on: { nodes: ["model.shop.rpt_copy"] }, + } + const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.uncertain.map((m) => m.name)).toEqual(expect.arrayContaining(["rpt_counts", "rpt_copy"])) + expect(reach.uncertain_exposures.map((e) => e.name)).toEqual(expect.arrayContaining(["ops", "copy_board"])) + } + }) + + test("IDENTIFIER() with the column, quoted or qualified, or a bind variable counts; another column does not", () => { + for (const sql of [ + "select identifier('revenue') from t", + `select IDENTIFIER('o."REVENUE"') from t`, + "select identifier(/* source column */ 'revenue') as copied from t", + `select identifier('o."net.revenue"') from t`, + "select identifier /* c */ ( -- why\n 'revenue') from t", + "select identifier($col) from t", + "select identifier(:col) from t", + ]) + expect(namesThroughIdentifier(sql, ["revenue"])).toBe(true) + // A tracked column whose quoted name holds a dot. + expect(namesThroughIdentifier(`select identifier('"net.amount"') from t`, ["net.amount"])).toBe(true) + // Anything but one plain string literal counts: concatenations, variables, functions. + for (const sql of [ + "select identifier('o.' || :col) from t", + "select identifier('rev' || 'enue') from t", + "select identifier(session.col_name) from t", + "select identifier(col_var) from t", + "select identifier(concat('rev', 'enue')) from t", + "select identifier('order_id'", + ]) + expect(namesThroughIdentifier(sql, ["revenue"])).toBe(true) + // `identifier` as a plain name is not the function. + expect(namesThroughIdentifier("select identifier, other from t", ["revenue"])).toBe(false) + // An escape, or any text that is not a plain name, is not decoded: it counts. + for (const sql of [ + String.raw`select identifier('\x72evenue') from t`, + String.raw`select identifier('it\'s.revenue') from t`, + `select identifier('"my col"') from t`, + ]) + expect(namesThroughIdentifier(sql, ["revenue"])).toBe(true) + // A quote in a column's name, doubled inside the quoted identifier. + expect(namesThroughIdentifier(`select identifier('"a""b"') from t`, ['a"b'])).toBe(true) + expect(namesThroughIdentifier("select identifier('order_id') from t", ["revenue"])).toBe(false) + }) +}) diff --git a/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts b/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts index 31bfcd5ec0..6fc28fbb6b 100644 --- a/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts +++ b/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts @@ -316,6 +316,24 @@ describe("dbtLineage: error handling", () => { expect(result.confidence_factors).toContain("No compiled SQL found — run `dbt compile` first") }) + test("reports the model's name, not the path used to find it, when there is no compiled SQL", () => { + const dir = makeTmpDir() + const manifest = { + nodes: { + "model.proj.uncompiled": { + resource_type: "model", + name: "uncompiled", + original_file_path: "models/uncompiled.sql", + depends_on: { nodes: [] }, + columns: {}, + }, + }, + sources: {}, + } + const result = dbtLineage({ manifest_path: writeManifest(dir, manifest), model: "models/uncompiled.sql" }) + expect(result.model_name).toBe("uncompiled") + }) + test("handles manifest with no nodes key at all", () => { const dir = makeTmpDir() const manifestPath = writeManifest(dir, { metadata: {} }) diff --git a/packages/opencode/test/altimate/dbt-lineage-structured.test.ts b/packages/opencode/test/altimate/dbt-lineage-structured.test.ts index 38975cb577..e73007fec8 100644 --- a/packages/opencode/test/altimate/dbt-lineage-structured.test.ts +++ b/packages/opencode/test/altimate/dbt-lineage-structured.test.ts @@ -4,7 +4,7 @@ // adapter is up, and the manifest's lineage otherwise. The dbt adapter is replaced // through `dbtAdapterInternals`, so these tests need no Python or dbt. import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, spyOn, test } from "bun:test" -import { mkdtemp, rm, writeFile } from "node:fs/promises" +import { mkdir, mkdtemp, rm, symlink, writeFile } from "node:fs/promises" import os from "node:os" import path from "node:path" import { Effect } from "effect" @@ -94,9 +94,9 @@ function fakeAdapter(initialized: boolean, lineage?: (params: { model: string; c } } -function useConfig(autoDiscovered: boolean, dbtIntegration = "corecommand") { +function useConfig(autoDiscovered: boolean, dbtIntegration = "corecommand", projectRoot = project) { dbtAdapterInternals.readConfig = async () => ({ - projectRoot: project, + projectRoot, pythonPath: "python3", dbtIntegration, queryLimit: 500, @@ -152,6 +152,107 @@ describe("dbt_lineage with the project's dbt adapter", () => { expect(at).toEqual([...at].sort((a, b) => a - b)) }) + test("a manifest named outside the adapter's project wins over the adapter", async () => { + const dir = await mkdtemp(path.join(os.tmpdir(), "dbt-lineage-project-")) + const mine = path.join(dir, "mine") + const other = path.join(dir, "other") + await Promise.all([mine, other].map((d) => mkdir(path.join(d, "target"), { recursive: true }))) + const empty = JSON.stringify({ metadata: { adapter_type: "duckdb" }, nodes: {}, sources: {} }) + await writeFile(path.join(mine, "dbt_project.yml"), "name: mine\n") + await writeFile(path.join(mine, "target", "manifest.json"), empty) + await writeFile(path.join(other, "target", "manifest.json"), empty) + // The adapter itself is configured with `mine`, so the tool reads the root the adapter cached. + resetDbtAdapter() + useConfig(false, "corecommand", mine) + try { + const tool = await initTool(DbtLineageTool) + const fallback = await tool.execute({ model: "customers", manifest_path: path.join(other, "target", "manifest.json") }, ctx) + expect(fallback.metadata.structuredContent).toBeUndefined() + expect(fallback.output).toContain(`is not in the dbt project at ${mine}`) + const kept = await tool.execute({ model: "customers", manifest_path: path.join(mine, "target", "manifest.json") }, ctx) + expect(kept.metadata.structuredContent).toEqual(TRACE) + // A dbt project nested in this one's folder is another project. + const nested = path.join(mine, "subproject") + await mkdir(path.join(nested, "target"), { recursive: true }) + await writeFile(path.join(nested, "dbt_project.yml"), "name: sub\n") + await writeFile(path.join(nested, "target", "manifest.json"), empty) + const inner = await tool.execute({ model: "customers", manifest_path: path.join(nested, "target", "manifest.json") }, ctx) + expect(inner.metadata.structuredContent).toBeUndefined() + expect(inner.output).toContain("belongs to the dbt project at") + // A folder whose name starts with `..` is still inside the project. + const dotted = path.join(mine, "..data") + await mkdir(path.join(dotted, "target"), { recursive: true }) + await writeFile(path.join(dotted, "target", "manifest.json"), empty) + const own = await tool.execute({ model: "customers", manifest_path: path.join(dotted, "target", "manifest.json") }, ctx) + expect(own.metadata.structuredContent).toEqual(TRACE) + // A sibling whose name only starts with this project's is outside it. + const sibling = `${mine}-2` + await mkdir(path.join(sibling, "target"), { recursive: true }) + await writeFile(path.join(sibling, "target", "manifest.json"), empty) + const next = await tool.execute({ model: "customers", manifest_path: path.join(sibling, "target", "manifest.json") }, ctx) + expect(next.metadata.structuredContent).toBeUndefined() + // An uncompiled project's own default manifest, not written yet, keeps the adapter's answer. + const uncompiled = path.join(mine, "fresh", "target", "manifest.json") + const unwritten = await tool.execute({ model: "customers", manifest_path: path.join(mine, "target", "nope", "manifest.json") }, ctx) + expect(unwritten.metadata.structuredContent).toEqual(TRACE) + expect((await tool.execute({ model: "customers", manifest_path: uncompiled }, ctx)).metadata.structuredContent).toEqual(TRACE) + // One not written yet in a folder no dbt project is above (a default path resolved against + // another folder) names no other project: the adapter answers. + const stray = await tool.execute({ model: "customers", manifest_path: path.join(other, "missing", "manifest.json") }, ctx) + expect(stray.metadata.structuredContent).toEqual(TRACE) + // One not written yet in another dbt project is that project's. + await writeFile(path.join(other, "dbt_project.yml"), "name: other\n") + const elsewhere = await tool.execute({ model: "customers", manifest_path: path.join(other, "missing", "manifest.json") }, ctx) + expect(elsewhere.metadata.structuredContent).toBeUndefined() + // A link to a manifest not written yet is placed where it points: in another project, that one's; + // in this one, the adapter answers. + await symlink(path.join(other, "target", "gone.json"), path.join(mine, "target", "linked.json")) + const linked = await tool.execute({ model: "customers", manifest_path: path.join(mine, "target", "linked.json") }, ctx) + expect(linked.metadata.structuredContent).toBeUndefined() + await symlink(path.join(mine, "target", "real.json"), path.join(mine, "target", "own.json")) + const own2 = await tool.execute({ model: "customers", manifest_path: path.join(mine, "target", "own.json") }, ctx) + expect(own2.metadata.structuredContent).toEqual(TRACE) + // A relative link climbs from its real folder: through a linked folder, `..` leaves this project. + const realOther = path.join(dir, "real") + await mkdir(path.join(realOther, "proj", "target"), { recursive: true }) + await mkdir(path.join(realOther, "otherproj"), { recursive: true }) + await writeFile(path.join(realOther, "otherproj", "dbt_project.yml"), "name: o\n") + await symlink(path.join(realOther, "proj", "target"), path.join(mine, "linked_target")) + await symlink("../../otherproj/target/manifest.json", path.join(realOther, "proj", "target", "manifest.json")) + const climbed = await tool.execute({ model: "customers", manifest_path: path.join(mine, "linked_target", "manifest.json") }, ctx) + expect(climbed.metadata.structuredContent).toBeUndefined() + // `..` after a link is collapsed first, as the manifest reader opens it: `/out/..` is read + // as this project's manifest, so the adapter answers, matching the graph include_graph reads. + await mkdir(path.join(other, "deep"), { recursive: true }) + await symlink(path.join(other, "deep"), path.join(mine, "out")) + // The other project's manifest is made unreadable: a graph read from it would fail to parse. + await writeFile(path.join(other, "target", "manifest.json"), "{ not json") + const viaDots = await tool.execute({ model: "customers", manifest_path: `${mine}/out/../target/manifest.json`, include_graph: true }, ctx) + expect(viaDots.metadata.structuredContent).toEqual(TRACE) + expect(String(viaDots.metadata.graph_error ?? "")).not.toMatch(/json|parse/i) + // Many folders not written yet below the project are climbed, not counted as links. + const deep = path.join(mine, ...Array.from({ length: 80 }, (_, i) => `d${i}`), "manifest.json") + expect((await tool.execute({ model: "customers", manifest_path: deep }, ctx)).metadata.structuredContent).toEqual(TRACE) + // A link loop cannot be placed. + await symlink(path.join(mine, "target", "loop-b"), path.join(mine, "target", "loop-a")) + await symlink(path.join(mine, "target", "loop-a"), path.join(mine, "target", "loop-b")) + expect((await tool.execute({ model: "customers", manifest_path: path.join(mine, "target", "loop-a") }, ctx)).metadata.structuredContent).toBeUndefined() + // A path through a file is not a folder: it is placed, not thrown on. + const through = await tool.execute({ model: "customers", manifest_path: path.join(mine, "dbt_project.yml", "manifest.json") }, ctx) + expect(through.metadata.structuredContent).toEqual(TRACE) + } finally { + await rm(dir, { recursive: true, force: true }) + } + }) + + test("include_graph with no manifest to read says so in graph_error", async () => { + const tool = await initTool(DbtLineageTool) + const result = await tool.execute({ model: "customers", include_graph: true }, ctx) + expect(result.metadata.structuredContent).toEqual(TRACE) + expect(result.metadata.graph).toBeUndefined() + expect(result.metadata.graph_error).toBe(`no manifest at ${path.join(project, "target", "manifest.json")}`) + }) + test("is handed to an MCP client as structuredContent", async () => { const server = await createServer({ allowWrite: true }) const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair() @@ -164,7 +265,8 @@ describe("dbt_lineage with the project's dbt adapter", () => { arguments: { model: "customers", column: "customer_lifetime_value" }, }) expect(answer.isError).toBe(false) - expect(answer.structuredContent).toEqual(TRACE) + // The MCP server adds the text report beside the trace (clients may show structured content in place of the text). + expect(answer.structuredContent).toEqual({ ...TRACE, report: expect.any(String) }) expect((answer.content as { text: string }[])[0].text).toContain("Trace of customers.customer_lifetime_value") } finally { await client.close() diff --git a/packages/opencode/test/altimate/impact-analysis.test.ts b/packages/opencode/test/altimate/impact-analysis.test.ts index 2ce470e672..e93083aa84 100644 --- a/packages/opencode/test/altimate/impact-analysis.test.ts +++ b/packages/opencode/test/altimate/impact-analysis.test.ts @@ -107,6 +107,18 @@ describe("impact_analysis: DAG traversal", () => { expect(result.output).toContain("BREAKING") }) + test("the structured downstream entries carry no path", async () => { + mockDispatcher({ "dbt.manifest": linearDAG }) + const tool = await initTool(ImpactAnalysisTool) + const result = await tool.execute( + { model: "stg_orders", change_type: "remove", manifest_path: "target/manifest.json", dialect: "snowflake" }, + ctx, + ) + const downstream = result.metadata.downstream as Record[] + expect(downstream.map((d) => d.name)).toEqual(["int_orders", "fct_orders", "rpt_daily"]) + expect(downstream.every((d) => !("path" in d))).toBe(true) + }) + test("SAFE severity when no downstream models exist", async () => { mockDispatcher({ "dbt.manifest": linearDAG }) const tool = await initTool(ImpactAnalysisTool) diff --git a/packages/opencode/test/altimate/mcp-server/server.test.ts b/packages/opencode/test/altimate/mcp-server/server.test.ts index f0060c7045..ed4dde9b83 100644 --- a/packages/opencode/test/altimate/mcp-server/server.test.ts +++ b/packages/opencode/test/altimate/mcp-server/server.test.ts @@ -1657,3 +1657,57 @@ mcpTest("startup skips a local catalog over the table gate with a short note", a expect(executed[0]!.sql).toContain("LIMIT 41") }) // altimate_change end + +// altimate_change start — lineage tools return structured content beside the text +mcpTest("dbt_lineage and impact_analysis declare an output schema and return their metadata as structured content", async () => { + const { client, call } = await fixture() + const tools = (await client.listTools()).tools + for (const name of ["dbt_lineage", "impact_analysis"]) expect(tools.find((tool) => tool.name === name)?.outputSchema).toBeDefined() + expect(tools.find((tool) => tool.name === "sql_execute")?.outputSchema).toBeUndefined() + + const dir = await mkdtemp(path.join(os.tmpdir(), "mcp-lineage-test-")) + cleanups.push(() => rm(dir, { recursive: true, force: true })) + const columns = (...names: string[]) => Object.fromEntries(names.map((name) => [name, { name, data_type: "double" }])) + const node = (name: string, deps: string[], sql: string, cols: string[]) => ({ + resource_type: "model", unique_id: `model.shop.${name}`, name, alias: name, original_file_path: `models/${name}.sql`, + depends_on: { nodes: deps }, compiled_code: sql, columns: columns(...cols), + }) + const manifestPath = path.join(dir, "manifest.json") + await Bun.write(manifestPath, JSON.stringify({ + metadata: { adapter_type: "duckdb" }, + sources: {}, + nodes: { + "model.shop.fct_orders": node("fct_orders", [], "select order_id, revenue from raw.orders", ["order_id", "revenue"]), + "model.shop.fct_daily": node("fct_daily", ["model.shop.fct_orders"], "select sum(revenue) as revenue from fct_orders", ["revenue"]), + }, + exposures: { "exposure.shop.exec": { name: "exec", label: "Exec Revenue", type: "dashboard", depends_on: { nodes: ["model.shop.fct_daily"] } } }, + })) + + const lineage = await call("dbt_lineage", { manifest_path: manifestPath, model: "fct_daily" }) + const structured = lineage.structuredContent as Record + // The structured report keeps the text's header; the mappings and edges travel as data. + const text = (lineage.content as { text: string }[])[0]!.text + expect(structured.report).toBe(text.split("\nColumn Mappings:")[0]!.trim()) + expect(text).toContain("Column Mappings:") + expect(structured.upstream).toEqual(["model.shop.fct_orders"]) + expect(structured.columns.revenue[0]).toContain("revenue") + expect(structured.edges[0].target).toBe("revenue") + + const impact = await call("impact_analysis", { manifest_path: manifestPath, model: "fct_orders", column: "revenue", change_type: "modify", dialect: "duckdb" }) + const reach = impact.structuredContent as Record + expect(reach.downstream.map((d: { name: string }) => d.name)).toEqual(["fct_daily"]) + expect(reach.column_reach.models.map((m: { name: string }) => m.name)).toEqual(["fct_daily"]) + expect(reach.exposures).toEqual([{ unique_id: "exposure.shop.exec", name: "exec", label: "Exec Revenue", type: "dashboard" }]) +}) +// altimate_change end + +// altimate_change start — a failing lineage call still answers with structured content its schema accepts +mcpTest("dbt_lineage and impact_analysis errors still return a structured report", async () => { + const { call } = await fixture() + for (const name of ["dbt_lineage", "impact_analysis"]) { + const result = await call(name, {}) + expect(result.isError).toBe(true) + expect(typeof (result.structuredContent as { report?: unknown } | undefined)?.report).toBe("string") + } +}) +// altimate_change end diff --git a/packages/opencode/test/altimate/tools/impact-analysis.test.ts b/packages/opencode/test/altimate/tools/impact-analysis.test.ts index 85c71a29df..709407295a 100644 --- a/packages/opencode/test/altimate/tools/impact-analysis.test.ts +++ b/packages/opencode/test/altimate/tools/impact-analysis.test.ts @@ -2,6 +2,40 @@ import { describe, test, expect } from "bun:test" import { findDownstream, formatImpactReport } from "../../../src/altimate/tools/impact-analysis" import type { DownstreamModel } from "../../../src/altimate/tools/impact-analysis" +describe("findDownstream: depth", () => { + test("a direct dependent also reachable through another dependent stays direct", () => { + const models = [ + { name: "y", depends_on: ["model.p.t"] }, + { name: "x", depends_on: ["model.p.y", "model.p.t"] }, + ] + const result = findDownstream("t", models) + expect(result.find((d) => d.name === "x")!.depth).toBe(1) + expect(result.find((d) => d.name === "x")!.path).toEqual(["t", "x"]) + }) +}) + +describe("formatImpactReport: exposures", () => { + const base = { model: "m", changeType: "remove", direct: [], transitive: [], affectedTestCount: 0, totalModels: 1 } + test("says when exposures could not be read", () => { + expect(formatImpactReport({ ...base, exposuresError: "bad json" })).toContain("Exposures: not read (bad json)") + }) + test("keeps a multi-line read error on one line", () => { + expect(formatImpactReport({ ...base, exposuresError: "a\nb" })).toContain("Exposures: not read (a b)") + }) + test("lists a dashboard once when column reach already printed it", () => { + const dash = { unique_id: "exposure.p.d", name: "d", label: "Dash", type: "dashboard" } + const out = formatImpactReport({ + ...base, + column: "c", + exposures: [dash, { unique_id: "exposure.q.d", name: "d", label: "Other", type: "dashboard" }], + columnReach: { models: [], uncertain: [], untraced: [], exposures: [dash], uncertain_exposures: [], truncated: false }, + }) + expect(out.match(/Dash \[dashboard\]/g)).toHaveLength(1) + expect(out).toContain("Other exposures (1)") + expect(out).toContain("Other [dashboard]") + }) +}) + describe("findDownstream: DAG traversal", () => { test("returns empty for leaf model with no dependents", () => { const models = [ @@ -114,7 +148,6 @@ describe("formatImpactReport", () => { direct: [], transitive: [], affectedTestCount: 0, - columnImpact: [], totalModels: 10, }) expect(report).toContain("REMOVE stg_temp") @@ -130,7 +163,6 @@ describe("formatImpactReport", () => { direct: [{ name: "fct_orders", depth: 1, path: ["stg_orders", "fct_orders"] }], transitive: [], affectedTestCount: 0, - columnImpact: [], totalModels: 20, }) expect(report).toContain("WARNING: This is a BREAKING change") @@ -146,7 +178,6 @@ describe("formatImpactReport", () => { direct: [{ name: "fct_orders", depth: 1, path: ["stg_orders", "fct_orders"] }], transitive: [], affectedTestCount: 5, - columnImpact: [], totalModels: 10, }) expect(report).toContain("WARNING: Rename requires updating all downstream references.") @@ -154,7 +185,7 @@ describe("formatImpactReport", () => { expect(report).toContain("Affected tests: 5") }) - test("column-level impact shows affected columns", () => { + test("column-level impact names the column and the change", () => { const report = formatImpactReport({ model: "stg_orders", column: "order_id", @@ -162,14 +193,45 @@ describe("formatImpactReport", () => { direct: [{ name: "fct_orders", depth: 1, path: ["stg_orders", "fct_orders"] }], transitive: [], affectedTestCount: 0, - columnImpact: ["total_amount", "order_count"], totalModels: 10, }) expect(report).toContain("RETYPE stg_orders.order_id") expect(report).toContain("CAUTION: Type change may cause implicit casts") - expect(report).toContain("Affected Output Columns (2)") - expect(report).toContain("total_amount") - expect(report).toContain("order_count") + }) + + test("column reach lists certain readers, possible readers, untraced models and exposures", () => { + const report = formatImpactReport({ + model: "fct_orders", + column: "revenue", + changeType: "modify", + direct: [{ name: "fct_daily", depth: 1, path: ["fct_orders", "fct_daily"] }], + transitive: [], + affectedTestCount: 0, + totalModels: 5, + exposures: [ + { unique_id: "exposure.p.exec", name: "exec", label: "Exec Revenue", type: "dashboard" }, + { unique_id: "exposure.p.far", name: "far", label: "Far Board", type: "dashboard" }, + ], + columnReach: { + models: [{ name: "fct_daily", unique_id: "model.p.fct_daily", columns: ["revenue"] }], + uncertain: [{ name: "rpt_big", unique_id: "model.p.rpt_big", reason: "names it only outside its select list (WHERE, JOIN, GROUP BY or HAVING)" }], + untraced: [{ name: "mart_ltv", unique_id: "model.p.mart_ltv", reason: "No compiled SQL found — run `dbt compile` first" }], + exposures: [{ unique_id: "exposure.p.exec", name: "exec", label: "Exec Revenue", type: "dashboard" }], + uncertain_exposures: [{ unique_id: "exposure.p.ops", name: "ops", label: "Ops", type: "dashboard" }], + truncated: true, + }, + }) + expect(report).toContain( + "Column Reach: fct_orders.revenue is read by at least 1 model and at least 1 exposure; at least 1 more may read it; at least 1 could not be traced (stopped early)", + ) + expect(report).toContain(" fct_daily → revenue") + expect(report).toContain(" ◆ Exec Revenue [dashboard]") + expect(report).toContain(" ? rpt_big: names it only outside its select list (WHERE, JOIN, GROUP BY or HAVING)") + expect(report).toContain(" ! mart_ltv not traced: No compiled SQL found — run `dbt compile` first") + expect(report).toContain(" ? ◆ Ops [dashboard] on a model that may read it") + expect(report).not.toContain("Exposures (") + // Stopped early: the other exposures are not claimed as non-readers. + expect(report).toContain("Other exposures (1): column reach stopped early, so some of these may not have been checked") }) test("percentage calculation with 0 total models does not produce NaN or Infinity", () => { @@ -179,7 +241,6 @@ describe("formatImpactReport", () => { direct: [], transitive: [], affectedTestCount: 0, - columnImpact: [], totalModels: 0, }) expect(report).not.toContain("NaN") @@ -194,7 +255,6 @@ describe("formatImpactReport", () => { direct: [{ name: "fct_orders", depth: 1, path: ["stg_orders", "fct_orders"] }], transitive: [{ name: "report", depth: 2, materialized: "table", path: ["stg_orders", "fct_orders", "report"] }], affectedTestCount: 0, - columnImpact: [], totalModels: 50, }) expect(report).toContain("Direct Dependents (1)") @@ -203,3 +263,24 @@ describe("formatImpactReport", () => { expect(report).toContain("Blast radius: 2/50 models (4.0%)") }) }) + +describe("findDownstream with versioned models", () => { + test("a versioned parent is matched by unique_id, whose last segment is the version", () => { + const models = [ + { name: "orders", unique_id: "model.p.orders.v2", depends_on: [] }, + { name: "rpt", unique_id: "model.p.rpt", depends_on: ["model.p.orders.v2"] }, + ] + expect(findDownstream("orders", models, "model.p.orders.v2").map((d) => [d.name, d.unique_id])).toEqual([["rpt", "model.p.rpt"]]) + expect(findDownstream("orders", models)).toEqual([]) + }) + + test("with an exact id, another package's model of the same name is not matched", () => { + const models = [ + { name: "orders", unique_id: "model.a.orders", depends_on: [] }, + { name: "orders", unique_id: "model.b.orders", depends_on: [] }, + { name: "b_only", unique_id: "model.b.b_only", depends_on: ["model.b.orders"] }, + ] + expect(findDownstream("orders", models, "model.a.orders")).toEqual([]) + expect(findDownstream("orders", models, "model.b.orders").map((d) => d.name)).toEqual(["b_only"]) + }) +})