diff --git a/packages/opencode/src/altimate/mcp-server/server.ts b/packages/opencode/src/altimate/mcp-server/server.ts index 8ed203ac74..c89ebee902 100644 --- a/packages/opencode/src/altimate/mcp-server/server.ts +++ b/packages/opencode/src/altimate/mcp-server/server.ts @@ -61,8 +61,11 @@ const STRUCTURED_FIELDS: Record = { "test_count", "downstream", "exposures", + "exposures_declared", "column_reach", "column_reach_error", + "column_answer", + "column_answer_error", "exposures_error", "error", ], @@ -96,7 +99,15 @@ export function structuredResult(tool: string, metadata: Record // 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) + if (Buffer.byteLength(JSON.stringify(full)) <= STRUCTURED_MAX_BYTES) return full + // The column answer is the largest field: past the cap it goes first, so the model-level fields + // and the reach stay. + if ("column_answer" in full) { + const { column_answer: _, ...rest } = full + const lighter = { ...rest, column_answer_error: "Left out of the structured result for size; the CLI's --json output carries it." } + if (Buffer.byteLength(JSON.stringify(lighter)) <= STRUCTURED_MAX_BYTES) return lighter + } + return 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 { diff --git a/packages/opencode/src/altimate/native/dbt/column-answer.ts b/packages/opencode/src/altimate/native/dbt/column-answer.ts new file mode 100644 index 0000000000..e6761689f0 --- /dev/null +++ b/packages/opencode/src/altimate/native/dbt/column-answer.ts @@ -0,0 +1,366 @@ +// Column impact answer: where a column's value goes downstream (or comes from upstream), hop by +// hop, from the project's column resolver (project-lineage.ts), with what the engine could not +// settle kept visible as tiers rather than dropped. +import * as fs from "fs" +import * as path from "path" +import { findModel } from "./helpers" +import { describeExposure, exposureReads, type ColumnReachExposure } from "./column-reach" +import { compareText, missingColumn, projectLineage, type How, type HowFrom, type ProjectLineage, type ShapeFlag } from "./project-lineage" + +export type { How, HowFrom, ShapeFlag } from "./project-lineage" + +/** Must match the pinned `@altimateai/altimate-core` version (a test checks it). */ +export const ENGINE = "altimate-core 0.9.0" +export const MAX_ANSWER_COLUMNS = 2000 + +export const WORKAROUNDS = [ + "relations keyed database.schema.name", + "ephemeral parents expanded", + "parent columns = documented + own lineage targets + SELECT list", + "aggregate FILTER clause dropped before analysis", + "WITH nested inside a CTE reported unresolved", + "reads with zero traced columns reported unresolved", +] + +export interface ColumnAnswerNode { + id: string + name: string + folder: string + materialized: string | null + contract: boolean + tests: number + exposures: string[] +} + +export interface ColumnAnswerColumn { + /** Index into `nodes`. */ + model: number + column: string + hop: number + /** Downstream: no reader of its own. Upstream: no source of its own. */ + end: boolean + /** How it was reached; null on the focus column. */ + how: How | null + how_from: HowFrom + type?: string +} + +export type ConsumerTier = "carries" | "may_carry" | "cleared" + +/** + * An exposure on the focus model or anything downstream of it. An exposure names the models it reads, + * not their columns, so "carries" means it reads a model the value reaches, not that it uses the value. + */ +export interface ColumnAnswerConsumer extends ColumnReachExposure { + tier: ConsumerTier + /** The models it reads in that tier, by name. */ + via: string[] +} + +/** [from column, to column, how], plus "name" when `how` was inferred from matching names. */ +export type ColumnAnswerEdge = [number, number, How] | [number, number, How, "name"] + +export interface ColumnAnswer { + version: 1 + focus: { model: string; column: string } + direction: "downstream" | "upstream" + inputs: { + manifest: { generated_at: string | null; older_than_sql: boolean } + sql: "compiled_code" + catalog: null + engine: string + workarounds: string[] + } + counts: { + columns: number + models: number + hops: number + contracts: number + ends: number + may_carry_models: number + cleared_models: number + model_level: number + } + nodes: ColumnAnswerNode[] + columns: ColumnAnswerColumn[] + edges: ColumnAnswerEdge[] + also_reads: { model: number; columns: string[]; downstream: number }[] + unresolved: { model: number; reason: string; downstream?: number }[] + flags: { model: number; flag: ShapeFlag }[] + /** Model-level downstream (upstream: ancestors) in none of the tiers, by name. Not in `nodes`. */ + cleared: { id: string; name: string }[] + /** Downstream only: upstream, `exposures` is empty. `declared` counts every exposure in the project. */ + consumers: { declared: number; exposures: ColumnAnswerConsumer[] } + truncated: boolean + next: null +} + + +// --------------------------------------------------------------------------------------------- +// The answer + +export function columnAnswer( + manifestPath: string, + model: string, + column: string, + direction: "downstream" | "upstream" = "downstream", + dialect?: string, +): ColumnAnswer { + const p = projectLineage(manifestPath, dialect) + const focusNode = + findModel(p.nodes, model) ?? + (p.traced(model) ? p.nodes[model] : null) ?? + Object.values(p.nodes).find((n) => n?.resource_type === "snapshot" && n.name === model) + if (!focusNode) throw new Error(`Model '${model}' not found in manifest`) + const focus: string = focusNode.unique_id ?? Object.keys(p.nodes).find((id) => p.nodes[id] === focusNode)! + const missing = missingColumn(p, focus, column) + if (missing) throw new Error(missing) + const want = column.toLowerCase() + + const tests: Record = {} + for (const n of Object.values(p.nodes)) + if (n?.resource_type === "test") for (const d of n.depends_on?.nodes ?? []) tests[d] = (tests[d] ?? 0) + 1 + const exposures: Record = {} + for (const x of Object.values(p.manifest.exposures ?? {})) + for (const d of x?.depends_on?.nodes ?? []) (exposures[d] ??= []).push(String(x.label || x.name)) + + const nodes: ColumnAnswerNode[] = [] + const nodeIndex = new Map() + const nodeIdx = (id: string) => { + let i = nodeIndex.get(id) + if (i !== undefined) return i + const n = p.node(id) ?? {} + i = nodes.length + nodeIndex.set(id, i) + const dir = String(n.original_file_path ?? "") + .replace(/\\/g, "/") + .split("/") + .slice(1, -1) + .join("/") + nodes.push({ + id, + name: String(n.name ?? id), + folder: dir || String(n.resource_type ?? ""), + materialized: n.config?.materialized ?? null, + contract: !!n.config?.contract?.enforced, + tests: tests[id] ?? 0, + exposures: exposures[id] ?? [], + }) + return i + } + + const columns: ColumnAnswerColumn[] = [] + const columnIndex = new Map() + const edges: ColumnAnswerEdge[] = [] + let truncated = false + const typeOf = (id: string, col: string) => { + const docs = p.node(id)?.columns ?? {} + const hit: any = docs[col] ?? Object.values(docs).find((c) => String(c?.name ?? "").toLowerCase() === col) + return hit?.data_type || undefined + } + /** The column's index, adding it when new (null when the answer is full). */ + const colIdx = (id: string, col: string, hop: number, how: How | null, howFrom: HowFrom): number | null => { + const k = `${id}|${col}` + const had = columnIndex.get(k) + if (had !== undefined) return had + if (columns.length >= MAX_ANSWER_COLUMNS) { + truncated = true + return null + } + const i = columns.length + columnIndex.set(k, i) + const type = typeOf(id, col) + columns.push({ model: nodeIdx(id), column: col, hop, end: false, how, how_from: howFrom, ...(type && { type }) }) + return i + } + const edge = (from: number, to: number, how: How, howFrom: HowFrom): ColumnAnswerEdge => + howFrom === "name" ? [from, to, how, "name"] : [from, to, how] + + const start = colIdx(focus, want, 0, null, null)! + const queue: [string, string, number][] = [[focus, want, 0]] + const also = new Map>() + const unresolved = new Map() + // Models with a column the cap left out: what sits past them was never checked. + const capped = new Set() + + if (direction === "downstream") { + while (queue.length) { + const [u, c, h] = queue.shift()! + const from = columnIndex.get(`${u}|${c}`)! + for (const reader of p.readers(u)) { + const lin = p.lineage(reader) + if (lin.unresolved) { + unresolved.set(reader, lin.unresolved) + continue + } + // Read in part: what it traced stands, and what sits past it may carry the value. + if (lin.partial) unresolved.set(reader, `read in part: ${lin.partial}`) + let hit = false + for (const e of lin.edges) { + if (e.from !== u || e.fromCol !== c) continue + hit = true + const fresh = !columnIndex.has(`${reader}|${e.to}`) + const to = colIdx(reader, e.to, h + 1, e.how, e.howFrom) + if (to === null) { + capped.add(reader) + continue + } + edges.push(edge(from, to, e.how, e.howFrom)) + if (fresh) queue.push([reader, e.to, h + 1]) + } + if (hit) continue + // An ephemeral focus is inlined as a CTE: its readers' edges name its parents' columns, not its own. + if (u === focus && p.ephemeral(focus)) { + unresolved.set(reader, `reads ${String(p.node(focus)?.name ?? focus)}, an ephemeral model inlined as a CTE, whose columns the trace can't follow`) + continue + } + const reads = p.reads(reader) + if (reads === null) unresolved.set(reader, "the engine couldn't list the columns its SQL reads") + else if (reads.has(c)) { + const set = also.get(reader) ?? new Set() + set.add(c) + also.set(reader, set) + } + } + } + } else { + while (queue.length) { + const [u, c, h] = queue.shift()! + if (!p.traced(u)) continue + const to = columnIndex.get(`${u}|${c}`)! + const lin = p.lineage(u) + if (lin.unresolved) { + unresolved.set(u, lin.unresolved) + continue + } + if (lin.partial) unresolved.set(u, `read in part: ${lin.partial}`) + for (const e of lin.edges) { + if (e.to !== c) continue + const fresh = !columnIndex.has(`${e.from}|${e.fromCol}`) + const from = colIdx(e.from, e.fromCol, h + 1, e.how, e.howFrom) + if (from === null) { + capped.add(e.from) + continue + } + edges.push(edge(from, to, e.how, e.howFrom)) + if (fresh) queue.push([e.from, e.fromCol, h + 1]) + } + } + } + + // Ends: downstream, a reached column nothing reads; upstream, one nothing feeds. + const linked = new Set(edges.map((e) => (direction === "downstream" ? e[0] : e[1]))) + for (const [i, c] of columns.entries()) c.end = i !== start && !linked.has(i) + + const valueModels = new Set(columns.filter((_, i) => i !== start).map((c) => nodes[c.model]!.id)) + valueModels.delete(focus) + // A value model that also reads the column in a filter, join or grouping stays listed: its other + // outputs (and the models past it) may change with the column too. + + const beyond = (id: string) => (direction === "downstream" ? p.descendants(id) : p.ancestors(id)) + const modelLevel = beyond(focus) + const mayCarry = new Set() + for (const id of [...also.keys(), ...unresolved.keys(), ...capped]) { + mayCarry.add(id) + for (const d of beyond(id)) mayCarry.add(d) + } + for (const id of [...valueModels, focus]) mayCarry.delete(id) + const cleared = [...modelLevel] + .filter((id) => !valueModels.has(id) && !mayCarry.has(id)) + .map((id) => ({ id, name: String(p.node(id)?.name ?? id) })) + .sort((a, b) => compareText(a.name, b.name) || compareText(a.id, b.id)) + + const clearedIds = new Set(cleared.map((one) => one.id)) + const consumers: ColumnAnswerConsumer[] = [] + if (direction === "downstream") { + const carries = (id: string) => id === focus || valueModels.has(id) + for (const [id, node] of Object.entries(p.manifest.exposures ?? {})) { + const reads = [...exposureReads(p.manifest, node)] + const tiers: [ConsumerTier, (one: string) => boolean][] = [ + ["carries", carries], + ["may_carry", (one) => mayCarry.has(one)], + ["cleared", (one) => clearedIds.has(one)], + ] + const tier = tiers.find(([, has]) => reads.some(has)) + if (!tier) continue + const via = reads.filter(tier[1]).map((one) => String(p.node(one)?.name ?? one)) + consumers.push({ ...describeExposure(id, node), tier: tier[0], via: via.sort(compareText) }) + } + const rank: ConsumerTier[] = ["carries", "may_carry", "cleared"] + consumers.sort((a, b) => rank.indexOf(a.tier) - rank.indexOf(b.tier) || compareText(a.label, b.label) || compareText(a.unique_id, b.unique_id)) + } + + const alsoReads = [...also].map(([id, cols]) => ({ + model: nodeIdx(id), + columns: [...cols], + downstream: p.descendants(id).size, + })) + const unresolvedOut = [...unresolved].map(([id, reason]) => { + const n = beyond(id).size + return { model: nodeIdx(id), reason, ...(n && { downstream: n }) } + }) + const flags: ColumnAnswer["flags"] = [] + for (const n of nodes) + if (p.traced(n.id)) for (const flag of p.lineage(n.id).flags) flags.push({ model: nodeIndex.get(n.id)!, flag }) + + return { + version: 1, + focus: { model: focus, column: want }, + direction, + inputs: { + manifest: { + generated_at: p.manifest.metadata?.generated_at ?? null, + older_than_sql: olderThanSql(manifestPath, p, nodes), + }, + sql: "compiled_code", + catalog: null, + engine: ENGINE, + workarounds: WORKAROUNDS, + }, + counts: { + columns: columns.length - 1, + models: valueModels.size, + hops: Math.max(0, ...columns.map((c) => c.hop)), + contracts: [...valueModels].filter((id) => !!p.node(id)?.config?.contract?.enforced).length, + ends: columns.filter((c) => c.end).length, + may_carry_models: mayCarry.size, + cleared_models: cleared.length, + model_level: modelLevel.size, + }, + nodes, + columns, + edges, + also_reads: alsoReads, + unresolved: unresolvedOut, + flags, + cleared, + consumers: { declared: Object.keys(p.manifest.exposures ?? {}).length, exposures: consumers }, + truncated, + next: null, + } +} + +/** Whether a model file in the answer changed after the manifest was written. */ +function olderThanSql(manifestPath: string, p: ProjectLineage, nodes: ColumnAnswerNode[]): boolean { + if (!p.projectDir) return false + try { + const written = fs.statSync(manifestPath).mtimeMs + for (const n of nodes) { + const file = p.nodes[n.id]?.original_file_path + if (!file || !p.traced(n.id)) continue + try { + if (fs.statSync(path.join(p.projectDir, file)).mtimeMs > written) return true + } catch {} + } + } catch {} + return false +} + +/** One line for the text report. */ +export function describeAnswer(a: ColumnAnswer): string { + const c = a.counts + const what = a.direction === "downstream" ? "Column impact" : "Column trace (upstream)" + const reach = `${c.columns} column${c.columns === 1 ? "" : "s"} in ${c.models} model${c.models === 1 ? "" : "s"}, ${c.hops} hop${c.hops === 1 ? "" : "s"}, ${c.ends} end${c.ends === 1 ? "" : "s"}` + const tiers = `${c.may_carry_models} more may carry it, ${c.cleared_models} cleared of ${c.model_level} at model level` + return `${what}: ${reach}; ${tiers}${a.truncated ? " (truncated)" : ""} (metadata.column_answer)` +} diff --git a/packages/opencode/src/altimate/native/dbt/column-reach.ts b/packages/opencode/src/altimate/native/dbt/column-reach.ts index 0f1b4cbb40..78a6b109af 100644 --- a/packages/opencode/src/altimate/native/dbt/column-reach.ts +++ b/packages/opencode/src/altimate/native/dbt/column-reach.ts @@ -8,6 +8,7 @@ // 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" +import { missingColumn, projectLineage, selectsStar, stripSql } from "./project-lineage" export interface ColumnReachModel { name: string @@ -30,6 +31,9 @@ export interface ColumnReachExposure { /** The exposure's label when it has one, else its name. */ label: string type: string + url?: string + /** The owner's name, as the project declares it. */ + owner?: string } export interface ColumnReach { @@ -113,44 +117,6 @@ function documentedColumns(node: any): Set { 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 @@ -256,6 +222,11 @@ export function columnReach(manifestPath: string, model: string, column: string, 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) + // The same parent columns dbtLineage reads each model against. + const project = projectLineage(manifestPath, dialect) + const missing = missingColumn(project, targetId, column) + if (missing) throw new Error(missing) + const known = (id: string) => new Set(project.columns(id).map((col) => col.name)) const children = childMapOf(manifest) // Everything downstream, then a topological order over it, so each node is traced after all its @@ -327,8 +298,10 @@ export function columnReach(manifestPath: string, model: string, column: string, 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") { + // dbtLineage rates a model low when the engine threw or answered SQL it cannot parse (`errors`, no + // edges), or when a `*` reads a parent with no known column. The edges it traced stand in the last + // case: a traced parent's columns are known, and a `*` over the target with none is caught below. + if (result.confidence === "low" && !result.unexpanded_stars?.length) { untraced.push(entry(id, result.confidence_factors[0] ?? "the lineage engine could not trace it")) unsure.add(id) continue @@ -356,7 +329,7 @@ export function columnReach(manifestPath: string, model: string, column: string, 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))) + parents.some((p) => [...tracked.get(p)!].some((col) => !known(p).has(col))) if (collides || wildcard) { unsure.add(id) uncertain.push( @@ -364,7 +337,7 @@ export function columnReach(manifestPath: string, model: string, column: string, id, collides ? "reads a parent whose table name another parent shares" - : "selects * from a parent whose schema.yml does not list the column", + : "selects * from a parent whose known columns don't include the column", ), ) continue @@ -490,10 +463,38 @@ export function exposuresOn(manifestPath: string, ids: Iterable): Column } 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", - })) + .map(([id, node]) => describeExposure(id, node)) +} + +/** An exposure as a manifest declares it. */ +export function describeExposure(id: string, node: any): ColumnReachExposure { + return { + unique_id: id, + name: node?.name ?? id, + label: node?.label || node?.name || id, + type: node?.type ?? "exposure", + ...(node?.url && { url: String(node.url) }), + ...(node?.owner?.name && { owner: String(node.owner.name) }), + } +} + +/** How many exposures the project declares, whatever they read. */ +export function exposuresDeclared(manifestPath: string): number { + return Object.keys(loadRawManifest(manifestPath)?.exposures ?? {}).length +} + +/** What an exposure reads, by unique_id: through metrics and semantic models, down to the models and sources. */ +export function exposureReads(manifest: any, exposure: any): Set { + const reads = new Set() + const seen = new Set() + const queue: string[] = [...(exposure?.depends_on?.nodes ?? [])] + while (queue.length) { + const id = String(queue.shift()) + if (seen.has(id)) continue + seen.add(id) + const layer = manifest?.metrics?.[id] ?? manifest?.semantic_models?.[id] + if (layer) queue.push(...(layer.depends_on?.nodes ?? [])) + else reads.add(id) + } + return reads } diff --git a/packages/opencode/src/altimate/native/dbt/helpers.ts b/packages/opencode/src/altimate/native/dbt/helpers.ts index 3454c61e2d..4fd0b70d8d 100644 --- a/packages/opencode/src/altimate/native/dbt/helpers.ts +++ b/packages/opencode/src/altimate/native/dbt/helpers.ts @@ -86,13 +86,15 @@ export function detectDialect(manifest: any): string { } /** - * Build a schema context object from upstream dependency nodes. + * Build a schema context object from upstream dependency nodes, keyed by bare table name. * Returns the { tables, version } format expected by core.Schema.fromJson(). */ export function buildSchemaContext( nodes: Record, sources: Record, upstreamIds: string[], + /** Columns found beyond the declared ones (e.g. from the node's own SQL), added after them. */ + moreColumns: (uid: string) => { name: string; type: string }[] = () => [], ): Record | null { const tables: Record = {} @@ -103,13 +105,12 @@ export function buildSchemaContext( const tableName = node.alias || node.name || "" if (!tableName) continue - const columnsDict = node.columns || {} - if (Object.keys(columnsDict).length === 0) continue - - const cols = Object.entries(columnsDict).map(([colName, col]: [string, any]) => ({ + const cols = Object.entries(node.columns || {}).map(([colName, col]: [string, any]) => ({ name: col.name || colName, type: col.data_type || col.type || "", })) + const declared = new Set(cols.map((col) => col.name.toLowerCase())) + for (const col of moreColumns(uid)) if (!declared.has(col.name.toLowerCase())) cols.push(col) if (cols.length > 0) { tables[tableName] = { columns: cols } diff --git a/packages/opencode/src/altimate/native/dbt/lineage.ts b/packages/opencode/src/altimate/native/dbt/lineage.ts index eac0cff688..b9692aa9e3 100644 --- a/packages/opencode/src/altimate/native/dbt/lineage.ts +++ b/packages/opencode/src/altimate/native/dbt/lineage.ts @@ -8,6 +8,55 @@ import * as core from "@altimateai/altimate-core" import type { DbtLineageParams, DbtLineageResult } from "../types" import { loadRawManifest, findModel, detectDialect, buildSchemaContext } from "./helpers" import { lineageGraph } from "./graph" +import { projectLineage } from "./project-lineage" + +/** + * The lineage with the placeholder columns taken out, and the parents whose placeholder reached the + * output: those a `*` reads. + */ +function withoutPlaceholders( + lineage: Record, + placeholders: Map, +): { lineage: Record; starred: string[] } { + if (!placeholders.size) return { lineage, starred: [] } + const parentOf = (text: unknown) => { + const lower = String(text ?? "").toLowerCase() + return [...placeholders].find(([placeholder]) => lower.includes(placeholder))?.[1] + } + const named = (text: unknown) => parentOf(text) !== undefined + const starred = new Set() + const dict: Record = {} + for (const [target, sources] of Object.entries(lineage.column_dict ?? {})) { + const parent = placeholders.get(target.toLowerCase()) + if (parent) { + starred.add(parent) + continue + } + if (!Array.isArray(sources)) { + dict[target] = sources + continue + } + // A `*` inside a function (hash(*), object_construct(*)) reads every column of the parent too. + for (const source of sources) { + const reads = parentOf(source) + if (reads) starred.add(reads) + } + dict[target] = sources.filter((source) => !named(source)) + } + const edges = Array.isArray(lineage.column_lineage) + ? lineage.column_lineage.filter((edge: any) => !named(edge?.source) && !named(edge?.target)) + : lineage.column_lineage + return { + lineage: { ...lineage, ...(lineage.column_dict && { column_dict: dict }), ...(lineage.column_lineage && { column_lineage: edges }) }, + starred: [...starred].sort(), + } +} + +/** Up to three names, then how many more. */ +function describeParents(names: string[]): string { + const shown = names.slice(0, 3).join(", ") + return names.length > 3 ? `${shown} and ${names.length - 3} more` : shown +} function unparsed(lineage: Record): boolean { const errors = Array.isArray(lineage.errors) ? lineage.errors : [] @@ -89,33 +138,60 @@ export function dbtLineage(params: DbtLineageParams): DbtLineageResult { // Detect dialect const dialect = params.dialect || detectDialect(manifest) - // Build schema context from upstream dependencies + // The parents' columns from the project's column resolver, as impact_analysis reads them: declared, + // else worked out from each parent's own SQL in DAG order. An ephemeral parent is inlined as a CTE, + // so its own parents are the tables the SQL reads. const upstreamIds: string[] = modelNode.depends_on?.nodes || [] - const schemaContext = buildSchemaContext(nodes, sources, upstreamIds) + const project = projectLineage(params.manifest_path, dialect) + const schemaContext = buildSchemaContext(nodes, sources, project.parents(uniqueId), (uid) => project.columns(uid)) + // A parent with no known column still gets a table entry, with one placeholder column: without one + // the engine drops the columns the SQL names from that parent, or gives them to another parent with + // a column of the same name. Where a placeholder reaches the output, a `*` reads that parent and the + // columns it carries are missing. Placeholders are taken out of the lineage returned. + const tables: Record = { ...(schemaContext?.tables ?? {}) } + const placeholders = new Map() + for (const id of project.unknownParents(uniqueId)) { + const node = project.node(id) + const table = node?.alias || node?.name || "" + if (!table || tables[table]) continue + const placeholder = `__altimate_unknown_${placeholders.size}__` + tables[table] = { columns: [{ name: placeholder, type: "" }] } + placeholders.set(placeholder, String(node.name ?? table)) + } // Delegate to altimate-core column_lineage let rawLineage: Record + let missed: string[] = [] try { - const schema = schemaContext ? core.Schema.fromJson(JSON.stringify(schemaContext)) : undefined + const schema = Object.keys(tables).length ? core.Schema.fromJson(JSON.stringify({ version: "1", tables })) : undefined const result = core.columnLineage(sql, dialect, schema) - rawLineage = JSON.parse(JSON.stringify(result)) + ;({ lineage: rawLineage, starred: missed } = withoutPlaceholders(JSON.parse(JSON.stringify(result)), placeholders)) } catch (e) { rawLineage = { error: String(e) } } + // The engine reports SQL it cannot parse in `errors` with no edges, rather than throwing. + const failure = rawLineage.error + ? String(rawLineage.error) + : unparsed(rawLineage) + ? String((rawLineage.errors as unknown[])[0]) + : undefined + const unexpanded = !failure && missed.length ? missed : [] return { model_name: modelNode.name || params.model, model_unique_id: uniqueId, compiled_sql: sql, raw_lineage: rawLineage, - // 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])] + confidence: failure || unexpanded.length ? "low" : "high", + confidence_factors: failure + ? [failure] + : unexpanded.length + ? [ + `select * over ${describeParents(unexpanded)}, whose columns aren't declared and can't be worked out from its SQL: the columns read through it are missing`, + ] : [], upstream: upstreamIds, + ...(unexpanded.length && { unexpanded_stars: unexpanded }), ...graph, } } diff --git a/packages/opencode/src/altimate/native/dbt/project-lineage.ts b/packages/opencode/src/altimate/native/dbt/project-lineage.ts new file mode 100644 index 0000000000..c48476927b --- /dev/null +++ b/packages/opencode/src/altimate/native/dbt/project-lineage.ts @@ -0,0 +1,489 @@ +// The project's column resolver: each model's (or snapshot's) column lineage from the manifest's +// compiled SQL and the lineage engine, and each node's columns, shared by dbt_lineage and the +// column impact answer so both read a model against the same parent columns. Without a dbt process +// or a warehouse: +// - relations keyed `database.schema.name`, so two parents sharing a table name stay apart; +// - ephemeral parents expanded to their own parents (the child's SQL inlines them as CTEs); +// - a parent's columns are its documented ones, the targets of its own lineage and its SELECT list, +// resolved in DAG order, so `select *` over an undocumented parent still resolves; +// - aggregate `FILTER (WHERE …)` clauses dropped before analysis (the engine loses every aggregate +// column of a SELECT that has one); +// - a WITH nested inside a CTE reported unresolved (the engine loses it and attaches outputs to any +// parent with a column of the same name); +// - SQL that reads tables but yields no edge reported unresolved, never as "not affected". +import * as fs from "fs" +import * as path from "path" +import { createHash } from "crypto" +import * as core from "@altimateai/altimate-core" +import { loadRawManifest, detectDialect } from "./helpers" + +/** = same · ~ renamed · ƒ calculated · Σ aggregated · ? unknown */ +export type How = "=" | "~" | "ƒ" | "Σ" | "?" +/** "engine": the engine's own label; "name": inferred from matching names. */ +export type HowFrom = "engine" | "name" | null +export type ShapeFlag = "unnest" | "pivot" | "except" | "nested_with" | "filter_rewritten" + +export interface LineageEdge { + from: string + fromCol: string + to: string + how: How + howFrom: HowFrom +} + +export interface ModelLineage { + edges: LineageEdge[] + /** Why the engine's answer for this model can't be used, when it can't. */ + unresolved?: string + /** The engine read only part of the SQL: its edges stand, and what it missed may carry a value. */ + partial?: string + flags: ShapeFlag[] +} + +// --------------------------------------------------------------------------------------------- +// SQL rewrites and shape checks + +/** + * `agg(x) filter (where c)` reads like `agg(x)` for lineage: drop the clause (parenthesis-matched, + * quote-aware) so the engine keeps the aggregate columns. + */ +export function stripAggFilter(sql: string): string { + let out = "" + let at = 0 + const re = /\)\s*filter\s*\(\s*where\b/gi + for (let m = re.exec(sql); m; m = re.exec(sql)) { + const open = sql.indexOf("(", m.index + 1) + let depth = 0 + let quote: string | null = null + let j = open + for (; j < sql.length; j++) { + const ch = sql[j] + if (quote) { + if (ch === quote) quote = null + continue + } + if (ch === "'" || ch === '"') quote = ch + else if (ch === "(") depth++ + else if (ch === ")" && --depth === 0) break + } + out += sql.slice(at, m.index + 1) + at = j + 1 + re.lastIndex = at + } + return out + sql.slice(at) +} + +/** + * SQL with comments blanked and string literals (quoted or dollar-quoted) emptied, in one pass so + * none can swallow another. + */ +export 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("$") ? "''" : " ", + ) +} + +/** A WITH opened inside a parenthesis: a CTE (or subquery) with its own WITH. */ +export function hasNestedWith(sql: string): boolean { + return /\(\s*with\s+(?:recursive\s+)?[a-z_][a-z0-9_]*\s+as\s*\(/i.test(stripSql(sql)) +} + +/** + * 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 +} + +export function shapeFlags(sql: string): ShapeFlag[] { + const code = stripSql(sql) + const flags: ShapeFlag[] = [] + if (/\b(?:unnest|flatten|explode(?:_outer)?|posexplode)\s*\(/i.test(code)) flags.push("unnest") + if (/\b(?:un)?pivot\b/i.test(code)) flags.push("pivot") + if (/\bexcept\b/i.test(code)) flags.push("except") + return flags +} + +const AGGREGATE = + /\b(?:sum|count|avg|min|max|median|approx_\w+|array_agg|listagg|string_agg|stddev\w*|variance|var_\w+|any_value|bool_and|bool_or|count_if|arbitrary|max_by|min_by|percentile\w*)\s*\(/i + +/** The engine's lens when it gives one; otherwise "=" by matching names, or "?". */ +export function howOf(lens: unknown, code: unknown, source: string, target: string): { how: How; howFrom: HowFrom } { + if (lens === "Original") return { how: "=", howFrom: "engine" } + if (lens === "Alias") return { how: "~", howFrom: "engine" } + if (lens === "Transformation") { + const steps = Array.isArray(code) ? code.map((s: any) => String(s?.expression ?? "")).join(" ") : "" + return { how: AGGREGATE.test(steps) ? "Σ" : "ƒ", howFrom: "engine" } + } + return source === target ? { how: "=", howFrom: "name" } : { how: "?", howFrom: null } +} + +export const compareText = (a: string, b: string) => (a < b ? -1 : a > b ? 1 : 0) + +/** Quoted or bare parts of `"db"."schema"."table"."col"` or `orders.col`. */ +function idents(text: string): string[] { + return [...text.matchAll(/"((?:[^"]|"")*)"|([^."]+)/g)].map((m) => (m[1] ?? m[2] ?? "").replace(/""/g, '"')) +} + +// --------------------------------------------------------------------------------------------- +// Caches: engine results keyed by a hash of the SQL and the schema it was read against, so a +// recompiled manifest reuses every model whose SQL and parents' columns did not change. + +const CACHE_LIMIT = 20_000 +const lineageCache = new Map() +const outputColumnsCache = new Map() +const readsCache = new Map | null>() + +function remember(cache: Map, key: string, make: () => T): T { + const hit = cache.get(key) + if (hit !== undefined) return hit + const value = make() + if (cache.size >= CACHE_LIMIT) cache.delete(cache.keys().next().value!) + cache.set(key, value) + return value +} + +const hash = (...parts: string[]) => createHash("sha1").update(parts.join("\0")).digest("hex") + +/** Clears the engine result caches (tests). */ +export function clearLineageCaches() { + lineageCache.clear() + outputColumnsCache.clear() + readsCache.clear() +} + +// --------------------------------------------------------------------------------------------- +// Per-manifest lineage + +const TRACED = new Set(["model", "snapshot"]) + +export class ProjectLineage { + readonly nodes: Record + readonly sources: Record + /** Model and snapshot children of each node, from depends_on. */ + readonly children: Record = {} + private readonly columnMemo = new Map() + private readonly lineageMemo = new Map() + private readonly settledMemo = new Map | undefined>() + private readonly busy = new Set() + + constructor( + readonly manifest: any, + readonly dialect: string, + readonly projectDir: string | null, + ) { + this.nodes = manifest.nodes ?? {} + this.sources = manifest.sources ?? {} + for (const [id, node] of Object.entries(this.nodes)) { + if (!TRACED.has(node?.resource_type)) continue + for (const parent of node.depends_on?.nodes ?? []) (this.children[parent] ??= []).push(id) + } + } + + node(id: string): any { + return this.nodes[id] ?? this.sources[id] + } + + traced(id: string): boolean { + return TRACED.has(this.nodes[id]?.resource_type) + } + + ephemeral(id: string): boolean { + return this.nodes[id]?.config?.materialized === "ephemeral" + } + + /** Parents with ephemeral ones replaced by their own parents. */ + parents(id: string): string[] { + const out: string[] = [] + const queue = [...(this.node(id)?.depends_on?.nodes ?? [])] + const seen = new Set() + while (queue.length) { + const p = queue.shift()! + if (seen.has(p)) continue + seen.add(p) + if (!this.node(p)) continue + if (this.ephemeral(p)) queue.push(...(this.node(p).depends_on?.nodes ?? [])) + else out.push(p) + } + return out + } + + /** Children, and the children reached through ephemeral children (they read this node inlined). */ + readers(id: string): string[] { + const out: string[] = [] + const queue = [...(this.children[id] ?? [])] + const seen = new Set() + while (queue.length) { + const c = queue.shift()! + if (seen.has(c)) continue + seen.add(c) + out.push(c) + if (this.ephemeral(c)) queue.push(...(this.children[c] ?? [])) + } + return out + } + + /** Every model and snapshot downstream, at model level. */ + descendants(id: string): Set { + const seen = new Set() + const queue = [id] + while (queue.length) + for (const c of this.children[queue.shift()!] ?? []) if (!seen.has(c)) (seen.add(c), queue.push(c)) + return seen + } + + /** Every node upstream, at model level. */ + ancestors(id: string): Set { + const seen = new Set() + const queue = [id] + while (queue.length) + for (const p of this.node(queue.shift()!)?.depends_on?.nodes ?? []) + if (this.node(p) && !seen.has(p)) (seen.add(p), queue.push(p)) + return seen + } + + private relationKey(n: any): string { + const name = n.resource_type === "source" ? n.identifier || n.name : n.alias || n.name + return [n.database, n.schema, name] + .filter((p) => typeof p === "string" && p !== "") + .join(".") + .toLowerCase() + } + + sqlOf(id: string): string { + const n = this.nodes[id] + return String(n?.compiled_code || n?.compiled_sql || "") + } + + private outputColumns(sql: string): string[] { + return remember(outputColumnsCache, hash(this.dialect, sql), () => { + try { + return (core.extractOutputColumns(sql, this.dialect) as string[]).map(String) + } catch { + return [] + } + }) + } + + /** Columns the SQL names anywhere (filters, joins, groupings included), lower-cased; null when the engine can't list them. */ + reads(id: string): Set | null { + const sql = this.sqlOf(id) + if (!sql) return new Set() + return remember(readsCache, hash(this.dialect, sql), () => { + try { + return new Set( + ((core.extractMetadata(sql, this.dialect) as any).columns ?? []).map((c: string) => String(c).toLowerCase()), + ) + } catch { + return null + } + }) + } + + private seedHeader(n: any): string[] { + if (!this.projectDir || !n.original_file_path) return [] + try { + const first = fs.readFileSync(path.join(this.projectDir, n.original_file_path), "utf8").split(/\r?\n/)[0] ?? "" + return first.split(",").map((c) => c.trim().replace(/^"|"$/g, "")) + } catch { + return [] + } + } + + /** A node's columns: documented, plus (models) its own lineage targets and SELECT list, (seeds) its CSV header. */ + columns(id: string): { name: string; type: string }[] { + const memo = this.columnMemo.get(id) + if (memo) return memo + const n = this.node(id) + const found = new Map() + const add = (name: unknown, type?: unknown) => { + if (typeof name !== "string" || !name || name === "*") return + const k = name.toLowerCase() + const t = typeof type === "string" ? type : "" + const had = found.get(k) + if (!had || (!had.type && t)) found.set(k, { name: k, type: t }) + } + for (const [k, c] of Object.entries(n?.columns ?? {})) add(c?.name || k, c?.data_type || c?.type) + if (n?.resource_type === "seed") for (const c of this.seedHeader(n)) add(c) + if (this.traced(id) && !this.busy.has(id)) { + for (const e of this.lineage(id).edges) add(e.to) + const sql = this.sqlOf(id) + if (sql) for (const c of this.outputColumns(sql)) add(c) + } + const out = [...found.values()] + this.columnMemo.set(id, out) + return out + } + + /** Parents (ephemeral ones expanded) for which the docs, a seed header and their own SQL give no column. */ + unknownParents(id: string): string[] { + return this.parents(id).filter((p) => this.columns(p).length === 0) + } + + /** + * The node's columns, lower-cased, when nothing can be missing from them; undefined otherwise. + * A model's are settled by a SELECT list with no `*`, or, with one, by lineage that resolved over + * parents whose own columns are settled. A source's or seed's are those it declares (or its header). + */ + settledColumns(id: string): Set | undefined { + if (this.settledMemo.has(id)) return this.settledMemo.get(id) + this.settledMemo.set(id, undefined) + const names = new Set(this.columns(id).map((c) => c.name)) + let settled = names.size > 0 + if (settled && this.traced(id)) { + const sql = this.sqlOf(id) + settled = !!sql && (selectsStar(sql) + ? !this.lineage(id).unresolved && this.parents(id).every((p) => this.settledColumns(p) !== undefined) + : this.outputColumns(sql).length > 0) + } + const out = settled ? names : undefined + this.settledMemo.set(id, out) + return out + } + + lineage(id: string): ModelLineage { + const memo = this.lineageMemo.get(id) + if (memo) return memo + this.busy.add(id) + try { + const result = this.computeLineage(id) + this.lineageMemo.set(id, result) + return result + } finally { + this.busy.delete(id) + } + } + + private computeLineage(id: string): ModelLineage { + const raw = this.sqlOf(id) + if (!raw) return { edges: [], unresolved: "no compiled SQL: run `dbt compile`", flags: [] } + const sql = stripAggFilter(raw) + const flags = shapeFlags(raw) + if (sql !== raw) flags.push("filter_rewritten") + if (hasNestedWith(sql)) { + flags.push("nested_with") + return { edges: [], unresolved: "a WITH nested inside a CTE: the engine's columns here are unreliable", flags } + } + + const tables: Record = {} + const byKey: Record = {} + const parents = this.parents(id) + for (const p of parents) { + const pn = this.node(p) + const k = this.relationKey(pn) + byKey[k] = p + const cols = this.columns(p) + if (!cols.length) continue + tables[k] = { + ...(pn.database && { database: pn.database }), + ...(pn.schema && { schema: pn.schema }), + columns: cols, + } + } + const schemaJson = JSON.stringify({ version: "1", tables }) + const result = remember(lineageCache, hash(this.dialect, sql, schemaJson), () => { + try { + const schema = Object.keys(tables).length ? core.Schema.fromJson(schemaJson) : undefined + return JSON.parse(JSON.stringify(core.columnLineage(sql, this.dialect, schema))) + } catch (e) { + return { exception: String(e instanceof Error ? e.message : e).slice(0, 200) } + } + }) + if (result.exception) return { edges: [], unresolved: result.exception, flags } + + const edges: LineageEdge[] = [] + const seen = new Set() + const keyOf = (text: string): [string, string] => { + const parts = idents(text) + const col = parts.pop() ?? "" + return [parts.join(".").toLowerCase(), col] + } + // The engine's edge order changes from process to process. A fixed order keeps the answer the same + // between runs, and settles which of two entries for one edge is used. + const ordered = (result.column_lineage ?? []) + .map((e: any) => ({ e, key: [e?.source, e?.target, e?.lens_type, JSON.stringify(e?.lens_code)].join("\0") })) + .sort((a: { key: string }, b: { key: string }) => compareText(a.key, b.key)) + for (const { e } of ordered) { + const text = String(e?.source ?? "") + let [k, col] = keyOf(text) + // A column the schema doesn't list comes back as ""db"".""schema"".""table"".col. + if (!byKey[k] && text.startsWith('""')) [k, col] = keyOf(text.replace(/""/g, '"')) + const from = byKey[k] + if (!from || !col) continue + const fromCol = col.toLowerCase() + const to = String(e.target ?? "").toLowerCase() + if (!to) continue + const dedupe = `${from}|${fromCol}|${to}` + if (seen.has(dedupe)) continue + seen.add(dedupe) + edges.push({ from, fromCol, to, ...howOfEdge(e, fromCol, to) }) + } + const errors: unknown[] = Array.isArray(result.errors) ? result.errors : [] + const firstError = (): string => { + const first: any = errors[0] + return String(first?.message ?? first).slice(0, 200) + } + if (!edges.length && errors.length) return { edges, unresolved: firstError(), flags } + if (!edges.length && parents.length) + return { + edges, + unresolved: `reads ${parents.length} table${parents.length === 1 ? "" : "s"} but the engine traced no column (select * over columns it can't see?)`, + flags, + } + return { edges, flags, ...(errors.length && { partial: firstError() }) } + } +} + +function howOfEdge(e: any, fromCol: string, to: string) { + return howOf(e?.lens_type, e?.lens_code, fromCol, to) +} + +/** + * Why a model can't have the column, when its columns are settled and don't include it (any case); + * undefined when it has it or its columns may be incomplete. + */ +export function missingColumn(p: ProjectLineage, id: string, column: string): string | undefined { + const settled = p.settledColumns(id) + if (!settled || settled.has(column.toLowerCase())) return undefined + const names = [...settled].sort(compareText) + const shown = names.length > 12 ? `${names.slice(0, 12).join(", ")} and ${names.length - 12} more` : names.join(", ") + return `${String(p.node(id)?.name ?? id)} has no column ${column}; its columns are ${shown}` +} + +const projects = new WeakMap>() + +export function projectLineage(manifestPath: string, dialect?: string): ProjectLineage { + const manifest = loadRawManifest(manifestPath) + if (!manifest) throw new Error(`Manifest not found at ${manifestPath}`) + const d = dialect || detectDialect(manifest) + let byDialect = projects.get(manifest) + if (!byDialect) projects.set(manifest, (byDialect = new Map())) + let p = byDialect.get(d) + if (!p) { + const target = path.dirname(path.resolve(manifestPath)) + byDialect.set(d, (p = new ProjectLineage(manifest, d, path.dirname(target)))) + } + return p +} diff --git a/packages/opencode/src/altimate/native/types.ts b/packages/opencode/src/altimate/native/types.ts index 5de10c591f..09334260f7 100644 --- a/packages/opencode/src/altimate/native/types.ts +++ b/packages/opencode/src/altimate/native/types.ts @@ -1089,6 +1089,8 @@ export interface DbtLineageResult { confidence_factors: string[] /** unique_ids the model depends on (`depends_on.nodes`). */ upstream?: string[] + /** The parents a `*` reads whose columns aren't known, by name: the only reason for "low" when set. */ + unexpanded_stars?: string[] /** With include_graph: the graph a lineage viewer draws for the model. */ graph?: import("./dbt/graph").LineageGraph } diff --git a/packages/opencode/src/altimate/tools/impact-analysis.ts b/packages/opencode/src/altimate/tools/impact-analysis.ts index 5a5e3de299..7585749257 100644 --- a/packages/opencode/src/altimate/tools/impact-analysis.ts +++ b/packages/opencode/src/altimate/tools/impact-analysis.ts @@ -6,7 +6,8 @@ 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" +import { columnReach, exposuresDeclared, exposuresOn, type ColumnReach, type ColumnReachExposure } from "../native/dbt/column-reach" +import { columnAnswer, describeAnswer, type ColumnAnswer } from "../native/dbt/column-answer" export const ImpactAnalysisTool = Tool.define("impact_analysis", { description: [ @@ -18,6 +19,7 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { "Examples:", '- impact_analysis({ model: "stg_orders", change_type: "remove" })', '- impact_analysis({ model: "stg_orders", column: "order_id", change_type: "rename" })', + '- impact_analysis({ model: "fct_orders", column: "revenue", change_type: "modify", direction: "upstream" })', '- impact_analysis({ manifest_path: "target/manifest.json", model: "dim_customers", change_type: "retype" })', ].join("\n"), parameters: z.object({ @@ -29,6 +31,13 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { 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().describe("SQL dialect for column lineage (auto-detected from the manifest if omitted)"), + direction: z + .enum(["downstream", "upstream"]) + .optional() + .default("downstream") + .describe( + "With a column: 'downstream' (default) answers where its value goes; 'upstream' traces where it comes from. Returned as metadata.column_answer.", + ), }), async execute(args, ctx) { try { @@ -80,11 +89,15 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { // 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 declaredExposures: number | undefined let reach: ColumnReach | undefined let reachError: string | undefined + let answer: ColumnAnswer | undefined + let answerError: string | undefined let exposuresError: string | undefined try { exposures = exposuresOn(args.manifest_path, affectedModelIds) + declaredExposures = exposuresDeclared(args.manifest_path) } 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) @@ -96,10 +109,21 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { // The model-level impact still stands; the reason is reported beside it. reachError = error instanceof Error ? error.message : String(error) } + try { + answer = columnAnswer( + args.manifest_path, + targetModel.unique_id ?? args.model, + args.column, + args.direction ?? "downstream", + args.dialect, + ) + } catch (error) { + answerError = error instanceof Error ? error.message : String(error) + } } // Step 6: Format the impact report - const output = formatImpactReport({ + const report = formatImpactReport({ model: targetModel.name, column: args.column, changeType: args.change_type, @@ -112,6 +136,7 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { columnReach: reach, columnReachError: reachError, }) + const output = answer ? `${report}\n${describeAnswer(answer)}` : report const totalAffected = downstream.length const severity = @@ -144,8 +169,11 @@ export const ImpactAnalysisTool = Tool.define("impact_analysis", { 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, + ...(declaredExposures !== undefined && { exposures_declared: declaredExposures }), ...(reach && { column_reach: reach }), ...(reachError && { column_reach_error: reachError }), + ...(answer && { column_answer: answer }), + ...(answerError && { column_answer_error: answerError }), ...(exposuresError && { exposures_error: exposuresError }), ...(findings.length > 0 && { findings }), }, diff --git a/packages/opencode/test/altimate/column-answer-order.test.ts b/packages/opencode/test/altimate/column-answer-order.test.ts new file mode 100644 index 0000000000..e9d8cdf223 --- /dev/null +++ b/packages/opencode/test/altimate/column-answer-order.test.ts @@ -0,0 +1,90 @@ +import { afterAll, afterEach, describe, expect, mock, test } from "bun:test" +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import * as realCore from "@altimateai/altimate-core" + +// The engine's edge order changes from process to process, which one process can't show: these tests +// reorder its answer themselves. +let reorder: (edges: any[]) => any[] = (edges) => edges +const real: any = { ...realCore } +// Bun's mock.module leaks across test files: put the genuine engine back for the files after this one. +afterAll(() => { + mock.module("@altimateai/altimate-core", () => real) +}) +mock.module("@altimateai/altimate-core", () => ({ + ...real, + columnLineage: (...args: unknown[]) => { + const result = JSON.parse(JSON.stringify(real.columnLineage(...args))) + return { ...result, column_lineage: reorder(result.column_lineage ?? []) } + }, +})) +const { columnAnswer } = await import("../../src/altimate/native/dbt/column-answer") +const { clearLineageCaches } = await import("../../src/altimate/native/dbt/project-lineage") + +const dirs: string[] = [] +afterEach(() => { + reorder = (edges) => edges + clearLineageCaches() + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) +}) + +const model = (name: string, deps: string[], sql: string, columns: string[]) => ({ + resource_type: "model", + unique_id: `model.shop.${name}`, + name, + alias: name, + database: "shop", + schema: "main", + original_file_path: `models/${name}.sql`, + depends_on: { nodes: deps.map((d) => `model.shop.${d}`) }, + config: { materialized: "table" }, + compiled_code: sql, + columns: Object.fromEntries(columns.map((c) => [c, { name: c }])), +}) + +/** The answer from a fresh project and empty caches, so the engine is asked again. */ +function answer() { + const nodes = [ + model("stg", [], "select 1 as id, 2 as amount", ["id", "amount"]), + model("mart", ["stg"], "select amount as a, amount * 2 as b, amount + 1 as c from shop.main.stg", []), + model("rollup", ["mart"], "select a as a2, b as b2, c as c2 from shop.main.mart", []), + ] + const dir = mkdtempSync(join(tmpdir(), "column-answer-order-")) + dirs.push(dir) + mkdirSync(join(dir, "target")) + const file = join(dir, "target", "manifest.json") + writeFileSync( + file, + JSON.stringify({ + metadata: { adapter_type: "duckdb", generated_at: "2026-01-01T00:00:00Z" }, + nodes: Object.fromEntries(nodes.map((n) => [n.unique_id, n])), + sources: {}, + exposures: {}, + }), + ) + clearLineageCaches() + return columnAnswer(file, "stg", "amount") +} + +describe("column answer: order", () => { + test("the same whichever order the engine returns its edges in", () => { + const first = answer() + expect(first.counts.columns).toBe(6) + reorder = (edges) => [...edges].reverse() + const second = answer() + expect(second.columns).toEqual(first.columns) + expect(second.edges).toEqual(first.edges) + expect(second.nodes).toEqual(first.nodes) + }) + + test("two entries for one edge: the same one is used whichever comes first", () => { + const other = (e: any) => ({ ...e, lens_type: "Transformation", lens_code: [{ expression: "sum(amount)" }] }) + reorder = (edges) => edges.flatMap((e) => [e, other(e)]) + const first = answer() + reorder = (edges) => edges.flatMap((e) => [other(e), e]) + const second = answer() + expect(second.columns).toEqual(first.columns) + expect(second.edges).toEqual(first.edges) + }) +}) diff --git a/packages/opencode/test/altimate/column-answer.test.ts b/packages/opencode/test/altimate/column-answer.test.ts new file mode 100644 index 0000000000..f66ce07869 --- /dev/null +++ b/packages/opencode/test/altimate/column-answer.test.ts @@ -0,0 +1,534 @@ +import { afterEach, describe, expect, test } from "bun:test" +import { mkdtempSync, readFileSync, rmSync, writeFileSync, mkdirSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { ENGINE, MAX_ANSWER_COLUMNS, columnAnswer, type ColumnAnswer } from "../../src/altimate/native/dbt/column-answer" +import { clearLineageCaches, hasNestedWith, howOf, shapeFlags, stripAggFilter } from "../../src/altimate/native/dbt/project-lineage" + +const dirs: string[] = [] +afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) + clearLineageCaches() +}) + +const model = ( + name: string, + deps: string[], + sql: string, + columns: string[] = [], + extra: Record = {}, +) => ({ + resource_type: "model", + unique_id: `model.shop.${name}`, + name, + alias: name, + database: "shop", + schema: "main", + original_file_path: `models/marts/${name}.sql`, + depends_on: { nodes: deps.map((d) => (d.includes(".") ? d : `model.shop.${d}`)) }, + config: { materialized: "table" }, + compiled_code: sql, + columns: Object.fromEntries(columns.map((c) => [c, { name: c, data_type: "double" }])), + ...extra, +}) + +const source = (name: string, columns: string[]) => ({ + resource_type: "source", + unique_id: `source.shop.raw.${name}`, + name, + identifier: name, + database: "shop", + schema: "raw", + columns: Object.fromEntries(columns.map((c) => [c, { name: c }])), +}) + +const T = (name: string) => `shop.main.${name}` + +// fct_orders.revenue (renamed from stg_orders.amount) is: +// - summed under a FILTER into fct_daily.paid_revenue, +// - renamed and calculated by mart_ltv (an enforced contract), then passed through a CTE by rpt_ltv, +// - only filtered on by rpt_big_orders (whose child may carry it), +// - read by rpt_nested through a WITH inside a CTE (unresolved; its child may carry it), +// - never read by rpt_counts (cleared). +function shop() { + const nodes: Record = { + "model.shop.stg_orders": model( + "stg_orders", + ["source.shop.raw.orders"], + "select order_id, customer_id, amount, status from shop.raw.orders", + ), + "model.shop.fct_orders": model( + "fct_orders", + ["stg_orders"], + `select order_id, customer_id, amount as revenue, status from ${T("stg_orders")}`, + ["order_id", "revenue"], + ), + "model.shop.fct_daily": model( + "fct_daily", + ["fct_orders"], + `select customer_id, sum(revenue) filter (where status = 'paid') as paid_revenue, count(*) as orders from ${T("fct_orders")} group by customer_id`, + ), + "model.shop.mart_ltv": model( + "mart_ltv", + ["fct_orders"], + `select order_id, revenue as ltv, revenue * 2 as double_rev from ${T("fct_orders")}`, + [], + { config: { materialized: "table", contract: { enforced: true } } }, + ), + "model.shop.rpt_ltv": model( + "rpt_ltv", + ["mart_ltv"], + `with x as (select order_id, ltv from ${T("mart_ltv")}) select order_id, ltv from x`, + ), + "model.shop.rpt_big_orders": model( + "rpt_big_orders", + ["fct_orders"], + `select order_id from ${T("fct_orders")} where revenue > 100`, + ), + "model.shop.rpt_big_child": model( + "rpt_big_child", + ["rpt_big_orders"], + `select order_id from ${T("rpt_big_orders")}`, + ), + "model.shop.rpt_counts": model( + "rpt_counts", + ["fct_orders"], + `select count(order_id) as orders from ${T("fct_orders")}`, + ), + "model.shop.rpt_nested": model( + "rpt_nested", + ["fct_orders"], + `with a as (with b as (select order_id, revenue from ${T("fct_orders")}) select order_id, revenue from b) select order_id, revenue from a`, + ), + "model.shop.rpt_nested_child": model( + "rpt_nested_child", + ["rpt_nested"], + `select order_id, revenue from ${T("rpt_nested")}`, + ), + "test.shop.not_null_fct_orders_revenue": { + resource_type: "test", + depends_on: { nodes: ["model.shop.fct_orders"] }, + }, + } + return { + metadata: { adapter_type: "duckdb", generated_at: "2026-01-01T00:00:00Z" }, + nodes, + sources: { "source.shop.raw.orders": source("orders", ["order_id", "customer_id", "amount", "status"]) }, + exposures: { + "exposure.shop.ltv_board": { + name: "ltv_board", + label: "LTV board", + depends_on: { nodes: ["model.shop.rpt_ltv"] }, + }, + }, + } +} + +function write(manifest: unknown): string { + const dir = mkdtempSync(join(tmpdir(), "column-answer-")) + dirs.push(dir) + mkdirSync(join(dir, "target")) + const file = join(dir, "target", "manifest.json") + writeFileSync(file, JSON.stringify(manifest)) + return file +} + +const at = (a: ColumnAnswer, model: string, column: string) => + a.columns.findIndex((c) => a.nodes[c.model]!.name === model && c.column === column) +const col = (a: ColumnAnswer, model: string, column: string) => a.columns[at(a, model, column)] +const named = (a: ColumnAnswer, idx: number) => a.nodes[idx]!.name + +describe("column answer: downstream", () => { + test("follows the value through a rename, a calculation, an aggregate FILTER and a CTE", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + const reached = a.columns + .slice(1) + .map((c) => `${a.nodes[c.model]!.name}.${c.column}`) + .sort() + expect(reached).toEqual(["fct_daily.paid_revenue", "mart_ltv.double_rev", "mart_ltv.ltv", "rpt_ltv.ltv"]) + expect(a.counts).toMatchObject({ columns: 4, models: 3, hops: 2, contracts: 1, ends: 3 }) + }) + + test("an aggregate FILTER clause no longer hides the aggregate, and is flagged", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + expect(at(a, "fct_daily", "paid_revenue")).toBeGreaterThan(0) + expect(a.flags).toContainEqual({ + model: a.nodes.findIndex((n) => n.name === "fct_daily"), + flag: "filter_rewritten", + }) + }) + + test("tiers: also reads, unresolved and what sits past them may carry it; the rest is cleared", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + expect( + a.also_reads.map((r) => ({ name: named(a, r.model), columns: r.columns, downstream: r.downstream })), + ).toEqual([{ name: "rpt_big_orders", columns: ["revenue"], downstream: 1 }]) + expect(a.unresolved.map((u) => [named(a, u.model), u.downstream])).toEqual([["rpt_nested", 1]]) + // value 3 + may carry 4 (rpt_big_orders, rpt_big_child, rpt_nested, rpt_nested_child) + cleared 1 (rpt_counts) + expect(a.counts).toMatchObject({ may_carry_models: 4, cleared_models: 1, model_level: 8 }) + expect(a.cleared).toEqual([{ id: "model.shop.rpt_counts", name: "rpt_counts" }]) + expect(a.nodes.map((n) => n.name)).not.toContain("rpt_counts") + }) + + test("a WITH nested inside a CTE is unresolved with the reason, never cleared, and flagged", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + const nested = a.unresolved.find((u) => named(a, u.model) === "rpt_nested")! + expect(nested.reason).toContain("WITH nested inside a CTE") + expect(a.flags).toContainEqual({ model: nested.model, flag: "nested_with" }) + expect(at(a, "rpt_nested", "revenue")).toBe(-1) + }) + + test("a union all of select * over columns the engine can't see is unresolved, not cleared", () => { + const m = { + metadata: { adapter_type: "duckdb" }, + sources: { "source.shop.raw.a": source("a", []), "source.shop.raw.b": source("b", []) }, + nodes: { + "model.shop.m_a": model("m_a", ["source.shop.raw.a"], "select * from shop.raw.a"), + "model.shop.m_b": model("m_b", ["source.shop.raw.b"], "select * from shop.raw.b"), + "model.shop.m_union": model( + "m_union", + ["m_a", "m_b"], + `select * from ${T("m_a")} union all select * from ${T("m_b")}`, + ), + }, + } + const a = columnAnswer(write(m), "m_a", "id") + expect(a.unresolved.map((u) => named(a, u.model))).toEqual(["m_union"]) + expect(a.unresolved[0]!.reason).toContain("traced no column") + expect(a.counts).toMatchObject({ columns: 0, may_carry_models: 1, cleared_models: 0, model_level: 1 }) + }) + + test("an ephemeral focus's readers are unresolved, not cleared: their SQL names its parents' columns", () => { + const m = { + metadata: { adapter_type: "duckdb" }, + sources: {}, + nodes: { + "model.shop.base": model("base", [], "select 1 as id, 2 as amount", ["id", "amount"]), + "model.shop.eph": model("eph", ["base"], `select * from ${T("base")}`, ["id", "amount"], { config: { materialized: "ephemeral" } }), + "model.shop.r_star": model("r_star", ["eph"], `with __dbt__cte__eph as (select * from ${T("base")}) select * from __dbt__cte__eph`), + }, + } + const a = columnAnswer(write(m), "eph", "amount") + expect(a.cleared.map((one) => one.name)).toEqual([]) + expect(a.unresolved.map((u) => named(a, u.model))).toEqual(["r_star"]) + expect(a.unresolved[0]!.reason).toContain("ephemeral") + }) + + test("a reader the engine reads only in part, or whose columns it can't list, is unresolved, never cleared", () => { + // Snowflake path access the engine parses only in part (`::variant:`), in a filter and in an output. + const m = { + metadata: { adapter_type: "snowflake" }, + sources: {}, + nodes: { + "model.shop.f": model("f", [], "select 1 as a, 2 as b", ["a", "b"]), + "model.shop.r_filter": model("r_filter", ["f"], `select a from ${T("f")} where b::variant:ok::boolean`), + "model.shop.r_value": model("r_value", ["f"], `select a, b::variant:path::string as c from ${T("f")}`), + "model.shop.after_filter": model("after_filter", ["r_filter"], `select a from ${T("r_filter")}`), + }, + } + const a = columnAnswer(write(m), "f", "b") + expect(a.cleared.map((one) => one.name)).toEqual([]) + expect(a.unresolved.map((u) => named(a, u.model)).sort()).toEqual(["r_filter", "r_value"]) + expect(a.unresolved.every((u) => u.reason.length > 0)).toBe(true) + // What the engine did trace still stands. + expect(a.columns.map((c) => `${named(a, c.model)}.${c.column}`)).toContain("r_value.c") + expect(a.counts.may_carry_models).toBeGreaterThanOrEqual(2) + }) + + test("a reader through an ephemeral model is reached", () => { + const m = shop() + m.nodes["model.shop.eph"] = model("eph", ["fct_orders"], `select order_id, revenue from ${T("fct_orders")}`, [], { + config: { materialized: "ephemeral" }, + }) + m.nodes["model.shop.rpt_eph"] = model( + "rpt_eph", + ["eph"], + `with __dbt__cte__eph as (select order_id, revenue from ${T("fct_orders")}) select revenue as eph_revenue from __dbt__cte__eph`, + ) + const a = columnAnswer(write(m), "fct_orders", "revenue") + expect(at(a, "rpt_eph", "eph_revenue")).toBeGreaterThan(0) + }) + + test("stops at the column limit, says so, and leaves what it never traced as may carry it, never cleared", () => { + // One wide model fills the answer; a reader after it, and that reader's child, are never traced. + const wide = Array.from({ length: MAX_ANSWER_COLUMNS + 5 }, (_, i) => `revenue as c${i}`).join(", ") + const m = shop() + m.nodes["model.shop.rpt_wide"] = model("rpt_wide", ["fct_orders"], `select ${wide} from ${T("fct_orders")}`) + m.nodes["model.shop.rpt_late"] = model("rpt_late", ["fct_orders"], `select order_id, revenue from ${T("fct_orders")}`) + m.nodes["model.shop.rpt_late_child"] = model("rpt_late_child", ["rpt_late"], "") + const a = columnAnswer(write(m), "fct_orders", "revenue") + expect(a.columns).toHaveLength(MAX_ANSWER_COLUMNS) + expect(a.truncated).toBe(true) + const cleared = a.cleared.map((c) => c.name) + expect(cleared).not.toContain("rpt_late") + expect(cleared).not.toContain("rpt_late_child") + expect(a.counts.may_carry_models).toBeGreaterThanOrEqual(4) + }) +}) + +describe("column answer: upstream", () => { + test("traces a column back to the source through a CTE, a rename and a staging model", () => { + const a = columnAnswer(write(shop()), "rpt_ltv", "ltv", "upstream") + expect(a.direction).toBe("upstream") + const reached = a.columns.slice(1).map((c) => [`${a.nodes[c.model]!.name}.${c.column}`, c.hop]) + expect(reached).toEqual([ + ["mart_ltv.ltv", 1], + ["fct_orders.revenue", 2], + ["stg_orders.amount", 3], + ["orders.amount", 4], + ]) + expect(a.counts).toMatchObject({ columns: 4, models: 4, hops: 4, ends: 1 }) + expect(col(a, "orders", "amount")!.end).toBe(true) + // Edges point from the source column to the column it feeds. + expect(a.edges).toContainEqual([at(a, "fct_orders", "revenue"), at(a, "mart_ltv", "ltv"), "~"]) + }) + + test("an unresolved model on the way is reported, and what's above it may carry", () => { + const a = columnAnswer(write(shop()), "rpt_nested_child", "revenue", "upstream") + // rpt_nested's SELECT list names revenue, so it is reached; its own SQL is unresolved. + expect(a.columns.slice(1).map((c) => `${named(a, c.model)}.${c.column}`)).toEqual(["rpt_nested.revenue"]) + expect( + a.unresolved.map((u) => [named(a, u.model), u.reason.includes("nested inside a CTE"), u.downstream]), + ).toEqual([["rpt_nested", true, 3]]) + // Everything above it may carry the value: fct_orders, stg_orders and the source. + expect(a.counts).toMatchObject({ may_carry_models: 3, cleared_models: 0, model_level: 4 }) + }) + + test("ancestors in none of the tiers are listed as cleared, by name, outside nodes", () => { + const m: any = shop() + m.sources["source.shop.raw.customers"] = source("customers", ["customer_id", "name"]) + m.nodes["model.shop.stg_customers"] = model( + "stg_customers", + ["source.shop.raw.customers"], + "select customer_id, name from shop.raw.customers", + ) + m.nodes["model.shop.fct_mix"] = model( + "fct_mix", + ["fct_orders", "stg_customers"], + `select o.revenue from ${T("fct_orders")} o join ${T("stg_customers")} c on o.customer_id = c.customer_id`, + ) + const a = columnAnswer(write(m), "fct_mix", "revenue", "upstream") + expect(a.cleared).toEqual([ + { id: "source.shop.raw.customers", name: "customers" }, + { id: "model.shop.stg_customers", name: "stg_customers" }, + ]) + expect(a.counts).toMatchObject({ models: 3, cleared_models: 2, model_level: 5 }) + expect(a.nodes.map((n) => n.name)).not.toContain("stg_customers") + }) +}) + +describe("column answer: how", () => { + test("uses the engine's label, and marks a name match as inferred", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + const edge = (m: string, c: string) => a.edges.find((e) => e[1] === at(a, m, c)) + expect(edge("mart_ltv", "ltv")).toEqual([0, at(a, "mart_ltv", "ltv"), "~"]) + expect(edge("mart_ltv", "double_rev")![2]).toBe("ƒ") + // Through a CTE the engine gives no label: "=" by matching names, never as certain. + expect(edge("rpt_ltv", "ltv")).toEqual([at(a, "mart_ltv", "ltv"), at(a, "rpt_ltv", "ltv"), "=", "name"]) + expect(col(a, "rpt_ltv", "ltv")).toMatchObject({ how: "=", how_from: "name" }) + expect(col(a, "mart_ltv", "ltv")).toMatchObject({ how: "~", how_from: "engine" }) + expect(a.columns[0]).toMatchObject({ how: null, how_from: null, hop: 0, end: false }) + }) + + test("maps each lens to a glyph", () => { + expect(howOf("Original", [], "a", "a")).toEqual({ how: "=", howFrom: "engine" }) + expect(howOf("Alias", [], "a", "b")).toEqual({ how: "~", howFrom: "engine" }) + expect(howOf("Transformation", [{ expression: "a * 2 AS b" }], "a", "b")).toEqual({ how: "ƒ", howFrom: "engine" }) + expect(howOf("Transformation", [{ expression: "SUM(a) AS b" }], "a", "b")).toEqual({ how: "Σ", howFrom: "engine" }) + expect(howOf("Not sure", [], "a", "a")).toEqual({ how: "=", howFrom: "name" }) + expect(howOf("Not sure", [], "a", "b")).toEqual({ how: "?", howFrom: null }) + }) +}) + +describe("column answer: SQL rewrites and shapes", () => { + test("drops aggregate FILTER clauses, parentheses and quotes included", () => { + expect(stripAggFilter("select sum(x) filter (where (a) = ')') as s, count(*) FILTER(WHERE b) n from t")).toBe( + "select sum(x) as s, count(*) n from t", + ) + expect(stripAggFilter("select x from t where y in (select 1)")).toBe("select x from t where y in (select 1)") + }) + + test("finds a WITH nested inside a CTE, not one in a comment", () => { + expect(hasNestedWith("with a as (with b as (select 1) select * from b) select * from a")).toBe(true) + expect(hasNestedWith("with a as (select 1), b as (select 2) select * from a, b")).toBe(false) + expect(hasNestedWith("with a as (select 1 -- (with b as (\n) select * from a")).toBe(false) + }) + + test("flags unnest, pivot and except", () => { + expect(shapeFlags("select v from t cross join unnest(t.arr) as u(v)")).toEqual(["unnest"]) + expect(shapeFlags("select * from t pivot (sum(x) for k in ('a'))")).toEqual(["pivot"]) + expect(shapeFlags("select a from t except select a from u")).toEqual(["except"]) + expect(shapeFlags("select 'unnest(' as s -- pivot")).toEqual([]) + }) +}) + +describe("column answer: consumers", () => { + const exposure = (name: string, deps: string[], extra: Record = {}) => ({ + name, + type: "dashboard", + depends_on: { nodes: deps }, + ...extra, + }) + const withExposures = (exposures: Record, more: Record = {}) => ({ + ...shop(), + exposures, + ...more, + }) + + test("each exposure lands in the tier of the models it reads, the closest tier first", () => { + const a = columnAnswer( + write( + withExposures({ + "exposure.shop.z_cleared": exposure("z_cleared", ["model.shop.rpt_counts"]), + "exposure.shop.m_maybe": exposure("m_maybe", ["model.shop.rpt_big_child"], { type: "notebook" }), + "exposure.shop.c_carries": exposure("c_carries", ["model.shop.rpt_ltv", "model.shop.rpt_counts"], { + label: "Carries board", + url: "https://bi.example.com/1", + owner: { name: "Data team", email: "data@example.com" }, + }), + "exposure.shop.a_focus": exposure("a_focus", ["model.shop.fct_orders"]), + "exposure.shop.upstream_only": exposure("upstream_only", ["model.shop.stg_orders"]), + }), + ), + "fct_orders", + "revenue", + ) + expect(a.consumers.declared).toBe(5) + expect(a.consumers.exposures.map((c) => [c.label, c.tier, c.via])).toEqual([ + ["Carries board", "carries", ["rpt_ltv"]], + ["a_focus", "carries", ["fct_orders"]], + ["m_maybe", "may_carry", ["rpt_big_child"]], + ["z_cleared", "cleared", ["rpt_counts"]], + ]) + expect(a.consumers.exposures[0]).toMatchObject({ type: "dashboard", url: "https://bi.example.com/1", owner: "Data team" }) + expect(JSON.stringify(a.consumers)).not.toContain("data@example.com") + }) + + test("an exposure that reads a metric reaches the model behind it", () => { + const a = columnAnswer( + write( + withExposures( + { "exposure.shop.via_metric": exposure("via_metric", ["metric.shop.ltv"]) }, + { + metrics: { "metric.shop.ltv": { depends_on: { nodes: ["semantic_model.shop.ltv"] } } }, + semantic_models: { "semantic_model.shop.ltv": { depends_on: { nodes: ["model.shop.mart_ltv"] } } }, + }, + ), + ), + "fct_orders", + "revenue", + ) + expect(a.consumers.exposures.map((c) => [c.name, c.tier, c.via])).toEqual([["via_metric", "carries", ["mart_ltv"]]]) + }) + + test("declared tells a project with no exposures from one with none downstream", () => { + const none = columnAnswer(write(withExposures({})), "fct_orders", "revenue") + expect(none.consumers).toEqual({ declared: 0, exposures: [] }) + const elsewhere = columnAnswer( + write(withExposures({ "exposure.shop.up": exposure("up", ["model.shop.stg_orders"]) })), + "fct_orders", + "revenue", + ) + expect(elsewhere.consumers).toEqual({ declared: 1, exposures: [] }) + }) + + test("upstream lists none", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue", "upstream") + expect(a.consumers).toEqual({ declared: 1, exposures: [] }) + }) +}) + +describe("column answer: a column the model doesn't have", () => { + test("is an error when the model's columns are settled, naming them", () => { + expect(() => columnAnswer(write(shop()), "fct_orders", "not_a_column")).toThrow( + "fct_orders has no column not_a_column; its columns are customer_id, order_id, revenue, status", + ) + expect(() => columnAnswer(write(shop()), "fct_orders", "not_a_column", "upstream")).toThrow("has no column") + }) + + test("matches the column in any case", () => { + expect(columnAnswer(write(shop()), "fct_orders", "Revenue").counts.models).toBe(3) + }) + + test("answers when a select * leaves the model's columns open", () => { + const value: any = shop() + value.sources["source.shop.raw.events"] = source("events", []) + value.nodes["model.shop.stg_events"] = model("stg_events", ["source.shop.raw.events"], "select * from shop.raw.events") + const a = columnAnswer(write(value), "stg_events", "kind") + expect(a.focus).toEqual({ model: "model.shop.stg_events", column: "kind" }) + }) +}) + +describe("column answer: contract", () => { + test("has the fixed shape", () => { + const a = columnAnswer(write(shop()), "fct_orders", "revenue") + expect(Object.keys(a)).toEqual([ + "version", + "focus", + "direction", + "inputs", + "counts", + "nodes", + "columns", + "edges", + "also_reads", + "unresolved", + "flags", + "cleared", + "consumers", + "truncated", + "next", + ]) + expect(a.version).toBe(1) + expect(a.focus).toEqual({ model: "model.shop.fct_orders", column: "revenue" }) + expect(a.direction).toBe("downstream") + expect(a.inputs).toMatchObject({ + manifest: { generated_at: "2026-01-01T00:00:00Z", older_than_sql: false }, + sql: "compiled_code", + catalog: null, + engine: ENGINE, + }) + expect(a.inputs.workarounds).toContain("aggregate FILTER clause dropped before analysis") + expect(Object.keys(a.counts).sort()).toEqual( + ["cleared_models", "columns", "contracts", "ends", "hops", "may_carry_models", "model_level", "models"].sort(), + ) + expect(a.nodes[0]).toEqual({ + id: "model.shop.fct_orders", + name: "fct_orders", + folder: "marts", + materialized: "table", + contract: false, + tests: 1, + exposures: [], + }) + expect(a.nodes.find((n) => n.name === "mart_ltv")!.contract).toBe(true) + expect(a.nodes.find((n) => n.name === "rpt_ltv")!.exposures).toEqual(["LTV board"]) + expect(a.columns[0]).toMatchObject({ model: 0, column: "revenue", hop: 0, type: "double" }) + for (const e of a.edges) { + expect(typeof e[0]).toBe("number") + expect(typeof e[1]).toBe("number") + expect(["=", "~", "ƒ", "Σ", "?"]).toContain(e[2]) + } + expect(a.truncated).toBe(false) + expect(a.next).toBeNull() + }) + + test("the engine label matches the pinned altimate-core", () => { + const pkg = JSON.parse(readFileSync(join(import.meta.dir, "../../package.json"), "utf8")) + expect(ENGINE).toBe(`altimate-core ${pkg.dependencies["@altimateai/altimate-core"]}`) + }) + + test("says when a model file changed after the manifest was written", () => { + const file = write(shop()) + const dir = join(file, "..", "..") + mkdirSync(join(dir, "models", "marts"), { recursive: true }) + const later = new Date(Date.now() + 60_000) + writeFileSync(join(dir, "models", "marts", "mart_ltv.sql"), "select 1") + require("node:fs").utimesSync(join(dir, "models", "marts", "mart_ltv.sql"), later, later) + expect(columnAnswer(file, "fct_orders", "revenue").inputs.manifest.older_than_sql).toBe(true) + }) + + test("an unknown model throws", () => { + expect(() => columnAnswer(write(shop()), "nope", "x")).toThrow("not found") + }) +}) diff --git a/packages/opencode/test/altimate/column-reach.test.ts b/packages/opencode/test/altimate/column-reach.test.ts index e16a60030f..6362ae9bb8 100644 --- a/packages/opencode/test/altimate/column-reach.test.ts +++ b/packages/opencode/test/altimate/column-reach.test.ts @@ -9,14 +9,16 @@ import { parentsNamed, joinsNaturally, namesThroughIdentifier, - selectsStar, } from "../../src/altimate/native/dbt/column-reach" +import { selectsStar } from "../../src/altimate/native/dbt/project-lineage" 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" +import { clearLineageCaches } from "../../src/altimate/native/dbt/project-lineage" const dirs: string[] = [] afterEach(() => { + clearLineageCaches() for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) }) @@ -311,8 +313,20 @@ describe("parentsNamed", () => { }) }) +describe("columnReach of a column the model doesn't have", () => { + test("is an error, not a reach with the model's dashboards as readers", () => { + expect(() => columnReach(manifestFile(manifest()), "fct_orders", "not_a_column")).toThrow( + "fct_orders has no column not_a_column; its columns are gross_revenue, order_id, revenue", + ) + }) + + test("is no error for the column in another case", () => { + expect(columnReach(manifestFile(manifest()), "fct_orders", "REVENUE").models.length).toBeGreaterThan(0) + }) +}) + 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", () => { + test("select * from a parent whose schema.yml leaves the column out reaches it through the parent's own SQL", () => { const value = manifest() value.nodes["model.shop.fct_orders"].columns = { order_id: column("order_id") } value.nodes["model.shop.rpt_star"] = model( @@ -328,8 +342,29 @@ describe("columnReach where the engine's schema falls short", () => { ["total"], ) const reach = columnReach(manifestFile(value), "fct_orders", "revenue") + expect(reach.models.find((m) => m.name === "rpt_star")?.columns).toEqual(["revenue"]) + expect(reach.models.find((m) => m.name === "rpt_after_star")?.columns).toEqual(["total"]) + expect(reach.uncertain.map((m) => m.name)).not.toContain("rpt_star") + }) + + test("select * from a target whose columns can't be known is a possible read, and its children may read it", () => { + const value = manifest() + value.nodes["model.shop.fct_orders"] = model("fct_orders", [], "select * from raw.orders", []) + 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", + "selects * from a parent whose known columns don't include the column", ) expect(reach.uncertain.map((m) => m.name)).toContain("rpt_after_star") }) @@ -341,7 +376,7 @@ describe("columnReach where the engine's schema falls short", () => { '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.fct_orders"] = model("fct_orders", [], "select * from raw.orders", []) value.nodes["model.shop.rpt_star"] = model("rpt_star", ["model.shop.fct_orders"], sql, []) value.exposures["exposure.shop.star_board"] = { name: "star_board", @@ -843,6 +878,20 @@ describe("structuredResult over the cap", () => { expect(Buffer.byteLength(JSON.stringify(out))).toBeLessThanOrEqual(60_000) expect("x".repeat(300_000).startsWith(out.report as string)).toBe(true) }) + + test("a column answer too big for the cap goes first; the reach and the model-level fields stay", () => { + const columnAnswer = { columns: Array.from({ length: 3000 }, (_, i) => ({ model: 0, name: `column_${i}` })) } + const out = structuredResult( + "impact_analysis", + { severity: "HIGH", downstream: ["a"], column_reach: { models: [] }, column_answer: columnAnswer }, + "Impact", + ) as Record + expect(out.column_answer).toBeUndefined() + expect(out.column_answer_error).toContain("for size") + expect(out).toMatchObject({ severity: "HIGH", downstream: ["a"], column_reach: { models: [] }, report: "Impact" }) + expect(out.truncated).toBeUndefined() + expect(Buffer.byteLength(JSON.stringify(out))).toBeLessThanOrEqual(60_000) + }) }) describe("structuredError escaping", () => { diff --git a/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts b/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts index 6fc28fbb6b..7167d69f78 100644 --- a/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts +++ b/packages/opencode/test/altimate/dbt-lineage-helpers.test.ts @@ -13,6 +13,7 @@ import { dbtLineage } from "../../src/altimate/native/dbt/lineage" import { writeFileSync, mkdtempSync, rmSync } from "fs" import { tmpdir } from "os" import { join } from "path" +import { clearLineageCaches } from "../../src/altimate/native/dbt/project-lineage" // --------------------------------------------------------------------------- // Helpers @@ -33,6 +34,7 @@ function writeManifest(dir: string, manifest: Record): string { } afterEach(() => { + clearLineageCaches() for (const dir of tmpDirs) { rmSync(dir, { recursive: true, force: true }) } @@ -346,3 +348,206 @@ describe("dbtLineage: error handling", () => { expect(result.confidence).toBe("low") }) }) + +// --------------------------------------------------------------------------- +// 6. Parents with no declared columns +// --------------------------------------------------------------------------- + +// Jaffle shop's customers and orders over staging models that declare their columns, or don't. +function jaffle(declareStaging: boolean, more: Record = {}) { + const R = (name: string) => `"jaffle"."main"."${name}"` + const node = (type: string, name: string, deps: string[], sql: string, columns: string[]) => ({ + resource_type: type, + unique_id: `${type}.jaffle.${name}`, + name, + alias: name, + database: "jaffle", + schema: "main", + config: { materialized: type === "seed" ? "seed" : "view" }, + depends_on: { nodes: deps }, + ...(sql && { compiled_code: sql }), + columns: Object.fromEntries(columns.map((c) => [c, { name: c }])), + }) + const staging = (cols: string[]) => (declareStaging ? cols : []) + const nodes: Record = { + "seed.jaffle.raw_customers": node("seed", "raw_customers", [], "", ["id", "first_name", "last_name"]), + "seed.jaffle.raw_orders": node("seed", "raw_orders", [], "", ["id", "user_id", "order_date", "status"]), + "seed.jaffle.raw_payments": node("seed", "raw_payments", [], "", ["id", "order_id", "payment_method", "amount"]), + "model.jaffle.stg_customers": node( + "model", + "stg_customers", + ["seed.jaffle.raw_customers"], + `with source as (select * from ${R("raw_customers")}), renamed as (select id as customer_id, first_name, last_name from source) select * from renamed`, + staging(["customer_id", "first_name", "last_name"]), + ), + "model.jaffle.stg_orders": node( + "model", + "stg_orders", + ["seed.jaffle.raw_orders"], + `with source as (select * from ${R("raw_orders")}), renamed as (select id as order_id, user_id as customer_id, order_date, status from source) select * from renamed`, + staging(["order_id", "customer_id", "order_date", "status"]), + ), + "model.jaffle.stg_payments": node( + "model", + "stg_payments", + ["seed.jaffle.raw_payments"], + `with source as (select * from ${R("raw_payments")}), renamed as (select id as payment_id, order_id, payment_method, amount / 100 as amount from source) select * from renamed`, + staging(["payment_id", "order_id", "payment_method", "amount"]), + ), + "model.jaffle.customers": node( + "model", + "customers", + ["model.jaffle.stg_customers", "model.jaffle.stg_orders", "model.jaffle.stg_payments"], + `with customers as (select * from ${R("stg_customers")}), orders as (select * from ${R("stg_orders")}), payments as (select * from ${R("stg_payments")}), +customer_orders as (select customer_id, min(order_date) as first_order, max(order_date) as most_recent_order, count(order_id) as number_of_orders from orders group by customer_id), +customer_payments as (select orders.customer_id, sum(amount) as total_amount from payments left join orders on payments.order_id = orders.order_id group by orders.customer_id), +final as (select customers.customer_id, customers.first_name, customers.last_name, customer_orders.first_order, customer_orders.most_recent_order, customer_orders.number_of_orders, customer_payments.total_amount as customer_lifetime_value +from customers left join customer_orders on customers.customer_id = customer_orders.customer_id left join customer_payments on customers.customer_id = customer_payments.customer_id) +select * from final`, + [], + ), + "model.jaffle.orders": node( + "model", + "orders", + ["model.jaffle.stg_orders", "model.jaffle.stg_payments"], + `with orders as (select * from ${R("stg_orders")}), payments as (select * from ${R("stg_payments")}), +order_payments as (select order_id, sum(case when payment_method = 'credit_card' then amount else 0 end) as credit_card_amount, sum(amount) as total_amount from payments group by order_id), +final as (select orders.order_id, orders.customer_id, orders.order_date, orders.status, order_payments.credit_card_amount, order_payments.total_amount as amount from orders left join order_payments on orders.order_id = order_payments.order_id) +select * from final`, + [], + ), + ...more, + } + const sources = { + "source.jaffle.raw.events": { + resource_type: "source", + name: "events", + source_name: "raw", + database: "jaffle", + schema: "raw", + columns: {}, + }, + } + return { metadata: { adapter_type: "duckdb" }, nodes, sources, exposures: {} } +} + +const outputs = (result: ReturnType) => + Object.keys((result.raw_lineage as any).column_dict ?? {}).sort() + +describe("dbtLineage: parents with no declared columns", () => { + test("a select * over undeclared staging models finds every column, as with them declared", () => { + for (const declared of [true, false]) { + const result = dbtLineage({ manifest_path: writeManifest(makeTmpDir(), jaffle(declared)), model: "customers" }) + expect(outputs(result)).toEqual([ + "customer_id", + "customer_lifetime_value", + "first_name", + "first_order", + "last_name", + "most_recent_order", + "number_of_orders", + ]) + expect(result.confidence).toBe("high") + } + }) + + test("orders.credit_card_amount traces to the payment amount with undeclared staging columns", () => { + const result = dbtLineage({ manifest_path: writeManifest(makeTmpDir(), jaffle(false)), model: "orders" }) + expect((result.raw_lineage as any).column_dict.credit_card_amount).toEqual([ + '"stg_payments"."amount"', + '"stg_payments"."payment_method"', + ]) + }) + + test("an ephemeral parent's own parents give the columns its inlined SQL reads", () => { + const manifest = jaffle(false, { + "model.jaffle.eph": { + resource_type: "model", + name: "eph", + alias: "eph", + config: { materialized: "ephemeral" }, + depends_on: { nodes: ["model.jaffle.stg_customers"] }, + compiled_code: 'select * from "jaffle"."main"."stg_customers"', + columns: {}, + }, + "model.jaffle.after_eph": { + resource_type: "model", + name: "after_eph", + alias: "after_eph", + config: { materialized: "view" }, + depends_on: { nodes: ["model.jaffle.eph"] }, + compiled_code: + 'with __dbt__cte__eph as (select * from "jaffle"."main"."stg_customers") select * from __dbt__cte__eph', + columns: {}, + }, + }) + const result = dbtLineage({ manifest_path: writeManifest(makeTmpDir(), manifest), model: "after_eph" }) + expect(outputs(result)).toEqual(["customer_id", "first_name", "last_name"]) + expect(result.upstream).toEqual(["model.jaffle.eph"]) + }) + + // stg_events selects * from a source that declares no columns, so neither its docs nor its SQL say + // what its columns are. + const unknownEvents = (child: string) => + jaffle(false, { + "model.jaffle.stg_events": { + resource_type: "model", + name: "stg_events", + alias: "stg_events", + config: { materialized: "view" }, + depends_on: { nodes: ["source.jaffle.raw.events"] }, + compiled_code: 'select * from "jaffle"."raw"."events"', + columns: {}, + }, + "model.jaffle.events_child": { + resource_type: "model", + name: "events_child", + alias: "events_child", + config: { materialized: "view" }, + depends_on: { nodes: ["model.jaffle.stg_events", "model.jaffle.stg_customers"] }, + compiled_code: child, + columns: {}, + }, + }) + + test("a select * over a parent whose columns can't be worked out is low, naming the parent", () => { + const result = dbtLineage({ + manifest_path: writeManifest( + makeTmpDir(), + unknownEvents( + 'select e.*, c.first_name from "jaffle"."main"."stg_events" e join "jaffle"."main"."stg_customers" c on e.customer_id = c.customer_id', + ), + ), + model: "events_child", + }) + expect(result.confidence).toBe("low") + expect(result.confidence_factors).toEqual([ + "select * over stg_events, whose columns aren't declared and can't be worked out from its SQL: the columns read through it are missing", + ]) + expect(result.unexpanded_stars).toEqual(["stg_events"]) + expect(outputs(result)).toContain("first_name") + }) + + test("a * inside a function over such a parent is low too, naming it", () => { + const result = dbtLineage({ + manifest_path: writeManifest(makeTmpDir(), unknownEvents('select hash(*) as h from "jaffle"."main"."stg_events"')), + model: "events_child", + }) + expect(result.confidence).toBe("low") + expect(result.unexpanded_stars).toEqual(["stg_events"]) + expect((result.raw_lineage as any).column_dict.h).toEqual([]) + }) + + test("a select * over a CTE that lists such a parent's columns stays high", () => { + const result = dbtLineage({ + manifest_path: writeManifest( + makeTmpDir(), + unknownEvents('with e as (select event_id, kind from "jaffle"."main"."stg_events") select * from e'), + ), + model: "events_child", + }) + expect(result.confidence).toBe("high") + expect(result.unexpanded_stars).toBeUndefined() + expect(outputs(result)).toEqual(["event_id", "kind"]) + }) +}) diff --git a/packages/opencode/test/altimate/impact-analysis-column-answer.test.ts b/packages/opencode/test/altimate/impact-analysis-column-answer.test.ts new file mode 100644 index 0000000000..1c9c6a530d --- /dev/null +++ b/packages/opencode/test/altimate/impact-analysis-column-answer.test.ts @@ -0,0 +1,171 @@ +// impact_analysis returns the column answer (metadata.column_answer) beside column_reach, in both +// directions, and the MCP server passes it on as structured content. +import { afterAll, afterEach, beforeEach, describe, expect, spyOn, test } from "bun:test" +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs" +import { tmpdir } from "node:os" +import { join } from "node:path" +import { initTool } from "./tool-fixture" +import * as Dispatcher from "../../src/altimate/native/dispatcher" +import { ImpactAnalysisTool } from "../../src/altimate/tools/impact-analysis" +import { structuredResult } from "../../src/altimate/mcp-server/server" +import { SessionID, MessageID } from "../../src/session/schema" +import { clearLineageCaches } from "../../src/altimate/native/dbt/project-lineage" + +const ctx = { + sessionID: SessionID.make("ses_test"), + messageID: MessageID.make("msg_test"), + callID: "call_test", + agent: "build", + abort: AbortSignal.any([]), + messages: [], + metadata: () => {}, + ask: async () => {}, +} + +const dirs: string[] = [] +let spy: ReturnType | undefined +const telemetryBefore = process.env.ALTIMATE_TELEMETRY_DISABLED +beforeEach(() => { + process.env.ALTIMATE_TELEMETRY_DISABLED = "true" +}) +afterEach(() => { + clearLineageCaches() + spy?.mockRestore() + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }) +}) +afterAll(() => { + if (telemetryBefore === undefined) delete process.env.ALTIMATE_TELEMETRY_DISABLED + else process.env.ALTIMATE_TELEMETRY_DISABLED = telemetryBefore +}) + +const model = (name: string, deps: string[], sql: string) => ({ + resource_type: "model", + unique_id: `model.shop.${name}`, + name, + alias: name, + database: "shop", + schema: "main", + original_file_path: `models/${name}.sql`, + depends_on: { nodes: deps.map((d) => `model.shop.${d}`) }, + config: { materialized: "table" }, + compiled_code: sql, + columns: {}, +}) + +// orders.amount → fct.revenue (renamed) → mart.total (summed); rpt filters on fct.revenue. +function setup(exposures: Record = {}) { + const nodes = { + "model.shop.orders": model("orders", [], "select 1 as order_id, 2.5 as amount"), + "model.shop.fct": model("fct", ["orders"], "select order_id, amount as revenue from shop.main.orders"), + "model.shop.mart": model("mart", ["fct"], "select order_id, revenue * 1.1 as total from shop.main.fct"), + "model.shop.rpt": model("rpt", ["fct"], "select order_id from shop.main.fct where revenue > 10"), + } + const dir = mkdtempSync(join(tmpdir(), "impact-answer-")) + dirs.push(dir) + mkdirSync(join(dir, "target")) + const file = join(dir, "target", "manifest.json") + writeFileSync(file, JSON.stringify({ metadata: { adapter_type: "duckdb" }, nodes, sources: {}, exposures })) + const models = Object.values(nodes).map((n) => ({ + name: n.name, + unique_id: n.unique_id, + depends_on: n.depends_on.nodes, + materialized: "table", + })) + spy = spyOn(Dispatcher, "call").mockImplementation((async (method: string) => { + if (method === "dbt.manifest") return { models, model_count: models.length, test_count: 0, tests: [] } + throw new Error(`No mock for ${method}`) + }) as any) + return file +} + +const run = async (args: Record) => { + const tool = await initTool(ImpactAnalysisTool) + return tool.execute({ change_type: "modify", ...args } as any, ctx) +} + +describe("impact_analysis: column answer", () => { + test("downstream by default, beside column_reach, with one line of text", async () => { + const manifest_path = setup() + const result = await run({ model: "fct", column: "revenue", manifest_path }) + const answer = result.metadata.column_answer + expect(answer.version).toBe(1) + expect(answer.direction).toBe("downstream") + expect(answer.counts).toMatchObject({ + columns: 1, + models: 1, + may_carry_models: 1, + cleared_models: 0, + model_level: 2, + }) + expect(answer.also_reads).toEqual([ + { model: answer.nodes.findIndex((n: any) => n.name === "rpt"), columns: ["revenue"], downstream: 0 }, + ]) + expect(result.metadata.column_reach.models.map((m: any) => m.name)).toEqual(["mart"]) + expect(result.output).toContain("Column Reach: fct.revenue") + expect(result.output.trimEnd().split("\n").at(-1)).toStartWith("Column impact: 1 column in 1 model") + }) + + test("upstream traces where the column comes from", async () => { + const manifest_path = setup() + const result = await run({ model: "mart", column: "total", manifest_path, direction: "upstream" }) + const answer = result.metadata.column_answer + expect(answer.direction).toBe("upstream") + const names = answer.columns.map((c: any) => `${answer.nodes[c.model].name}.${c.column}`) + expect(names).toEqual(["mart.total", "fct.revenue", "orders.amount"]) + expect(result.output.trimEnd().split("\n").at(-1)).toStartWith("Column trace (upstream): 2 columns in 2 models") + }) + + test("no column, no answer", async () => { + const manifest_path = setup() + const result = await run({ model: "fct", manifest_path }) + expect(result.metadata.column_answer).toBeUndefined() + }) + + test("says how many exposures the project declares, with or without a column", async () => { + const none = await run({ model: "fct", manifest_path: setup() }) + expect(none.metadata.exposures_declared).toBe(0) + expect(none.metadata.exposures).toEqual([]) + const manifest_path = setup({ + "exposure.shop.board": { + name: "board", + label: "Board", + type: "dashboard", + url: "https://bi.example.com/1", + owner: { name: "Data team" }, + depends_on: { nodes: ["model.shop.mart"] }, + }, + "exposure.shop.up": { name: "up", depends_on: { nodes: ["model.shop.orders"] } }, + }) + const modelLevel = await run({ model: "fct", manifest_path }) + expect(modelLevel.metadata.exposures_declared).toBe(2) + expect(modelLevel.metadata.exposures).toEqual([ + { unique_id: "exposure.shop.board", name: "board", label: "Board", type: "dashboard", url: "https://bi.example.com/1", owner: "Data team" }, + ]) + const column = await run({ model: "fct", column: "revenue", manifest_path }) + expect(column.metadata.exposures_declared).toBe(2) + expect(column.metadata.column_answer.consumers.exposures.map((c: any) => [c.name, c.tier])).toEqual([["board", "carries"]]) + }) + + test("a column the model doesn't have is a column error, with no column reach or answer", async () => { + const manifest_path = setup({ + "exposure.shop.fct_board": { name: "fct_board", label: "Fct board", depends_on: { nodes: ["model.shop.fct"] } }, + }) + const result = await run({ model: "fct", column: "not_a_column", manifest_path }) + const error = "fct has no column not_a_column; its columns are order_id, revenue" + expect(result.metadata.column_reach).toBeUndefined() + expect(result.metadata.column_answer).toBeUndefined() + expect(result.metadata.column_reach_error).toBe(error) + expect(result.metadata.column_answer_error).toBe(error) + expect(result.output).toContain(`Column Reach: not traced (${error}); the model-level impact above stands`) + }) + + test("the MCP structured result carries it", async () => { + const manifest_path = setup() + const result = await run({ model: "fct", column: "revenue", manifest_path }) + const structured = structuredResult("impact_analysis", result.metadata, result.output) + expect(structured.truncated).toBeUndefined() + expect((structured.column_answer as any)?.version).toBe(1) + expect(structured.column_answer).toEqual(result.metadata.column_answer) + expect(structured.column_reach).toEqual(result.metadata.column_reach) + }) +})