Data Governance — Lineage, Cataloging, Access Control, and Data Mesh

TABLE-LEVEL LINEAGE:
  silver.orders ← bronze.orders (via dbt stg_orders)
  gold.daily_revenue ← silver.orders + silver.stores

COLUMN-LEVEL LINEAGE (more precise):
  gold.daily_revenue.net_revenue
    ← silver.orders.order_amount
    ← silver.orders.discount_amount
    (transformation: order_amount - discount_amount)

  gold.daily_revenue.store_region
    ← silver.stores.region
    ← silver.stores.state
    (transformation: CASE WHEN state IN (...) THEN 'South' ...)

Column-level lineage tells you:
  - Exactly which source column populated each Gold column
  - Which transformations were applied along the way
  - Impact analysis: if orders.order_amount definition changes,
    which Gold columns are affected?
  - Audit: prove to a regulator which source fields produced a reported metric

{
  "eventType": "COMPLETE",
  "eventTime": "2026-03-17T06:14:32.000Z",
  "run": {
    "runId": "d7c7a7b8-3e1a-4a2c-9b4d-...",
    "facets": {
      "nominalTime": {"nominalStartTime": "2026-03-17T06:00:00Z"}
    }
  },
  "job": {
    "namespace": "freshcart.dbt",
    "name": "silver.orders"
  },
  "inputs": [
    {
      "namespace": "freshcart.bronze",
      "name": "orders",
      "facets": {
        "schema": {
          "fields": [
            {"name": "order_id",   "type": "INTEGER"},
            {"name": "customer_id","type": "INTEGER"},
            {"name": "amount",     "type": "DECIMAL"}
          ]
        }
      }
    }
  ],

  "outputs": [
    {
      "namespace": "freshcart.silver",
      "name": "orders",
      "facets": {
        "columnLineage": {
          "fields": {
            "order_id":   {"inputFields": [{"namespace":"freshcart.bronze","name":"orders","field":"order_id"}]},
            "net_revenue":{"inputFields": [
              {"namespace":"freshcart.bronze","name":"orders","field":"amount"},
              {"namespace":"freshcart.bronze","name":"orders","field":"discount_amount"}
            ]}
          }
        }
      }
    }
  ]
}

LINEAGE BACKENDS:
  Marquez:      open source, stores OpenLineage events, REST API, simple UI
  DataHub:      comprehensive catalog + lineage, ingestion connectors for dbt/Spark/Airflow
  Atlan:        managed, deep dbt/Airflow integration
  OpenMetadata: open source alternative to DataHub

dbt LINEAGE (automatic, no extra setup):
  dbt generates a DAG of all models and their dependencies.
  dbt docs generate produces a browsable lineage graph:
    dbt docs generate && dbt docs serve
  Shows: every model, its SQL, its upstream models, its downstream models.
  Column-level lineage: dbt-column-lineage package adds column mapping.

You make the change, run dbt, and discover 4 Gold models break
because they all had column-specific assertions on net_revenue.
You also broke a dashboard that a non-dbt BI tool was using.
Investigation time: 3 hours.

Step 1: Query the dbt graph for all downstream consumers of silver.orders:
$ dbt ls --select +silver.orders+
  silver.orders
  gold.daily_revenue         ← depends on silver.orders
  gold.customer_ltv          ← depends on silver.orders
  gold.fct_orders_wide       ← depends on silver.orders
  gold.store_performance     ← depends on silver.orders

Step 2: Check which downstream models use net_revenue specifically —
  all four use SUM(net_revenue) or the raw column directly. All affected.

Step 3: Check for non-dbt consumers (BI tools, APIs, external queries) via
  DataHub catalog → silver.orders.net_revenue → "Downstream consumers" tab
  Shows: Metabase dashboard "Daily Revenue" — uses this column directly

Step 4: Plan the migration — add tip_amount first (non-breaking), update
  net_revenue backward-compatibly, notify the Metabase dashboard owner,
  update all 4 Gold models together, deploy Silver → Gold in one dbt run.

