Where did this value come from?
Which job wrote this table?
Which source position was used?
Which transform version produced this field?
Which downstream reports depend on this column?
Which assets become invalid if this upstream dataset changes?
Which consumers see PII derived from this source?
Which pipeline run introduced the bad output?
Which backfill superseded the previous result?

What produced what, from what, when, using which code/config/schema, under which run context?

If this thing changes or fails, what else is affected, how badly, and what action should be taken?

If source.case.status changes type from string to enum, who breaks?
If case-normalizer v8 produced wrong event_time, which published assets are tainted?
If gold.case_sla_breach fails freshness SLO, which dashboards and APIs are degraded?

run identity
job identity
input datasets
output datasets
code version
config version
schema versions
source positions or snapshots
quality result
publication state
owner/service identity
timestamp

{
  "eventType": "COMPLETE",
  "eventTime": "2026-07-04T02:10:12Z",
  "run": {
    "runId": "run-20260704-0200-case-normalizer"
  },
  "job": {
    "namespace": "regulatory-data-platform",
    "name": "case-normalizer"
  },
  "inputs": [
    {
      "namespace": "kafka://prod-cluster",
      "name": "case-events.v3"
    }
  ],
  "outputs": [
    {
      "namespace": "iceberg://prod-catalog/enforcement",
      "name": "silver.case_event"
    }
  ]
}

Job    = abstract process that consumes/produces datasets
Run    = one execution instance of a job
Dataset = input or output data asset
Facet  = additional metadata attached to run/job/dataset

job A produced table B

runId=case-normalizer/2026-07-04T02:00
job=case-normalizer
codeVersion=git:abc123
artifact=case-normalizer:2.8.1
configVersion=normalizer-config:v17
input=topic case-events.v3 offsets {0: 8001-9110, 1: 7010-8022}
inputSchema=case-event-avro:3.2.0
output=iceberg silver.case_event snapshot 885
outputSchema=silver-case-event:5.1.0
qualityResult=PASS
rowCount=17_812
rejected=21
publishedAt=2026-07-04T02:07:18Z

public record DatasetRef(
        String namespace,
        String name,
        DatasetType type,
        Optional<String> version
) {}

public enum DatasetType {
    KAFKA_TOPIC,
    ICEBERG_TABLE,
    POSTGRES_TABLE,
    OBJECT_PREFIX,
    API_ENDPOINT,
    SEARCH_INDEX,
    DASHBOARD,
    REPORT,
    MODEL,
    UNKNOWN
}

public record JobRef(
        String namespace,
        String name,
        String owner,
        String service
) {}

public record RunRef(
        String runId,
        Instant startedAt,
        Optional<String> parentRunId,
        Optional<String> backfillId
) {}

public record CodeRef(
        String gitSha,
        String artifactName,
        String artifactVersion,
        Optional<String> containerDigest
) {}

public record LineageEvent(
        LineageEventType type,
        Instant eventTime,
        RunRef run,
        JobRef job,
        List<DatasetRef> inputs,
        List<DatasetRef> outputs,
        CodeRef code,
        Map<String, Object> facets
) {}

public enum LineageEventType {
    START,
    RUNNING,
    COMPLETE,
    FAIL,
    ABORT
}

Internal LineageEvent
    -> OpenLineage JSON
    -> platform lineage store
    -> audit store
    -> observability event

case_event
prod.case_event
silver.case_event
s3://bucket/path/case_event
iceberg.case_event
Case Event Silver

namespace = iceberg://prod-catalog/enforcement
name      = silver.case_event

namespace = kafka://prod-msk-a
name      = regulatory.case-events.v3

namespace = postgres://case-db-prod
name      = public.case

namespace = s3://regulatory-prod-landing
name      = vendor-a/cases/YYYY/MM/DD/

{
  "kafka": {
    "topic": "case-events.v3",
    "partitions": {
      "0": {"from": 882001, "toExclusive": 883120},
      "1": {"from": 450010, "toExclusive": 451004}
    }
  }
}

{
  "cdc": {
    "connector": "case-db-postgres",
    "sourceLsnFrom": "16/B374D848",
    "sourceLsnTo": "16/B389FF20",
    "snapshot": false
  }
}

{
  "iceberg": {
    "table": "silver.case_event",
    "snapshotId": 884,
    "sequenceNumber": 1901
  }
}

{
  "api": {
    "endpoint": "GET /cases",
    "cursorFrom": "eyJwYWdlIjo1MDA...",
    "cursorTo": "eyJwYWdlIjo1MjA...",
    "watermark": "2026-07-04T01:59:59Z"
  }
}

