diff --git a/backend/.sqlx/query-3162283a5066660e2f31575344083bf1585004f9fddee904ab176ac85ba3c094.json b/backend/.sqlx/query-3162283a5066660e2f31575344083bf1585004f9fddee904ab176ac85ba3c094.json new file mode 100644 index 0000000000..0937a312d1 --- /dev/null +++ b/backend/.sqlx/query-3162283a5066660e2f31575344083bf1585004f9fddee904ab176ac85ba3c094.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000',\n 'model.p.stock', 'sku', 'model.p.stock_daily', 'sku', 'copy')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "3162283a5066660e2f31575344083bf1585004f9fddee904ab176ac85ba3c094" +} diff --git a/backend/.sqlx/query-a87dcb8f812b555564d71afcb7141bdac5c52e50e697c6714b91b9fd3155572c.json b/backend/.sqlx/query-a87dcb8f812b555564d71afcb7141bdac5c52e50e697c6714b91b9fd3155572c.json new file mode 100644 index 0000000000..313dec3743 --- /dev/null +++ b/backend/.sqlx/query-a87dcb8f812b555564d71afcb7141bdac5c52e50e697c6714b91b9fd3155572c.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, tags)\n VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.stock',\n 'model', 'stock', 'u/a/wh/analytics/stock', '{}'),\n ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.stock_daily',\n 'model', 'stock_daily', 'u/a/wh/analytics/stock_daily', '{}')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "a87dcb8f812b555564d71afcb7141bdac5c52e50e697c6714b91b9fd3155572c" +} diff --git a/backend/.sqlx/query-cec32d42bceaf500ffaf47102cd14db3228451391ce11b8a14d7c11d4ee304ba.json b/backend/.sqlx/query-cec32d42bceaf500ffaf47102cd14db3228451391ce11b8a14d7c11d4ee304ba.json new file mode 100644 index 0000000000..2ef26b366b --- /dev/null +++ b/backend/.sqlx/query-cec32d42bceaf500ffaf47102cd14db3228451391ce11b8a14d7c11d4ee304ba.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n SELECT $1, $2, $3, '00000000-0000-0000-0000-000000000000',\n 'model.p.raw_orders', 'c' || i, 'model.p.orders', 'c' || i, 'copy'\n FROM generate_series(1, 100000) i", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "cec32d42bceaf500ffaf47102cd14db3228451391ce11b8a14d7c11d4ee304ba" +} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 4b86013e7d..89e3c7c6b2 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -1037,6 +1037,11 @@ impl ColumnLineageQuery { "at least one asset_path is required".to_string(), )); } + if asset_paths.len() > MAX_ASKED_RELATIONS { + return Err(windmill_common::error::Error::BadRequest(format!( + "at most {MAX_ASKED_RELATIONS} asset_path values may be asked about at once" + ))); + } Ok(ColumnLineageQuery { asset_paths, dbt_script_hash }) } } @@ -1066,14 +1071,22 @@ const MAX_TRACE_EDGES: usize = 5_000; /// bound says. const MAX_OWNER_ROUNDS: usize = 8; -/// How many edges one trace may HOLD while walking, as opposed to answer with. -/// A project's edges arrive whole — the walk decides what is in the component, -/// so a `LIMIT` on the fetch would cut an arbitrary set that need not even -/// contain the asked-for relation — and a project is bounded at ingest by -/// `MAX_COLUMN_EDGES`, which is 200k. Rounds are what this bounds: reading a -/// second project's worth on top of an already outsized first one buys nothing, -/// since the walk is going to stop at `MAX_TRACE_EDGES` regardless. -const MAX_HELD_EDGES: usize = 100_000; +/// How many edges a trace may already hold before it stops looking for projects +/// it has not read. NOT a bound on what one trace fetches: the seeds' own +/// projects are read whole whatever their size, because reading them IS the +/// answer, and a project's edges arrive whole in any case — the walk is what +/// decides which of them are in the component, so a `LIMIT` would cut a set that +/// need not contain the asked-for relation at all. What a single fetch is +/// bounded by is the ingest's `MAX_COLUMN_EDGES` per version. This bounds the +/// EXPANSION on top of that: reading a further project's worth once a trace is +/// already this size buys nothing the walk will not cut at `MAX_TRACE_EDGES`. +const EXPANSION_EDGE_BUDGET: usize = 100_000; + +/// How many relations one request may ask about. Generous: the pipeline page +/// sends every dbt relation the selection's own producer lineage reaches, which +/// is a handful even in a large folder. It exists so a crafted request cannot +/// hand `= ANY($2)` an arbitrarily long array. +const MAX_ASKED_RELATIONS: usize = 1_000; /// A stored project graph: a deployed version, or one job's snapshot of it. /// `script_hash` is NULL for an editor buffer's parse, which names no version. @@ -1140,8 +1153,9 @@ pub async fn dbt_column_lineage_for( let mut answer: Vec = Vec::new(); let mut pending: Vec = seeds.iter().cloned().collect(); let mut truncated = false; + let mut rounds = 0usize; - for _ in 0..MAX_OWNER_ROUNDS { + loop { if pending.is_empty() { break; } @@ -1232,9 +1246,22 @@ pub async fn dbt_column_lineage_for( .map(|o| (o.script_path, o.script_hash, o.job_id)) .filter(|k| read.insert(k.clone())) .collect(); + // Nothing left this caller may read and has not read: the component is + // whole, however many relations were still waiting to be asked about. + // Their owners are projects already in hand. if fresh.is_empty() { break; } + // From here a project exists that this answer will not contain, so the + // two stops below are cuts and are reported as such. Deciding it after + // the owners query rather than before is what keeps a big project's + // small component from being called truncated: `pending` alone only says + // a relation has not been ASKED about, not that anything was left out. + rounds += 1; + if rounds > MAX_OWNER_ROUNDS || edges.len() >= EXPANSION_EDGE_BUDGET { + truncated = true; + break; + } let fresh_paths: Vec = fresh.iter().map(|k| k.0.clone()).collect(); let fresh_hashes: Vec> = fresh.iter().map(|k| k.1).collect(); let fresh_jobs: Vec = fresh.iter().map(|k| k.2).collect(); @@ -1313,20 +1340,9 @@ pub async fn dbt_column_lineage_for( .collect::>() .into_iter() .collect(); - // Stop discovering projects once the held set is outsized. The tail - // below reports what that leaves unresolved, and reports nothing when - // the walk had already reached everything. - if edges.len() >= MAX_HELD_EDGES { - break; - } } tx.commit().await?; - // Out of rounds with relations still unresolved: more of the component - // exists, which is what hitting the edge bound also means. - Ok(Json(ColumnLineageResponse { - truncated: truncated || !pending.is_empty(), - edges: answer, - })) + Ok(Json(ColumnLineageResponse { truncated, edges: answer })) } struct WalkedComponent { diff --git a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs index 75299f4d9f..ebeee49ecc 100644 --- a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs +++ b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs @@ -940,3 +940,74 @@ async fn a_component_past_the_bound_is_cut_and_says_so(db: Pool) { assert_eq!(body["edges"].as_array().unwrap().len(), 5000); assert_eq!(body["truncated"], serde_json::json!(true)); } + +/// The other side of the same flag: a project big enough to stop the expansion +/// still answers a small component WHOLE, and must not claim it was cut. +/// +/// Reaching the size that stops expansion says nothing on its own — nor does a +/// relation whose owners have not been asked about, since those owners are +/// usually the project already in hand. Only a project this caller may read and +/// this answer does not contain is a cut. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn a_whole_component_in_a_big_project_is_not_called_cut(db: Pool) { + seed(&db, uuid::Uuid::from_u128(7)).await; + seed_deployed_orders(&db).await; + // A second family, two relations and one edge, sharing no column with the + // first — the whole of what a selection on it should return. + sqlx::query!( + "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id, + resource_type, name, asset_path, tags) + VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.stock', + 'model', 'stock', 'u/a/wh/analytics/stock', '{}'), + ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.stock_daily', + 'model', 'stock_daily', 'u/a/wh/analytics/stock_daily', '{}')", + WS, + PATH, + HASH, + ) + .execute(&db) + .await + .unwrap(); + sqlx::query!( + "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, + parent_unique_id, parent_column, child_unique_id, + child_column, lineage_kind) + VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', + 'model.p.stock', 'sku', 'model.p.stock_daily', 'sku', 'copy')", + WS, + PATH, + HASH, + ) + .execute(&db) + .await + .unwrap(); + // And enough unrelated families beside them to pass the expansion budget. + sqlx::query!( + "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, + parent_unique_id, parent_column, child_unique_id, + child_column, lineage_kind) + SELECT $1, $2, $3, '00000000-0000-0000-0000-000000000000', + 'model.p.raw_orders', 'c' || i, 'model.p.orders', 'c' || i, 'copy' + FROM generate_series(1, 100000) i", + WS, + PATH, + HASH, + ) + .execute(&db) + .await + .unwrap(); + + let admin = ApiAuthed { is_admin: true, ..outsider() }; + let body = column_lineage(&db, &admin, &["u/a/wh/analytics/stock_daily"], None).await; + assert_eq!( + body["edges"], + serde_json::json!([{ + "from_asset_path": "u/a/wh/analytics/stock", + "from_column": "sku", + "to_asset_path": "u/a/wh/analytics/stock_daily", + "to_column": "sku", + "kind": "copy", + }]), + ); + assert_eq!(body["truncated"], serde_json::json!(false)); +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index e2bb8c6a1a..4abd5f4dff 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -24479,9 +24479,13 @@ paths: required: true description: > The `dbt://` relations whose lineage to return. Repeated, once per - relation, and answered as one union. + relation, and answered as one union. At least one, and at most 1000 — + a request naming none, or more than that, is refused rather than + answered with an empty component. schema: type: array + maxItems: 1000 + minItems: 1 items: type: string - name: dbt_script_hash @@ -24728,9 +24732,13 @@ paths: required: true description: > The `dbt://` relations whose lineage to return. Repeated, once per - relation, and answered as one union. + relation, and answered as one union. At least one, and at most 1000 — + a request naming none, or more than that, is refused rather than + answered with an empty component. schema: type: array + maxItems: 1000 + minItems: 1 items: type: string responses: diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 926948a676..5fec1e42f7 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1689,11 +1689,23 @@ served — a synthetic 3000-model project whose models share a column returns 58 direct edges and 7.3MB. The walk is breadth-first from the asked-for relations and stops at 5000 edges, so what survives is the part nearest the selection rather than an arbitrary slice, and `truncated` says so: a trace that stops short is -otherwise indistinguishable from one that ends. The FETCH is not bounded the same -way — a project's edges arrive whole, because the walk is what decides which of -them are in the component, and a `LIMIT` would cut a set that need not contain -the asked-for relation at all. What bounds it is the ingest's own cap per version -plus a stop on discovering further projects once the held set is outsized. +otherwise indistinguishable from one that ends. + +`truncated` is set only where a project this caller may read is left unread, or +where the walk itself was cut. Neither the expansion's size budget nor a relation +whose owners were never asked about is evidence on its own: a big project's small +component is answered whole, and the relations still waiting to be asked about at +the end are almost always owned by the project already in hand. The order the +rounds run in is what makes that decidable — the owners query comes before the +budget check, so the stop happens with a named unread project rather than a +suspicion of one. + +The FETCH is not bounded the way the answer is: a project's edges arrive whole, +because the walk is what decides which of them are in the component, and a +`LIMIT` would cut a set that need not contain the asked-for relation at all. The +seeds' own projects are therefore read whatever their size — reading them is the +answer. What bounds a single fetch is the ingest's own cap per version; the +budget bounds only the expansion on top of it. The walk is in Rust rather than a recursive CTE. `EXPLAIN ANALYZE` on that same project measured 1243ms against 59ms for the query alone: a CTE has no index to diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 005e3a3ffb..4b252b9338 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -163,6 +163,9 @@ /** The lineage reaches past what the graph holds: the API cut it at the * part nearest the selection. */ selectionColumnTruncated?: boolean + /** That trace could not be fetched. Distinguished from an empty one: a + * project without the analysis pass draws nothing either. */ + selectionColumnFailed?: boolean /** dbt provenance of the selected relation, when a dbt project * materializes it — carries the model's own SQL. */ selectionDbt?: DbtAssetProvenance @@ -298,6 +301,7 @@ selectionColumnGraph, selectionColumnLoading = false, selectionColumnTruncated = false, + selectionColumnFailed = false, selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, @@ -1250,13 +1254,14 @@ targetLabel={selection.path} loading={selectionColumnLoading} truncated={selectionColumnTruncated} + failed={selectionColumnFailed} />
{/key} - {:else if selectionDbt && (selectionDbt.raw_code || selectionDbtHasColumns || selectionColumnNodes.length > 0 || selectionColumnLoading)} + {:else if selectionDbt && (selectionDbt.raw_code || selectionDbtHasColumns || selectionColumnNodes.length > 0 || selectionColumnLoading || selectionColumnFailed)} +{#snippet failure()} +
+ Part of this column lineage could not be loaded, so the trace may be incomplete. +
+{/snippet} + {#if loading && nodes.length === 0}
Loading column lineage
-{:else if graph && nodes.length > 0} +{:else if nodes.length === 0} + {#if failed} +
{@render failure()}
+ {/if} +{:else if graph}
+ {#if failed} + {@render failure()} + {/if} {#if truncated}
Showing the part of the trace nearest this relation. The lineage reaches further than one diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 721fafb6be..b71a51ea0a 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -80,6 +80,7 @@ selectionColumnGraph, selectionColumnLoading = false, selectionColumnTruncated = false, + selectionColumnFailed = false, selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, @@ -191,6 +192,9 @@ selectionColumnLoading?: boolean /** That trace cut at the part nearest the selection. */ selectionColumnTruncated?: boolean + /** That trace could not be fetched. Distinguished from an empty one: a + * project without the analysis pass draws nothing either. */ + selectionColumnFailed?: boolean /** dbt provenance of the selected relation — carries its SQL. */ selectionDbt?: DbtAssetProvenance schemaCanEvolve?: boolean @@ -525,6 +529,7 @@ {selectionColumnGraph} {selectionColumnLoading} {selectionColumnTruncated} + {selectionColumnFailed} {selectionDbt} {schemaCanEvolve} {selectionForkMaterialization} diff --git a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts index 189ba05b5d..477ed630b1 100644 --- a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts +++ b/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts @@ -20,6 +20,9 @@ export type DbtColumnLineageState = { /** The component reaches past what `graph` holds — the API cut it at the * part nearest the selection. */ readonly truncated: boolean + /** The request failed, so `graph` is empty for a reason that is not "this + * project has no column lineage". */ + readonly failed: boolean } function fetchLineage( @@ -56,10 +59,18 @@ export function useDbtColumnLineage(args: { assetPaths: () => string[] /** The graph on screen, so the lineage describes the same project. */ pin?: () => DbtGraphPin | undefined + /** Which fetch of that graph is on screen. It changes when the view goes and + * gets the graph again — a Refresh, a deploy — and asking again is the whole + * point: the relation, the pin and the seeds are all unchanged by a + * redeploy, so without this the pane would pair a freshly fetched model's + * SQL and columns with the edges of the version before it. It is also what + * retries a request that failed. */ + generation?: () => unknown }): DbtColumnLineageState { let graph = $state(EMPTY_COLUMN_GRAPH) let loading = $state(false) let truncated = $state(false) + let failed = $state(false) // The question the state in hand answers, and a counter deciding which answer // is still wanted. Neither is a cache of edges: the API returns a whole @@ -71,36 +82,55 @@ export function useDbtColumnLineage(args: { const workspace = args.workspace() const paths = [...new Set(args.assetPaths())].sort() const pin = args.pin?.() - const question = JSON.stringify([workspace, pin?.jobId, pin?.scriptHash, paths]) + const question = JSON.stringify([ + workspace, + pin?.jobId, + pin?.scriptHash, + args.generation?.() ?? null, + paths + ]) // A selection re-derived from a graph that polled is the same question. Not // asking it again is what keeps a run page from refetching a component's - // worth of edges every poll to redraw what is already on screen. + // worth of edges every poll to redraw what is already on screen — while a + // graph the view deliberately went and fetched moves `generation`, so that + // IS a new question. if (question === asked) return asked = question const id = ++latest if (!workspace || paths.length === 0) { graph = EMPTY_COLUMN_GRAPH truncated = false + failed = false loading = false return } + // The previous question's failure is not this one's. Cleared as the request + // goes out rather than when it lands, or a retry keeps saying the trace is + // incomplete while it is being fetched, and a new selection inherits the + // last one's failure until its own answer arrives. + failed = false loading = true fetchLineage(workspace, paths, pin).then( (r) => { if (id !== latest) return graph = buildDbtColumnGraph(r?.edges ?? []) truncated = r?.truncated ?? false + failed = false loading = false }, // Lineage annotates a graph that renders without it, so a failed fetch // leaves that branch unexpanded rather than putting an error over the - // model. Not retried on its own: the effect reruns whenever the canvas - // does, and a failing endpoint would then be asked once per redraw. - // Selecting another node and back asks again. + // model — but it SAYS so, because an empty trace is what a project + // without the analysis pass looks like, and the two must not read + // alike. Not retried on its own: the effect reruns whenever the canvas + // redraws, and a failing endpoint would then be asked once per redraw. + // Selecting another node and back asks again, and where a caller + // passes `generation`, so does going and fetching the graph. () => { if (id !== latest) return graph = EMPTY_COLUMN_GRAPH truncated = false + failed = true loading = false } ) @@ -115,6 +145,9 @@ export function useDbtColumnLineage(args: { }, get truncated() { return truncated + }, + get failed() { + return failed } } } diff --git a/frontend/src/lib/components/dbt/DbtEditor.svelte b/frontend/src/lib/components/dbt/DbtEditor.svelte index 469536f305..4b30f8fa28 100644 --- a/frontend/src/lib/components/dbt/DbtEditor.svelte +++ b/frontend/src/lib/components/dbt/DbtEditor.svelte @@ -586,6 +586,7 @@ columnGraph={selectionColumnGraph} columnLoading={columnLineage.loading} columnTruncated={columnLineage.truncated} + columnFailed={columnLineage.failed} onOpenFile={open} onClose={() => (graphSelection = undefined)} /> diff --git a/frontend/src/lib/components/dbt/DbtModelDetails.svelte b/frontend/src/lib/components/dbt/DbtModelDetails.svelte index d3dd25d04a..5863d00a59 100644 --- a/frontend/src/lib/components/dbt/DbtModelDetails.svelte +++ b/frontend/src/lib/components/dbt/DbtModelDetails.svelte @@ -43,6 +43,7 @@ columnGraph, columnLoading = false, columnTruncated = false, + columnFailed = false, onOpenFile, onClose }: { @@ -57,6 +58,7 @@ columnGraph?: ColumnLineageGraph columnLoading?: boolean columnTruncated?: boolean + columnFailed?: boolean onOpenFile?: (path: string) => void onClose?: () => void } = $props() @@ -273,6 +275,7 @@ targetLabel={dbt.unique_id} loading={columnLoading} truncated={columnTruncated} + failed={columnFailed} /> {#if showRows && preview} diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 4dc68cd9ab..b4ddb8350c 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -2019,9 +2019,20 @@ } return [...paths] }) + // Which fetch of the folder graph is on screen. `graphRes.current` is a new + // object per fetch, so this moves on a Refresh, a deploy and a folder switch + // — and on nothing else, which is what keeps a keystroke in the editor from + // re-asking. Without it a redeploy would leave the pane pairing the new + // version's SQL and columns with the old version's edges, since the relation, + // the pin and the seeds are all unchanged by one. + let graphGeneration = $state(0) + $effect(() => { + if (graphRes.current) untrack(() => graphGeneration++) + }) const dbtColumnLineage = useDbtColumnLineage({ workspace: () => $workspaceStore, - assetPaths: () => dbtSeedPaths + assetPaths: () => dbtSeedPaths, + generation: () => graphGeneration }) // One graph across both, so a trace crosses the dbt/ducklake boundary in // either direction rather than stopping at it. @@ -2630,6 +2641,7 @@ selectionColumnGraph={pe.activeDraft ? EMPTY_COLUMN_GRAPH : columnGraph} selectionColumnLoading={dbtColumnLineage.loading} selectionColumnTruncated={dbtColumnLineage.truncated} + selectionColumnFailed={dbtColumnLineage.failed} {schemaCanEvolve} {selectionForkMaterialization} {schemaContractContext}