From 6c8e0217a1371317cad799974a361872a6a744ad Mon Sep 17 00:00:00 2001 From: Paul Sattaur Date: Wed, 12 Aug 2026 22:53:37 -0400 Subject: [PATCH] =?UTF-8?q?Phase=208.2=20=E2=80=94=20Field-Level=20Lineage?= =?UTF-8?q?=20+=20Column=20Impact?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 22 +- apps/api/skydata_studio/api/routes/lineage.py | 36 +- .../api/skydata_studio/api/routes/platform.py | 4 +- apps/api/skydata_studio/schemas/dbt.py | 1 + apps/api/skydata_studio/schemas/lineage.py | 63 +++ .../services/dbt_transformations.py | 41 +- apps/api/skydata_studio/services/lineage.py | 368 +++++++++++++++++- apps/web/src/components/FieldLineagePanel.jsx | 220 +++++++++++ apps/web/src/components/Sidebar.jsx | 2 +- apps/web/src/pages/Lineage.jsx | 19 +- apps/web/src/pages/StudioOverview.jsx | 16 +- apps/web/src/styles.css | 149 +++++++ docs/ARCHITECTURE.md | 17 +- docs/ROADMAP.md | 33 +- docs/SkyData_RepoMap.md | 1 + docs/repository-map.txt | 1 + tests/test_lineage.py | 303 +++++++++++++- tests/test_platform.py | 2 +- .../models/intermediate/schema.yml | 33 ++ .../skydata_studio/models/marts/schema.yml | 34 ++ .../skydata_studio/models/staging/schema.yml | 6 + 21 files changed, 1333 insertions(+), 38 deletions(-) create mode 100644 apps/web/src/components/FieldLineagePanel.jsx diff --git a/README.md b/README.md index 744ad79..d4b0c32 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ SkyData Studio is the post-ingestion data engineering application in the Sky ecosystem. It begins where SkyCommand's ingestion responsibility ends and prepares trusted data for analytical consumption by SkyWeb Analytics, Power BI, and future client-facing applications. -**Current status:** Phases 0 through 7 are complete. Phase 7 closed with TRUSTED dbt quality evidence, a COMPLIANT 5/5 consumer gate, durable incident reconciliation, and a 30-day/99% quality SLO reporting MEETING at 100% observed compliance; the final Phase 7.4 validation passed 72 tests. **Phase 8.1 is now in progress:** Studio metadata mappings, dbt dependencies, semantic models, and governed metrics are stitched into one federated lineage graph with transitive downstream impact analysis. +**Current status:** Phases 0 through 7 and Phase 8.1 are complete. Phase 8.1 closed with a READY 11-node/10-edge federated graph, a DFF impact radius reaching all 10 downstream nodes, and final validation passing 76 tests. **Phase 8.2 is now in progress:** source-to-target field mappings, dbt column annotations, and governed metric expressions are stitched into a field-level impact graph without creating a second lineage authority. --- @@ -226,21 +226,27 @@ The first clean proof should show `MEETING`, 100% observed compliance, a 99% tar --- -## Phase 8.1 quick start +## Phase 8.2 quick start -Phase 8.1 composes a read-only lineage graph from existing authorities rather than creating another lineage metadata store. The first proof joins the READY `DFF → FED_FUNDS_RATE_MART` Studio mapping to the dbt source/model DAG and then continues through the semantic model into all governed metrics. +Phase 8.2 preserves the Phase 8.1 asset graph and adds field-level lineage without asking the Studio API to guess arbitrary SQL. The two registered Studio field mappings remain the source-to-target authority, while dbt model columns declare their upstream inputs beside the model definitions in `schema.yml`; `dbt build` carries those annotations into `manifest.json`. Governed metric expressions then bind mart fields to business measures. -Start Studio PostgreSQL and ensure the current dbt artifacts exist, then inspect the graph: +Regenerate the dbt artifacts after overlaying the Phase 8.2 model annotations: + +```powershell +.\scripts\dbt.ps1 build +``` + +Inspect the field graph: ```powershell Invoke-RestMethod ` - "http://localhost:8100/api/v1/lineage/summary" | + "http://localhost:8100/api/v1/lineage/fields/summary" | ConvertTo-Json -Depth 12 ``` -The current Federal Funds Rate proof should report one metadata mapping, 3 dbt business models, 1 semantic model, 4 metrics, 11 total nodes, and 10 directed edges. The default impact radius begins at `DFF` and should reach 10 downstream nodes including all 3 dbt models, the semantic model, and all 4 governed metrics. +The Federal Funds Rate proof should report 2 Studio field mappings, 18 annotated dbt business-model columns, 4 metric bindings, 28 field/metric nodes, and 27 directed edges. Selecting `DFF.value` should reach 15 downstream nodes and exactly the average, minimum, and maximum Federal Funds Rate metrics while leaving the observation-count metric outside that radius. -Open **Quality & Lineage → Lineage** and select any node to recompute its transitive downstream radius. You can query the same impact directly with `GET /api/v1/lineage/impact?nodeId=...`. Phase 8.1 remains a federated read model: field-level lineage, quality overlays, and report/consumer impact stay in later Phase 8 slices. +Open **Quality & Lineage → Lineage** to inspect both the Phase 8.1 asset graph and the Phase 8.2 field graph. Field impact is also available through `GET /api/v1/lineage/fields/impact?fieldId=...`. The field graph remains a read model: Studio-owned field mappings, dbt-owned column annotations, and dbt-owned metric expressions stay at their existing authorities. ## Validation contract @@ -496,4 +502,4 @@ The repository currently includes: - a Phase 7.1 artifact-backed Data Quality surface that joins dbt test definitions to latest run outcomes and computes a consumer-facing trust posture; - backend/frontend validation, repository map, and compact handoff tooling. -The active implementation target is **Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation**. Phases 0 through 7 are complete: the Studio now has governed ingestion handoffs, metadata and mappings, replay-safe pipelines, Airflow orchestration, dbt models and semantics, quality gates, durable incidents, and observation-backed reliability history. Phase 8.1 federates those existing dependency authorities into one directed graph so downstream impact can be explained without creating a second lineage store. +The active implementation target is **Phase 8.2 — Field-Level Lineage and Column Impact Foundation**. Phases 0 through 7 are complete and Phase 8.1 established the federated asset/model/semantic/metric graph. Phase 8.2 carries the same ownership model down to individual fields so schema-change impact can be traced from DFF source columns through Studio mappings, dbt derivations, and governed metrics. diff --git a/apps/api/skydata_studio/api/routes/lineage.py b/apps/api/skydata_studio/api/routes/lineage.py index cb7d67e..d35488a 100644 --- a/apps/api/skydata_studio/api/routes/lineage.py +++ b/apps/api/skydata_studio/api/routes/lineage.py @@ -4,8 +4,18 @@ from sqlalchemy.exc import SQLAlchemyError from skydata_studio.db.session import SessionDependency -from skydata_studio.schemas.lineage import LineageImpactSummary, LineageSummary -from skydata_studio.services.lineage import lineage_impact, lineage_summary +from skydata_studio.schemas.lineage import ( + FieldLineageImpactSummary, + FieldLineageSummary, + LineageImpactSummary, + LineageSummary, +) +from skydata_studio.services.lineage import ( + field_lineage_impact, + field_lineage_summary, + lineage_impact, + lineage_summary, +) router = APIRouter() @@ -40,3 +50,25 @@ def lineage_node_impact( return lineage_impact(session, node_id) except (SQLAlchemyError, OSError, ValueError, TypeError) as error: raise _unavailable(error) from error + + +@router.get("/fields/summary", response_model=FieldLineageSummary) +def field_lineage_graph( + session: SessionDependency, + focus_field_id: Annotated[str | None, Query(alias="focusFieldId")] = None, +) -> FieldLineageSummary: + try: + return field_lineage_summary(session, focus_field_id=focus_field_id) + except (SQLAlchemyError, OSError, ValueError, TypeError) as error: + raise _unavailable(error) from error + + +@router.get("/fields/impact", response_model=FieldLineageImpactSummary) +def field_lineage_field_impact( + session: SessionDependency, + field_id: Annotated[str, Query(alias="fieldId", min_length=1)], +) -> FieldLineageImpactSummary: + try: + return field_lineage_impact(session, field_id) + except (SQLAlchemyError, OSError, ValueError, TypeError) as error: + raise _unavailable(error) from error diff --git a/apps/api/skydata_studio/api/routes/platform.py b/apps/api/skydata_studio/api/routes/platform.py index 39a4ccb..69d0648 100644 --- a/apps/api/skydata_studio/api/routes/platform.py +++ b/apps/api/skydata_studio/api/routes/platform.py @@ -60,7 +60,7 @@ code="LINEAGE_IMPACT", name="Lineage and Impact Analysis", description=( - "Stitches metadata mappings, dbt dependencies, semantic models, " + "Stitches metadata mappings, dbt dependencies, field lineage, semantic models, " "and metrics into one impact graph." ), status="FOUNDATION", @@ -96,7 +96,7 @@ def platform_summary() -> PlatformSummary: product="SkyData Studio", subtitle="Data Engineering Workbench", theme="Aurora Foundry", - current_phase="Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation", + current_phase="Phase 8.2 — Field-Level Lineage and Column Impact Foundation", boundary=( "SkyData Studio starts after SkyCommand ingestion and publishes governed " "analytical products for SkyWeb Analytics, Power BI, and future consumers." diff --git a/apps/api/skydata_studio/schemas/dbt.py b/apps/api/skydata_studio/schemas/dbt.py index 377de28..2a86cc2 100644 --- a/apps/api/skydata_studio/schemas/dbt.py +++ b/apps/api/skydata_studio/schemas/dbt.py @@ -32,6 +32,7 @@ class DbtTransformationSummary(BaseModel): class DbtModelColumnSummary(BaseModel): name: str description: str | None = None + lineage_inputs: list[str] = Field(default_factory=list) class DbtModelDependencySummary(BaseModel): diff --git a/apps/api/skydata_studio/schemas/lineage.py b/apps/api/skydata_studio/schemas/lineage.py index 989e5aa..9daedb4 100644 --- a/apps/api/skydata_studio/schemas/lineage.py +++ b/apps/api/skydata_studio/schemas/lineage.py @@ -74,3 +74,66 @@ class LineageSummary(BaseModel): nodes: list[LineageNode] edges: list[LineageEdge] default_impact: LineageImpactSummary + + +FieldLineageNodeType = Literal[ + "SOURCE_FIELD", + "CURATED_FIELD", + "DBT_SOURCE_FIELD", + "DBT_MODEL_FIELD", + "METRIC", +] +FieldLineageEdgeType = Literal[ + "FIELD_MAPPING", + "PUBLISHES_AS", + "DERIVES", + "FEEDS_METRIC", +] + + +class FieldLineageNode(BaseModel): + id: str + label: str + field_name: str + node_type: FieldLineageNodeType + layer: LineageLayer + system: str + relation: str | None = None + parent_node_id: str | None = None + parent_label: str | None = None + description: str | None = None + status: LineageStatus = "UNKNOWN" + metadata: dict[str, str | int | float | bool | None] = Field(default_factory=dict) + + +class FieldLineageEdge(BaseModel): + id: str + upstream_id: str + downstream_id: str + edge_type: FieldLineageEdgeType + label: str + + +class FieldLineageImpactSummary(BaseModel): + selected_field_id: str | None = None + selected_field_label: str | None = None + downstream_node_count: int = Field(ge=0) + affected_field_count: int = Field(ge=0) + affected_metric_count: int = Field(ge=0) + affected_relations: list[str] + affected_layers: list[str] + nodes: list[FieldLineageNode] + + +class FieldLineageSummary(BaseModel): + project_name: str = "skydata_studio" + phase: str = "Phase 8.2 — Field-Level Lineage and Column Impact Foundation" + artifact_status: Literal["READY", "PARTIAL", "MISSING"] + field_mapping_count: int = Field(ge=0) + dbt_annotated_column_count: int = Field(ge=0) + metric_binding_count: int = Field(ge=0) + node_count: int = Field(ge=0) + edge_count: int = Field(ge=0) + nodes: list[FieldLineageNode] + edges: list[FieldLineageEdge] + default_impact: FieldLineageImpactSummary diff --git a/apps/api/skydata_studio/services/dbt_transformations.py b/apps/api/skydata_studio/services/dbt_transformations.py index d1d36bc..9033b2c 100644 --- a/apps/api/skydata_studio/services/dbt_transformations.py +++ b/apps/api/skydata_studio/services/dbt_transformations.py @@ -261,12 +261,26 @@ def dbt_model_catalogue(target_dir: Path | None = None) -> DbtModelCatalogueSumm if isinstance(columns_payload, dict): for column_name, column_payload in columns_payload.items(): description = None + lineage_inputs: list[str] = [] if isinstance(column_payload, dict): raw_description = column_payload.get("description") if isinstance(raw_description, str) and raw_description.strip(): description = raw_description + raw_meta = column_payload.get("meta") + if isinstance(raw_meta, dict): + raw_lineage_inputs = raw_meta.get("lineage_inputs") + if isinstance(raw_lineage_inputs, list): + lineage_inputs = [ + str(item) + for item in raw_lineage_inputs + if isinstance(item, str) and item.strip() + ] columns.append( - DbtModelColumnSummary(name=str(column_name), description=description) + DbtModelColumnSummary( + name=str(column_name), + description=description, + lineage_inputs=lineage_inputs, + ) ) test_count = 0 @@ -538,12 +552,31 @@ def dbt_semantic_layer(target_dir: Path | None = None) -> DbtSemanticLayerSummar metric_type = str(metric.get("type") or "simple").upper() if metric_type not in {"SIMPLE", "RATIO", "CUMULATIVE", "DERIVED", "CONVERSION"}: metric_type = "SIMPLE" - aggregation_value = metric.get("agg", measure.get("agg")) - expression_value = metric.get("expr", measure.get("expr")) + raw_type_params = metric.get("type_params") + type_params = raw_type_params if isinstance(raw_type_params, dict) else {} + raw_aggregation_params = type_params.get("metric_aggregation_params") + aggregation_params = ( + raw_aggregation_params if isinstance(raw_aggregation_params, dict) else {} + ) + aggregation_value = metric.get( + "agg", + measure.get("agg", aggregation_params.get("agg")), + ) + expression_value = metric.get( + "expr", + measure.get("expr", type_params.get("expr")), + ) time_dimension_value = metric.get( "agg_time_dimension", - measure.get("agg_time_dimension", default_time_dimension), + measure.get( + "agg_time_dimension", + aggregation_params.get("agg_time_dimension", default_time_dimension), + ), ) + if semantic_name is None: + raw_semantic_name = aggregation_params.get("semantic_model") + if isinstance(raw_semantic_name, str) and raw_semantic_name: + semantic_name = raw_semantic_name aggregation = ( str(aggregation_value) if isinstance(aggregation_value, str) else None ) diff --git a/apps/api/skydata_studio/services/lineage.py b/apps/api/skydata_studio/services/lineage.py index 25bffe3..34e01d2 100644 --- a/apps/api/skydata_studio/services/lineage.py +++ b/apps/api/skydata_studio/services/lineage.py @@ -7,13 +7,20 @@ from skydata_studio.schemas.dbt import DbtModelCatalogueSummary, DbtSemanticLayerSummary from skydata_studio.schemas.lineage import ( + FieldLineageEdge, + FieldLineageImpactSummary, + FieldLineageNode, + FieldLineageSummary, LineageEdge, LineageImpactSummary, LineageNode, LineageSummary, ) from skydata_studio.services.dbt_transformations import dbt_model_catalogue, dbt_semantic_layer -from skydata_studio.services.metadata_registry import list_metadata_mappings +from skydata_studio.services.metadata_registry import ( + get_metadata_mapping, + list_metadata_mappings, +) def _canonical_name(value: str) -> str: @@ -352,3 +359,362 @@ def lineage_summary( def lineage_impact(session: Session, node_id: str) -> LineageImpactSummary: summary = lineage_summary(session, focus_node_id=node_id) return summary.default_impact + + + +def _field_node_id(parent_id: str, field_name: str) -> str: + return f"field:{parent_id}:{field_name.strip().lower()}" + + +def _field_impact_from_graph( + nodes: list[FieldLineageNode], + edges: list[FieldLineageEdge], + selected_field_id: str | None, +) -> FieldLineageImpactSummary: + node_by_id = {node.id: node for node in nodes} + if selected_field_id not in node_by_id: + selected_field_id = None + + if selected_field_id is None: + preferred = next( + (node.id for node in nodes if node.node_type == "SOURCE_FIELD"), + None, + ) + selected_field_id = preferred or (nodes[0].id if nodes else None) + + if selected_field_id is None: + return FieldLineageImpactSummary( + downstream_node_count=0, + affected_field_count=0, + affected_metric_count=0, + affected_relations=[], + affected_layers=[], + nodes=[], + ) + + adjacency: dict[str, list[str]] = {} + for edge in edges: + adjacency.setdefault(edge.upstream_id, []).append(edge.downstream_id) + + visited: set[str] = set() + queue: deque[str] = deque(adjacency.get(selected_field_id, [])) + while queue: + node_id = queue.popleft() + if node_id in visited: + continue + visited.add(node_id) + queue.extend(adjacency.get(node_id, [])) + + impacted = [node for node in nodes if node.id in visited] + return FieldLineageImpactSummary( + selected_field_id=selected_field_id, + selected_field_label=node_by_id[selected_field_id].label, + downstream_node_count=len(impacted), + affected_field_count=sum(node.node_type != "METRIC" for node in impacted), + affected_metric_count=sum(node.node_type == "METRIC" for node in impacted), + affected_relations=sorted( + {node.relation for node in impacted if node.relation is not None} + ), + affected_layers=sorted({str(node.layer) for node in impacted}), + nodes=impacted, + ) + + +def _parse_lineage_input(reference: str) -> tuple[str, str, str] | None: + kind, separator, remainder = reference.partition(":") + if not separator or kind not in {"source", "model"}: + return None + parent, dot, field_name = remainder.rpartition(".") + if not dot or not parent or not field_name: + return None + return kind, parent, field_name + + +def compose_field_lineage_summary( + session: Session, + *, + catalogue: DbtModelCatalogueSummary, + semantic: DbtSemanticLayerSummary, + focus_field_id: str | None = None, +) -> FieldLineageSummary: + nodes: list[FieldLineageNode] = [] + edges: list[FieldLineageEdge] = [] + node_ids: set[str] = set() + edge_ids: set[str] = set() + + def add_node(node: FieldLineageNode) -> None: + if node.id in node_ids: + return + node_ids.add(node.id) + nodes.append(node) + + def add_edge(edge: FieldLineageEdge) -> None: + if edge.id in edge_ids: + return + edge_ids.add(edge.id) + edges.append(edge) + + mappings = list_metadata_mappings(session, status="READY", limit=500, offset=0) + target_fields_by_canonical: dict[tuple[str, str], str] = {} + field_mapping_count = 0 + + for mapping_item in mappings.items: + mapping = get_metadata_mapping(session, mapping_item.id) + source_parent_id = _metadata_node_id(mapping.source_asset.id) + target_parent_id = _metadata_node_id(mapping.target_asset.id) + target_canonical = _canonical_name(mapping.target_asset.code) + + for field_mapping in mapping.field_mappings: + if field_mapping.source_field_code is None: + continue + field_mapping_count += 1 + source_field = field_mapping.source_field_code.lower() + target_field = field_mapping.target_field_code.lower() + source_id = _field_node_id(source_parent_id, source_field) + target_id = _field_node_id(target_parent_id, target_field) + + add_node( + FieldLineageNode( + id=source_id, + label=f"{mapping.source_asset.code}.{source_field}", + field_name=source_field, + node_type="SOURCE_FIELD", + layer="RAW", + system=mapping.source_asset.system_code, + relation=mapping.source_asset.code, + parent_node_id=source_parent_id, + parent_label=mapping.source_asset.code, + status="READY", + metadata={ + "mapping_code": mapping.code, + "transformation_type": field_mapping.transformation_type, + }, + ) + ) + add_node( + FieldLineageNode( + id=target_id, + label=f"{mapping.target_asset.code}.{target_field}", + field_name=target_field, + node_type="CURATED_FIELD", + layer="MART", + system=mapping.target_asset.system_code, + relation=mapping.target_asset.code, + parent_node_id=target_parent_id, + parent_label=mapping.target_asset.code, + description=field_mapping.description, + status="READY", + metadata={ + "target_data_type": field_mapping.target_data_type, + "transformation_type": field_mapping.transformation_type, + }, + ) + ) + add_edge( + FieldLineageEdge( + id=f"field-mapping:{field_mapping.id}", + upstream_id=source_id, + downstream_id=target_id, + edge_type="FIELD_MAPPING", + label=field_mapping.transformation_type, + ) + ) + target_fields_by_canonical[(target_canonical, target_field)] = target_id + + source_field_ids: dict[tuple[str, str], str] = {} + source_unique_ids: dict[str, str] = {} + for model in catalogue.models: + for dependency in model.upstream: + if dependency.resource_type == "SOURCE": + source_unique_ids.setdefault(dependency.name, dependency.unique_id) + + for source_name, unique_id in source_unique_ids.items(): + canonical = _canonical_name(source_name) + for (target_canonical, field_name), upstream_id in target_fields_by_canonical.items(): + if target_canonical != canonical: + continue + parent_id = _dbt_node_id(unique_id) + node_id = _field_node_id(parent_id, field_name) + source_field_ids[(source_name, field_name)] = node_id + add_node( + FieldLineageNode( + id=node_id, + label=f"{source_name}.{field_name}", + field_name=field_name, + node_type="DBT_SOURCE_FIELD", + layer="SOURCE", + system="dbt", + relation=f"mart.{source_name}", + parent_node_id=parent_id, + parent_label=source_name, + status="READY" if catalogue.artifact_status == "READY" else "MISSING", + metadata={"unique_id": unique_id}, + ) + ) + add_edge( + FieldLineageEdge( + id=f"field-publish:{upstream_id}:{node_id}", + upstream_id=upstream_id, + downstream_id=node_id, + edge_type="PUBLISHES_AS", + label="curated field → dbt source field", + ) + ) + + model_by_name = {model.name: model for model in catalogue.models} + model_field_ids: dict[tuple[str, str], str] = {} + dbt_annotated_column_count = 0 + + for model in catalogue.models: + parent_id = _dbt_node_id(model.unique_id) + for column in model.columns: + if not column.lineage_inputs: + continue + dbt_annotated_column_count += 1 + field_name = column.name.lower() + node_id = _field_node_id(parent_id, field_name) + model_field_ids[(model.name, field_name)] = node_id + add_node( + FieldLineageNode( + id=node_id, + label=f"{model.name}.{field_name}", + field_name=field_name, + node_type="DBT_MODEL_FIELD", + layer=model.layer, + system="dbt", + relation=model.relation, + parent_node_id=parent_id, + parent_label=model.name, + description=column.description, + status="READY" if model.build_status == "READY" else "UNKNOWN", + metadata={ + "materialization": model.materialization, + "lineage_inputs": len(column.lineage_inputs), + }, + ) + ) + + for model in catalogue.models: + for column in model.columns: + downstream_id = model_field_ids.get((model.name, column.name.lower())) + if downstream_id is None: + continue + for reference in column.lineage_inputs: + parsed = _parse_lineage_input(reference) + if parsed is None: + continue + kind, parent_name, field_name = parsed + upstream_id: str | None = None + if kind == "source": + source_name = parent_name.split(".")[-1] + upstream_id = source_field_ids.get((source_name, field_name.lower())) + else: + upstream_model = model_by_name.get(parent_name) + if upstream_model is not None: + upstream_id = model_field_ids.get( + (upstream_model.name, field_name.lower()) + ) + if upstream_id is None: + continue + add_edge( + FieldLineageEdge( + id=f"field-derive:{upstream_id}:{downstream_id}", + upstream_id=upstream_id, + downstream_id=downstream_id, + edge_type="DERIVES", + label="dbt column lineage", + ) + ) + + semantic_model_by_name = {model.name: model for model in semantic.semantic_models} + model_by_relation = {model.relation.lower(): model for model in catalogue.models} + metric_binding_count = 0 + for metric in semantic.metrics: + semantic_model = ( + semantic_model_by_name.get(metric.semantic_model) + if metric.semantic_model + else None + ) + if semantic_model is None or semantic_model.relation is None or not metric.expression: + continue + mart_model = model_by_relation.get(semantic_model.relation.lower()) + if mart_model is None: + continue + upstream_id = model_field_ids.get((mart_model.name, metric.expression.lower())) + if upstream_id is None: + continue + metric_binding_count += 1 + metric_parent_id = _metric_node_id(metric.unique_id) + metric_id = _field_node_id(metric_parent_id, metric.name) + add_node( + FieldLineageNode( + id=metric_id, + label=metric.label or metric.name, + field_name=metric.name, + node_type="METRIC", + layer="METRIC", + system="dbt semantic", + parent_node_id=metric_parent_id, + parent_label=metric.semantic_model, + description=metric.description, + status="READY" if semantic.artifact_status == "READY" else "MISSING", + metadata={ + "expression": metric.expression, + "aggregation": metric.aggregation, + "time_dimension": metric.time_dimension, + }, + ) + ) + add_edge( + FieldLineageEdge( + id=f"field-metric:{upstream_id}:{metric_id}", + upstream_id=upstream_id, + downstream_id=metric_id, + edge_type="FEEDS_METRIC", + label=f"{metric.aggregation or metric.metric_type}({metric.expression})", + ) + ) + + artifact_status: Literal["READY", "PARTIAL", "MISSING"] + if ( + catalogue.artifact_status == "READY" + and semantic.artifact_status == "READY" + and field_mapping_count > 0 + and dbt_annotated_column_count > 0 + ): + artifact_status = "READY" + elif nodes: + artifact_status = "PARTIAL" + else: + artifact_status = "MISSING" + + default_impact = _field_impact_from_graph(nodes, edges, focus_field_id) + return FieldLineageSummary( + artifact_status=artifact_status, + field_mapping_count=field_mapping_count, + dbt_annotated_column_count=dbt_annotated_column_count, + metric_binding_count=metric_binding_count, + node_count=len(nodes), + edge_count=len(edges), + nodes=nodes, + edges=edges, + default_impact=default_impact, + ) + + +def field_lineage_summary( + session: Session, + *, + focus_field_id: str | None = None, +) -> FieldLineageSummary: + return compose_field_lineage_summary( + session, + catalogue=dbt_model_catalogue(), + semantic=dbt_semantic_layer(), + focus_field_id=focus_field_id, + ) + + +def field_lineage_impact(session: Session, field_id: str) -> FieldLineageImpactSummary: + summary = field_lineage_summary(session, focus_field_id=field_id) + return summary.default_impact diff --git a/apps/web/src/components/FieldLineagePanel.jsx b/apps/web/src/components/FieldLineagePanel.jsx new file mode 100644 index 0000000..ffaf2d5 --- /dev/null +++ b/apps/web/src/components/FieldLineagePanel.jsx @@ -0,0 +1,220 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import StatusPill from './StatusPill.jsx'; +import { getJson } from '../services/api.js'; + +const EMPTY_IMPACT = { + selected_field_id: null, + selected_field_label: null, + downstream_node_count: 0, + affected_field_count: 0, + affected_metric_count: 0, + affected_relations: [], + affected_layers: [], + nodes: [], +}; + +const EMPTY = { + artifact_status: 'MISSING', + field_mapping_count: 0, + dbt_annotated_column_count: 0, + metric_binding_count: 0, + node_count: 0, + edge_count: 0, + nodes: [], + edges: [], + default_impact: EMPTY_IMPACT, +}; + +function fieldOrder(node) { + if (node.node_type === 'SOURCE_FIELD') return 1; + if (node.node_type === 'CURATED_FIELD') return 2; + if (node.node_type === 'DBT_SOURCE_FIELD') return 3; + if (node.layer === 'STAGING') return 4; + if (node.layer === 'INTERMEDIATE') return 5; + if (node.layer === 'MART') return 6; + if (node.node_type === 'METRIC') return 7; + return 99; +} + +function statusTone(status) { + if (status === 'READY') return 'READY'; + if (status === 'PARTIAL') return 'WARNING'; + return 'PLANNED'; +} + +function fieldKind(node) { + if (node.node_type === 'SOURCE_FIELD') return 'Source field'; + if (node.node_type === 'CURATED_FIELD') return 'Curated field'; + if (node.node_type === 'DBT_SOURCE_FIELD') return 'dbt source field'; + if (node.node_type === 'DBT_MODEL_FIELD') return 'dbt model field'; + return 'Governed metric'; +} + +function FieldLineagePanel() { + const [summary, setSummary] = useState(EMPTY); + const [impact, setImpact] = useState(EMPTY_IMPACT); + const [loading, setLoading] = useState(true); + const [impactLoading, setImpactLoading] = useState(false); + const [error, setError] = useState(''); + + const applySummary = useCallback((payload) => { + setSummary(payload); + setImpact(payload.default_impact || EMPTY_IMPACT); + }, []); + + const load = useCallback(async () => { + setLoading(true); + setError(''); + try { + applySummary(await getJson('/api/v1/lineage/fields/summary')); + } catch (requestError) { + setError(requestError.message); + } finally { + setLoading(false); + } + }, [applySummary]); + + useEffect(() => { + const controller = new AbortController(); + getJson('/api/v1/lineage/fields/summary', { signal: controller.signal }) + .then(applySummary) + .catch((requestError) => { + if (requestError.name !== 'AbortError') setError(requestError.message); + }) + .finally(() => { + if (!controller.signal.aborted) setLoading(false); + }); + return () => controller.abort(); + }, [applySummary]); + + const groups = useMemo(() => { + const grouped = new Map(); + [...summary.nodes] + .sort((left, right) => ( + fieldOrder(left) - fieldOrder(right) || left.label.localeCompare(right.label) + )) + .forEach((node) => { + const parent = node.parent_label || node.relation || node.layer; + const key = `${fieldOrder(node)}:${parent}`; + const current = grouped.get(key) || { key, label: parent, layer: node.layer, nodes: [] }; + current.nodes.push(node); + grouped.set(key, current); + }); + return [...grouped.values()]; + }, [summary.nodes]); + + const impactedIds = useMemo( + () => new Set((impact.nodes || []).map((node) => node.id)), + [impact.nodes], + ); + + async function selectField(node) { + setImpactLoading(true); + setError(''); + try { + const path = `/api/v1/lineage/fields/impact?fieldId=${encodeURIComponent(node.id)}`; + setImpact(await getJson(path)); + } catch (requestError) { + setError(requestError.message); + } finally { + setImpactLoading(false); + } + } + + return ( + <> +
+
+
+ FIELD LINEAGE · PHASE 8.2 +