{
  "file": {
    "manifestId": "vendor-a-20260704-0200",
    "files": [
      {"path": "landing/vendor-a/cases/2026-07-04/file-01.csv", "sha256": "..."}
    ]
  }
}

If input column X changes, which output columns break?
Which output column contains derived PII?
Which metric depends on policy table field response_days?

public record ColumnLineage(
        DatasetColumn output,
        List<DatasetColumn> inputs,
        TransformationKind kind,
        String expressionId,
        String description
) {}

public record DatasetColumn(
        DatasetRef dataset,
        String column
) {}

public enum TransformationKind {
    COPY,
    RENAME,
    CAST,
    HASH,
    MASK,
    DERIVE,
    AGGREGATE,
    FILTER_DEPENDENCY,
    JOIN_DEPENDENCY,
    CONSTANT,
    UNKNOWN
}

List<ColumnLineage> lineage = List.of(
    new ColumnLineage(
        col(silverCaseEvent, "sla_deadline"),
        List.of(col(sourceCase, "accepted_at"), col(policyTable, "response_days")),
        TransformationKind.DERIVE,
        "sla_deadline_v2",
        "accepted_at plus policy response days using business calendar"
    ),
    new ColumnLineage(
        col(goldBreach, "is_breached"),
        List.of(col(silverCaseEvent, "sla_deadline"), col(silverCaseEvent, "closed_at")),
        TransformationKind.DERIVE,
        "breach_flag_v4",
        "closed_at after sla_deadline or still open after deadline"
    )
);

public record RecordProvenance(
        String sourceSystem,
        String sourceEntity,
        String sourceRecordId,
        Optional<String> sourceVersion,
        Optional<String> sourcePosition,
        String producingRunId,
        String transformVersion,
        List<String> inputEventIds
) {}

{
  "outputRecordId": "case-sla-breach:CASE-9001:2026-07-04",
  "sourceRecordId": "CASE-9001",
  "inputEventIds": [
    "case-accepted:CASE-9001:9",
    "policy-updated:POLICY-ENF-2026:2"
  ],
  "producingRunId": "breach-detector/20260704T0200",
  "transformVersion": "breach-detector-v4.3.1"
}

emit COMPLETE lineage
write output fails

write staging
validate staging
publish output atomically
observe published version
emit COMPLETE lineage with published version

write data files
commit snapshot
record snapshot id
emit COMPLETE lineage with snapshot id

produce events
transaction commit succeeds
record output topic + partition offset range if available
emit COMPLETE lineage

write effect ledger
perform idempotent API call
confirm outcome or mark unknown
emit lineage with side effect status

public interface LineageEmitter {
    void emit(LineageEvent event) throws LineageEmissionException;
}

public final class CompositeLineageEmitter implements LineageEmitter {
    private final List<LineageEmitter> delegates;

    public CompositeLineageEmitter(List<LineageEmitter> delegates) {
        this.delegates = List.copyOf(delegates);
    }

    @Override
    public void emit(LineageEvent event) {
        RuntimeException failure = null;

        for (LineageEmitter delegate : delegates) {
            try {
                delegate.emit(event);
            } catch (RuntimeException e) {
                if (failure == null) failure = e;
                else failure.addSuppressed(e);
            }
        }

        if (failure != null) {
            throw failure;
        }
    }
}

CREATE TABLE pipeline_lineage_outbox (
    id UUID PRIMARY KEY,
    run_id TEXT NOT NULL,
    event_type TEXT NOT NULL,
    event_time TIMESTAMPTZ NOT NULL,
    payload JSONB NOT NULL,
    status TEXT NOT NULL,
    attempts INT NOT NULL DEFAULT 0,
    last_error TEXT,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
    sent_at TIMESTAMPTZ
);

commit output
insert lineage outbox event in same durable control transaction if possible
relay sends to lineage sink

{
  "runId": "gold-sla-breach/20260704T0200",
  "job": "gold-sla-breach",
  "status": "PUBLISHED",
  "startedAt": "2026-07-04T02:00:00Z",
  "finishedAt": "2026-07-04T02:06:42Z",
  "inputs": [
    {
      "dataset": "iceberg://prod/enforcement/silver.case_event",
      "snapshotId": 884
    }
  ],
  "outputs": [
    {
      "dataset": "iceberg://prod/enforcement/gold.case_sla_breach",
      "snapshotId": 885
    }
  ],
  "code": {
    "gitSha": "abc123",
    "artifact": "breach-detector:4.3.1"
  },
  "config": {
    "version": "breach-config:17"
  },
  "quality": {
    "status": "PASS",
    "failedCriticalChecks": 0
  },
  "metrics": {
    "inputRows": 1200041,
    "outputRows": 9904,
    "rejectedRows": 31
  }
}