Total time with lineage:    30 minutes of planning, zero surprises.
Total time without lineage: 3 hours of debugging broken things after the fact.

# datahub_recipes/dbt_freshcart.yml
source:
  type: dbt
  config:
    manifest_path:       /path/to/dbt/target/manifest.json
    catalog_path:        /path/to/dbt/target/catalog.json
    sources_path:        /path/to/dbt/target/sources.json
    target_platform:     snowflake
    include_column_lineage: true  # column-level lineage from dbt
    owner_extraction_pattern: "^Team:(?P<owner>.*)$"
sink:
  type: datahub-rest
  config:
    server: "http://datahub-gms:8080"

# datahub_recipes/snowflake_freshcart.yml
source:
  type: snowflake
  config:
    account_id:   freshcart.snowflake.com
    username:     datahub_service_account
    database:     freshcart_prod
    warehouse:    analyst_wh
    include_table_lineage: true    # query log-based lineage
    include_column_lineage: true
    include_usage_stats:   true    # who queried what and when

# RUN: datahub ingest -c datahub_recipes/dbt_freshcart.yml
# RUN: datahub ingest -c datahub_recipes/snowflake_freshcart.yml

WHAT DATAHUB PROVIDES AFTER INGESTION:
Search:       "net revenue" → finds gold.daily_revenue.net_revenue
Description:  "Net revenue after discounts. Source: silver.orders."
Owner:        data-team@freshcart.com
Lineage:      upstream: silver.orders, downstream: Metabase dashboard
Usage:        queried by priya@freshcart.com, 48 times in last 7 days
Quality:      last dbt test run: all 12 tests passed, 2026-03-17
Tags:         [PII-free, Gold, Finance, SLA-monitored]

# models/gold/_schema.yml
models:
  - name: daily_revenue
    description: >
      Pre-aggregated daily revenue by store and date.
      Updated daily at 06:30 ET. SLA: complete by 08:00 ET.
      Source of truth for Finance dashboard.
    meta:
      owner: data-team@freshcart.com
      sla: "08:00 ET daily"
      consumers: [finance-dashboard, cfo-report]
    columns:
      - name: net_revenue
        description: >
          Total order revenue after discounts.
          Calculation: SUM(order_amount - discount_amount)
          from silver.orders WHERE status = 'delivered'.
        meta:
          is_pii: false
          business_owner: finance@freshcart.com

# When dbt docs generate runs → DataHub ingestion picks up descriptions
# Descriptions become visible in DataHub search and table pages

EVERYDAY GOVERNANCE WORKFLOW FOR A DATA ENGINEER:
  Before adding a new Gold table:
    1. Create dbt model with complete description in schema.yml
    2. Add owner meta field (team email)
    3. Add column descriptions for all columns
    4. Add relevant tags (PII-free, Gold, team)
    5. Add to DataHub business glossary if it defines a new term

  After deployment:
    DataHub ingestion runs → table discoverable in catalog within 1 hour
    Analyst can find it, read the description, understand the grain
    Without Slack DMs to the data team

Finance:    "revenue" = sum of all order amounts including cancelled
Operations: "revenue" = delivered orders only
Product:    "revenue" = GMV (gross merchandise value, all statuses)
Three teams, three definitions, three different numbers in the same meeting.

Term: "Net Revenue"
Definition: Sum of (order_amount - discount_amount) for orders with
            status = 'delivered'. Excludes cancelled and in-progress orders.
            Approved by: CFO on 2026-01-15.
Owners: finance@freshcart.com (business), data-team@freshcart.com (technical)
Linked assets: gold.daily_revenue.net_revenue, gold.customer_ltv.net_revenue

Term: "GMV (Gross Merchandise Value)"
Definition: Sum of order_amount for all orders regardless of status.
            Used for investor reporting. Does NOT subtract discounts.
Owners: product@freshcart.com
Linked assets: gold.investor_metrics.gmv

Once defined: every table column tagged with "Net Revenue" gets the
canonical definition. Analysts see it when they hover the column — the
SAME definition appears in every BI tool, in DataHub, and in dbt docs.

