#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Additivity {
Additive,
SemiAdditive,
NonAdditive,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetricInfo {
pub name: String,
pub owner: String,
pub declared: bool,
pub additivity: Option<Additivity>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DimensionInfo {
pub name: String,
pub owner: String,
pub is_temporal: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ModelInfo {
pub name: String,
pub has_timestamp: bool,
pub columns: Vec<String>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum LineageKind {
Model,
Column,
Relationship,
Cube,
Metric,
Dimension,
View,
Query,
Dashboard,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LineageNode {
pub id: String,
pub kind: LineageKind,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LineageEdge {
pub from: String,
pub to: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Severity {
None,
Compatibility,
Structural,
Semantic,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct BlastRadius {
pub seed: String,
pub downstream: Vec<String>,
pub severity: Severity,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct LineageGraph {
pub nodes: Vec<LineageNode>,
pub edges: Vec<LineageEdge>,
}
impl LineageGraph {
pub fn is_resolvable(&self) -> bool {
self.edges
.iter()
.all(|e| self.contains(&e.from) && self.contains(&e.to))
}
pub fn contains(&self, id: &str) -> bool {
self.nodes.iter().any(|n| n.id == id)
}
pub fn node(&self, id: &str) -> Option<&LineageNode> {
self.nodes.iter().find(|n| n.id == id)
}
pub fn blast_radius(&self, seed: &str) -> BlastRadius {
let mut downstream: std::collections::BTreeSet<String> = std::collections::BTreeSet::new();
let mut stack = vec![seed.to_string()];
while let Some(current) = stack.pop() {
for edge in self.edges.iter().filter(|e| e.from == current) {
if edge.to != seed && downstream.insert(edge.to.clone()) {
stack.push(edge.to.clone());
}
}
}
let severity = downstream
.iter()
.map(|id| self.node_severity(id))
.max()
.unwrap_or(Severity::None);
BlastRadius {
seed: seed.to_string(),
downstream: downstream.into_iter().collect(),
severity,
}
}
fn node_severity(&self, id: &str) -> Severity {
match self.node(id).map(|n| n.kind) {
Some(LineageKind::Metric | LineageKind::Query | LineageKind::Dashboard) => {
Severity::Semantic
}
Some(LineageKind::Model | LineageKind::View | LineageKind::Column) => {
Severity::Structural
}
Some(LineageKind::Relationship | LineageKind::Cube | LineageKind::Dimension) => {
Severity::Compatibility
}
None => Severity::None,
}
}
}
pub trait ContextLoader {
fn is_parseable(&self) -> bool;
fn parse_error(&self) -> Option<&str> {
None
}
fn metrics(&self) -> &[MetricInfo];
fn dimensions(&self) -> &[DimensionInfo];
fn time_dimensions(&self) -> &[DimensionInfo];
fn models(&self) -> &[ModelInfo];
fn model(&self, name: &str) -> Option<&ModelInfo> {
self.models().iter().find(|m| m.name == name)
}
fn lineage(&self) -> &LineageGraph;
fn source_introspectable(&self) -> Option<bool> {
None
}
fn raw_docs_readable(&self) -> Option<bool> {
None
}
fn lineage_diagnostics(&self) -> &[String] {
&[]
}
fn metric_additivity(&self, metric: &str) -> Option<Additivity> {
self.metrics()
.iter()
.find(|m| m.name == metric)
.and_then(|m| m.additivity)
}
fn can_answer(&self, predicate: &str) -> bool {
match predicate {
"metric_additive" => self.metrics().iter().any(|m| m.declared),
"source_introspectable" => self.source_introspectable().is_some(),
"raw_docs_readable" => self.raw_docs_readable().is_some(),
"mdl_parseable"
| "wren_project_exists"
| "has_metric"
| "has_queryable_dimension"
| "has_time_dimension"
| "has_groupable_dimension"
| "model_has_timestamp"
| "lineage_resolvable" => true,
_ => false,
}
}
}
#[derive(Debug, Default)]
pub struct ExternalContext {
lineage: LineageGraph,
}
impl ExternalContext {
pub fn new() -> Self {
Self::default()
}
}
impl ContextLoader for ExternalContext {
fn is_parseable(&self) -> bool {
true
}
fn metrics(&self) -> &[MetricInfo] {
&[]
}
fn dimensions(&self) -> &[DimensionInfo] {
&[]
}
fn time_dimensions(&self) -> &[DimensionInfo] {
&[]
}
fn models(&self) -> &[ModelInfo] {
&[]
}
fn lineage(&self) -> &LineageGraph {
&self.lineage
}
fn can_answer(&self, _predicate: &str) -> bool {
false
}
}
#[cfg(test)]
mod tests {
use super::*;
struct FakeContext {
metrics: Vec<MetricInfo>,
dimensions: Vec<DimensionInfo>,
models: Vec<ModelInfo>,
lineage: LineageGraph,
parseable: bool,
}
impl ContextLoader for FakeContext {
fn is_parseable(&self) -> bool {
self.parseable
}
fn metrics(&self) -> &[MetricInfo] {
&self.metrics
}
fn dimensions(&self) -> &[DimensionInfo] {
&self.dimensions
}
fn time_dimensions(&self) -> &[DimensionInfo] {
&self.dimensions
}
fn models(&self) -> &[ModelInfo] {
&self.models
}
fn lineage(&self) -> &LineageGraph {
&self.lineage
}
}
fn metric(name: &str, declared: bool, additivity: Option<Additivity>) -> MetricInfo {
MetricInfo {
name: name.to_string(),
owner: "m".to_string(),
declared,
additivity,
}
}
#[test]
fn metric_additivity_looks_up_declared_metric() {
let ctx = FakeContext {
metrics: vec![
metric("total_revenue", true, Some(Additivity::Additive)),
metric("avg_order", true, Some(Additivity::NonAdditive)),
metric("amount", false, None),
],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: true,
};
assert_eq!(
ctx.metric_additivity("total_revenue"),
Some(Additivity::Additive)
);
assert_eq!(
ctx.metric_additivity("avg_order"),
Some(Additivity::NonAdditive)
);
assert_eq!(ctx.metric_additivity("amount"), None);
assert_eq!(ctx.metric_additivity("nope"), None);
}
#[test]
fn can_answer_metric_additive_needs_a_declared_metric() {
let cubeless = FakeContext {
metrics: vec![metric("amount", false, None)],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: true,
};
assert!(!cubeless.can_answer("metric_additive"));
let with_cube = FakeContext {
metrics: vec![metric("total_revenue", true, Some(Additivity::Additive))],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: true,
};
assert!(with_cube.can_answer("metric_additive"));
}
#[test]
fn can_answer_existence_predicates_always_true() {
let ctx = FakeContext {
metrics: vec![],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: false,
};
for p in [
"mdl_parseable",
"wren_project_exists",
"has_metric",
"has_queryable_dimension",
"has_time_dimension",
"has_groupable_dimension",
"model_has_timestamp",
"lineage_resolvable",
] {
assert!(ctx.can_answer(p), "{p} should be answerable");
}
assert!(!ctx.can_answer("some_future_unknown_predicate"));
}
#[test]
fn mdl_only_adapter_cannot_answer_constitutive_raw_shape() {
let ctx = FakeContext {
metrics: vec![],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: true,
};
assert_eq!(ctx.source_introspectable(), None);
assert_eq!(ctx.raw_docs_readable(), None);
assert!(!ctx.can_answer("source_introspectable"));
assert!(!ctx.can_answer("raw_docs_readable"));
}
#[test]
fn raw_source_adapter_answers_constitutive_raw_shape() {
struct RawFake;
impl ContextLoader for RawFake {
fn is_parseable(&self) -> bool {
true
}
fn metrics(&self) -> &[MetricInfo] {
&[]
}
fn dimensions(&self) -> &[DimensionInfo] {
&[]
}
fn time_dimensions(&self) -> &[DimensionInfo] {
&[]
}
fn models(&self) -> &[ModelInfo] {
&[]
}
fn lineage(&self) -> &LineageGraph {
static EMPTY: std::sync::OnceLock<LineageGraph> = std::sync::OnceLock::new();
EMPTY.get_or_init(LineageGraph::default)
}
fn source_introspectable(&self) -> Option<bool> {
Some(true)
}
fn raw_docs_readable(&self) -> Option<bool> {
Some(false)
}
}
let ctx = RawFake;
assert!(ctx.can_answer("source_introspectable"));
assert!(ctx.can_answer("raw_docs_readable"));
assert_eq!(ctx.source_introspectable(), Some(true));
assert_eq!(ctx.raw_docs_readable(), Some(false));
}
#[test]
fn lineage_resolvable_detects_dangling_edges() {
let good = LineageGraph {
nodes: vec![
LineageNode {
id: "model:orders".into(),
kind: LineageKind::Model,
},
LineageNode {
id: "metric:revenue".into(),
kind: LineageKind::Metric,
},
],
edges: vec![LineageEdge {
from: "model:orders".into(),
to: "metric:revenue".into(),
}],
};
assert!(good.is_resolvable());
let dangling = LineageGraph {
nodes: vec![LineageNode {
id: "model:orders".into(),
kind: LineageKind::Model,
}],
edges: vec![LineageEdge {
from: "model:orders".into(),
to: "metric:missing".into(),
}],
};
assert!(!dangling.is_resolvable());
}
fn node(id: &str, kind: LineageKind) -> LineageNode {
LineageNode {
id: id.into(),
kind,
}
}
fn edge(from: &str, to: &str) -> LineageEdge {
LineageEdge {
from: from.into(),
to: to.into(),
}
}
fn sample_graph() -> LineageGraph {
LineageGraph {
nodes: vec![
node("model:orders", LineageKind::Model),
node("cube:revenue", LineageKind::Cube),
node("metric:revenue.total", LineageKind::Metric),
node("dim:revenue.status", LineageKind::Dimension),
node("view:orders_view", LineageKind::View),
],
edges: vec![
edge("model:orders", "cube:revenue"),
edge("cube:revenue", "metric:revenue.total"),
edge("cube:revenue", "dim:revenue.status"),
edge("model:orders", "view:orders_view"),
],
}
}
#[test]
fn blast_radius_is_the_transitive_downstream_closure() {
let graph = sample_graph();
let radius = graph.blast_radius("model:orders");
assert_eq!(
radius.downstream,
vec![
"cube:revenue".to_string(),
"dim:revenue.status".to_string(),
"metric:revenue.total".to_string(),
"view:orders_view".to_string(),
],
"changing the base model reaches the cube, its members, and the view"
);
assert_eq!(radius.severity, Severity::Semantic);
}
#[test]
fn blast_radius_of_a_leaf_is_empty() {
let graph = sample_graph();
let radius = graph.blast_radius("metric:revenue.total");
assert!(radius.downstream.is_empty());
assert_eq!(radius.severity, Severity::None);
}
#[test]
fn blast_radius_severity_is_the_max_over_downstream() {
let graph = sample_graph();
assert_eq!(
graph.blast_radius("cube:revenue").severity,
Severity::Semantic
);
let structural = LineageGraph {
nodes: vec![
node("model:m", LineageKind::Model),
node("view:v", LineageKind::View),
],
edges: vec![edge("model:m", "view:v")],
};
assert_eq!(
structural.blast_radius("model:m").severity,
Severity::Structural
);
}
#[test]
fn consumer_nodes_are_semantic_severity() {
let graph = LineageGraph {
nodes: vec![
node("metric:revenue.total", LineageKind::Metric),
node("query:monthly-revenue", LineageKind::Query),
node("dashboard:exec-weekly", LineageKind::Dashboard),
],
edges: vec![
edge("metric:revenue.total", "query:monthly-revenue"),
edge("metric:revenue.total", "dashboard:exec-weekly"),
],
};
let radius = graph.blast_radius("metric:revenue.total");
assert_eq!(
radius.downstream,
vec![
"dashboard:exec-weekly".to_string(),
"query:monthly-revenue".to_string(),
],
"a metric with consumers is no longer a leaf"
);
assert_eq!(radius.severity, Severity::Semantic);
}
#[test]
fn lineage_diagnostics_default_to_empty() {
let ctx = FakeContext {
metrics: vec![],
dimensions: vec![],
models: vec![],
lineage: LineageGraph::default(),
parseable: true,
};
assert!(ctx.lineage_diagnostics().is_empty());
}
#[test]
fn blast_radius_is_cycle_safe() {
let cyclic = LineageGraph {
nodes: vec![node("a", LineageKind::Model), node("b", LineageKind::Model)],
edges: vec![edge("a", "b"), edge("b", "a")],
};
let radius = cyclic.blast_radius("a");
assert_eq!(radius.downstream, vec!["b".to_string()]);
}
#[test]
fn severity_ordering() {
assert!(Severity::Semantic > Severity::Structural);
assert!(Severity::Structural > Severity::Compatibility);
assert!(Severity::Compatibility > Severity::None);
}
}