Field case.priority renamed to risk_priority.
Who breaks?

Normalizer v8 parsed timezone incorrectly for accepted_at.
Which outputs are tainted?

silver.case_event is 4 hours late.
Which SLA dashboards or alerting services are degraded?

Source field subject_national_id is reclassified as restricted PII.
Which downstream assets contain direct or derived values?

Backfill campaign BF-20260704 will rewrite 180 days of gold assets.
Who must be notified?

public record ImpactRequest(
        AssetNode changedAsset,
        Optional<String> changedColumn,
        ChangeType changeType,
        Severity sourceSeverity,
        Instant effectiveAt
) {}

public enum ChangeType {
    SCHEMA_BREAKING,
    SCHEMA_COMPATIBLE,
    DATA_INCORRECT,
    FRESHNESS_BREACH,
    PRIVACY_RECLASSIFICATION,
    ASSET_DEPRECATED,
    BACKFILL_RESTATEMENT
}

public record ImpactResult(
        AssetNode asset,
        ImpactPath path,
        ImpactSeverity severity,
        RecommendedAction action,
        String reason
) {}

public final class ImpactAnalyzer {
    private final LineageGraph graph;
    private final PolicyEvaluator policyEvaluator;

    public List<ImpactResult> downstreamImpact(ImpactRequest request) {
        var results = new ArrayList<ImpactResult>();
        var queue = new ArrayDeque<ImpactPath>();
        var visited = new HashSet<AssetNode>();

        queue.add(ImpactPath.start(request.changedAsset()));

        while (!queue.isEmpty()) {
            ImpactPath path = queue.removeFirst();
            AssetNode current = path.last();

            if (!visited.add(current)) {
                continue;
            }

            for (LineageEdge edge : graph.outgoing(current)) {
                var next = edge.to();
                var nextPath = path.extend(edge, next);

                ImpactResult result = policyEvaluator.evaluate(request, nextPath);
                results.add(result);

                if (result.severity().shouldPropagate()) {
                    queue.addLast(nextPath);
                }
            }
        }

        return results;
    }
}

public enum ImpactSeverity {
    INFO,
    WARNING,
    BLOCKING,
    TAINTED,
    RESTRICTED,
    CRITICAL;

    public boolean shouldPropagate() {
        return this != INFO;
    }
}

public boolean edgeUsesColumn(LineageEdge edge, String column) {
    return edge.columnLineage().stream()
        .anyMatch(cl -> cl.inputs().stream()
            .anyMatch(input -> input.column().equals(column)));
}

public record QualityFacet(
        String suiteId,
        String suiteVersion,
        QualityStatus status,
        int totalChecks,
        int failedChecks,
        List<QualityFailure> failures
) {}

Quality gate failed on silver.case_event partition 2026-07-04.
Which gold assets depend on that partition?
Which dashboards should be marked degraded?

public record PartitionRef(
        DatasetRef dataset,
        Map<String, String> spec
) {}

{"dataset":"silver.case_event", "partition":{"event_date":"2026-07-04"}}
{"dataset":"gold.case_sla_breach", "partition":{"report_date":"2026-07-04"}}
{"dataset":"case_events.v3", "partition":{"kafka_partition":"4"}}

job=case-normalizer
run=stream-run-20260704
interval=2026-07-04T02:00:00Z/PT5M
input topic offsets:
  p0: 8001-9200
  p1: 7010-8112
output topic offsets:
  p0: 10001-11202
checkpoint=chk-9002
watermark=2026-07-04T01:58:00Z
quality=pass-with-warnings

source dataset(s) -> producer job -> output topic offset range

input topic offset range -> consumer job -> output dataset(s)

input table snapshot(s)
output table snapshot
operation: append / overwrite / merge / delete / compact
manifest/run id

{
  "inputSnapshots": [
    {"table": "silver.case_event", "snapshotId": 884},
    {"table": "reference.policy", "snapshotId": 102}
  ],
  "outputSnapshots": [
    {"table": "gold.case_sla_breach", "snapshotId": 885}
  ],
  "operation": "REPLACE_PARTITIONS",
  "partitions": ["report_date=2026-07-04"]
}

Which input snapshots produced output snapshot 885?

job: breach-detector
version: 4.3.1
inputs:
  - dataset: iceberg://prod/enforcement/silver.case_event
  - dataset: iceberg://prod/reference/policy_calendar