import datahub.emitter.mce_builder as builder
from datahub.emitter.mcp import MetadataChangeProposalWrapper
from datahub.metadata.schema_classes import GlossaryTermInfoClass

term_urn = builder.make_glossary_term_urn("NetRevenue")
term_info = GlossaryTermInfoClass(
    definition="Sum of (order_amount - discount_amount) for status='delivered'.",
    termSource="INTERNAL",
    sourceRef="Finance/MetricsDefinitions.pdf",
    sourceUrl="https://docs.freshcart.internal/metrics/net-revenue",
)
emitter.emit_mcp(MetadataChangeProposalWrapper(entityUrn=term_urn, aspect=term_info))

SYSADMIN
└── DATA_PLATFORM_ADMIN      ← full platform access (data engineering lead)
    ├── PIPELINE_ROLE         ← transformation pipelines (read bronze, write silver/gold)
    ├── ANALYST_ROLE          ← read silver/gold, no PII, no bronze
    ├── DATA_SCIENTIST_ROLE   ← read bronze/silver/gold, no PII columns
    ├── BI_SERVICE_ROLE       ← read gold only, used by Metabase service account
    ├── FINANCE_ROLE          ← read gold finance schema only
    └── OPERATIONS_ROLE       ← read gold operations schema only

-- PIPELINE_ROLE: runs dbt, reads bronze, writes silver and gold
GRANT USAGE ON DATABASE freshcart_prod TO ROLE pipeline_role;
GRANT USAGE ON SCHEMA freshcart_prod.bronze TO ROLE pipeline_role;
GRANT SELECT ON ALL TABLES IN SCHEMA freshcart_prod.bronze TO ROLE pipeline_role;
GRANT USAGE, CREATE TABLE ON SCHEMA freshcart_prod.silver TO ROLE pipeline_role;
GRANT USAGE, CREATE TABLE ON SCHEMA freshcart_prod.gold   TO ROLE pipeline_role;

-- ANALYST_ROLE: read silver and gold, no bronze (has raw PII), no write
GRANT USAGE ON DATABASE freshcart_prod TO ROLE analyst_role;
GRANT USAGE ON SCHEMA freshcart_prod.silver TO ROLE analyst_role;
GRANT USAGE ON SCHEMA freshcart_prod.gold   TO ROLE analyst_role;
GRANT SELECT ON ALL TABLES IN SCHEMA freshcart_prod.silver TO ROLE analyst_role;
GRANT SELECT ON ALL TABLES IN SCHEMA freshcart_prod.gold   TO ROLE analyst_role;
-- Explicitly block: no access to bronze (raw PII in landing/bronze)

-- BI_SERVICE_ROLE: Metabase service account, read gold only
GRANT USAGE ON SCHEMA freshcart_prod.gold TO ROLE bi_service_role;
GRANT SELECT ON ALL TABLES IN SCHEMA freshcart_prod.gold TO ROLE bi_service_role;

-- FUTURE GRANTS: apply to tables created after the GRANT statement
GRANT SELECT ON FUTURE TABLES IN SCHEMA freshcart_prod.silver TO ROLE analyst_role;
GRANT SELECT ON FUTURE TABLES IN SCHEMA freshcart_prod.gold   TO ROLE analyst_role;
-- Without FUTURE GRANTS: every new Silver/Gold table must be manually granted

CREATE OR REPLACE MASKING POLICY mask_email_pii
AS (val STRING) RETURNS STRING ->
    CASE
        WHEN CURRENT_ROLE() IN ('DATA_PLATFORM_ADMIN', 'PIPELINE_ROLE')
        THEN val                              -- engineers see raw email
        WHEN CURRENT_ROLE() = 'DATA_SCIENTIST_ROLE'
        THEN SHA2(val, 256)                   -- scientists see hash
        ELSE '***MASKED***'                   -- everyone else sees mask
    END;

ALTER TABLE freshcart_prod.silver.customers
ALTER COLUMN customer_email
SET MASKING POLICY mask_email_pii;