DFF fields → curated columns → dbt derivations → governed metrics

+
+
+ + +
+
+ + {error ?
!{error}
: null} + +
+
FIELD NODES{summary.node_count}{summary.edge_count} directed edges
+
FIELD MAPS{summary.field_mapping_count}Studio mapping evidence
+
DBT COLUMNS{summary.dbt_annotated_column_count}Manifest annotations
+
METRIC BINDINGS{summary.metric_binding_count}Mart field consumers
+
+ + {groups.length ? ( +
+ {groups.map((group) => ( +
+
{group.layer}{group.label}
+
+ {group.nodes.map((node) => { + const selected = impact.selected_field_id === node.id; + const impacted = impactedIds.has(node.id); + return ( + + ); + })} +
+
+ ))} +
+ ) : ( +
+ No field-level lineage annotations are available yet. + Run .\scripts\dbt.ps1 build to regenerate the dbt manifest. +
+ )} +
+ +
+
+
+
FIELD IMPACT RADIUS

{impact.selected_field_label || 'Select a field'}

+ 8.2 +
+
+
DOWNSTREAM{impact.downstream_node_count}
+
FIELDS{impact.affected_field_count}
+
METRICS{impact.affected_metric_count}
+
RELATIONS{impact.affected_relations.length}
+
+

