Lineage First: How CoreModels Maps an Airflow Estate into the Governed Graph

Airflow is not a schema estate. A warehouse table has columns with types; a dbt model has a contract; an Avro subject has fields. A DAG has none of those — it has tasks, a schedule, an owner, and (with data-aware scheduling) declared relationships to the data it produces and consumes. So when we built the Apache Airflow connector for CoreModels, the design question was not "how do we pretend DAGs are tables?" but "what is orchestration's actual governance value?" The answer we committed to: the DAG-to-asset dependency graph, and pipeline hygiene. This connector is lineage-first, and every mapping decision below follows from that.

This article is the deep dive: exactly how dags, tasks, and datasets artifacts become governed graph structure, what rides the vendor-metadata mixin, how the audit rules compute, and where the honest edges of the mapping are.

The estate model, bent honestly

Every CoreModels vendor connector parses into one neutral estate model: datasets with fields and checks, plus lineage edges, with vendor-specific detail riding metadata bags — never new top-level concepts. Airflow bends that model in three declared ways:

• A DAG becomes a dataset with materialization dag. Its identity is the dag_id; its description comes from the DAG's description; its file location (fileloc) is recorded as the physical name; its tags are kept as tags.
• A data-aware-scheduling asset ("Asset" in Airflow 3, "Dataset" in Airflow 2's API) also becomes a dataset, with materialization asset. Its identity is its URI; the display name is the last path segment of that URI (so s3://lake/orders.parquet reads as orders.parquet); the full URI is kept as the physical name.
• A task becomes a field on its DAG's dataset — the task id is the field name, and the operator class is recorded as the field's native type, resolved as class_ref.class_name, falling back to operator_name, falling back to the literal task. A task-level owner rides along as metadata.

The materialization marker is what keeps the two dataset flavors distinct downstream: the hygiene rules and the estate facts recorded at import both discriminate on dag versus asset.

Everything that is orchestration-specific — and therefore homeless in a schema-shaped model — rides airflow.* metadata keys on the dataset:

Key	Source	Notes
airflow.paused	is_paused	"true" / "false"; feeds the paused-producer rule
airflow.owners	owners array	comma-joined; feeds the dag-no-owner rule
airflow.schedule	schedule_interval or timetable_summary	for object-shaped schedules, the value field or the __type marker (e.g. a cron expression or a timetable type name)
airflow.owner (on a task)	the task's owner	per-task accountability

That schedule extraction is worth a sentence: Airflow's REST API has represented schedules as plain strings, typed objects ({"__type": "CronExpression", "value": "0 6 * * *"}), and summary fields across versions. The connector accepts all three shapes and records whichever is present — version drift in the source API becomes a non-event.

Into the graph: two write paths

An import writes through two deliberate paths. Schema shape — the DAG and asset Types, the task Elements — flows through the same hardened writer every CoreModels import uses, which is what makes the estate exportable through every schema format the platform speaks. Estate facts that a schema IR deliberately cannot hold — lineage, provenance, state — are written natively using the graph's own extensibility. Concretely, after an Airflow import your project contains:

• A Type node per DAG and per asset, with the vendor identity written as a queryable mapsTo mixin value (standard: the vendor key airflow, URI: the dag_id or asset URI). This identity round-trips: it is how a later audit matches artifacts back to governed nodes.
• Elements per task on the DAG's Type, each carrying the operator class as its native type.
• A vendor-metadata mixin value per node — materialization, physical name, description, tags and the airflow.* bag above on each Type; the operator class as native type, plus its own tags and meta bag, on each Element — refreshed on re-import, because it is estate bookkeeping, not governed meaning. That distinction is the import posture in one line: governed nodes are never mutated by import; metadata bookkeeping is.
• Lineage relations in the Depends On relation group — the same relation group the dbt and warehouse connectors write. This is the payoff of lineage-first design: an Airflow asset, a dbt model, and a Snowflake table can participate in one dependency graph, and cross-estate reconciliation can then assert that two vendors' datasets are the same physical thing.
• A state node recording the last import: versions, timestamps, artifact fingerprint (a SHA-256-derived 16-hex-character content hash), counts, and estate facts — the connector records four: total DAGs, paused DAGs, assets, and tasks.

The lineage mechanics

The datasets artifact is where the graph earns its keep. For each asset, Airflow reports producing_tasks and consuming_dags, and the connector turns them into directed dependency edges with a precise orientation:

• producing_tasks → the asset depends on the producing DAG (the asset's freshness is downstream of the producer running).
• consuming_dags → the consuming DAG depends on the asset (the consumer's correctness is downstream of the asset).

Chain those through a shared asset and you get cross-DAG lineage without either DAG referencing the other: daily_revenue → s3://lake/orders.parquet → load_orders. Duplicate producer/consumer declarations are deduplicated, and an edge that references a DAG absent from the dags artifact is kept in the parsed snapshot but flagged external — rather than dropping the fact, the parse records that your deployment depends on orchestration it did not describe. Import does not write those edges into the graph (there is no node to attach them to); they surface in the response as lineageEdgesSkipped.

The audit rules, and how they compute

Beyond the shared coverage codes (dataset-unmapped, field-unmapped) and drift codes (dataset-removed, field-removed, field-type-drift, contract-drift, enum changes), the Airflow connector contributes four conformance rules. They are cheap to state and precise in implementation:

• dag-no-description (Info) — the DAG's description is empty: "DAG has no description — pipelines without stated intent resist governance."
• dag-no-owner (Warning) — airflow.owners is empty or exactly the default airflow. A default owner is treated as no owner: nobody is accountable.
• asset-unproduced (Warning) — an asset has at least one consuming edge but appears in no producing edge in this deployment: an ungoverned upstream dependency whose freshness orchestration cannot see.
• paused-producer (Warning) — the sharpest rule. For each produced asset, take its producer set and consumer set from the lineage edges; if every producer carries airflow.paused: "true" while at least one consumer does not, the consumers will silently run on stale data. The finding names both sides:

{ "section": "Conformance", "severity": "Warning", "code": "paused-producer",
  "subject": "s3://lake/orders.parquet",
  "message": "Every producer of this asset is paused while active DAGs still consume it — consumers will silently run on stale data.",
  "detail": "producers: load_orders; active consumers: daily_revenue" }

Note what made that rule possible: pausing state rode the metadata bag, direction rode the lineage edges, and the rule is a pure function over the parsed snapshot. No live Airflow connection was consulted — which also means the rule works identically in CI, in a re-audit against a stored snapshot, and over MCP.

Type mapping: one honest approximation

Warehouse connectors map native types (NUMBER(38,0), VARCHAR) onto schema primitives. Airflow's "native types" are operator class names — PythonOperator is not a data type. The connector therefore maps every native type to String and flags every single mapping as an approximation. You will see this surfaced rather than hidden: String is a carrier for the operator name, not a claim that a task "is a string." We consider a visibly approximate mapping more useful than a fabricated type system.

Lossiness, limits, and gotchas

The connector parses open-world and reserves hard failure for genuinely unusable input:

• Only dags can fail the parse. Missing, invalid JSON, an unrecognized shape (it accepts {"dags": [...]} or a bare array), or zero DAGs — each returns an explicit error naming the problem.
• tasks and datasets never fail an import. Invalid JSON in either, or a datasets payload that is not a recognizable asset list, produces a lossiness record stating the artifact was ignored — the import proceeds with what it has. Check the lossiness array; a "successful" import with an ignored datasets artifact has no cross-DAG lineage.
• The tasks artifact accepts two shapes: the documented {"<dag_id>": <tasks response>} object, or an array of objects each carrying dag_id and tasks. Tasks for a DAG not present in dags are skipped; duplicate task ids within a DAG are not double-added.
• The snapshot cap. Import persists the parsed snapshot for later re-audits. When the encoded snapshot exceeds the storage cap (~1.5 MB), import reports snapshotStored: false with a lossiness record; fresh-artifact audits still work, but re-audit has no stored snapshot to run against.
• No live sync, by design. CoreModels never holds Airflow credentials; live connection is a declared-but-deferred capability. You upload API responses you produced yourself.
• No Generate, by design. Capabilities are Import and Audit; a generate call is refused by the capability gate with an explicit error — "Connector 'airflow' does not support generation." — before any emission code runs, because orchestration code is not derivable from schema. This connector is intentionally the proof that a capability flag can say no cleanly.

The shape of the whole design is visible in that list: everything the connector cannot represent is declared — as a capability flag, a lossiness record, or a String-approximation flag — rather than silently absorbed. For the practical walkthrough that puts this model to work, see the Apache Airflow quickstart in the CoreModels integration docs.