Data Lineage: A Food Safety Analogy
================================================

Olive Oil Traceability:
  Olive tree (Andalusia, Spain, Plot #A17)
    -> Harvested (2024-10-15, Worker Team #3)
    -> Pressed (cold-press, Machine #M7, 27 C)
    -> Filtered (triple filtration, Batch #B2024-1015)
    -> Bottled (500ml, Line #L2, Batch #F-88721)
    -> QA tested (acidity 0.3%, passed)
    -> Shipped (cold chain, Vehicle #T-9981)
    -> Supermarket shelf (the bottle you see)

Data Lineage:
  Raw data (ERP system, table sales_orders)
    -> Cleaned (deduplicated, null handling, Pipeline #P-001)
    -> Transformed (currency conversion, unit normalization, Transform #T-023)
    -> Aggregated (by region, monthly cumulative, Transform #T-024)
    -> Joined (with customer dimension table, Transform #T-025)
    -> Modeled (calculate customer LTV, Model #M-012)
    -> Displayed (CEO dashboard: "Monthly Revenue: $12.5M")

Entity-Level Lineage (Dataset-Level)
================================================

Tracks "relationships between datasets":

  Dataset A ------> Transform 1 ------> Dataset C
                                           |
  Dataset B ------> Transform 2 -----------+

  Question: "Which upstream datasets does Dataset C depend on?"
  Answer: "Dataset A and Dataset B"

  Granularity: Coarse  |  Cost: Low  |  Value: Moderate


Field-Level Lineage (Column-Level)
================================================

Tracks "relationships between fields":

  Dataset A                    Dataset C
  +----------------+          +----------------+
  | order_id       |--------->| order_id       |
  | raw_amount     |--+       |                |
  | currency       |--+-Xform>| amount_usd     |
  +----------------+  |       |                |
                      |       |                |
  Dataset B           |       |                |
  +----------------+  |       |                |
  | exchange_rate  |--+       |                |
  | customer_name  |--------->| customer_name  |
  +----------------+          +----------------+

  Question: "How was amount_usd calculated?"
  Answer: "raw_amount x exchange_rate, where raw_amount comes from
           Dataset A and exchange_rate comes from Dataset B"

  Granularity: Fine  |  Cost: High  |  Value: Very High

Compliance Scenario
================================================

Scenario: Banking Regulatory Review

Regulator: "The non-performing loan ratio of 2.3% you reported --
            how was this number calculated?"

Without lineage:
  Analyst: "Uh... let me check... probably from System A..."
  (3 days later) "We confirmed it's from System A, but there was
   manual processing by Department B. We need to verify the logic..."
  (1 week later) "Sorry, the person who owned this process left..."

  Result: Regulatory penalty, reputation damage

With lineage:
  Analyst: (clicks the number) "Here's the full calculation chain:
  1. Source: Core banking system, loan_master table
  2. Cleaning: Removed test accounts (Pipeline #P-1234, last run 2024-03-15)
  3. Classification: Overdue >90 days = non-performing (Transform #T-5678)
  4. Aggregation: NPL total / Total loans (Transform #T-5679)
  5. Result: 2.3% (Updated: 2024-03-20 08:00)"

  Result: Audit completed in 30 minutes

Debugging Scenario
================================================

Problem: "Why did the monthly sales report drop 30% suddenly?"

Without lineage:
  Must manually check every step:
  1. Report query logic -> seems fine
  2. Data warehouse table -> row counts look right
  3. ETL jobs -> which ETL? There are 47 of them...
  4. Source systems -> which source? There are 12...

  Time spent: 2-3 days

With lineage:
  1. Click report number -> see dependency chain
  2. Discover Transform #T-456's input dataset
     was last updated 3 days ago (previously daily)
  3. Trace upstream from Transform #T-456 ->
     find that CRM system data sync job failed
  4. Root cause: CRM system upgrade changed API format

  Time spent: 15 minutes

Impact Analysis Scenario
================================================

Question: "If I change the customer_address field format,
           what will be affected?"

Without lineage:
  "I don't know, let me ask every team..."
  (might miss some, causing downstream systems to crash)

With lineage:
  +---------------------+
  | customer_address    |
  | (source: CRM)       |
  +----------+----------+
             |
    +--------+--------------------+
    |        |                    |
    v        v                    v
  Logistics  Customer Profile   Compliance Reports
  +-Address   +-Customer 360     +-AML Report
  | parsing   +-Regional         +-Customer Due
  +-Delivery  | analysis         | Diligence
  | routing   +-Email            +
  +-Freight     marketing
    calc

  Impact scope: 3 systems, 7 downstream transforms, 12 reports
  Risk level: HIGH (includes compliance reports)
  Recommendation: Notify teams, schedule compatibility change window

# Foundry Transform Example -- Automatic Lineage Capture

from transforms.api import Transform, Input, Output

@transform(
    order_input=Input("/datasets/raw/sales_orders"),
    rate_input=Input("/datasets/ref/exchange_rates"),
    output=Output("/datasets/clean/orders_usd")
)
def compute_orders_usd(order_input, rate_input, output):
    """
    Foundry automatically captures lineage:

    Input datasets:
      - /datasets/raw/sales_orders
      - /datasets/ref/exchange_rates

    Output dataset:
      - /datasets/clean/orders_usd

    Transform code:
      - Stored in version control
      - Each execution records code version
    """
    orders = order_input.dataframe()
    rates = rate_input.dataframe()

    # This JOIN is automatically recorded as field-level lineage
    result = orders.join(rates, on="currency")

    # This computation is automatically recorded:
    # amount_usd = amount * rate
    result = result.withColumn(
        "amount_usd",
        result["amount"] * result["rate"]
    )

    output.write_dataframe(result)

# Metadata automatically recorded by Foundry:
# {
#   "transform_rid": "ri.foundry.main.transform.abc123",
#   "inputs": ["ri.foundry.main.dataset.orders", "ri.foundry.main.dataset.rates"],
#   "output": "ri.foundry.main.dataset.orders_usd",
#   "code_version": "git:abc123def",
#   "execution_time": "2024-03-20T08:00:00Z",
#   "field_lineage": {
#     "order_id": {"source": "orders.order_id", "transform": "passthrough"},
#     "amount_usd": {"source": ["orders.amount", "rates.rate"], "transform": "multiply"}
#   }
# }

Ontology Lineage Tracking
================================================

Raw Data Layer:
  +---------+   +----------+   +---------+
  | CRM     |   | ERP      |   | IoT     |
  | System  |   | System   |   | Platform|
  +----+----+   +-----+----+   +----+----+
       |              |             |
       v              v             v
Cleaning / Transform Layer:
  +---------+   +----------+   +---------+
  |Customer |   |Order     |   |Device   |
  |Data     |   |Data      |   |Data     |
  |(cleaned)|   |(cleaned) |   |(cleaned)|
  +----+----+   +-----+----+   +----+----+
       |              |             |
       v              v             v
Ontology Object Layer:
  +------------------------------------------+
  |                                          |
  |  Customer Object       Order Object      |
  |  +----------------+   +---------------+  |
  |  | name           |   | order_id      |  |
  |  | address        |   | amount        |  |
  |  | lifetime_value |   | status        |  |
  |  | risk_score     |   | customer_ref  |  |
  |  +----------------+   +---------------+  |
  |         |                    |           |
  |         +---- places -------+           |
  |                                          |
  |  lifetime_value lineage:                 |
  |  = SUM(Order.amount)                     |
  |    WHERE Order.customer_ref = this       |
  |    Source: ERP.sales_orders.total_amount  |
  |    Transforms: FX conversion -> monthly  |
  |                aggregation -> cumulative |
  |                                          |
  +------------------------------------------+

Foundry Lineage Visualization UI (Conceptual)
================================================

+------------------------------------------------------+
|  Lineage Explorer                            [X]     |
|                                                      |
|  Current Node: "Monthly Revenue Report"              |
|  +------------------------------------------------+ |
|  |                                                | |
|  |  [Upstream]                                    | |
|  |                                                | |
|  |  CRM --> Clean --> Customer --+                | |
|  |                               +--> Revenue --> | |
|  |  ERP --> Clean --> Orders ----+     Calc       | |
|  |                                      |         | |
|  |  Rates ------------------------------|         | |
|  |                                                | |
|  |  [Downstream]                                  | |
|  |                                                | |
|  |  --> Executive Dashboard                       | |
|  |  --> Board Report                              | |
|  |  --> SEC Filing (10-K)                         | |
|  |                                                | |
|  +------------------------------------------------+ |
|                                                      |
|  Depth: [1] [2] [3] [All]  Dir: [Up] [Down] [Both]  |
|                                                      |
|  +- Node Details -----------------------------------+|
|  | Revenue Calc (Transform #T-789)                  ||
|  | Last run: 2024-03-20 08:00 UTC                   ||
|  | Code version: main@abc123d                       ||
|  | Duration: 4m 32s                                 ||
|  | Output rows: 1,247,831                           ||
|  | Owner: data-engineering@company.com              ||
|  +--------------------------------------------------+|
+------------------------------------------------------+

Approach Comparison
================================================

Static Analysis (parse code/SQL):

  SQL: SELECT a.id, a.amount * b.rate AS amount_usd
       FROM orders a JOIN rates b ON a.currency = b.code

  Parse Result:
    amount_usd <- orders.amount, rates.rate  (multiplication)
    id         <- orders.id                   (passthrough)

  Pros: No execution needed, lineage available upfront
  Cons: Complex logic hard to parse (UDFs, dynamic SQL)

Runtime Capture (record during execution):

  Instrument the DataFrame engine with tracing:
    Every column operation records its sources

  result["amount_usd"] = df["amount"] * df["rate"]
  -> Record: amount_usd = multiply(orders.amount, rates.rate)

  Pros: Precise, supports complex logic
  Cons: Runtime overhead, requires actual execution

Foundry's Approach: Combines Both
  1. Static analysis for initial lineage (fast)
  2. Runtime validation and enrichment (precise)
  3. Cache results to avoid recomputation

Lineage Data Model
================================================

// Node (data entity)
Node {
    id:        string            // unique identifier
    type:      enum              // DATASET | FIELD | TRANSFORM | MODEL
    name:      string            // human-readable name
    namespace: string            // project/space
    metadata:  map<string, any>  // extension properties
}

// Edge (lineage relationship)
Edge {
    source_id:   string      // upstream node
    target_id:   string      // downstream node
    type:        enum        // DERIVES_FROM | TRANSFORMS_TO | COMPUTES
    transform:   string      // description (e.g., "SUM", "JOIN", "FILTER")
    confidence:  float       // confidence (static analysis may be imprecise)
    created_at:  timestamp   // creation time
    created_by:  string      // Pipeline or User ID
}

// Graph Storage Selection
+------------------------------------------------+
|  Storage Options                               |
|                                                |
|  Neo4j / JanusGraph:                           |
|    [+] Excellent graph traversal performance   |
|    [+] Native path query support               |
|    [-] Complex operations                      |
|                                                |
|  PostgreSQL + Recursive CTE:                   |
|    [+] Simple operations                       |
|    [+] Good transaction support                |
|    [-] Limited deep traversal performance      |
|                                                |
|  Foundry's choice: Custom graph store          |
|  coomia-dip choice: PostgreSQL + cache layer    |
+------------------------------------------------+

Upstream Tracing
================================================

Question: "Where does 'quarterly_revenue' on the CEO
           dashboard come from?"

Query: trace_upstream("dashboard.quarterly_revenue")

Result (bottom to top):

  Layer 4 (Display): CEO Dashboard -> quarterly_revenue
       ^
  Layer 3 (Aggregation): Revenue Aggregation Transform
                  = SUM(monthly_revenue) WHERE quarter = Q1
       ^
  Layer 2 (Computation): Monthly Revenue Transform
                  = SUM(order_amount_usd) GROUP BY month
       ^
  Layer 1 (Cleaning): Order Cleaning Pipeline
                  = orders WHERE status != 'cancelled'
                    AND test_account = false
       ^
  Layer 0 (Source):   ERP System -> sales_orders table
                  Source: SAP S/4HANA
                  Update frequency: Real-time CDC
                  Last sync: 2024-03-20 07:55:00

Downstream Impact Analysis
================================================

Question: "If I change the update frequency of exchange_rates,
           what is affected?"

Query: trace_downstream("ref.exchange_rates")

Result (top to bottom):

  ref.exchange_rates
       |
       +--> orders_usd (Transform #T-023)
       |    +--> monthly_revenue (Transform #T-024)
       |    |    +--> CEO Dashboard (Report)
       |    |    +--> Board Report (Report)
       |    |    +--> SEC Filing (Report) [!] Compliance Critical
       |    |
       |    +--> customer_ltv (Model #M-012)
       |         +--> Customer Risk Score (Model #M-015)
       |         +--> Marketing Campaign Targeting (App)
       |
       +--> fx_pnl_report (Transform #T-067)
            +--> Treasury Dashboard (Report)

  Impact Summary:
  - Direct downstream: 2 Transforms
  - Indirect downstream: 7 datasets, 4 reports, 1 ML model
  - Compliance impact: SEC Filing (high priority)
  - Recommendation: Notify finance and compliance teams before change

Data Lineage Tool Ecosystem
================================================

                    Automation Level
                    ^
                    |
        Foundry ----+  <- Fully automatic, deep platform integration
                    |
       DataHub -----+  <- Requires configuring Ingestion Sources
                    |
    OpenLineage ----+  <- Requires integrating SDK into code
                    |
   Apache Atlas ----+  <- Requires Hooks + manual configuration
                    |
      Manual -------+  <- Excel or Wiki records (not recommended)
                    |
                    +------------------------------------> Flexibility

  Foundry: "Fully automatic but locked into the platform"
  DataHub: "Semi-automatic, multi-source, active community"
  OpenLineage: "Standardized protocol, good for heterogeneous envs"
  Atlas: "Hadoop ecosystem native, big data scenarios"

coomia-dip LineageService RPC Endpoints
================================================

+-----------------------------------------------+
|  LineageService (gRPC)                        |
|                                               |
|  Basic CRUD:                                  |
|  1. RecordLineage       - Record lineage      |
|  2. GetLineage          - Query entity lineage|
|  3. DeleteLineage       - Delete lineage      |
|                                               |
|  Graph Traversal:                             |
|  4. TraceUpstream       - Upstream tracing    |
|  5. TraceDownstream     - Downstream tracing  |
|  6. GetFullGraph        - Full lineage graph  |
|                                               |
|  Analysis:                                    |
|  7. AnalyzeImpact       - Impact analysis     |
|  8. FindRootCause       - Root cause analysis |
|                                               |
|  Field-Level:                                 |
|  9. RecordFieldLineage  - Record field lineage|
|  10. GetFieldLineage    - Query field lineage |
|                                               |
|  Workflow:                                    |
|  11. GetWorkflowLineage - Workflow lineage    |
|                                               |
+-----------------------------------------------+

// coomia-dip Lineage Protobuf Definition (Simplified)

syntax = "proto3";
package onto.lineage.v1;

// Lineage Node
message LineageNode {
    string id = 1;
    string name = 2;
    NodeType type = 3;
    string namespace = 4;
    map<string, string> properties = 5;
    google.protobuf.Timestamp created_at = 6;
}

enum NodeType {
    NODE_TYPE_UNSPECIFIED = 0;
    DATASET = 1;
    FIELD = 2;
    TRANSFORM = 3;
    MODEL = 4;
    ONTOLOGY_TYPE = 5;
    ONTOLOGY_PROPERTY = 6;
    WORKFLOW = 7;
}

// Lineage Edge
message LineageEdge {
    string source_id = 1;
    string target_id = 2;
    EdgeType type = 3;
    string transform_description = 4;
    float confidence = 5;
    google.protobuf.Timestamp recorded_at = 6;
    string recorded_by = 7;  // Pipeline or User ID
}

enum EdgeType {
    EDGE_TYPE_UNSPECIFIED = 0;
    DERIVES_FROM = 1;       // Data derivation
    TRANSFORMS_TO = 2;      // Transformation
    COMPUTES = 3;            // Computation
    REFERENCES = 4;          // Reference
    AGGREGATES = 5;          // Aggregation
}

// Impact Analysis Result
message ImpactAnalysisResult {
    string source_id = 1;
    repeated ImpactedEntity impacted_entities = 2;
    int32 total_impacted = 3;
    RiskLevel risk_level = 4;
}

message ImpactedEntity {
    LineageNode node = 1;
    int32 distance = 2;          // Distance from source
    repeated string path = 3;    // Impact path
    string impact_type = 4;      // Impact type description
}

enum RiskLevel {
    RISK_UNSPECIFIED = 0;
    LOW = 1;
    MEDIUM = 2;
    HIGH = 3;
    CRITICAL = 4;
}

# coomia-dip LineageService Core Implementation (Simplified)

from typing import List, Optional, Set
from dataclasses import dataclass, field

@dataclass
class LineageGraph:
    """In-memory lineage graph structure"""
    nodes: dict[str, LineageNode] = field(default_factory=dict)
    forward_edges: dict[str, list[LineageEdge]] = field(default_factory=dict)
    backward_edges: dict[str, list[LineageEdge]] = field(default_factory=dict)

class LineageServiceImpl:
    """Lineage service implementation"""

    def __init__(self, db_session, cache_client):
        self.db = db_session
        self.cache = cache_client

    async def trace_upstream(
        self,
        entity_id: str,
        max_depth: int = 10,
        include_field_lineage: bool = False
    ) -> LineageGraph:
        """
        Upstream tracing: trace all upstream dependencies
        from the target entity.

        Algorithm: BFS graph traversal
        Optimization: Redis cache for hot paths
        """
        visited: Set[str] = set()
        queue = [(entity_id, 0)]
        graph = LineageGraph()

        while queue:
            current_id, depth = queue.pop(0)

            if current_id in visited or depth > max_depth:
                continue
            visited.add(current_id)

            node = await self._get_node(current_id)
            if node:
                graph.nodes[current_id] = node

            upstream_edges = await self._get_upstream_edges(current_id)
            for edge in upstream_edges:
                graph.backward_edges.setdefault(
                    current_id, []
                ).append(edge)
                queue.append((edge.source_id, depth + 1))

                if include_field_lineage:
                    field_edges = await self._get_field_lineage(
                        edge.source_id, current_id
                    )
                    for fe in field_edges:
                        graph.backward_edges.setdefault(
                            fe.target_id, []
                        ).append(fe)

        return graph

    async def analyze_impact(
        self,
        entity_id: str,
        change_type: str = "SCHEMA_CHANGE"
    ) -> ImpactAnalysisResult:
        """
        Impact analysis: assess downstream impact
        of modifying an entity.

        Returns: list of impacted entities + risk level
        """
        downstream_graph = await self.trace_downstream(
            entity_id, max_depth=20
        )

        impacted = []
        max_risk = RiskLevel.LOW

        for node_id, node in downstream_graph.nodes.items():
            if node_id == entity_id:
                continue

            distance = self._calculate_distance(
                downstream_graph, entity_id, node_id
            )
            path = self._find_path(
                downstream_graph, entity_id, node_id
            )

            risk = self._assess_risk(node, change_type)
            if risk.value > max_risk.value:
                max_risk = risk

            impacted.append(ImpactedEntity(
                node=node,
                distance=distance,
                path=path,
                impact_type=f"{change_type} propagation"
            ))

        return ImpactAnalysisResult(
            source_id=entity_id,
            impacted_entities=sorted(
                impacted, key=lambda x: x.distance
            ),
            total_impacted=len(impacted),
            risk_level=max_risk
        )

    def _assess_risk(
        self, node: LineageNode, change_type: str
    ) -> RiskLevel:
        """Assess risk based on node type and change type"""
        if "compliance" in node.properties.get("tags", ""):
            return RiskLevel.CRITICAL
        if node.type in (NodeType.MODEL, NodeType.WORKFLOW):
            return RiskLevel.HIGH
        if change_type == "SCHEMA_CHANGE":
            return RiskLevel.MEDIUM
        return RiskLevel.LOW

Data Quality + Lineage Integration
================================================

Scenario: Automatically detect data quality issue propagation

  Quality Check Failed:
  +--------------------------------------+
  |  Dataset: customer_addresses         |
  |  Check: Address completeness         |
  |  Result: FAILED (23% missing zip)    |
  |  Time: 2024-03-20 08:00              |
  +------------------+-------------------+
                     | Auto-trigger impact analysis
                     v
  +--------------------------------------+
  |  Impact Report                       |
  |                                      |
  |  Direct Impact:                      |
  |  +- customer_geo_analysis (Transform)|
  |  +- delivery_routing (Model)         |
  |  +- regional_revenue (Report)        |
  |                                      |
  |  Indirect Impact:                    |
  |  +- territory_planning (App)         |
  |  +- logistics_optimization (Model)   |
  |                                      |
  |  Recommended Actions:                |
  |  1. Pause delivery_routing retraining|
  |  2. Add warning to regional_revenue  |
  |  3. Notify logistics team            |
  +--------------------------------------+

GDPR Data Subject Request + Lineage
================================================

Scenario: User requests data deletion (Right to be Forgotten)

Request: "Delete all data for user-12345"

Lineage Trace:
  user-12345's data appears in:

  1. CRM.customers (source)
     +- user_profile (cleaned)
        +- customer_360 (Ontology Object)
        |  +- customer_name: "John Smith"
        |  +- customer_email: "john@example.com"
        |  +- customer_phone: "+1-xxx"
        |
        +- marketing_targets (aggregated dataset)
        |  +- contains user-12345's behavioral data
        |
        +- ml_training_data (model training set)
           +- contains user-12345's features

  Locations to delete/anonymize: 4 datasets + 1 Ontology object
  Models to retrain: 1
  Compliance: Record complete audit log of deletion

Workflow-Level Lineage
================================================

A complete data processing workflow may span multiple systems:

+-----------------------------------------------------+
|  Workflow: "Daily Revenue Pipeline"                  |
|  ID: WF-2024-0320-001                               |
|                                                     |
|  Step 1: Data Ingestion (Kafka -> Raw Storage)       |
|  +-------------------------------------+             |
|  |  Source: ERP CDC Stream              |             |
|  |  Target: raw.sales_orders            |             |
|  |  Records: 15,234                     |             |
|  |  Duration: 2m 15s                    |             |
|  +----------------------+--------------+             |
|                         |                            |
|  Step 2: Data Cleaning (Spark Transform)             |
|  +-------------------------------------+             |
|  |  Input: raw.sales_orders             |             |
|  |  Output: clean.sales_orders          |             |
|  |  Removed: 123 duplicates, 45 nulls   |             |
|  |  Duration: 5m 42s                    |             |
|  +----------------------+--------------+             |
|                         |                            |
|  Step 3: Currency Conversion (Python Transform)      |
|  +-------------------------------------+             |
|  |  Inputs: clean.sales_orders,         |             |
|  |          ref.exchange_rates           |             |
|  |  Output: enriched.orders_usd         |             |
|  |  Duration: 1m 03s                    |             |
|  +----------------------+--------------+             |
|                         |                            |
|  Step 4: Ontology Sync (Ontology Updater)            |
|  +-------------------------------------+             |
|  |  Input: enriched.orders_usd          |             |
|  |  Target: Order objects (Ontology)    |             |
|  |  Created: 234, Updated: 1,890        |             |
|  |  Duration: 3m 21s                    |             |
|  +-------------------------------------+             |
|                                                     |
|  Total Duration: 12m 21s                             |
|  Status: SUCCESS                                     |
+-----------------------------------------------------+

Data Trust Pyramid
================================================

            /\
           /  \
          /Decision\       "I can make decisions based on this"
         / Trust    \
        /------------\
       / Data         \    "I understand what this data means"
      / Understanding  \
     /------------------\
    / Source              \  "I know where it came from"
   / Transparency         \
  /------------------------\
 / Quality                  \  "Data has been quality-checked"
/ Assurance                  \
+-----------------------------+
| Traceability                |  "Every step is recorded"
+-----------------------------+

No lineage = pyramid foundation missing = cannot build trust