+ {impactLoading + ? 'Tracing column-level impact…' + : impact.downstream_node_count + ? `This field can affect ${impact.affected_layers.join(' → ')}.` + : 'This field has no downstream consumers in the current graph.'} +

+
+ +
+
+
FIELD IMPACT INVENTORY

Dependent fields and metrics

+ {impact.nodes.length} affected +
+
+ {impact.nodes.map((node) => ( +
{node.layer}{node.label}{fieldKind(node)}
+ ))} + {!impact.nodes.length ?

No downstream field impact for the selected field.

: null} +
+
+
+ + ); +} + +export default FieldLineagePanel; diff --git a/apps/web/src/components/Sidebar.jsx b/apps/web/src/components/Sidebar.jsx index 7756616..c15d392 100644 --- a/apps/web/src/components/Sidebar.jsx +++ b/apps/web/src/components/Sidebar.jsx @@ -122,7 +122,7 @@ function Sidebar({ mobileOpen, onClose }) { Development - Phase 8.1 cross-layer lineage + Phase 8.2 field-level lineage diff --git a/apps/web/src/pages/Lineage.jsx b/apps/web/src/pages/Lineage.jsx index 38377d3..e35c805 100644 --- a/apps/web/src/pages/Lineage.jsx +++ b/apps/web/src/pages/Lineage.jsx @@ -1,4 +1,5 @@ import { useCallback, useEffect, useMemo, useState } from 'react'; +import FieldLineagePanel from '../components/FieldLineagePanel.jsx'; import StatusPill from '../components/StatusPill.jsx'; import { getJson } from '../services/api.js'; @@ -118,11 +119,11 @@ function Lineage() {
- PHASE 8.1 · CROSS-LAYER LINEAGE + IMPACT RADIUS + PHASE 8.2 · FIELD-LEVEL LINEAGE + COLUMN IMPACT

Lineage & Impact

- Stitch Studio metadata mappings, dbt dependency artifacts, semantic models, and governed - metrics into one read-only graph, then trace what sits downstream before a change lands. + Keep the federated asset graph from Phase 8.1, then trace field-level impact from registered + mappings through dbt derivations and into governed metrics before a schema change lands.

+ +
-
ACCEPTANCE CONTRACT

What Phase 8.1 proves

8.1
+
ACCEPTANCE CONTRACT

What Phase 8.2 proves

8.2
    -
  1. 01
    Authorities remain separateStudio mappings, dbt artifacts, and semantic definitions stay at their existing owners while one graph projects them together.
    2. -
  2. 02
    Physical seams connectThe curated Studio mart bridges the registered source mapping to the dbt source and model DAG.
    3. -
  3. 03
    Impact is transitiveSelecting any node walks the directed graph to expose downstream models, semantic models, metrics, and affected layers.
    4. +
  4. 01
    Field mappings stay authoritativeStudio reuses the two registered source-to-target field maps rather than reconstructing ingestion semantics.
    5. +
  5. 02
    dbt owns derived columnsColumn lineage declarations live beside dbt models and are carried through the generated manifest.
    6. +
  6. 03
    Metrics become field consumersMetric expressions connect mart fields to business measures so field-level downstream impact is explicit.
BOUNDARY

Read the graph; do not rewrite it

⌁

- Phase 8.1 is a federated read model. It does not copy dbt dependency metadata into Studio PostgreSQL or make lineage its own new authority. Field-level impact, quality overlays, and consumer/report lineage remain later Phase 8 slices. + Phase 8.2 remains a federated read model. Studio does not infer arbitrary SQL or persist a second lineage store; mapping fields, dbt column annotations, and metric expressions stay at their existing authorities. Quality overlays and consumer/report lineage remain later Phase 8 slices.

Federated evidenceDirected impactNo duplicate authority
diff --git a/apps/web/src/pages/StudioOverview.jsx b/apps/web/src/pages/StudioOverview.jsx index eb28678..27bdae1 100644 --- a/apps/web/src/pages/StudioOverview.jsx +++ b/apps/web/src/pages/StudioOverview.jsx @@ -5,7 +5,7 @@ const FALLBACK = { product: 'SkyData Studio', subtitle: 'Data Engineering Workbench', theme: 'Aurora Foundry', - current_phase: 'Phase 7.4 — Quality SLO and Reliability History', + current_phase: 'Phase 8.2 — Field-Level Lineage and Column Impact Foundation', boundary: 'SkyData Studio starts after SkyCommand ingestion and publishes governed analytical products for downstream consumers.', capabilities: [], }; @@ -41,7 +41,7 @@ function StudioOverview() {
- AURORA FOUNDRY · PHASE 8.1 + AURORA FOUNDRY · PHASE 8.2

Shape trusted data into analytical products.

{summary.boundary}

@@ -61,7 +61,7 @@ function StudioOverview() {
Platform API{apiState}FastAPI contract service
-
Current Phase8.1cross-layer lineage + impact radius
+
Current Phase8.2field-level lineage + column impact
Pipeline ProofREADYCurated mart materialization proven
OrchestratorAirflow 3Batch and asset workflows
@@ -96,7 +96,7 @@ function StudioOverview() { { code: 'AIRFLOW', name: 'Apache Airflow Orchestration', description: 'DAGs, assets, schedules, retries, and backfills.', status: 'FOUNDATION', phase: 5 }, { code: 'DBT', name: 'dbt Transformation Layer', description: 'Tested staging, intermediate, mart, and semantic models.', status: 'FOUNDATION', phase: 6 }, { code: 'QUALITY_LINEAGE', name: 'Data Quality and Reliability', description: 'Trust evidence, contracts, incidents, and reliability history.', status: 'READY', phase: 7 }, - { code: 'LINEAGE_IMPACT', name: 'Lineage and Impact Analysis', description: 'Federated metadata, dbt, semantic, and metric dependency graph.', status: 'FOUNDATION', phase: 8 }, + { code: 'LINEAGE_IMPACT', name: 'Lineage and Impact Analysis', description: 'Federated asset and field lineage from mappings through dbt and metrics.', status: 'FOUNDATION', phase: 8 }, { code: 'ANALYTICS_DELIVERY', name: 'Analytics Delivery', description: 'Governed products for SkyWeb Analytics and Power BI.', status: 'PLANNED', phase: 9 }, ]).map((capability) => (
@@ -111,11 +111,11 @@ function StudioOverview() {
-
CURRENT IMPLEMENTATION

cross-layer lineage and impact radius

PHASE 8.1
+
CURRENT IMPLEMENTATION

field-level lineage and column impact

PHASE 8.2
    -
  1. 01
    Federated lineage evidenceStudio mappings, dbt dependencies, semantic models, and metrics are projected into one graph without duplicating authority.
    2. -
  2. 02
    Curated seam stitchingThe registered DFF → mart mapping connects cleanly into the dbt source and transformation DAG.
    3. -
  3. 03
    Transitive impact radiusSelect a node to see every downstream model, semantic model, metric, and affected engineering layer.
    4. +
  4. 01
    Field mapping evidenceThe registered DFF field maps seed column lineage without duplicating ingestion ownership.
    5. +
  5. 02
    dbt column annotationsDerived-column lineage lives beside dbt model definitions and is projected from the generated manifest.
    6. +
  6. 03
    Metric-level impactSelect a field to trace dependent dbt fields and governed business metrics before a schema change lands.
diff --git a/apps/web/src/styles.css b/apps/web/src/styles.css index cb661f7..c8b4ddb 100644 --- a/apps/web/src/styles.css +++ b/apps/web/src/styles.css @@ -1290,3 +1290,152 @@ a { color: inherit; } grid-template-columns: 1fr 1fr; } } + +/* Phase 8.2 — field-level lineage + column impact */ +.field-lineage-actions { + display: flex; + align-items: center; + gap: 10px; +} + +.field-lineage-metrics { + display: grid; + grid-template-columns: repeat(4, minmax(0, 1fr)); + gap: 9px; + margin: 12px 0 16px; +} + +.field-lineage-metrics > div { + display: grid; + gap: 5px; + padding: 12px; + background: rgba(4, 17, 27, 0.55); + border: 1px solid rgba(116, 171, 186, 0.12); + border-radius: 9px; +} + +.field-lineage-metrics span { + color: var(--text-muted); + font-size: 9px; + font-weight: 850; + letter-spacing: 0.1em; +} + +.field-lineage-metrics strong { + color: var(--accent-bright); + font-size: 18px; +} + +.field-lineage-metrics small { + color: var(--text-muted); + font-size: 10px; +} + +.field-lineage-groups { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + gap: 12px; +} + +.field-lineage-group { + display: grid; + gap: 10px; + padding: 13px; + background: rgba(4, 17, 27, 0.5); + border: 1px solid rgba(116, 171, 186, 0.13); + border-radius: 11px; +} + +.field-lineage-group header { + display: flex; + align-items: center; + justify-content: space-between; + gap: 10px; +} + +.field-lineage-group header span { + color: var(--text-muted); + font-size: 9px; + font-weight: 850; + letter-spacing: 0.1em; +} + +.field-lineage-group header strong { + color: var(--accent-bright); + font-size: 12px; + overflow-wrap: anywhere; +} + +.field-lineage-items { + display: flex; + flex-wrap: wrap; + gap: 7px; +} + +.field-lineage-node { + display: grid; + gap: 3px; + min-width: 118px; + padding: 9px 10px; + text-align: left; + color: var(--text-primary); + background: rgba(8, 26, 37, 0.88); + border: 1px solid rgba(84, 211, 198, 0.16); + border-radius: 8px; + cursor: pointer; + transition: border-color 140ms ease, transform 140ms ease, box-shadow 140ms ease; +} + +.field-lineage-node:hover { + transform: translateY(-1px); + border-color: rgba(84, 211, 198, 0.48); +} + +.field-lineage-node.is-impacted { + border-color: rgba(151, 118, 255, 0.4); + box-shadow: inset 0 0 0 1px rgba(151, 118, 255, 0.07); +} + +.field-lineage-node.is-selected { + border-color: rgba(84, 231, 210, 0.88); + box-shadow: 0 0 0 1px rgba(84, 231, 210, 0.15); +} + +.field-lineage-node strong { + color: var(--text-primary); + font-size: 11px; + overflow-wrap: anywhere; +} + +.field-lineage-node small { + color: var(--text-muted); + font-size: 9px; +} + +.field-impact-list { + max-height: 390px; +} + +@media (max-width: 1180px) { + .field-lineage-metrics, + .field-lineage-groups { + grid-template-columns: repeat(2, minmax(0, 1fr)); + } +} + +@media (max-width: 760px) { + .field-lineage-actions { + align-items: flex-start; + flex-direction: column; + } + + .field-lineage-metrics, + .field-lineage-groups { + grid-template-columns: 1fr; + } + + .field-lineage-node { + min-width: 0; + flex: 1 1 130px; + } +} diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 52f2a45..3074191 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -147,7 +147,22 @@ DFF → FED_FUNDS_RATE_MART → fed_funds_rate source → fed_funds_ra The bridge from the Studio curated asset to the dbt source is a read-time seam. Metadata mappings remain Studio-owned; dbt `manifest.json` remains the transformation dependency authority; semantic definitions remain dbt-owned. The lineage service normalizes those identifiers into directed nodes and edges and uses a downstream breadth-first traversal for impact radius. No graph rows are persisted in Phase 8.1. -This first slice answers asset/model/semantic/metric impact. Field mappings, quality/incident overlays, pipeline/Airflow execution impact, and report/Power BI consumers remain later Phase 8 boundaries. +This first slice answers asset/model/semantic/metric impact. Quality/incident overlays, pipeline/Airflow execution impact, and report/Power BI consumers remain later Phase 8 boundaries. + +## Phase 8.2 field-level lineage boundary + +Phase 8.2 carries the same federated ownership model down to columns. Studio does not parse arbitrary SQL or create a new column-lineage table. The source-to-curated field seam comes from existing `metadata_field_mapping` rows, while dbt model columns declare their upstream fields beside the model definitions through `meta.lineage_inputs`. `dbt build` carries those annotations into `manifest.json`, so Studio reads the generated artifact it already trusts for model metadata. + +```text +DFF.OBSERVATION_DATE → FED_FUNDS_RATE_MART.observation_date → dbt date fields + └→ observation_key → count metric + +DFF.VALUE → FED_FUNDS_RATE_MART.rate → dbt rate fields + ├→ previous/change/direction fields + └→ average/minimum/maximum metrics +``` + +Metric dependencies come from dbt metric expressions, not a Studio-owned semantic mapping. The field-lineage service normalizes these authorities into ephemeral field nodes and edges and performs the same downstream breadth-first traversal used by the Phase 8.1 asset graph. ## Product blueprint and mapping boundary diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index 633955d..6c3a8c8 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -573,7 +573,7 @@ Closure evidence: ## Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation -**Status:** In progress. +**Status:** Complete. Live graph proof, transitive impact evidence, and final local validation are green. Scope: @@ -596,3 +596,34 @@ Acceptance proof: - the Lineage workbench renders the federated graph and recomputes downstream impact when a node is selected; - local validation remains green with 76 pytest tests plus Ruff, mypy, npm audit, ESLint, and Vite build. +Closure evidence: + +- `GET /api/v1/lineage/summary` returned `artifact_status=READY`, 1 metadata mapping, 3 dbt models, 1 semantic model, 4 metrics, 11 nodes, and 10 edges; +- the default DFF impact radius returned 10 downstream nodes, 3 affected dbt models, 1 semantic model, and 4 governed metrics; +- the Phase 8.1 Ruff line-length cleanup changed formatting only; +- final local validation passed Ruff, mypy across 57 source files, 76 pytest tests, npm audit with zero vulnerabilities, ESLint, and Vite production build. + +## Phase 8.2 — Field-Level Lineage and Column Impact Foundation + +**Status:** In progress. + +Scope: + +- preserve the Phase 8.1 asset/model/semantic/metric graph and add a separate field-level read model rather than persisting a second lineage authority; +- use existing Studio field-mapping rows as the source-to-curated field authority for `OBSERVATION_DATE → OBSERVATION_DATE` and `VALUE → RATE`; +- declare dbt derived-column inputs beside model definitions through column `meta.lineage_inputs`, then read those declarations from generated `manifest.json`; +- expand intermediate and mart column documentation so every field in the proof has an explicit lineage declaration; +- bind final mart fields to governed metrics through dbt metric expressions rather than introducing a Studio-owned semantic dependency table; +- expose `GET /api/v1/lineage/fields/summary` and `GET /api/v1/lineage/fields/impact?fieldId=...`; +- extend the Lineage workbench with grouped field nodes and transitive field-impact evidence; +- keep quality overlays, Airflow runtime lineage, and report/Power BI consumers outside this field-level foundation. + +Acceptance proof: + +- a fresh `dbt build` remains green and regenerates manifest column annotations while retaining 4 models, 14 data tests, 4 metrics, and 1 semantic model; +- the field lineage API reports `artifact_status=READY`, 2 Studio field mappings, 18 annotated dbt business-model columns, 4 metric bindings, 28 nodes, and 27 directed edges; +- selecting `DFF.value` reaches 15 downstream nodes including rate-derived columns and exactly 3 rate metrics; +- selecting `DFF.observation_date` reaches the observation-key path and observation-count metric without falsely reporting rate metrics; +- the Phase 8.1 asset graph remains READY at 11 nodes / 10 edges; +- local validation remains green with 80 pytest tests plus Ruff, mypy, npm audit, ESLint, and Vite build. + diff --git a/docs/SkyData_RepoMap.md b/docs/SkyData_RepoMap.md index d2f4349..e8c4446 100644 --- a/docs/SkyData_RepoMap.md +++ b/docs/SkyData_RepoMap.md @@ -87,6 +87,7 @@ SkyDataStudio/ │ ├── styles.css │ ├── components/ │ │ ├── BrandMark.jsx +│ │ ├── FieldLineagePanel.jsx │ │ ├── Sidebar.jsx │ │ ├── StatusPill.jsx │ │ └── Topbar.jsx diff --git a/docs/repository-map.txt b/docs/repository-map.txt index 9792aba..a69b29a 100644 --- a/docs/repository-map.txt +++ b/docs/repository-map.txt @@ -74,6 +74,7 @@ SkyDataStudio │ ├── src │ │ ├── components │ │ │ ├── BrandMark.jsx +│ │ │ ├── FieldLineagePanel.jsx │ │ │ ├── Sidebar.jsx │ │ │ ├── StatusPill.jsx │ │ │ └── Topbar.jsx diff --git a/tests/test_lineage.py b/tests/test_lineage.py index a66ada6..144b62c 100644 --- a/tests/test_lineage.py +++ b/tests/test_lineage.py @@ -5,13 +5,17 @@ from skydata_studio.schemas.dbt import ( DbtModelCatalogueItem, DbtModelCatalogueSummary, + DbtModelColumnSummary, DbtModelDependencySummary, DbtSemanticLayerSummary, DbtSemanticMetricSummary, DbtSemanticModelSummary, ) from skydata_studio.schemas.metadata import MetadataAssetCreate, MetadataMappingCreate -from skydata_studio.services.lineage import compose_lineage_summary +from skydata_studio.services.lineage import ( + compose_field_lineage_summary, + compose_lineage_summary, +) from skydata_studio.services.metadata_registry import ( create_metadata_mapping, register_metadata_asset, @@ -304,3 +308,300 @@ def test_lineage_metadata_only_state_is_partial(lineage_session: Session) -> Non assert summary.node_count == 2 assert summary.edge_count == 1 assert summary.default_impact.downstream_node_count == 1 + + +def _field_catalogue() -> DbtModelCatalogueSummary: + source = DbtModelDependencySummary( + unique_id="source.skydata_studio.studio_curated.fed_funds_rate", + name="fed_funds_rate", + resource_type="SOURCE", + ) + staging = DbtModelDependencySummary( + unique_id="model.skydata_studio.stg_fed_funds_rate", + name="stg_fed_funds_rate", + resource_type="MODEL", + ) + intermediate = DbtModelDependencySummary( + unique_id="model.skydata_studio.int_fed_funds_rate_changes", + name="int_fed_funds_rate_changes", + resource_type="MODEL", + ) + mart = DbtModelDependencySummary( + unique_id="model.skydata_studio.fct_fed_funds_rate_daily", + name="fct_fed_funds_rate_daily", + resource_type="MODEL", + ) + models = [ + DbtModelCatalogueItem( + unique_id=staging.unique_id, + name=staging.name, + layer="STAGING", + relation="dbt_staging.stg_fed_funds_rate", + materialization="VIEW", + build_status="READY", + path="models/staging/stg_fed_funds_rate.sql", + tags=[], + columns=[ + DbtModelColumnSummary( + name="observation_date", + lineage_inputs=["source:fed_funds_rate.observation_date"], + ), + DbtModelColumnSummary( + name="rate", + lineage_inputs=["source:fed_funds_rate.rate"], + ), + ], + upstream=[source], + downstream=[intermediate], + test_count=3, + ), + DbtModelCatalogueItem( + unique_id=intermediate.unique_id, + name=intermediate.name, + layer="INTERMEDIATE", + relation="dbt_intermediate.int_fed_funds_rate_changes", + materialization="VIEW", + build_status="READY", + path="models/intermediate/int_fed_funds_rate_changes.sql", + tags=[], + columns=[ + DbtModelColumnSummary( + name="observation_date", + lineage_inputs=["model:stg_fed_funds_rate.observation_date"], + ), + DbtModelColumnSummary( + name="observation_month", + lineage_inputs=["model:stg_fed_funds_rate.observation_date"], + ), + DbtModelColumnSummary( + name="observation_year", + lineage_inputs=["model:stg_fed_funds_rate.observation_date"], + ), + DbtModelColumnSummary( + name="rate", + lineage_inputs=["model:stg_fed_funds_rate.rate"], + ), + DbtModelColumnSummary( + name="previous_rate", + lineage_inputs=["model:stg_fed_funds_rate.rate"], + ), + DbtModelColumnSummary( + name="rate_change", + lineage_inputs=["model:stg_fed_funds_rate.rate"], + ), + DbtModelColumnSummary( + name="rate_change_bps", + lineage_inputs=["model:stg_fed_funds_rate.rate"], + ), + ], + upstream=[staging], + downstream=[mart], + test_count=3, + ), + DbtModelCatalogueItem( + unique_id=mart.unique_id, + name=mart.name, + layer="MART", + relation="dbt_mart.fct_fed_funds_rate_daily", + materialization="TABLE", + build_status="READY", + path="models/marts/fct_fed_funds_rate_daily.sql", + tags=[], + columns=[ + DbtModelColumnSummary( + name="observation_key", + lineage_inputs=["model:int_fed_funds_rate_changes.observation_date"], + ), + DbtModelColumnSummary( + name="observation_date", + lineage_inputs=["model:int_fed_funds_rate_changes.observation_date"], + ), + DbtModelColumnSummary( + name="observation_month", + lineage_inputs=["model:int_fed_funds_rate_changes.observation_month"], + ), + DbtModelColumnSummary( + name="observation_year", + lineage_inputs=["model:int_fed_funds_rate_changes.observation_year"], + ), + DbtModelColumnSummary( + name="rate", + lineage_inputs=["model:int_fed_funds_rate_changes.rate"], + ), + DbtModelColumnSummary( + name="previous_rate", + lineage_inputs=["model:int_fed_funds_rate_changes.previous_rate"], + ), + DbtModelColumnSummary( + name="rate_change", + lineage_inputs=["model:int_fed_funds_rate_changes.rate_change"], + ), + DbtModelColumnSummary( + name="rate_change_bps", + lineage_inputs=["model:int_fed_funds_rate_changes.rate_change_bps"], + ), + DbtModelColumnSummary( + name="rate_direction", + lineage_inputs=[ + "model:int_fed_funds_rate_changes.rate", + "model:int_fed_funds_rate_changes.previous_rate", + ], + ), + ], + upstream=[intermediate], + downstream=[], + test_count=5, + ), + ] + return DbtModelCatalogueSummary( + artifact_status="READY", + model_count=3, + ready_model_count=3, + source_count=1, + test_count=14, + models=models, + ) + + +def _field_semantic() -> DbtSemanticLayerSummary: + semantic_name = "fed_funds_rate_daily" + semantic_model = DbtSemanticModelSummary( + unique_id="semantic_model.skydata_studio.fed_funds_rate_daily", + name=semantic_name, + relation="dbt_mart.fct_fed_funds_rate_daily", + entities=[], + dimensions=[], + metric_names=[ + "average_federal_funds_rate", + "minimum_federal_funds_rate", + "maximum_federal_funds_rate", + "federal_funds_observation_count", + ], + ) + metrics = [ + DbtSemanticMetricSummary( + unique_id="metric.skydata_studio.average_federal_funds_rate", + name="average_federal_funds_rate", + label="Average Federal Funds Rate", + metric_type="SIMPLE", + aggregation="AVERAGE", + expression="rate", + semantic_model=semantic_name, + ), + DbtSemanticMetricSummary( + unique_id="metric.skydata_studio.minimum_federal_funds_rate", + name="minimum_federal_funds_rate", + label="Minimum Federal Funds Rate", + metric_type="SIMPLE", + aggregation="MIN", + expression="rate", + semantic_model=semantic_name, + ), + DbtSemanticMetricSummary( + unique_id="metric.skydata_studio.maximum_federal_funds_rate", + name="maximum_federal_funds_rate", + label="Maximum Federal Funds Rate", + metric_type="SIMPLE", + aggregation="MAX", + expression="rate", + semantic_model=semantic_name, + ), + DbtSemanticMetricSummary( + unique_id="metric.skydata_studio.federal_funds_observation_count", + name="federal_funds_observation_count", + label="Federal Funds Observation Count", + metric_type="SIMPLE", + aggregation="COUNT_DISTINCT", + expression="observation_key", + semantic_model=semantic_name, + ), + ] + return DbtSemanticLayerSummary( + artifact_status="READY", + semantic_model_count=1, + metric_count=4, + entity_count=0, + dimension_count=0, + semantic_models=[semantic_model], + metrics=metrics, + ) + + +def test_field_lineage_composes_mapping_dbt_columns_and_metrics( + lineage_session: Session, +) -> None: + _register_proof_mapping(lineage_session) + summary = compose_field_lineage_summary( + lineage_session, + catalogue=_field_catalogue(), + semantic=_field_semantic(), + ) + assert summary.artifact_status == "READY" + assert summary.field_mapping_count == 2 + assert summary.dbt_annotated_column_count == 18 + assert summary.metric_binding_count == 4 + assert summary.node_count == 28 + assert summary.edge_count == 27 + + +def test_field_lineage_value_impact_reaches_three_rate_metrics( + lineage_session: Session, +) -> None: + _register_proof_mapping(lineage_session) + summary = compose_field_lineage_summary( + lineage_session, + catalogue=_field_catalogue(), + semantic=_field_semantic(), + ) + value_node = next(node for node in summary.nodes if node.label == "DFF.value") + focused = compose_field_lineage_summary( + lineage_session, + catalogue=_field_catalogue(), + semantic=_field_semantic(), + focus_field_id=value_node.id, + ) + impact = focused.default_impact + assert impact.downstream_node_count == 15 + assert impact.affected_metric_count == 3 + labels = {node.label for node in impact.nodes} + assert "fct_fed_funds_rate_daily.rate_direction" in labels + assert "Federal Funds Observation Count" not in labels + + +def test_field_lineage_date_impact_reaches_observation_count( + lineage_session: Session, +) -> None: + _register_proof_mapping(lineage_session) + summary = compose_field_lineage_summary( + lineage_session, + catalogue=_field_catalogue(), + semantic=_field_semantic(), + ) + date_node = next( + node for node in summary.nodes if node.label == "DFF.observation_date" + ) + focused = compose_field_lineage_summary( + lineage_session, + catalogue=_field_catalogue(), + semantic=_field_semantic(), + focus_field_id=date_node.id, + ) + labels = {node.label for node in focused.default_impact.nodes} + assert focused.default_impact.affected_metric_count == 1 + assert "Federal Funds Observation Count" in labels + assert "Average Federal Funds Rate" not in labels + + +def test_field_lineage_without_dbt_annotations_is_partial(lineage_session: Session) -> None: + _register_proof_mapping(lineage_session) + summary = compose_field_lineage_summary( + lineage_session, + catalogue=_catalogue(), + semantic=_semantic(), + ) + assert summary.artifact_status == "PARTIAL" + assert summary.field_mapping_count == 2 + assert summary.dbt_annotated_column_count == 0 + assert summary.metric_binding_count == 0 + assert summary.node_count == 6 + assert summary.edge_count == 4 diff --git a/tests/test_platform.py b/tests/test_platform.py index bedc68d..7c31834 100644 --- a/tests/test_platform.py +++ b/tests/test_platform.py @@ -13,7 +13,7 @@ def test_platform_summary_exposes_product_boundary() -> None: assert payload["theme"] == "Aurora Foundry" assert "SkyCommand" in payload["boundary"] assert payload["current_phase"] == ( - "Phase 8.1 — Cross-Layer Lineage Graph and Impact Radius Foundation" + "Phase 8.2 — Field-Level Lineage and Column Impact Foundation" ) assert {item["code"] for item in payload["capabilities"]} >= { "AIRFLOW", diff --git a/transformations/dbt/skydata_studio/models/intermediate/schema.yml b/transformations/dbt/skydata_studio/models/intermediate/schema.yml index de1f6b1..97aac95 100644 --- a/transformations/dbt/skydata_studio/models/intermediate/schema.yml +++ b/transformations/dbt/skydata_studio/models/intermediate/schema.yml @@ -5,9 +5,42 @@ models: description: Reusable Federal Funds Rate model with prior-day and change attributes. columns: - name: observation_date + description: Observation date preserved from the staged daily grain. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.observation_date data_tests: - not_null - unique + - name: observation_month + description: Calendar month derived from the observation date. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.observation_date + - name: observation_year + description: Calendar year derived from the observation date. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.observation_date - name: rate + description: Federal Funds Effective Rate preserved from staging. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.rate data_tests: - not_null + - name: previous_rate + description: Prior available Federal Funds Effective Rate calculated with a window lag. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.rate + - name: rate_change + description: Change from the prior available rate in percentage points. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.rate + - name: rate_change_bps + description: Change from the prior available rate expressed in basis points. + meta: + lineage_inputs: + - model:stg_fed_funds_rate.rate diff --git a/transformations/dbt/skydata_studio/models/marts/schema.yml b/transformations/dbt/skydata_studio/models/marts/schema.yml index e5da61f..f028256 100644 --- a/transformations/dbt/skydata_studio/models/marts/schema.yml +++ b/transformations/dbt/skydata_studio/models/marts/schema.yml @@ -19,11 +19,17 @@ models: columns: - name: observation_key description: Stable one-row-per-day semantic entity key derived from the observation date. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.observation_date entity: name: fed_funds_observation type: primary - name: observation_date description: "Grain and business key: one row per observation date." + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.observation_date granularity: day dimension: type: time @@ -32,19 +38,47 @@ models: - unique - name: observation_month description: Calendar month containing the daily observation. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.observation_month granularity: month dimension: type: time - name: observation_year description: Calendar year containing the daily observation. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.observation_year dimension: type: categorical - name: rate description: Federal Funds Effective Rate for the observation date. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.rate data_tests: - not_null + - name: previous_rate + description: Prior available Federal Funds Effective Rate carried into the fact model. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.previous_rate + - name: rate_change + description: Change from the prior available rate in percentage points. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.rate_change + - name: rate_change_bps + description: Change from the prior available rate expressed in basis points. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.rate_change_bps - name: rate_direction description: Direction of movement compared with the prior available observation. + meta: + lineage_inputs: + - model:int_fed_funds_rate_changes.rate + - model:int_fed_funds_rate_changes.previous_rate dimension: type: categorical data_tests: diff --git a/transformations/dbt/skydata_studio/models/staging/schema.yml b/transformations/dbt/skydata_studio/models/staging/schema.yml index a2f63a3..ee3ce23 100644 --- a/transformations/dbt/skydata_studio/models/staging/schema.yml +++ b/transformations/dbt/skydata_studio/models/staging/schema.yml @@ -6,10 +6,16 @@ models: columns: - name: observation_date description: Unique observation date. + meta: + lineage_inputs: + - source:fed_funds_rate.observation_date data_tests: - not_null - unique - name: rate description: Federal Funds Effective Rate normalized to NUMERIC(10,4). + meta: + lineage_inputs: + - source:fed_funds_rate.rate data_tests: - not_null