Lineage from pipelines

By the end of this chapter you can capture lineage from your pipelines as OpenLineage run events, with column lineage parsed from their SQL, including the columns that only join, filter and group. You'll use the graph those events describe to say what a column change breaks downstream, and to propose classifications for tables nobody has classified. You'll also have checks for the three ways lineage fails in production: events that never arrive, a job that sends none, and a copy of personal data classified as less than personal.

The platform, the online learning platform that chapter 1 introduces and this book follows, sells professional courses to adult learners, and four SQL pipelines build reporting tables from its eight source tables. Each is one create table ... as select ... in examples/pipelines/, and the book's shared module lib.pipelines runs it on DuckDB and writes the result back to the catalog:

• course_performance writes analytics.course_performance: enrollments, completions, ratings and revenue per course.
• learner_activity writes analytics.learner_activity: each learner's events and last activity.
• marketing_audience writes marketing.audience: the learners marketing may email, with their name and email.
• review_corpus writes ai.review_corpus: the review text the platform's course assistant is grounded on.

Chapter 2's metadata check found that two of those outputs, marketing.audience and ai.review_corpus, have no metadata at all: no owner, and no classification for any column. Yet one holds learners' names and emails, and the other holds the free text of reviews, where chapter 3 found names and emails too. Two questions follow, and the catalog can't answer either. Which columns of those tables are personal data? And if the CRM team renames or reformats crm.learners.email, what breaks? Both are questions about lineage: where each column's data came from, and where it goes.

This chapter adds the book's shared module for lineage, lib.lineage. It runs each pipeline, tells OpenLineage what the run read and wrote, column by column, and reads the result back as a graph that answers both questions.

Send a run event from every pipeline

OpenLineage, an open specification for lineage, describes it with three objects (object model). A job is "a process that consumes or produces Datasets", such as a pipeline. A run is one occurrence of a job, identified by a UUID the client generates, and the spec recommends UUIDv7. A dataset is a table or file, named by a namespace and a name that together identify it. A run reports itself through run events: a START when it begins, then one terminal event, COMPLETE, FAIL or ABORT, and a server treats the events for one run as accumulating (run cycle). Lineage comes mostly from the work itself: in the spec's words, a backend "learns about Datasets primarily by receiving information about Jobs".

The Python client, openlineage-python 1.54.0, builds events from RunEvent, Run, Job, InputDataset and OutputDataset, which come from openlineage.client.event_v2, and sends them through a transport (client docs). It ships HTTP and Kafka transports for servers, and a file transport "designed mainly for integration testing", which with append set writes one JSON event per line to a single file (configuration). The client computes no lineage. It serializes what the caller passes in, and adds only metadata of its own: a run facet with its version, the job's source location when it runs in a Git checkout, and any environment variables you configure it to collect (emit in client/client.py, openlineage-python). So the lineage that reaches a server is the lineage your code put in the event.

For the platform, lib.lineage.run wraps lib.pipelines.run, which runs one of the platform's four SQL pipelines and writes its table to the catalog. It sends START with the tables the pipeline's SQL reads as inputs, runs the pipeline, and then sends COMPLETE with the output dataset and two facets on it. A facet is OpenLineage's name for a block of metadata attached to a run, a job or a dataset, and these two hold the output's schema and its column lineage, which the next section builds. If the pipeline raises, it sends FAIL instead. lib.lineage.run_all does that for all four pipelines into one events file. The datasets need a namespace, and OpenLineage's naming conventions have no entry for an Iceberg table in a local SQL catalog. The nearest is the local file system, named by its path, which would put a folder on your machine into every event. So lib.lineage uses the placeholder iceberg://platform, with the catalog's identifiers, such as crm.learners, as names.

The examples need the same bytes every time, and two things in an event change by default: its time, and a random run ID. So lib.lineage stamps every event with the nominal start of the nightly run on the book's "today", and derives the run ID with the client's generate_static_uuid, which, its docstring says, "always produces the same result" for the same time and data. Here the data is the job's name. Run the four pipelines with lineage twice, into two files, and print what the first file's events say, never the raw JSON:

1"""Run the platform's four pipelines with OpenLineage run events, then read back what the events say."""23from openlineage.client.uuid import generate_new_uuid, generate_static_uuid45from lib import lineage6from lib.catalog import open_catalog7from lib.platform import scratch89folder = scratch()10catalog = open_catalog()11lineage.run_all(catalog, folder / "first.jsonl")12lineage.run_all(catalog, folder / "second.jsonl")          # the same nightly run, sent again13events = lineage.read_events(folder / "first.jsonl")1415print("events:", len(events))16for event in events:17    job, state = event["job"]["name"], event["eventType"]18    if state == "START":19        print(f"{job} START, reads:", ", ".join(dataset["name"] for dataset in event["inputs"]))20    else:21        output = event["outputs"][0]22        lineage_fields = output["facets"]["columnLineage"]["fields"]23        print(f"{job} {state}, writes:", output["name"], f"({len(lineage_fields)} columns with lineage)")2425print("event times:", ", ".join(sorted({event["eventTime"] for event in events})))26print("distinct run IDs:", len({event["run"]["runId"] for event in events}))27print("run facets the client added:", ", ".join(events[0]["run"]["facets"]))28print("same bytes when sent again:", (folder / "first.jsonl").read_bytes() == (folder / "second.jsonl").read_bytes())29print("two random run IDs equal:", generate_new_uuid() == generate_new_uuid())30print("two static run IDs equal:", generate_static_uuid(lineage.NIGHTLY, b"marketing_audience")31      == generate_static_uuid(lineage.NIGHTLY, b"marketing_audience"))

