From 3dcd3949a14199b106506994ea31ca3de7e636b3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 6 Jul 2026 20:12:12 +0200 Subject: [PATCH] feat(pipelines): auto-derive cascade edges from ducklake/s3 reads (+ muted-read badge) (#9963) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(pipelines): auto-derive cascade trigger edges from ducklake/s3 reads Within a `// pipeline`, a read of a ducklake table or s3 object now auto-wires its cascade trigger edge straight from the FROM clause, so `// on ` is only needed for edges inference can't see (dynamic SQL) or to carry per-edge opts. Two opt-outs: `// mute ` suppresses a single derived edge (a lookup / SCD input read every run but not cascaded on), and `// mute all` opts the script out of derivation entirely (back to explicit-`// on`-only). Explicit `// on` still wins the dedup. Scoped to ducklake + s3 reads; resource/datatable/volume stay explicit. Read-write (RW) and write inputs are excluded so a self-referential merge can't loop-trigger itself; ambiguous (None) access is skipped. - parser: `mute` / `mute_all` in PipelineAnnotations (Rust + TS mirror) - deploy: derive_pipeline_asset_trigger_refs → script_trigger rows - frontend: resolveGraph mirrors derivation for the live edit-mode canvas - tests: shared parity corpus + derive-helper units + resolveGraph overlays Co-Authored-By: Claude Opus 4.8 (1M context) * feat(pipelines): mark auto-derived cascade edges with a persisted derived flag + "auto" badge Persist script_trigger.derived (deploy: true for ducklake/s3-read derivation, false for explicit // on) and return it from the asset-graph endpoint so the canvas renders a Sparkles "auto" badge on auto-wired edges — the inference is now visible on both the deployed graph and the live edit canvas, not just implied. Dispatch (fetch_subscribers) ignores the flag, so a derived edge fires identically to an explicit // on. Also copy derived in the workspace-clone trigger copy, and backfill muteAssets/muteAll into two empty PipelineAnnotations literals the base commit left stale (check:fast). Co-Authored-By: Claude Opus 4.8 (1M context) * fix(pipelines): derive cascade edge from effective (alt-fallback) asset access derive_pipeline_asset_trigger_refs gated on the raw parser access_type, but the persisted asset.usage_access_type and the frontend canvas both use access_type.or(alt_access_type). An ambiguous parse with a manual read override was persisted/drawn as a read yet derived no edge, so the auto edge silently vanished on deploy. Gate on the effective access type for parity. Co-Authored-By: Claude Opus 4.8 (1M context) * feat(pipelines): badge muted reads instead of auto-derived edges Auto-derivation is the default now, so badging every derived cascade edge is noise. Drop the "auto" badge and the persisted `script_trigger.derived` flag (migration + insert param + graph field + clone copy) that only powered it, and instead badge the exception: a ducklake/s3 asset a script reads but does NOT cascade — `// mute ` / `// mute all`. `computeMutedReadKeys` marks a read-only ('r') supported read with no cascade trigger and no self-write; the canvas renders a bell-off "muted" badge on that read edge. Also fixes two review parity nits: - TS `// on` parser now strips trailing `key=value` opts (e.g. `debounce=60s`) like the Rust `split_trailing_kv_opts`, so the ref dedups against inference. - A `// materialize` producer reading its own target is upgraded to `rw` (deploy) / excluded via the materialize write refs (canvas), so it neither self-cascades nor shows as a muted read. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(pipelines): drop redundant // on for auto-derived reads; gate muted badge to pipeline scripts - Templates no longer scaffold `// on ` for a ducklake/s3 input the body reads — the read auto-wires the cascade now that derivation is the default. Kept for datatable/resource (not auto-derived) and native triggers. The discoverability hint now mentions `// mute` (the newly relevant annotation). - computeMutedReadKeys only badges reads by `// pipeline` scripts. A plain script or flow reading a ducklake/s3 asset never had an auto trigger to suppress, so it must render as ordinary lineage, not "muted" (Codex review). Co-Authored-By: Claude Opus 4.8 (1M context) * fix(pipelines): only drop template // on when the body actually reads the input The redundant-`// on` removal assumed the generated body reads the ducklake/s3 input, but postgres/bash/generic bodies (and `data_upload`, which reads the picker file) ignore `input` — dropping `// on` there left the asset-created script with no cascade at all. Gate the drop on READS_INPUT_LANGS (bun/deno/python/duckdb) so non-reading templates keep the explicit trigger. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../windmill-parser/src/asset_parser.rs | 27 +++ .../tests/fixtures/pipeline_annotations.json | 104 ++++++++++ .../tests/pipeline_annotations_parity.rs | 20 ++ backend/windmill-api-scripts/src/scripts.rs | 77 ++++++- backend/windmill-common/src/assets.rs | 190 ++++++++++++++++++ .../assets/AssetGraph/AssetGraphCanvas.svelte | 22 +- .../AssetGraph/AssetGraphDetailsPane.svelte | 4 +- .../assets/AssetGraph/AssetGraphEdge.svelte | 31 ++- .../parsePipelineAnnotations.parity.test.ts | 16 +- .../parsePipelineAnnotations.test.ts | 7 + .../AssetGraph/parsePipelineAnnotations.ts | 49 ++++- .../AssetGraph/pipelineTemplates.test.ts | 67 ++++++ .../assets/AssetGraph/pipelineTemplates.ts | 30 ++- .../assets/AssetGraph/resolveGraph.test.ts | 168 +++++++++++++++- .../assets/AssetGraph/resolveGraph.ts | 153 +++++++++++++- .../(logged)/pipeline/[folder]/+page.svelte | 4 +- 16 files changed, 950 insertions(+), 19 deletions(-) diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 045cac87e6..bdce9ec764 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -528,6 +528,16 @@ pub struct PipelineAnnotations { pub column_lineage: Vec, pub macros: bool, pub use_libs: Vec, + // `// mute ` — suppress the auto-derived cascade edge for a read + // that would otherwise trigger this script (a lookup / slowly-changing + // dimension you read every run but don't want to re-run on). Only Asset + // specs are stored; native trigger kinds are never auto-derived, so + // muting them is meaningless. + pub mute: Vec, + // `// mute all` — opt out of auto-derivation entirely for this script. + // Falls back to explicit-`// on`-only semantics. Explicit `// on` edges + // are unaffected. + pub mute_all: bool, } impl ParseAssetsOutput { @@ -956,6 +966,23 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } + // `// mute all` opts out of auto-derived cascade edges entirely; + // `// mute ` suppresses the one edge. Only asset refs are + // muteable — native trigger kinds are never auto-derived. Checked + // before the generic `on`/asset shorthand (a complete word, so + // prose like `// muted for now` never matches). + if let Some(after_kw) = consume_keyword(rest, "mute") { + let arg = after_kw.trim(); + if arg == "all" { + out.mute_all = true; + } else if let Some(spec @ TriggerSpec::Asset { .. }) = parse_trigger_spec(arg) { + if !out.mute.contains(&spec) { + out.mute.push(spec); + } + } + continue; + } + // `data_test` is checked before `on`/asset shorthands and is a complete // word (so it never collides with the `// test:` CI annotation, which // has no whitespace after `test`). Accumulates — every well-formed line diff --git a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json index cabfe20ddf..226f32914c 100644 --- a/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json +++ b/backend/parsers/windmill-parser/tests/fixtures/pipeline_annotations.json @@ -769,5 +769,109 @@ "tag": null, "retry": null } + }, + { + "name": "mute suppresses a single ducklake read edge", + "code": "// pipeline\n// mute ducklake://main.orders\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "mute": [ + "ducklake:main.orders" + ] + } + }, + { + "name": "mute all opts out of all auto-derivation", + "code": "// pipeline\n// mute all\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "mute_all": true + } + }, + { + "name": "mute accumulates in order and dedups", + "code": "// pipeline\n// mute ducklake://main.a\n// mute s3://raw/b\n// mute ducklake://main.a\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "mute": [ + "ducklake:main.a", + "s3object:raw/b" + ] + } + }, + { + "name": "mute of a native trigger kind is dropped (only assets are muteable)", + "code": "// pipeline\n// mute kafka\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null + } + }, + { + "name": "mute prose without an asset ref never false-positives", + "code": "// pipeline\n// muted for now\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null + } + }, + { + "name": "mute all coexists with explicit on edges", + "code": "// pipeline\n// mute all\n// on ducklake://main.orders\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [ + "ducklake:main.orders" + ], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null, + "mute_all": true + } + }, + { + "name": "on asset ref strips trailing key=value opts", + "code": "// pipeline\n// on ducklake://main.orders debounce=60s\nselect 1", + "expected": { + "in_pipeline": true, + "asset_triggers": [ + "ducklake:main.orders" + ], + "native_triggers": [], + "partition": null, + "freshness": null, + "tag": null, + "retry": null + } } ] diff --git a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs index 7adb07b55c..fff818edb2 100644 --- a/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs +++ b/backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs @@ -53,6 +53,13 @@ struct Expected { // `// use ` accumulation, declaration order, deduped. Absent === []. #[serde(default)] use_libs: Vec, + // `// mute ` accumulation as `kind:path`, declaration order, deduped. + // Absent === []. + #[serde(default)] + mute: Vec, + // `// mute all` marker. Absent === false. + #[serde(default)] + mute_all: bool, } #[derive(Deserialize)] @@ -251,5 +258,18 @@ fn pipeline_annotation_fixtures_match() { assert_eq!(got.macros, f.expected.macros, "{ctx}: macros"); assert_eq!(got.use_libs, f.expected.use_libs, "{ctx}: use_libs"); + + let mute: Vec = got + .mute + .iter() + .filter_map(|t| match t { + TriggerSpec::Asset { asset_kind, path, .. } => { + Some(format!("{}:{}", kind_str(*asset_kind), path)) + } + _ => None, + }) + .collect(); + assert_eq!(mute, f.expected.mute, "{ctx}: mute"); + assert_eq!(got.mute_all, f.expected.mute_all, "{ctx}: mute_all"); } } diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index d7a461dafa..bfc7dd1ac7 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -45,8 +45,9 @@ use windmill_dep_map::scoped_dependency_map::ScopedDependencyMap; use windmill_common::{ assets::{ clear_script_triggers, clear_static_asset_usage, clear_static_asset_usage_by_script_hash, - insert_script_trigger, parse_duration_secs, parse_pipeline_annotations, - replace_static_asset_usage, trigger_spec_to_row, AssetUsageKind, TriggerSpec, + derive_pipeline_asset_trigger_refs, insert_script_trigger, parse_duration_secs, + parse_pipeline_annotations, replace_static_asset_usage, trigger_spec_to_row, + AssetUsageKind, ScriptTriggerKind, TriggerSpec, }, error::{self, to_anyhow}, min_version::{MIN_VERSION_SUPPORTS_DEBOUNCING, MIN_VERSION_SUPPORTS_DEBOUNCING_V2}, @@ -1457,11 +1458,22 @@ async fn create_script_internal<'c>( // fire and the view would be an orphan node in the lineage graph. for (target_kind, path) in m.write_targets() { let kind = windmill_common::assets::asset_kind_from_parser(target_kind); - if !a.iter().any(|x| x.kind == kind && x.path == path) { + use windmill_common::assets::AssetUsageAccessType; + if let Some(existing) = a.iter_mut().find(|x| x.kind == kind && x.path == path) { + // The body reads its own managed target (an incremental/merge + // model `SELECT`ing from the table it materializes). The runtime + // still generates the write, so the effective access is RW — mark + // it so, otherwise it stays a plain read and (a) auto-derives a + // self-cascade edge back to this producer and (b) shows as a muted + // read on the canvas. Both are wrong: it's the script's own output. + if existing.access_type != Some(AssetUsageAccessType::W) { + existing.access_type = Some(AssetUsageAccessType::RW); + } + } else { a.push(windmill_common::assets::AssetWithAltAccessType { path, kind, - access_type: Some(windmill_common::assets::AssetUsageAccessType::W), + access_type: Some(AssetUsageAccessType::W), alt_access_type: None, columns: None, }); @@ -2035,6 +2047,63 @@ async fn create_script_internal<'c>( .await?; } + // Auto-derived cascade edges: within a `// pipeline`, a read of a ducklake + // table or s3 object wires the cascade edge straight from the FROM clause, + // so `// on ` is only needed for edges inference can't see (dynamic + // SQL) or to carry per-edge opts. `// mute ` / `// mute all` opt out. + // Explicit `// on` asset edges (inserted just above) win the dedup — they + // carry the per-edge debounce, so a derived row must not shadow them. + if in_pipeline && !pipeline_annotations.mute_all { + let asset_ref = |spec: &TriggerSpec| { + trigger_spec_to_row(spec) + .filter(|(k, _)| *k == ScriptTriggerKind::Asset) + .map(|(_, r)| r) + }; + let explicit_refs: std::collections::HashSet = + pipeline_triggers.iter().filter_map(asset_ref).collect(); + let muted_refs: std::collections::HashSet = pipeline_annotations + .mute + .iter() + .filter_map(asset_ref) + .collect(); + let derived = derive_pipeline_asset_trigger_refs( + effective_assets.as_deref().unwrap_or(&[]), + &explicit_refs, + &muted_refs, + pipeline_annotations.mute_all, + ); + // Derived edges have no per-`// on` opts, so they take the script-level + // `// debounce` default and `// retry` policy — same as writing a bare + // `// on ` would. + let derived_debounce_s = pipeline_debounce_default + .as_deref() + .and_then(parse_duration_secs); + let derived_retry_count = pipeline_annotations + .retry + .as_ref() + .map(|r| r.count.min(i16::MAX as u32) as i16); + let derived_retry_delay_s = pipeline_annotations + .retry + .as_ref() + .and_then(|r| r.delay.as_deref()) + .and_then(parse_duration_secs); + for trigger_ref in derived { + insert_script_trigger( + &mut *tx, + &w_id, + AssetUsageKind::Script, + &ns.path, + ScriptTriggerKind::Asset, + &trigger_ref, + pipeline_join_all, + derived_debounce_s, + derived_retry_count, + derived_retry_delay_s, + ) + .await?; + } + } + // Schedule annotations (`// on schedule`) are marker-only — the binding // lives on the schedule row's own `script_path` field, which the user // creates separately via the schedule editor. No script-create-time diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 862bd645c3..62f6d32070 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -169,6 +169,68 @@ fn is_write_access(access: Option) -> bool { ) } +/// Kinds whose *read* usage auto-derives a cascade trigger edge inside a +/// `// pipeline`. Scoped to the two intra-pipeline data kinds — a ducklake +/// table read (the core case) and an s3 object read (file-ingestion +/// producers). Resource / datatable / volume reads stay explicit-`// on`: +/// a config/lookup read cascading is more often surprising than wanted. +fn is_auto_trigger_kind(kind: AssetKind) -> bool { + matches!(kind, AssetKind::Ducklake | AssetKind::S3Object) +} + +/// Trigger refs auto-derived from a pipeline script's inferred reads, so the +/// FROM clause alone wires the cascade edge (no redundant `// on `). +/// +/// Included: an input read read-*only* (`R`) of a supported kind +/// ([`is_auto_trigger_kind`]). The effective access type is +/// `access_type.or(alt_access_type)` — same precedence as the persisted +/// `asset.usage_access_type` and the frontend mirror's `access_type ?? +/// alt_access_type`, so a manual read override on an ambiguous parse still +/// derives an edge (and the live canvas and the deployed graph agree). +/// Excluded, each for a reason: +/// - `RW` / `W` — the script also writes the asset; an edge would be a +/// self-triggering loop. +/// - `None` access — usage is ambiguous (poisoned merge) with no override; +/// can't confirm a read, so fail safe and don't cascade. +/// - already in `explicit_refs` — the author wrote `// on `, which +/// wins (it carries the per-edge debounce/opts). +/// - in `muted_refs` — a `// mute ` opt-out (lookup / SCD input). +/// +/// `mute_all` (from `// mute all`) short-circuits to no derivation, leaving +/// only the explicit `// on` edges. Returns canonical refs (e.g. +/// `ducklake://main.orders`), deduped, in input order. +pub fn derive_pipeline_asset_trigger_refs( + assets: &[AssetWithAltAccessType], + explicit_refs: &HashSet, + muted_refs: &HashSet, + mute_all: bool, +) -> Vec { + if mute_all { + return vec![]; + } + let mut out = vec![]; + let mut seen = HashSet::new(); + for a in assets { + // Effective access mirrors the persisted `usage_access_type` and the + // frontend derivation: an explicit parse wins, else the manual override. + let access = a.access_type.or(a.alt_access_type); + if access != Some(AssetUsageAccessType::R) || !is_auto_trigger_kind(a.kind) { + continue; + } + let Some(prefix) = a.kind.canonical_prefix() else { + continue; + }; + let r = format!("{}{}", prefix, a.path); + if explicit_refs.contains(&r) || muted_refs.contains(&r) { + continue; + } + if seen.insert(r.clone()) { + out.push(r); + } + } + out +} + /// Clear and reinsert the full static-asset usage set of a script in one tx, /// invalidating the producer-writes cache at most once and only on a real /// change. The cache (asset_dispatch::ASSET_PRODUCER_WRITES_CACHE) keys a @@ -369,6 +431,134 @@ mod debounce_duration_tests { } } +#[cfg(test)] +mod derive_trigger_tests { + use super::{derive_pipeline_asset_trigger_refs, AssetKind, AssetUsageAccessType}; + use std::collections::HashSet; + use windmill_types::assets::AssetWithAltAccessType; + + fn asset( + kind: AssetKind, + path: &str, + at: Option, + ) -> AssetWithAltAccessType { + AssetWithAltAccessType { + path: path.to_string(), + kind, + access_type: at, + alt_access_type: None, + columns: None, + } + } + + fn derive(assets: &[AssetWithAltAccessType]) -> Vec { + derive_pipeline_asset_trigger_refs(assets, &HashSet::new(), &HashSet::new(), false) + } + + #[test] + fn read_only_ducklake_and_s3_derive_an_edge() { + use AssetUsageAccessType::R; + let a = [ + asset(AssetKind::Ducklake, "main.orders", Some(R)), + asset(AssetKind::S3Object, "raw/events", Some(R)), + ]; + assert_eq!( + derive(&a), + vec![ + "ducklake://main.orders".to_string(), + "s3://raw/events".to_string() + ] + ); + } + + #[test] + fn writes_and_rw_are_skipped_to_avoid_self_edges() { + use AssetUsageAccessType::*; + // W (pure producer) and RW (reads *and* writes the same table — a + // self-cascade if edged) both derive nothing. + let a = [ + asset(AssetKind::Ducklake, "main.out", Some(W)), + asset(AssetKind::Ducklake, "main.self", Some(RW)), + ]; + assert!(derive(&a).is_empty()); + } + + #[test] + fn ambiguous_access_and_unsupported_kinds_are_skipped() { + use AssetUsageAccessType::R; + let a = [ + asset(AssetKind::Ducklake, "main.ambiguous", None), // poisoned merge + asset(AssetKind::Resource, "f/db", Some(R)), // out of scope + asset(AssetKind::DataTable, "main.dt", Some(R)), // out of scope + ]; + assert!(derive(&a).is_empty()); + } + + #[test] + fn manual_read_override_on_ambiguous_parse_derives_an_edge() { + use AssetUsageAccessType::{R, W}; + // Parser can't confirm access (`access_type: None`) but the user manually + // overrode it. Effective access = `access_type.or(alt_access_type)`, the + // same value persisted to `asset.usage_access_type` and used by the + // frontend canvas — so a read override derives an edge (parity, no + // silently-vanishing edge on deploy) and a write override does not. + let read_override = AssetWithAltAccessType { + path: "main.override_r".to_string(), + kind: AssetKind::Ducklake, + access_type: None, + alt_access_type: Some(R), + columns: None, + }; + let write_override = AssetWithAltAccessType { + path: "main.override_w".to_string(), + kind: AssetKind::Ducklake, + access_type: None, + alt_access_type: Some(W), + columns: None, + }; + assert_eq!( + derive(&[read_override, write_override]), + vec!["ducklake://main.override_r".to_string()] + ); + } + + #[test] + fn explicit_and_muted_refs_are_excluded() { + use AssetUsageAccessType::R; + let a = [ + asset(AssetKind::Ducklake, "main.explicit", Some(R)), + asset(AssetKind::Ducklake, "main.muted", Some(R)), + asset(AssetKind::Ducklake, "main.keep", Some(R)), + ]; + let explicit: HashSet = ["ducklake://main.explicit".to_string()].into(); + let muted: HashSet = ["ducklake://main.muted".to_string()].into(); + assert_eq!( + derive_pipeline_asset_trigger_refs(&a, &explicit, &muted, false), + vec!["ducklake://main.keep".to_string()] + ); + } + + #[test] + fn mute_all_derives_nothing() { + use AssetUsageAccessType::R; + let a = [asset(AssetKind::Ducklake, "main.orders", Some(R))]; + assert!( + derive_pipeline_asset_trigger_refs(&a, &HashSet::new(), &HashSet::new(), true) + .is_empty() + ); + } + + #[test] + fn duplicate_reads_dedup() { + use AssetUsageAccessType::R; + let a = [ + asset(AssetKind::Ducklake, "main.orders", Some(R)), + asset(AssetKind::Ducklake, "main.orders", Some(R)), + ]; + assert_eq!(derive(&a), vec!["ducklake://main.orders".to_string()]); + } +} + #[cfg(test)] mod trigger_ref_roundtrip_tests { use super::{parse_asset_trigger_ref, trigger_spec_to_row, AssetKind, ScriptTriggerKind}; diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte index fbaf3ed6ed..64bd2790cb 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphCanvas.svelte @@ -18,6 +18,7 @@ import PanToNode from './PanToNode.svelte' import InitialFitView from './InitialFitView.svelte' import { layoutAssetGraph } from './assetGraphLayout' + import { computeMutedReadKeys } from './resolveGraph' import { buildDownstreamMap } from './graphTraversal' import { buildLineageDownstreamMap } from './boundedCascade' import type { AssetGraphResponse, AssetGraphSelection, NativeTriggerKind } from './types' @@ -228,6 +229,11 @@ | 'macro' | 'test-dependency' unsaved?: boolean + // Muted read edge: a ducklake/s3 input read every run whose (default) + // auto cascade trigger is suppressed by `// mute` / `// mute all`. + // Rendered with a bell-off badge — auto-wiring is the norm, so we mark + // the read that deliberately does NOT cascade, not every derived edge. + muted?: boolean // Edge from a missing-trigger placeholder — styled red dashed to // signal "this script declared `// on kafka` but no trigger row // targets it; create one or remove the annotation". @@ -490,6 +496,10 @@ }) } + // Read edges of a ducklake/s3 asset with no cascade trigger = muted + // (`// mute` / `// mute all` opted the default auto trigger out). Gated + // on pipeline scripts inside the helper (non-pipeline reads never derive). + const mutedReadKeys = computeMutedReadKeys(g.edges, g.triggers, g.runnables) for (const e of g.edges) { const runnableId = `${e.runnable_kind}:${e.runnable_path}` const assetId = `asset:${e.asset_kind}:${e.asset_path}` @@ -542,7 +552,13 @@ source: assetId, target: runnableId, kind: 'lineage-read', - unsaved: e.unsaved + unsaved: e.unsaved, + // Only a pure `'r'` read can be muted; `'rw'` is a self-read. + muted: + access === 'r' && + mutedReadKeys.has( + `${e.asset_kind}:${e.asset_path}->${e.runnable_kind}:${e.runnable_path}` + ) }) } } @@ -1079,7 +1095,9 @@ // Macro-edge badge: which of the library's macros the consumer // calls (all of them when pulled in via `// use`). macro_names: e.macro_names, - via_use: e.via_use + via_use: e.via_use, + // Muted read edge — bell-off badge on the read link. + muted: e.muted }, animated, label, diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index a097f09b6c..4e5acaee08 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -636,7 +636,9 @@ dataTests: [], columnLineage: [], macros: false, - useLibs: [] + useLibs: [], + muteAssets: [], + muteAll: false } ) // `// macros` library: the defined signatures for the strip above the diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte index 329fad6902..e16e11566d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphEdge.svelte @@ -1,7 +1,7 @@