outputs:
  - dataset: iceberg://prod/enforcement/gold.case_sla_breach
columnLineage:
  - output: gold.case_sla_breach.case_id
    inputs:
      - silver.case_event.case_id
    kind: COPY
  - output: gold.case_sla_breach.is_breached
    inputs:
      - silver.case_event.sla_deadline
      - silver.case_event.closed_at
    kind: DERIVE

Every published asset version must have a producing run.
Every producing run must have at least one code version.
Every certified asset must have owner and SLO.
Every sensitive output must have upstream sensitive lineage or documented derivation.
Every gold asset must have at least one quality result.
Every backfill output must link to a backfill campaign.
Every table write outside platform must be flagged.

public final class LineageQualityValidator {
    public List<LineageViolation> validate(AssetPublication publication) {
        var violations = new ArrayList<LineageViolation>();

        if (publication.producingRun().isEmpty()) {
            violations.add(error("MISSING_PRODUCING_RUN"));
        }
        if (publication.codeRef().isEmpty()) {
            violations.add(error("MISSING_CODE_REF"));
        }
        if (publication.asset().certified() && publication.owner().isEmpty()) {
            violations.add(error("CERTIFIED_ASSET_MISSING_OWNER"));
        }
        if (publication.asset().classification().isSensitive()
                && publication.privacyLineage().isEmpty()) {
            violations.add(error("SENSITIVE_ASSET_MISSING_PRIVACY_LINEAGE"));
        }

        return violations;
    }
}

Where did sensitive data go?
Was it copied, masked, hashed, aggregated, tokenized, or leaked?

public enum Sensitivity {
    PUBLIC,
    INTERNAL,
    CONFIDENTIAL,
    PII,
    RESTRICTED_PII,
    SECRET
}

public record PrivacyTransform(
        DatasetColumn output,
        List<DatasetColumn> inputs,
        Sensitivity outputSensitivity,
        PrivacyOperation operation
) {}

public enum PrivacyOperation {
    COPY,
    MASK,
    TOKENIZE,
    HASH,
    ENCRYPT,
    AGGREGATE,
    DROP,
    DERIVE,
    UNKNOWN
}

Hashed PII may still be sensitive.

source.subject_national_id reclassified as RESTRICTED_PII
-> case_subject_hash derives from it via HASH
-> gold.case_subject_risk contains derived restricted identifier
-> export.daily_case_risk is sent to external vendor
-> access/export policy review required

asset: iceberg://prod/enforcement/gold.case_sla_breach
consumers:
  - name: regulatory-sla-dashboard
    type: DASHBOARD
    owner: enforcement-analytics
    criticality: HIGH
    freshnessSlo: PT2H
  - name: case-risk-api
    type: API
    owner: case-platform
    criticality: CRITICAL
    freshnessSlo: PT15M
  - name: monthly-regulatory-report
    type: REPORT
    owner: compliance-reporting
    criticality: CRITICAL
    deadline: business-day-3

public record Supersession(
        DatasetVersion badVersion,
        DatasetVersion replacementVersion,
        String reason,
        String incidentId,
        Instant supersededAt
) {}

CREATE TABLE dataset (
    dataset_id UUID PRIMARY KEY,
    namespace TEXT NOT NULL,
    name TEXT NOT NULL,
    type TEXT NOT NULL,
    owner TEXT,
    classification TEXT,
    UNIQUE(namespace, name)
);

CREATE TABLE pipeline_job (
    job_id UUID PRIMARY KEY,
    namespace TEXT NOT NULL,
    name TEXT NOT NULL,
    owner TEXT,
    service TEXT,
    UNIQUE(namespace, name)
);

CREATE TABLE pipeline_run (
    run_id TEXT PRIMARY KEY,
    job_id UUID NOT NULL REFERENCES pipeline_job(job_id),
    status TEXT NOT NULL,
    started_at TIMESTAMPTZ NOT NULL,
    finished_at TIMESTAMPTZ,
    code_version TEXT,
    config_version TEXT,
    backfill_id TEXT
);

CREATE TABLE run_input (
    run_id TEXT NOT NULL REFERENCES pipeline_run(run_id),
    dataset_id UUID NOT NULL REFERENCES dataset(dataset_id),
    version_ref TEXT,
    position JSONB,
    PRIMARY KEY(run_id, dataset_id)
);