events: 8
course_performance START, reads: billing.payments, learning.courses, learning.enrollments, learning.reviews
course_performance COMPLETE, writes: analytics.course_performance (7 columns with lineage)
learner_activity START, reads: crm.learners, learning.events
learner_activity COMPLETE, writes: analytics.learner_activity (6 columns with lineage)
marketing_audience START, reads: crm.learners, learning.courses, learning.enrollments
marketing_audience COMPLETE, writes: marketing.audience (5 columns with lineage)
review_corpus START, reads: learning.courses, learning.reviews
review_corpus COMPLETE, writes: ai.review_corpus (5 columns with lineage)
event times: 2026-10-01T02:00:00+00:00
distinct run IDs: 4
run facets the client added: tags
same bytes when sent again: True
two random run IDs equal: False
two static run IDs equal: True

The file holds 8 events, a START and a COMPLETE for each of the 4 runs, all stamped 2026-10-01T02:00:00+00:00. The marketing_audience job's START names the tables its SQL reads, crm.learners, learning.courses and learning.enrollments, and its COMPLETE names marketing.audience, with lineage for each of its columns. Comparing the two files prints True: sending the same nominal run again wrote the same bytes, which random run IDs would never allow. Even so, the events hold something you didn't put there: a run facet named tags, which carries the client's own version (changelog, 1.38.0), and every facet's _producer URL names that version too (DEFAULT_PRODUCER in the client's client/constants.py). So the bytes change with every client release even when the lineage doesn't, and a test should compare what events say, as this example's summary does.

A production emitter stamps COMPLETE with the time the run finished. A derived run ID is still worth keeping there, made from the scheduled time, the job's name and the attempt number: an event sent twice then lands in the same run instead of creating a second one, and a retried job gets a run of its own. Production events should also carry the output's Iceberg snapshot ID, in the version dataset facet that the object model names for it. The examples leave it out, because PyIceberg 0.12.0 draws each snapshot ID from a random UUID (_generate_snapshot_id in its pyiceberg/table/metadata.py).

Working rule: Send a START and then a COMPLETE or FAIL event from the code that runs each job, so no run can happen without them, and name datasets by catalog identifiers under a namespace you choose, never a path. In tests, use the file transport with a fixed event time and static run IDs, and compare what events say, never their bytes. In production, send through a transport to your lineage server, and derive each run ID from the scheduled time, the job and the attempt, so a resent event never creates a duplicate run.

Trace each output column to its sources

A run event says which tables a job read and wrote. Column lineage says which input columns each output column came from, and that is what governance needs: a table that reads crm.learners may or may not copy its emails. OpenLineage puts column lineage in a dataset facet, columnLineage, on the output: fields maps each output column to its inputFields, and an optional dataset list holds inputs that affect the whole dataset (facet schema). Since the client computes none of it, something has to parse each job's SQL.

sqlglot, a SQL parser, does it with sqlglot.lineage.lineage. Given a statement and the schema of the tables it reads, it returns, for each output column, a tree whose leaves are source columns, and with column=None it returns every output column at once (sqlglot lineage). It needs the schema. Without one, SELECT * comes back as a single output named *, and an unqualified column in a join as a placeholder with no table. lib.lineage.schema builds the schema from the catalog's Iceberg metadata, so the parser sees the columns the warehouse has, at sqlglot 30.24.0.

For the platform, take marketing_audience. In words, it picks learners who agreed to marketing email and whose account is active, and adds each one's favourite course category, which a CTE computes by ranking categories by the learner's enrollments. Run sqlglot on it, and on two columns of course_performance:

1"""What sqlglot.lineage says the platform's pipeline outputs come from, and what it leaves out or doesn't mark."""2import sqlglot3from sqlglot import exp45from lib import lineage, pipelines6from lib.catalog import open_catalog78schema = lineage.schema(open_catalog(["crm.learners", "learning.courses", "learning.enrollments", "learning.reviews",9                                      "billing.payments"]))10statement = pipelines.sql("marketing_audience")11found = lineage.sqlglot_sources(statement, schema)12for column, sources in found.items():13    print(f"marketing.audience.{column} comes from:", ", ".join(sources))1415parsed = sqlglot.parse_one(statement, read="duckdb")16tables = {table.alias_or_name: f"{table.db}.{table.name}" for table in parsed.find_all(exp.Table)}17where = parsed.find(exp.Where)18print("the pipeline's WHERE:", where.this.sql(dialect="duckdb"))19in_lineage = {source for sources in found.values() for source in sources}20for column in where.find_all(exp.Column):21    source = f"{tables[column.table]}.{column.name}"22    print(f"{source} in any column's lineage:", source in in_lineage)2324course_performance = lineage.sqlglot_sources(pipelines.sql("course_performance"), schema)25for column in ("completions", "revenue_usd"):26    print(f"analytics.course_performance.{column} comes from:", ", ".join(course_performance[column]))

marketing.audience.learner_id comes from: crm.learners.learner_id
marketing.audience.full_name comes from: crm.learners.full_name
marketing.audience.email comes from: crm.learners.email
marketing.audience.country comes from: crm.learners.country
marketing.audience.favourite_category comes from: learning.courses.category
the pipeline's WHERE: l.marketing_consent AND l.status = 'active'
crm.learners.marketing_consent in any column's lineage: False
crm.learners.status in any column's lineage: False
analytics.course_performance.completions comes from: learning.enrollments.completed_at, learning.enrollments.enrollment_id
analytics.course_performance.revenue_usd comes from: billing.payments.amount, billing.payments.status

The values are right. email comes from crm.learners.email, and favourite_category is traced through the CTE to learning.courses.category. What's missing is everything that decides which learners are in the table. The pipeline's WHERE is l.marketing_consent AND l.status = 'active', and neither column is in any output's lineage: the example's check prints False for marketing_consent and False for status. The consent filter is the reason marketing may hold this table at all, and the lineage doesn't mention it. Nor does it mention the join keys from enrollments and courses.

The second problem is the reverse: an input that only filters is listed as if it were a value. course_performance sums only succeeded payments, with filter (where p.status = 'succeeded'), and sqlglot says revenue_usd comes from billing.payments.amount, billing.payments.status: the column summed and the column that only decides which payments count, listed alike. completions counts enrollments with filter (where e.completed_at is not null), and comes back the same way, from learning.enrollments.completed_at, learning.enrollments.enrollment_id. Both problems follow from how sqlglot builds the tree. It collects the columns in each output's expression, so it sees a filter inside an expression and nothing outside it, and its nodes carry no type to say how an input was used.

Working rule: Use sqlglot's lineage for the columns each output's value is computed from, and always pass the catalog's schema, generated from the table metadata, never typed. Don't read its answer as everything a column depends on: it misses the columns that join, filter and group, and it doesn't say which of its inputs are values.

Add the inputs that decide which rows exist

sqlglot's lineage lists the columns a value is computed from, and OpenLineage has a place for the rest. Each input field carries a list of transformations, each with a type, a subtype and a masking flag (column lineage facet). DIRECT means the "output column value was somehow derived from inputField value", as IDENTITY, TRANSFORMATION or AGGREGATION. INDIRECT means the "output column value is impacted by the value of inputField column, but it's not derived from it", as JOIN, GROUP_BY, FILTER, SORT, WINDOW or CONDITIONAL. Inputs that shape every row, such as a WHERE, belong in the facet's dataset list.

For governance, the INDIRECT side matters, because information flows through conditions as well as values. Denning calls it "implicit flow": a statement that assigns only when a condition holds still tells you the condition's value (Denning, 1976). A row in marketing.audience tells you that learner consented.

lib.lineage.column_lineage adds the two things sqlglot's answer lacks: a type for each value input, and the INDIRECT inputs. It types each value input by where the column sits in the output's expression: a bare column is IDENTITY, a column inside an aggregate is AGGREGATION, and anything else is TRANSFORMATION. A column inside an aggregate's FILTER, a window's PARTITION BY or ORDER BY, or a CASE's condition or operand is INDIRECT, and a hash on the way sets masking. COALESCE's arguments count as values: OpenLineage lists COALESCE under CONDITIONAL, but any of its arguments can become the output's value. Then it walks every SELECT in the statement, CTEs included, and collects the columns of its row clauses, following CTE columns back to the tables they come from:

ROW_CLAUSES = {"joins": "JOIN", "where": "FILTER", "group": "GROUP_BY", "having": "FILTER", "qualify": "FILTER",
               "order": "SORT"}

Run it on marketing_audience, and on the two course_performance columns again:

1"""The book's column lineage: each column's inputs as OpenLineage types them, plus the inputs that shape every row,2which sqlglot.lineage leaves out."""3from lib import lineage, pipelines4from lib.catalog import open_catalog56schema = lineage.schema(open_catalog(["crm.learners", "learning.courses", "learning.enrollments", "learning.reviews",7                                      "billing.payments"]))8fields, rows = lineage.column_lineage(pipelines.sql("marketing_audience"), schema)9for column, inputs in fields.items():10    print(f"marketing.audience.{column}:", ", ".join(f"{dependency.source} {dependency.type} {dependency.subtype}"11                                                    for dependency in inputs))12print("inputs that shape every row of marketing.audience:")13for source in sorted({dependency.source for dependency in rows}):14    print(f"  {source}:", "INDIRECT",15          ", ".join(dependency.subtype for dependency in rows if dependency.source == source))1617values = {dependency.source for inputs in fields.values() for dependency in inputs}18print("source columns, values only:", len(values))19print("source columns, with the rows' inputs:", len(values | {dependency.source for dependency in rows}))2021fields, _ = lineage.column_lineage(pipelines.sql("course_performance"), schema)22for column in ("completions", "revenue_usd"):23    print(f"analytics.course_performance.{column}:",24          ", ".join(f"{dependency.source} {dependency.type} {dependency.subtype}" for dependency in fields[column]))

marketing.audience.learner_id: crm.learners.learner_id DIRECT IDENTITY
marketing.audience.full_name: crm.learners.full_name DIRECT IDENTITY
marketing.audience.email: crm.learners.email DIRECT IDENTITY
marketing.audience.country: crm.learners.country DIRECT IDENTITY
marketing.audience.favourite_category: learning.courses.category DIRECT IDENTITY
inputs that shape every row of marketing.audience:
  crm.learners.learner_id: INDIRECT JOIN
  crm.learners.marketing_consent: INDIRECT FILTER
  crm.learners.status: INDIRECT FILTER
  learning.courses.category: INDIRECT GROUP_BY, JOIN
  learning.courses.course_id: INDIRECT JOIN
  learning.enrollments.course_id: INDIRECT JOIN
  learning.enrollments.learner_id: INDIRECT GROUP_BY, JOIN
source columns, values only: 5
source columns, with the rows' inputs: 10
analytics.course_performance.completions: learning.enrollments.completed_at INDIRECT FILTER, learning.enrollments.enrollment_id DIRECT AGGREGATION
analytics.course_performance.revenue_usd: billing.payments.amount DIRECT AGGREGATION, billing.payments.status INDIRECT FILTER

The values are as sqlglot found them, each an IDENTITY copy. The inputs that shape every row are new: crm.learners.marketing_consent and crm.learners.status as FILTER, the learner and course IDs as JOIN, and the enrollments' learner_id and the category as GROUP_BY. The lineage of marketing.audience grows from 5 source columns to 10. For revenue_usd, the two inputs are now told apart: amount is a DIRECT AGGREGATION, and status an INDIRECT FILTER. So are the two inputs of completions: enrollment_id is counted, and completed_at only filters.

Two details of the facet are worth knowing. The per-field transformationType and transformationDescription are deprecated, replaced by the per-input transformations that lib.lineage writes (changelog, 1.17.1). And OpenLineage 1.53.0 added a Lineage dataset facet for declared lineage, such as a view's, sent on a DatasetEvent, which consumers should prefer where both describe the same relationship (lineage facet). Run events from jobs carry columnLineage, as the platform's do.

Working rule: Record INDIRECT inputs with every job's column lineage: the join keys, filters and grouping columns of every SELECT, CTEs included, in the facet's dataset list, and type each value input with OpenLineage's subtypes. Expect more edges, many of them join keys. Keep them anyway: a filter on a consent column decides who may be emailed, and it reaches the table only as an INDIRECT input.

Fail on statements the parser can't read

The platform's column lineage is only as good as the parse behind it, and a parse can be wrong without failing. sqlglot's lineage accepts more than a SELECT: it finds the query inside a CREATE TABLE ... AS, an INSERT ... SELECT or a CREATE VIEW, and it reads a MERGE too (sqlglot, optimizer/scope.py). Three cases come back looking fine. An INSERT's outputs are named after its SELECT, never after the table's columns. A MERGE comes back as the last query sqlglot reaches in it, as if that query were what the MERGE writes. When USING names a table or a CTE and no subquery follows, sqlglot returns no lineage and no error. When USING is a subquery, or a subquery sits in a WHEN condition or a SET, it returns the columns of the last such subquery, whether the MERGE writes them or not. And a misspelt column gets through when it's unqualified in a query on one table: sqlglot assigns it to that table whether the table has the column or not.

All four of the platform's pipelines are CREATE TABLE ... AS statements. As tables grow, pipelines like these often become incremental: an INSERT that appends a day's revenue to analytics.daily_revenue, or a MERGE that updates analytics.learner_activity in place. Run sqlglot and lib.lineage.column_lineage on both, with the MERGE in two forms:

1"""Two incremental statements sqlglot.lineage answers without complaint and gets wrong: an INSERT, whose outputs it2names after the SELECT, and a MERGE, for which it returns nothing, or the columns of the subquery its USING names."""3import pyarrow as pa4from sqlglot.lineage import lineage as sqlglot_lineage56from lib import lineage7from lib.catalog import open_catalog89catalog = open_catalog(["billing.payments", "crm.learners"])10catalog.create_namespace("analytics")             # the incremental table, as it exists before the day's INSERT11catalog.create_table("analytics.daily_revenue", schema=pa.schema([("day", pa.date32()), ("revenue_usd", pa.float64())]))12schema = lineage.schema(catalog)13insert = """insert into analytics.daily_revenue14select cast(p.paid_at as date), sum(p.amount) from billing.payments p where p.status = 'succeeded' group by 1"""15merges = {"USING a table": """merge into analytics.learner_activity a using crm.learners l16on a.learner_id = l.learner_id when matched then update set country = l.country""",17          "USING a subquery": """merge into analytics.learner_activity a18using (select l.learner_id, l.country, l.email from crm.learners l) s on a.learner_id = s.learner_id19when not matched then insert (learner_id, country) values (s.learner_id, s.country)"""}2021print("analytics.daily_revenue columns:", ", ".join(schema["analytics"]["daily_revenue"]))22print("sqlglot, INSERT outputs:", ", ".join(sqlglot_lineage(None, insert, schema, dialect="duckdb")))23fields, rows = lineage.column_lineage(insert, schema)24for column, inputs in fields.items():25    print(f"book, INSERT {column}:", ", ".join(f"{dependency.source} {dependency.type} {dependency.subtype}"26                                               for dependency in inputs))27print("book, INSERT rows shaped by:", ", ".join(f"{dependency.source} {dependency.subtype}" for dependency in rows))28for label, merge in merges.items():29    outputs = sqlglot_lineage(None, merge, schema, dialect="duckdb")30    print(f"sqlglot, MERGE {label}, outputs:", ", ".join(outputs) or "none")3132refused = {"INSERT with one value too many": insert.replace("sum(p.amount)", "sum(p.amount), count(*)"),33           **{f"MERGE {label}": merge for label, merge in merges.items()}}34for label, statement in refused.items():35    try:36        lineage.column_lineage(statement, schema)37    except lineage.LineageError as error:38        print(f"book, {label}:", error)

analytics.daily_revenue columns: day, revenue_usd
sqlglot, INSERT outputs: paid_at, _col_1
book, INSERT day: billing.payments.paid_at DIRECT TRANSFORMATION
book, INSERT revenue_usd: billing.payments.amount DIRECT AGGREGATION
book, INSERT rows shaped by: billing.payments.paid_at GROUP_BY, billing.payments.status FILTER
sqlglot, MERGE USING a table, outputs: none
sqlglot, MERGE USING a subquery, outputs: learner_id, country, email
book, INSERT with one value too many: analytics.daily_revenue: the INSERT writes 2 columns, and its SELECT has 3 values
book, MERGE USING a table: analytics.learner_activity: lineage is read only from CREATE ... AS and INSERT ... SELECT, not from MERGE
book, MERGE USING a subquery: analytics.learner_activity: lineage is read only from CREATE ... AS and INSERT ... SELECT, not from MERGE

The table analytics.daily_revenue has the columns day, revenue_usd, and sqlglot names the INSERT's outputs paid_at, _col_1: the first after the column inside the cast, the second a generated name for an expression with no alias. Neither is a column of the table. An INSERT fills the table's columns by position, so lib.lineage matches by position too, and takes the names from the catalog: the columns the INSERT lists, or else all of the table's. So day comes from billing.payments.paid_at DIRECT TRANSFORMATION, and the INSERT's grouping and its filter on status shape every row: billing.payments.paid_at GROUP_BY, billing.payments.status FILTER. When the SELECT has more or fewer values than the columns written, lib.lineage raises: analytics.daily_revenue: the INSERT writes 2 columns, and its SELECT has 3 values.

For the MERGE whose USING names a table, sqlglot returns none, and no error. For the one whose USING is a subquery, it returns learner_id, country, email, though that MERGE writes only learner_id and country. A check for empty lineage catches the first and passes the second. So lib.lineage doesn't judge what sqlglot returns for a MERGE. Before it calls sqlglot, it refuses every statement that isn't a CREATE ... AS or an INSERT ... SELECT: analytics.learner_activity: lineage is read only from CREATE ... AS and INSERT ... SELECT, not from MERGE.

A misspelt column, the third case, depends on how the column is written. The platform's pipelines qualify every column with its table's alias, as in l.email, and sqlglot checks a qualified column against the schema itself:

1"""A misspelt column: sqlglot raises when the column is qualified, and when it isn't, in a query on one table, traces it2to that table anyway. lib.lineage checks every source against the schema."""3from sqlglot.errors import OptimizeError45from lib import lineage6from lib.catalog import open_catalog78schema = lineage.schema(open_catalog(["crm.learners"]))9qualified = "create table marketing.audience as select l.learner_id, l.emial from crm.learners l"10unqualified = "create table marketing.audience as select learner_id, emial from crm.learners"1112try:13    lineage.sqlglot_sources(qualified, schema)14except OptimizeError as error:15    print("sqlglot, qualified l.emial:", f"{type(error).__name__}: {error}")16print("sqlglot, unqualified emial comes from:", ", ".join(lineage.sqlglot_sources(unqualified, schema)["emial"]))17try:18    lineage.column_lineage(unqualified, schema)19except lineage.LineageError as error:20    print("book, unqualified emial:", error)

sqlglot, qualified l.emial: OptimizeError: Unknown column: emial
sqlglot, unqualified emial comes from: crm.learners.emial
book, unqualified emial: crm.learners has no column emial

Qualified, the typo stops sqlglot with OptimizeError: Unknown column: emial. Unqualified, in a query on one table, it comes back as crm.learners.emial, a confident answer about a column that doesn't exist. In a join, a misspelt unqualified column belongs to none of the tables, and sqlglot returns a placeholder with no table (sqlglot, lineage.py), which lib.lineage rejects as a source it can't place. For the single-table case, lib.lineage checks every source against the schema and raises crm.learners has no column emial.

lib.lineage.record, which run uses, reads the lineage before the job runs, so a pipeline whose lineage can't be read fails before it writes anything. That is strict, and deliberate: a job with empty lineage would vanish from every impact report and every classification check in this chapter, and one with wrong lineage would mislead them. Run the same check in CI on every pull request that changes a pipeline, so the failure shows up in review rather than at night.

Working rule: Treat a statement your parser can't read correctly as an error, never as empty or partial lineage. Refuse every kind of statement it gets wrong, such as MERGE, before you ask it, and fail on empty lineage, a source the parser can't place and a column the schema doesn't have. Match an INSERT's columns by position against the table's columns in the catalog, never the SELECT's names, and fail when the counts differ. For a MERGE, write a SELECT of the columns it writes, and record the job's lineage from that, as an INSERT ... SELECT into the MERGE's table, until your parser handles MERGE itself.

Find what a column change breaks

With column lineage in every COMPLETE event, the platform has a graph: columns are nodes, and each input field is an edge to the column or table it feeds. Impact analysis asks what a change to one column can break. It is a walk of that graph: start at the column, follow every edge out of it, and keep going from wherever you land, at any depth. No special method is needed; it's graph reachability. One rule is specific to lineage. An INDIRECT edge into a table means every row of it can change, so the walk continues from all of that table's columns.

lib.lineage.graph builds the graph from the events file alone, as a lineage server would, not from the SQL. That way it includes every job that sends events, whatever it's written in, and nothing that doesn't. lib.lineage.downstream does the walk.

For the platform, take three changes its teams might make. The CRM team lowercases every email, or renames crm.learners.email. A new sign-up flow changes what crm.learners.marketing_consent means for older accounts. The learning team renames a category in learning.courses. Before you run it, predict how many columns the consent change can reach:

1"""What a change to a source column can break downstream, read from the events the nightly run sent."""23from lib import lineage4from lib.catalog import open_catalog5from lib.platform import scratch67events_file = scratch() / "events.jsonl"8lineage.run_all(open_catalog(), events_file)9graph = lineage.graph(lineage.read_events(events_file))10print("edges in the column graph:", len(graph.edges))1112for column in ("crm.learners.email", "crm.learners.marketing_consent", "learning.courses.category"):13    reached = lineage.downstream(graph, column)14    print(f"{column} reaches:")15    for target in dict.fromkeys(edge.target for edge in reached):16        how = ", ".join(f"{edge.type} {edge.subtype}" for edge in reached if edge.target == target)17        print(f"  {target}{', every row' if target.count('.') == 1 else ''}:", how)18    values = [edge for edge in reached if edge.type == "DIRECT"]19    print(f"{column}, columns reached through values:", len(lineage.columns_reached(graph, values)))20    print(f"{column}, columns reached in all:", len(lineage.columns_reached(graph, reached)))

edges in the column graph: 51
crm.learners.email reaches:
  marketing.audience.email: DIRECT IDENTITY
crm.learners.email, columns reached through values: 1
crm.learners.email, columns reached in all: 1
crm.learners.marketing_consent reaches:
  marketing.audience, every row: INDIRECT FILTER
crm.learners.marketing_consent, columns reached through values: 0
crm.learners.marketing_consent, columns reached in all: 5
learning.courses.category reaches:
  analytics.course_performance, every row: INDIRECT GROUP_BY
  analytics.course_performance.category: DIRECT IDENTITY
  marketing.audience, every row: INDIRECT GROUP_BY, INDIRECT JOIN
  marketing.audience.favourite_category: DIRECT IDENTITY
learning.courses.category, columns reached through values: 2
learning.courses.category, columns reached in all: 12

The graph has 51 edges. A change to email reaches one column, marketing.audience.email, so renaming it breaks the marketing pipeline and no other job that sends lineage. The consent change reaches 0 columns through values, yet through the filter it reaches all 5 columns of marketing.audience: it changes who gets emailed. A lineage tool that follows values only would report a change to consent, the column that decides who may be emailed, as having no impact. A category rename reaches 2 columns through values, and 12 in all, because categories also group analytics.course_performance and rank each learner's favourite.

The same walk serves two later chapters: chapter 7 follows a learner's columns downstream to find every derived table an erasure must reach, and chapter 10 writes an impact report for every pull request that changes a schema.

Working rule: Run impact analysis on the graph built from events, INDIRECT edges included, and report two lists: columns whose values can change, and tables whose rows can change. Treat a change to a column used only in filters and joins as a change to every column of the tables it filters, and send the report to the owners of each table it reaches.

Propose classifications from lineage

A derived column holds data from its inputs, so its classification can be derived from theirs. The classic rule comes from Denning's lattice model of information flow: the class of a value computed from several inputs is the least upper bound, the "join", of their classes (Denning, 1976). The platform's classifications form a line, public, internal, personal and sensitive, so the join is the stricter of them, which stricter in lib.metadata, the book's metadata module, returns. Which inputs count is a judgement. We count INDIRECT inputs too, following Denning's implicit flows, because a table filtered on consent reveals each learner's consent. That choice lives in governance/lineage.toml, so a team can change it in one reviewed line.

Two more choices come from the same caution. OpenLineage's masking flag says only that a value was obfuscated, such as by a hash (column lineage facet). Whether a hashed identifier is still personal data is a legal question: chapter 2 quotes the GDPR recital on pseudonymized data and leaves the answer to lawyers. So the book's rule never lowers a classification because of masking. Catalog servers leave the same choice to configuration. Apache Atlas propagates classifications along lineage and lets you block propagation on an edge, for example where a view masks a PII column (Atlas), and DataHub propagates glossary terms along column lineage, a feature its 1.7.0 documentation describes as a public beta in DataHub Cloud (DataHub). And a proposal is a floor, not a verdict: like the drafts a local model writes in chapter 2, it's reviewed before it goes into governance/metadata.toml, the platform's file of owners and classifications.

For the platform, lib.lineage.propose proposes a classification for every column a job wrote, and lib.metadata.problems lists the tables that need one. Before trusting the rule on the two tables with no metadata, marketing.audience and ai.review_corpus, check it on the two that people did classify, analytics.course_performance and analytics.learner_activity:

1"""Propose classifications for the pipeline outputs that have no metadata, from their lineage, and check the rule2against the outputs people did classify."""34from lib import lineage, metadata5from lib.catalog import open_catalog6from lib.platform import scratch78catalog = open_catalog()9events_file = scratch() / "events.jsonl"10lineage.run_all(catalog, events_file)11for problem in metadata.problems(catalog, metadata.load()):12    print("metadata problem:", problem)1314graph, known = lineage.graph(lineage.read_events(events_file)), lineage.declared()15print("propagated through:", " and ".join(lineage.rules()["propagate_through"]))16proposed = lineage.propose(graph, known)17for column, (classification, decided_by) in proposed.items():18    if column not in known:19        first = decided_by[0]20        print(f"{column}: {classification}, from {first.source} ({first.type} {first.subtype})")21print("personal columns proposed:", sum(1 for column, (classification, _) in proposed.items()22                                        if column not in known and classification == "personal"))2324classified = [column for column in proposed if column in known]25print("declared classifications checked:", len(classified))26print("classes declared for them:", ", ".join(sorted({known[column] for column in classified})))27for through in (["DIRECT", "INDIRECT"], ["DIRECT"]):28    proposals = lineage.propose(graph, known, through)29    agree = sum(1 for column in classified if proposals[column][0] == known[column])30    print(f"declared classifications reproduced, through {' and '.join(through)}:", f"{agree} of {len(classified)}")31missed = [column for column in classified if proposals[column][0] != known[column]]    # with DIRECT inputs only32print("missed with DIRECT inputs only:", len(missed))33for column in missed:34    print(f"  {column}: declared {known[column]}, DIRECT inputs propose {proposals[column][0]}")35print("declared less protected than lineage proposes:", len(lineage.understated(proposed, known)))

metadata problem: ai.review_corpus: no metadata
metadata problem: marketing.audience: no metadata
propagated through: DIRECT and INDIRECT
marketing.audience.learner_id: internal, from crm.learners.learner_id (DIRECT IDENTITY)
marketing.audience.full_name: personal, from crm.learners.full_name (DIRECT IDENTITY)
marketing.audience.email: personal, from crm.learners.email (DIRECT IDENTITY)
marketing.audience.country: internal, from crm.learners.country (DIRECT IDENTITY)
marketing.audience.favourite_category: internal, from crm.learners.learner_id (INDIRECT JOIN)
ai.review_corpus.review_id: internal, from learning.reviews.review_id (DIRECT IDENTITY)
ai.review_corpus.course_id: internal, from learning.reviews.course_id (DIRECT IDENTITY)
ai.review_corpus.course_title: internal, from learning.reviews.course_id (INDIRECT JOIN)
ai.review_corpus.rating: internal, from learning.reviews.rating (DIRECT IDENTITY)
ai.review_corpus.body: personal, from learning.reviews.body (DIRECT IDENTITY)
personal columns proposed: 3
declared classifications checked: 13
classes declared for them: internal
declared classifications reproduced, through DIRECT and INDIRECT: 13 of 13
declared classifications reproduced, through DIRECT: 10 of 13
missed with DIRECT inputs only: 3
  analytics.course_performance.course_id: declared internal, DIRECT inputs propose public
  analytics.course_performance.title: declared internal, DIRECT inputs propose public
  analytics.course_performance.category: declared internal, DIRECT inputs propose public
declared less protected than lineage proposes: 0

The metadata check from chapter 2 still reports the two tables with no metadata, and lineage proposes a classification for each of their columns. It proposes 3 personal columns: full_name and email in marketing.audience, each an IDENTITY copy of a personal column in crm.learners, and body in ai.review_corpus, a copy of the reviews' free text. The rest come out internal, including favourite_category, for which the example prints internal, from crm.learners.learner_id (INDIRECT JOIN). Its values are category names from the public course catalog, but the rows they sit in are chosen by a join on learner IDs and a consent filter, and those are internal.

On the 13 columns people did classify, all of them internal, counting INDIRECT inputs reproduces 13 of 13, and DIRECT inputs alone reproduce 10 of 13. The 3 that DIRECT inputs miss are course_id, title and category of analytics.course_performance, which a DIRECT-only rule calls public, as in the course catalog, while the analytics team chose internal. So the rule proposes no less than people chose here, and lineage finds no declared classification lower than its proposal (0). That is weak evidence: a rule that proposed internal for every column would score as well. And it can't show what the rule does with a personal filter, because none of the platform's filters or joins reads a personal column: even marketing_consent is classified internal.

So build a table whose filter does. For a birthday offer, marketing wants the learners who consented and were born this month, with their ID and country, and nothing else:

1"""A table chosen by a personal column: the learners with a birthday this month, for a birthday offer. None of its2values is personal, yet each row reveals the month of a learner's birth. Only counting INDIRECT inputs proposes3personal."""45from lib import lineage6from lib.catalog import open_catalog7from lib.platform import scratch8from lib.warehouse import connect910catalog = open_catalog(["crm.learners"])11# The consenting learners born in the month of the nightly run, on the book's "today".12select = ("select l.learner_id, l.country from crm.learners l "13          f"where l.marketing_consent and month(l.date_of_birth) = {lineage.NIGHTLY.month}")141516def offers() -> None:17    rows = connect(catalog).sql(select).to_arrow_table()18    catalog.create_namespace_if_not_exists("marketing")19    catalog.create_table("marketing.birthday_offers", schema=rows.schema).append(rows)202122events_file = scratch() / "events.jsonl"23lineage.record(lineage.client(events_file), "birthday_offers", f"create table marketing.birthday_offers as {select}",24               catalog, offers)25offers_table = catalog.load_table("marketing.birthday_offers").scan().to_arrow()26print("learners in marketing.birthday_offers:", f"{len(offers_table):,}")2728graph, known = lineage.graph(lineage.read_events(events_file)), lineage.declared()29print("crm.learners.date_of_birth declared:", known["crm.learners.date_of_birth"])30for through in (["DIRECT", "INDIRECT"], ["DIRECT"]):31    for column, (classification, decided_by) in lineage.propose(graph, known, through).items():32        label = f"{column}, through {' and '.join(through)}"33        print(f"{label}:", classification)34        print(f"{label}, decided by:", f"{decided_by[0].source} ({decided_by[0].type} {decided_by[0].subtype})")

learners in marketing.birthday_offers: 165
crm.learners.date_of_birth declared: personal
marketing.birthday_offers.learner_id, through DIRECT and INDIRECT: personal
marketing.birthday_offers.learner_id, through DIRECT and INDIRECT, decided by: crm.learners.date_of_birth (INDIRECT FILTER)
marketing.birthday_offers.country, through DIRECT and INDIRECT: personal
marketing.birthday_offers.country, through DIRECT and INDIRECT, decided by: crm.learners.date_of_birth (INDIRECT FILTER)
marketing.birthday_offers.learner_id, through DIRECT: internal
marketing.birthday_offers.learner_id, through DIRECT, decided by: crm.learners.learner_id (DIRECT IDENTITY)
marketing.birthday_offers.country, through DIRECT: internal
marketing.birthday_offers.country, through DIRECT, decided by: crm.learners.country (DIRECT IDENTITY)

The table holds 165 learners, and neither of its columns copies anything personal: counting DIRECT inputs only proposes internal for learner_id and internal for country. Yet every row says that its learner was born in the month of the offer, part of a date of birth, which governance/metadata.toml classifies personal. Counting INDIRECT inputs proposes personal for both columns, decided by crm.learners.date_of_birth (INDIRECT FILTER). That is the case we count INDIRECT inputs for, and it shows the cost too: one personal filter makes every column of a table personal.

OpenLineage can carry classifications in the events themselves, with the Tags dataset facet, whose field key "identifies the field in a dataset if a tag applies to one" (tags facet, schema). We keep them in governance/metadata.toml instead, the one source that the platform's access policies also read, through lib.policy, the book's policy module (chapter 5).

Working rule: Propose each derived column's classification as the stricter of its inputs', counting INDIRECT inputs, and send the proposals to the table's owner for review before they become metadata. Never let lineage lower a classification someone declared, and never treat masking as declassification. Count DIRECT inputs only if your pipelines often join or filter on personal columns, such as matching on email, and every output would otherwise come out personal; then review INDIRECT personal inputs by hand.

What goes wrong in production

Lineage fails quietly: a missing edge looks exactly like a column with no dependencies, and every check built on the graph, from impact analysis to proposed classifications, then passes. Three failures are common, and each has a check.

Events that never arrive

The client reads OPENLINEAGE_DISABLED from the environment, and when it's true, events are dropped (configuration). In the client's source, that check comes first (openlineage-python, client/client.py): before it looks at the transport you passed explicitly, the client picks a transport that does nothing. Its only notice is a log line at INFO level: "OpenLineage is disabled. No events will be emitted." Python's root logger is set to WARNING by default, so the line never shows (logging HOWTO). A variable left in a shared environment file after a debugging session is enough. Set it, then run the platform's review_corpus pipeline with an explicit file transport:

1"""With OPENLINEAGE_DISABLED=true in the environment, an explicit file transport writes nothing and raises nothing: its2only notice is one INFO log line, which Python's default logging level hides."""3import os45from lib import lineage6from lib.catalog import open_catalog7from lib.platform import scratch89os.environ["OPENLINEAGE_DISABLED"] = "true"        # say, left in a shared environment file after a debugging session1011events_file = scratch() / "events.jsonl"12client = lineage.client(events_file)                # a file transport, passed in explicitly13catalog = open_catalog(["learning.courses", "learning.reviews"])14output = lineage.run(catalog, "review_corpus", client)1516print("transport the client chose:", client.transport.kind)17print("pipeline wrote:", output)18print("rows written:", f"{len(catalog.load_table(output).scan().to_arrow()):,}")19print("events file exists:", events_file.exists())20missing = lineage.missing_runs(["review_corpus"], lineage.read_events(events_file))21print("jobs that ran with no COMPLETE event:", ", ".join(missing))

transport the client chose: noop
pipeline wrote: ai.review_corpus
rows written: 3,000
events file exists: False
jobs that ran with no COMPLETE event: review_corpus

The transport the client chose is noop. The pipeline ran and wrote 3,000 rows to ai.review_corpus, the check for the events file prints False, and nothing raised an error. The check that catches it compares the jobs that ran, which the scheduler knows, with the jobs that sent a COMPLETE event: lib.lineage.missing_runs lists review_corpus.

Working rule: After every scheduled run, compare the jobs that ran with the jobs whose COMPLETE events arrived, and alert on any difference. Don't rely on the emitter to report its own silence.

A job nobody instrumented

Lineage covers only the jobs that send it. The platform's marketing team runs a weekly Python script that copies marketing.audience into marketing.campaign_list, a table its email tool reads. Nobody wired the script to OpenLineage:

1"""A Python step that copies marketing.audience for the email tool and tells OpenLineage nothing: impact analysis misses2the copy, and a check of the catalog against the lineage finds it."""34from lib import lineage5from lib.catalog import open_catalog6from lib.platform import scratch7from lib.warehouse import connect89catalog = open_catalog()10events_file = scratch() / "events.jsonl"11lineage.run_all(catalog, events_file)1213# The marketing team's weekly export, written straight to the catalog: no lineage event.14export = connect(catalog).sql("select learner_id, full_name, email from marketing.audience").to_arrow_table()15catalog.create_table("marketing.campaign_list", schema=export.schema).append(export)16print("rows exported:", f"{len(export):,}")1718events = lineage.read_events(events_file)19reached = lineage.downstream(lineage.graph(events), "crm.learners.email")20print("crm.learners.email reaches:", ", ".join(edge.target for edge in reached))21print("tables no job's lineage accounts for:", ", ".join(lineage.unaccounted(catalog, events)))

rows exported: 1,829
crm.learners.email reaches: marketing.audience.email
tables no job's lineage accounts for: marketing.campaign_list

The script copied 1,829 learners' names and emails, and impact analysis can't see the copy: crm.learners.email still reaches only marketing.audience.email. An erasure guided by this lineage would miss the copy. The catalog knows the table exists, though, so lib.lineage.unaccounted compares the catalog's tables with the lineage: every table must be one of the platform's sources or the output of a job that lineage knows. It finds marketing.campaign_list.

Working rule: Check the catalog against the lineage on a schedule, and treat a table no job's lineage accounts for as an incident: find the job, and make it send events.

Personal data in a table classified internal

Instrumenting the marketing team's export fixes the graph, and exposes a second problem. When the export's owner created marketing.campaign_list, they classified every column internal, because only the marketing team reads it. That's a statement about who reads a table, not what it holds. Wrap the script's work in lib.lineage.record, with the SELECT it runs as its lineage, and compare the declared classifications with lineage's proposal:

1"""The same export, now sending lineage, in a table its owner tagged internal: lineage finds the personal data in it."""23from lib import lineage4from lib.catalog import open_catalog5from lib.platform import scratch6from lib.warehouse import connect78catalog = open_catalog()9events_file = scratch() / "events.jsonl"10lineage.run_all(catalog, events_file)1112select = "select a.learner_id, a.full_name, a.email from marketing.audience a"131415def export() -> None:16    rows = connect(catalog).sql(select).to_arrow_table()17    catalog.create_table("marketing.campaign_list", schema=rows.schema).append(rows)181920statement = f"create table marketing.campaign_list as {select}"21lineage.record(lineage.client(events_file), "campaign_export", statement, catalog, export)2223# Its owner tagged every column internal: "only the marketing team reads it".24known = lineage.declared() | {f"marketing.campaign_list.{c}": "internal" for c in ("learner_id", "full_name", "email")}25graph = lineage.graph(lineage.read_events(events_file))26print("crm.learners.email reaches:", ", ".join(edge.target for edge in lineage.downstream(graph, "crm.learners.email")))27for problem in lineage.understated(lineage.propose(graph, known), known):28    print("understated:", problem)

crm.learners.email reaches: marketing.audience.email, marketing.campaign_list.email
understated: marketing.campaign_list.full_name: declared internal, lineage proposes personal
understated: marketing.campaign_list.email: declared internal, lineage proposes personal

crm.learners.email now reaches marketing.audience.email, marketing.campaign_list.email: two hops, the second through the export. And lib.lineage.understated reports full_name and email in marketing.campaign_list as declared internal while lineage proposes personal: copies of personal data that any policy reading the declared classification would treat as internal.

Working rule: Run lib.lineage.understated, or your own check like it, on every lineage update, and fail when a declared classification is lower than lineage proposes. Classify by what a table holds, never by who reads it, and fix such a finding by raising the classification, not by dropping the edge.

What this chapter decided

• Every pipeline run sends OpenLineage run events, START and then COMPLETE or FAIL, from the code that runs it (lib.lineage.run), with the output's schema and column lineage. Datasets are named by catalog identifiers under the placeholder namespace iceberg://platform. Examples fix event times and derive run IDs from the nominal run, and compare what events say, not their bytes.
• Column lineage is sqlglot's value lineage, typed as OpenLineage subtypes, plus INDIRECT inputs: the join, filter, grouping and sort columns of every SELECT, CTEs included, in the facet's dataset list.
• Lineage that can't be read correctly is an error, never empty or partial lineage. lib.lineage reads only CREATE ... AS and INSERT ... SELECT, and refuses every other statement, MERGE included, whatever its USING names. Empty lineage, an unplaceable source or an unknown column stops the job before it runs. An INSERT's columns come from the table's schema in the catalog, matched by position, and a count that differs is an error.
• Impact analysis walks the graph built from events, INDIRECT edges included, and reports columns whose values can change and tables whose rows can change.
• Classifications propagate as the stricter of all inputs', DIRECT and INDIRECT, as set in governance/lineage.toml, so a table chosen by a personal filter is proposed personal. Lineage never lowers a declared classification, and masking never declassifies. The proposals for the two unclassified tables, full_name, email and body personal and every other column internal, go to their owners for review before they enter governance/metadata.toml.
• Three checks guard the lineage itself: jobs that ran with no COMPLETE event, tables no job's lineage accounts for, and declared classifications lower than lineage proposes.