Analyst queries silver.customers:        customer_email = '***MASKED***'
Data engineer queries silver.customers:  customer_email = 'priya@example.com'
Data scientist queries silver.customers: customer_email = 'sha256hash...'
All three query the SAME table — masking is transparent, no separate views.

CREATE OR REPLACE ROW ACCESS POLICY store_data_isolation
AS (store_id VARCHAR) RETURNS BOOLEAN ->
    CURRENT_ROLE() IN ('DATA_PLATFORM_ADMIN', 'ANALYST_ROLE')   -- full access
    OR store_id = (
        SELECT assigned_store_id
        FROM governance.user_store_assignments
        WHERE username = CURRENT_USER()
    );

ALTER TABLE freshcart_prod.gold.store_performance
ADD ROW ACCESS POLICY store_data_isolation ON (store_id);
-- Store manager queries gold.store_performance:
-- automatically filtered to their assigned store only — no WHERE clause needed

import boto3
lf = boto3.client('lakeformation')

lf.grant_permissions(
    Principal={'DataLakePrincipalIdentifier': 'arn:aws:iam::123456:role/AnalystRole'},
    Resource={
        'TableWithColumns': {
            'DatabaseName': 'freshcart_silver',
            'Name': 'customers',
            'ColumnWildcard': {
                'ExcludedColumnNames': ['customer_email', 'phone_number', 'address']
            },
        }
    },
    Permissions=['SELECT'],
)

lf.create_data_cells_filter(
    TableData={
        'TableCatalogId': '123456789',
        'DatabaseName':   'freshcart_gold',
        'TableName':      'store_performance',
        'Name':           'south_region_filter',
        'RowFilter': {'FilterExpression': "store_region = 'South'"},
        'ColumnWildcard': {
            'ExcludedColumnNames': ['gross_margin_pct', 'operating_cost']
        },
    }
)

# Every SELECT on a Lake Formation-registered table generates a CloudTrail event.
# Query: which users accessed silver.customers in the last 30 days?
SELECT userIdentity.principalId, eventTime, requestParameters.tableName
FROM cloudtrail_logs
WHERE eventName = 'GetTable'
  AND requestParameters.tableName = 'customers'
  AND eventTime > CURRENT_DATE - 30

Level 1 — Direct identifiers (highest sensitivity):
  customer_email, phone_number, national_id (SSN), passport_number,
  bank_account_number, credit_card_number
Level 2 — Indirect identifiers (can identify combined with other data):
  full_name, address, date_of_birth, ip_address, device_id, GPS coordinates
Level 3 — Quasi-identifiers (alone not identifying, combined risky):
  city, gender, age_group, job_title, company_name
Level 4 — Non-PII (no restriction):
  order_amount, product_category, store_id, order_status

import re
from dataclasses import dataclass

@dataclass
class PIIDetectionResult:
    column: str
    pii_level: int
    pii_type: str
    confidence: float