CREATE TABLE run_output (
    run_id TEXT NOT NULL REFERENCES pipeline_run(run_id),
    dataset_id UUID NOT NULL REFERENCES dataset(dataset_id),
    version_ref TEXT,
    position JSONB,
    publication_status TEXT NOT NULL,
    PRIMARY KEY(run_id, dataset_id, version_ref)
);

CREATE TABLE dataset_edge (
    from_dataset_id UUID NOT NULL REFERENCES dataset(dataset_id),
    to_dataset_id UUID NOT NULL REFERENCES dataset(dataset_id),
    edge_type TEXT NOT NULL,
    confidence TEXT NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL,
    PRIMARY KEY(from_dataset_id, to_dataset_id, edge_type)
);

CREATE TABLE column_lineage (
    id UUID PRIMARY KEY,
    from_dataset_id UUID NOT NULL,
    from_column TEXT NOT NULL,
    to_dataset_id UUID NOT NULL,
    to_column TEXT NOT NULL,
    transform_kind TEXT NOT NULL,
    transform_version TEXT NOT NULL,
    expression_id TEXT,
    updated_at TIMESTAMPTZ NOT NULL
);

WITH RECURSIVE downstream(depth, from_dataset_id, to_dataset_id, path) AS (
    SELECT
        1,
        from_dataset_id,
        to_dataset_id,
        ARRAY[from_dataset_id, to_dataset_id]
    FROM dataset_edge
    WHERE from_dataset_id = :changed_dataset_id

    UNION ALL

    SELECT
        d.depth + 1,
        e.from_dataset_id,
        e.to_dataset_id,
        d.path || e.to_dataset_id
    FROM downstream d
    JOIN dataset_edge e ON e.from_dataset_id = d.to_dataset_id
    WHERE d.depth < 10
      AND NOT e.to_dataset_id = ANY(d.path)
)
SELECT * FROM downstream;

WITH RECURSIVE col_downstream AS (
    SELECT
        from_dataset_id,
        from_column,
        to_dataset_id,
        to_column,
        1 AS depth
    FROM column_lineage
    WHERE from_dataset_id = :dataset_id
      AND from_column = :column

    UNION ALL

    SELECT
        cl.from_dataset_id,
        cl.from_column,
        cl.to_dataset_id,
        cl.to_column,
        cd.depth + 1
    FROM col_downstream cd
    JOIN column_lineage cl
      ON cl.from_dataset_id = cd.to_dataset_id
     AND cl.from_column = cd.to_column
    WHERE cd.depth < 10
)
SELECT * FROM col_downstream;

GET /assets/{id}/upstream
GET /assets/{id}/downstream
GET /assets/{id}/runs?from=&to=
GET /runs/{runId}
GET /runs/{runId}/inputs
GET /runs/{runId}/outputs
POST /impact/analyze
POST /lineage/events
POST /lineage/reconcile
GET /columns/{asset}/{column}/downstream
GET /incidents/{incidentId}/affected-assets

{
  "asset": "iceberg://prod/enforcement/silver.case_event",
  "column": "accepted_at",
  "changeType": "DATA_INCORRECT",
  "effectiveWindow": {
    "from": "2026-07-01T00:00:00Z",
    "to": "2026-07-04T02:00:00Z"
  },
  "sourceSeverity": "TAINTED"
}

{
  "affected": [
    {
      "asset": "iceberg://prod/enforcement/gold.case_sla_breach",
      "severity": "CRITICAL",
      "path": ["silver.case_event", "gold.case_sla_breach"],
      "reason": "uses accepted_at to compute sla_deadline and breach flag",
      "recommendedAction": "block publication and reprocess affected partitions"
    }
  ]
}

System computes full blast radius.
User sees only assets they are authorized to view.
Privileged incident commander sees restricted paths.

Architecture diagram says A -> B -> C, but no run evidence exists.

Emit run-level lineage with input/output versions.

No downstream dashboard/API/report consumers are tracked.

Add consumer registry and non-table assets.

Only successful runs are recorded. Incident root cause disappears.

Emit START/FAIL/ABORT events with partial evidence.

Lineage service outage causes permanent metadata gap.

Use run manifest and lineage outbox.

Sensitive data is copied/derived downstream without visibility.

Track column lineage and privacy operations.

SQL parser misses UDF/API/dynamic transformation.

Combine automatic capture with versioned manual declarations.

a diagram generated by a catalog tool

a queryable evidence graph of dataset production, transformation, ownership, quality, code version, runtime execution, and downstream usage

manually asking five teams who might be affected

policy-driven traversal of the evidence graph to determine blast radius, severity, and required action

No important data output should be published without enough lineage to explain how it was produced and who may be affected if it is wrong.