lineage

etl_craft.services.lineage ¶

Column lineage: which columns each SQL task's target is made from, traced across tasks.

Every active SQL task's SELECT (inline or from its file, with the pipeline-id tokens replaced) is parsed in the warehouse's SQL dialect. Each column of the SELECT is traced back through CTEs, subqueries and joins to the source columns it is made from, and recorded against the task's TARGET_OBJECT:

a column copied as it is: copy;
a column computed from others, such as o.amount * c.rate: the expression, one edge per source column;
a column made from none, such as COUNT(*) or a constant: the expression, with no source.

Tables are named as the SQL names them, lower case, with the active warehouse database left off, so a task's target and a later task's source meet. Because one task's target is the next one's source, the edges join into one graph, which LineageGraph walks upstream to the first sources and downstream to the last consumers, across pipelines.

Lineage is stored per task in AUD_COLUMN_LINEAGE with a hash of the SELECT, target and dialect, and worked out again only when one of them changes. SQL that cannot be traced, such as SELECT * over a table whose columns are not known, is reported with the reason.

COPY `module-attribute` ¶

COPY = 'copy'

The transformation of a column copied from its source as it is.

SQLGLOT_DIALECTS `module-attribute` ¶

SQLGLOT_DIALECTS = {'postgres': 'postgres', 'duckdb': 'duckdb', 'duckdb_iceberg': 'duckdb', 'trino_iceberg': 'trino', 'databricks': 'databricks', 'databricks_iceberg': 'databricks', 'snowflake': 'snowflake', 'snowflake_iceberg': 'snowflake'}

The sqlglot dialect each warehouse dialect's SQL is parsed in.

Edge `dataclass` ¶

A target column and one column it is made from, written by one task.

TaskLineage `dataclass` ¶

One SQL task's lineage, or why it has none; cached when it was stored already.

sources are the tables its SELECT reads, joins and filters included, known whenever the SELECT parses, even when its columns cannot be traced.

LineageGraph `dataclass` ¶

Every task's edges, joined where one task's target is another's source.

__post_init__ ¶

__post_init__() -> None

Index the edges by the column they lead into and out of.

of `classmethod` ¶

of(lineages: Iterable[TaskLineage]) -> LineageGraph

Join the edges of every task's lineage.

columns ¶

columns(table: str) -> list[str]

Return the columns of table any edge mentions, by name.

upstream ¶

upstream(table: str, column: str, depth: int | None = None) -> list[tuple[int, Edge]]

Return the edges into table.column and, recursively, into their sources.

downstream ¶

downstream(table: str, column: str, depth: int | None = None) -> list[tuple[int, Edge]]

Return the edges out of table.column and, recursively, out of their targets.

upstream_tables ¶

upstream_tables(table: str) -> list[tuple[int, str, str]]

Return (level, table, task) for every table table is made from, recursively.

downstream_tables ¶

downstream_tables(table: str) -> list[tuple[int, str, str]]

Return (level, table, task) for every table made from table, recursively.

table_name ¶

table_name(parts: Iterable[str], catalog: str | None) -> str

Return a table's name as lineage compares it: lower case, without the active database.

extract ¶

extract(select_sql: str, target_object: str, *, dialect: str | None, catalog: str | None = None) -> list[Edge]

Trace each column of select_sql to its sources; MetadataError when it cannot.

output_columns ¶

output_columns(select_sql: str, *, dialect: str | None) -> list[str] | None

Return the names of the columns select_sql returns, or None when they are unknown.

They are unknown when the SELECT does not parse, or returns * or a column with no name.

referenced_tables ¶

referenced_tables(select_sql: str, *, dialect: str | None, catalog: str | None = None) -> tuple[str, ...]

Return every table select_sql reads, CTEs left out; empty when it does not parse.

lineage_key ¶

lineage_key(select_sql: str, target_object: str, dialect: str | None, catalog: str | None = None) -> str

Hash what a task's lineage is worked out from; any change means working it out again.

collect ¶

collect(engine: Engine, config: ConnectorConfig, *, refresh: bool = False, record: bool = True) -> list[TaskLineage]

Return the lineage of every active SQL task, working out and storing what changed.

refresh works everything out again. Without record nothing is stored.

dialect_and_catalog ¶

dialect_and_catalog(config: ConnectorConfig) -> tuple[str | None, str | None]

Return the sqlglot dialect and the catalog of the active warehouse; None without one.