PATTERNS = {
    'email':    (1, re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b')),
    'phone_us': (1, re.compile(r'\b(?:\+1[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}\b')),
    'ssn':      (1, re.compile(r'\b\d{3}-?\d{2}-?\d{4}\b')),
    'name':     (2, None),   # requires ML classifier — name detection is hard with regex
    'ip_addr':  (2, re.compile(r'\b(?:\d{1,3}\.){3}\d{1,3}\b')),
}

def detect_pii_in_column(sample_values: list[str], column_name: str) -> PIIDetectionResult:
    hits = {'email': 0, 'phone_us': 0, 'ssn': 0, 'ip_addr': 0}
    total = len(sample_values)
    for val in sample_values:
        if not isinstance(val, str):
            continue
        for pii_type, (level, pattern) in PATTERNS.items():
            if pattern and pattern.search(val):
                hits[pii_type] += 1
    for pii_type, count in hits.items():
        confidence = count / total if total > 0 else 0
        if confidence > 0.3:    # > 30% of sampled values match
            level = PATTERNS[pii_type][0]
            return PIIDetectionResult(column=column_name, pii_level=level, pii_type=pii_type, confidence=round(confidence, 2))
    return PIIDetectionResult(column=column_name, pii_level=4, pii_type='none', confidence=1.0)

>>> detect_pii_in_column(['priya@example.com', 'raj@example.com', 'not-an-email'], 'contact_field')
PIIDetectionResult(column='contact_field', pii_level=1, pii_type='email', confidence=0.67)

columns:
  - name: customer_email
    meta:
      pii_level: 1
      pii_type: email
      masking_required: true
      retain_in_layers: [bronze]   # raw email only in bronze
      transform_for_silver: "SHA2(customer_email, 256)"

  - name: phone_number
    meta:
      pii_level: 1
      pii_type: phone
      masking_required: true
      transform_for_silver: "REGEXP_REPLACE(phone_number, '[0-9]', 'X')"

# dbt macro that reads pii meta and generates masking transformations:
# {{ transform_pii_column('customer_email') }}
# Generates: SHA2(customer_email, 256) AS customer_email_hashed

from datetime import date, datetime
import hashlib

DELETION_SENTINEL = 'GDPR_ERASED'

def process_erasure_request(customer_id: int, request_date: date, conn) -> dict:
    """Anonymise a customer's PII across all layers. Does NOT delete rows —
    replaces PII values with a sentinel. Preserves non-PII columns for
    statistical use."""
    results = {}

    # BRONZE is the raw layer. This is the ONLY time we UPDATE Bronze —
    # for legal compliance, overriding the append-only design principle.
    bronze_rows = conn.execute("""
        UPDATE bronze.customers
        SET customer_email  = %s,
            phone_number    = %s,
            full_name       = %s,
            ip_address      = %s,
            gdpr_erased     = TRUE,
            gdpr_erased_at  = %s
        WHERE customer_id = %s
        RETURNING customer_id
    """, (DELETION_SENTINEL, DELETION_SENTINEL, DELETION_SENTINEL,
          DELETION_SENTINEL, datetime.utcnow(), customer_id)).fetchall()

    results['bronze_rows_updated'] = len(bronze_rows)
    return results

    # SILVER already has masked columns (email_hashed) but may still have
    # quasi-identifiers like full_name or address.
    conn.execute("""
        UPDATE silver.customers
        SET customer_name  = %s,
            address        = %s,
            gdpr_erased    = TRUE
        WHERE customer_id = %s
    """, (DELETION_SENTINEL, DELETION_SENTINEL, customer_id))

    # Update SCD2 dim_customer — ALL versions:
    conn.execute("""
        UPDATE gold.dim_customer
        SET customer_name = %s
        WHERE customer_id = %s
    """, (DELETION_SENTINEL, customer_id))

    # GOLD aggregates (revenue by store, etc.) do NOT contain PII rows — no
    # update needed for most Gold tables. EXCEPTION: fct_orders_wide has
    # customer city + region at order level, which can be quasi-identifying.
    conn.execute("""
        UPDATE gold.fct_orders_wide
        SET customer_tier   = NULL,
            customer_city   = NULL,
            customer_region = NULL
        WHERE customer_id = %s
    """, (customer_id,))

    conn.execute("""
        INSERT INTO governance.erasure_requests
            (customer_id, request_date, completed_at, layers_updated, status)
        VALUES (%s, %s, %s, %s, 'completed')
    """, (customer_id, request_date, datetime.utcnow(), str(results)))
    conn.commit()

    # DOWNSTREAM SYSTEMS:
    # - ML models trained on this customer's data: log for retraining audit
    # - Kafka topics: publish an erasure event for downstream consumers
    # - Data Vault: update satellite records for this hub key
    # All systems subscribed to the erasure event must handle PII deletion.

    return results

✓ PII inventory: every column tagged with pii_level in schema.yml
✓ Masking at Silver boundary: PII replaced before analyst access
✓ Access control: Bronze (raw PII) not accessible to analysts
✓ Audit log: all accesses to PII-containing tables logged
✓ Erasure pipeline: tested, runs within 30 days of request
✓ Data retention: Bronze PII tables have lifecycle policies (max 2 years)
✓ Data residency: customer data stays in approved region (state law requirement)
✓ Consent log: when consent was given/revoked, stored in governance schema

FreshCart’s Data Protection Officer conducts a GDPR audit. They find that customer emails and phone numbers in silver.customers are directly accessible to 12 analysts, the data engineer who quit last year still has active Snowflake credentials, and there is no audit log of who accessed customer data. You are given two weeks to fix all three.

Finding 1 — PII accessible to all analysts

-- Current: GRANT SELECT ON ALL TABLES IN SCHEMA silver TO ROLE analyst_role;
-- silver.customers has: customer_email, phone_number, full_name, address

CREATE MASKING POLICY mask_email AS (v VARCHAR) RETURNS VARCHAR ->
    CASE WHEN CURRENT_ROLE() IN ('PIPELINE_ROLE','DATA_PLATFORM_ADMIN')
         THEN v ELSE SHA2(v, 256) END;

CREATE MASKING POLICY mask_phone AS (v VARCHAR) RETURNS VARCHAR ->
    CASE WHEN CURRENT_ROLE() IN ('PIPELINE_ROLE','DATA_PLATFORM_ADMIN')
         THEN v ELSE REGEXP_REPLACE(v, '.', 'X') END;

CREATE MASKING POLICY mask_name AS (v VARCHAR) RETURNS VARCHAR ->
    CASE WHEN CURRENT_ROLE() IN ('PIPELINE_ROLE','DATA_PLATFORM_ADMIN')
         THEN v ELSE LEFT(v, 1) || '***' END;   -- 'P***' (initial only)

ALTER TABLE silver.customers ALTER COLUMN customer_email SET MASKING POLICY mask_email;
ALTER TABLE silver.customers ALTER COLUMN phone_number SET MASKING POLICY mask_phone;
ALTER TABLE silver.customers ALTER COLUMN full_name SET MASKING POLICY mask_name;
-- No data model changes needed. No view required. Masking is transparent.

Analyst queries silver.customers:
customer_email = 'a3f4...sha256hash...'
phone_number   = 'XXXXXXXXXX'
full_name      = 'P***'

Finding 2 — departed employee still has active credentials

SELECT user_name, last_success_login, login_history
FROM snowflake.account_usage.users
WHERE disabled = FALSE
ORDER BY last_success_login DESC;
-- Found: former_employee@freshcart.com, last login 47 days ago

ALTER USER former_employee@freshcart.com SET DISABLED = TRUE;

-- Preventive: automate offboarding via HR system integration —
-- when HR marks an employee departed: disable Snowflake user, revoke all
-- role assignments, log the access review, notify the security team.

Finding 3 — no audit log of customer data access

-- Snowflake retains access_history for 365 days in ACCOUNT_USAGE. Always on.
GRANT SELECT ON SNOWFLAKE.ACCOUNT_USAGE.ACCESS_HISTORY TO ROLE dpo_role;

-- DPO audit query: who accessed customer PII in the last 90 days?
SELECT user_name, query_start_time, query_text, base_objects_accessed
FROM snowflake.account_usage.access_history,
     LATERAL FLATTEN(base_objects_accessed) f
WHERE f.value:objectName::STRING ILIKE '%silver.customers%'
  AND query_start_time >= CURRENT_DATE - 90
ORDER BY query_start_time DESC;

-- Ongoing: schedule a weekly DPO report to governance.pii_access_log
-- for a permanent audit trail beyond Snowflake's 365-day retention.

OUTCOMES AFTER REMEDIATION:
PII masked for all analysts:            same-day fix via column masking
Departed employee disabled:             same day
Audit log available to DPO:             1 day (access grant + query setup)
Full PII inventory across all tables:   1 week (schema.yml audit)
Automated offboarding process:          2 weeks (HR integration)
GDPR compliance documentation updated:  2 weeks