Skip to content

Automatic SQL datasource metadata and OpenLineage table lineage - #1720

Open
Dev-iL wants to merge 1 commit into
apache:mainfrom
SummitSG-LLC:2609/richer-sql-metadata
Open

Dev-iL wants to merge 1 commit into
apache:mainfrom
SummitSG-LLC:2609/richer-sql-metadata

Conversation

@Dev-iL

@Dev-iL Dev-iL commented Sep 21, 2026

Copy link
Copy Markdown
Collaborator

Related: #1564

Problem

A team uses Hamilton's built-in SQL loaders to read orders and customers, transforms the data in Python, and uses a built-in SQL saver to write daily revenue into a reporting database. The data operation succeeds, but its metadata does not identify the database/server or reliably qualify the tables involved:

  • A query read (@load_from.sql(query_or_table=...)) records the SQL text but no table name at all.
  • A table write records a bare table name, with no database, server, or schema.
  • The OpenLineage adapter files SQL datasets under the job namespace rather than the datasource, so a report written by one job and read by another job cannot be connected, and two tables with the same name in different databases are indistinguishable.

Today, answering "which source tables fed this report?" requires writing a custom loader, duplicating connection details, or teaching each downstream integration (e.g. an OpenLineage consumer, an orchestrator) to inspect Hamilton internals.

Use case this PR is built around

A Hamilton graph reads a query joining sales.public.orders and sales.public.customers on a source PostgreSQL server, computes daily revenue totals in Python, and writes analytics.reporting.daily_revenue on a separate warehouse server:

@load_from.sql(query_or_table=value(REVENUE_QUERY), db_connection=source("sales_db"))
def order_lines(df: pd.DataFrame) -> pd.DataFrame:
    return df

def daily_revenue(order_lines: pd.DataFrame) -> pd.DataFrame:
    return order_lines.groupby(["order_date", "country"], as_index=False)["amount"].sum()

@save_to.sql(table_name=value("daily_revenue"), db_connection=source("warehouse_db"), ...)
def revenue_report(daily_revenue: pd.DataFrame) -> pd.DataFrame:
    return daily_revenue

No custom loader, no manually-supplied lineage mapping — just the normal @load_from.sql/@save_to.sql decorators and the connections the author already has. With this change, Hamilton's OpenLineage adapter reports:

Inputs
  postgres://source.example:5432     sales.public.orders
  postgres://source.example:5432     sales.public.customers
Output
  postgres://warehouse.example:5432  analytics.reporting.daily_revenue

examples/openlineage is rewritten around exactly this story (on two local SQLite files, so it runs with no external server) — see its README for the before/after metadata and a runnable walkthrough.

What changed

Core metadata (hamilton/io/utils.py)

  • get_sql_metadata() gains optional keyword-only db_connection and schema params. The sql_metadata dict it returns is now version 1.1.0: on top of the existing rows/query/table_name/timestamp, it adds schema (the writer's explicit schema, if any), operation ("read"/"write"/None for legacy calls), source (a plain dict of dialect/host/port/database/default_schema — scalars only, never a connection object or credentials), and notes explaining why source is None when it is.
  • New get_sql_source() inspects SQLAlchemy URL strings, Engines, Connections, and raw sqlite3.Connections, entirely read-only: it never opens a new connection, starts a transaction, or changes session state. In-memory SQLite databases (:memory:, sqlite://, sqlite:///:memory:) are deliberately left unidentified rather than given a shared or fabricated identity.
  • The original two-argument call form keeps its keys and row-count semantics unchanged.

Pandas SQL I/O (hamilton/plugins/pandas_extensions.py)

  • PandasSqlReader/PandasSqlWriter now pass their db_connection (and, for the writer, its explicit schema) into the new metadata call. The reader's db_connection type annotation is widened from str | sqlite3.Connection to Any, since it already accepted SQLAlchemy Engine/Connection objects in practice.

OpenLineage integration (hamilton/plugins/h_openlineage.py)

  • New reusable, side-effect-free sql_datasets(sql_metadata, operation=None) -> SqlDatasets converts Hamilton's SQL metadata into OpenLineage datasets, using the openlineage-sql parser to resolve every physical table a query reads or writes (aliases and CTEs are excluded). Datasets are named per the OpenLineage naming convention: postgres://{host}:{port} + {database}.{schema}.{table} for PostgreSQL (folding unquoted identifiers, as the server does), and sqlite://{absolute path} + {table} for SQLite. It emits no events and opens no connections, so it's a clean boundary for another integration (e.g. a future Airflow provider) to reuse without going through Hamilton's own event-emitting adapter.
  • OpenLineageAdapter now routes SQL loader/saver nodes through this function, so SQL dataset identity comes from the datasource rather than the job namespace — a later read of daily_revenue resolves to the same identity it was written under, regardless of job namespace. Anything that can't be fully identified (unknown source, unsupported dialect, missing schema, parser error, missing openlineage-sql install) is logged and left out rather than guessed. A conversion failure logs only the exception type (never a message that might quote connection details) and the run event is still emitted without datasets — SQL I/O that already succeeded is never turned into a failed node by a lineage problem.

Packaging / CI

  • The openlineage extra gains openlineage-sql, marked sys_platform != 'win32' since it publishes no Windows wheel; the base extra (and the OpenLineage client) still installs cleanly there, with SQL table resolution unavailable.
  • .github/workflows/hamilton-main.yml adds a disposable postgres:16 service and a "Test openlineage" step to the existing matrix.

Tests

  • tests/io/test_utils.py and tests/plugins/test_pandas_extensions.py extended for the new metadata fields, connection forms (URL string, Engine, Connection, raw sqlite3), and credential-leak negatives.
  • New tests/plugins/test_h_openlineage.py: dataset naming for SQLite and PostgreSQL (the latter env-gated via HAMILTON_TEST_POSTGRES_URL, skipped otherwise), schema precedence (SQL-qualified → writer schema= → connection default), an end-to-end revenue-reporting run with real FileTransport-captured OpenLineage events, and failure-injection tests proving metadata/parser errors never fail a node that already succeeded.
  • tests/conftest.py gains a postgres_schema fixture that creates and drops its own throwaway schema on whatever server
    HAMILTON_TEST_POSTGRES_URL names.

Docs

  • docs/concepts/materialization.rst: new "SQL metadata and lineage" section — field reference, supported connection forms, schema precedence, and what happens when a datasource can't be identified.
  • docs/reference/lifecycle-hooks/OpenLineageAdapter.rst: dataset naming table, the identity change from the old job-scoped SQL datasets (existing lineage history won't automatically connect to the new datasource-scoped identities — this is called out explicitly, there's no automatic backend-history migration), and the reusable sql_datasets() entry point.
  • examples/openlineage: rewritten end-to-end around the revenue-reporting use case above; runs locally with no external service via OpenLineageClient's FileTransport.

Compatibility

  • The legacy two-argument get_sql_metadata(query_or_table, results) call still works and returns the same keys with the same meanings.
  • Statement-vs-table-name detection changed from "contains the upper-case substring SELECT" to "contains whitespace", so a lower-case select ... statement is now correctly recognized as a query (documented as a behavior change).
  • Dataset identity change: SQL datasets emitted by OpenLineageAdapter were previously scoped to the job namespace with a bare table name; they are now scoped to the datasource namespace with a qualified name. A lineage backend will show these as new, disconnected datasets relative to history recorded under the old identity — this is called out in OpenLineageAdapter.rst rather than presented as a silent compatibility change.

Checklist

  • PR has an informative and human-readable title (this will be pulled into the release notes)
  • Changes are limited to a single goal (no scope creep)
  • Code passed the pre-commit check & code is left cleaner/nicer than when first encountered.
  • Any change in functionality is tested
  • New functions are documented (with a description, list of inputs, and expected output)
  • Placeholder code is flagged / future TODOs are captured in comments
  • Project documentation has been updated if adding/changing functionality.

AI Disclosure

Codex (GPT-6-Astra) was used to author this PR.

Hamilton's built-in SQL loaders/savers (`@load_from.sql`, `@save_to.sql`,
`from_.sql`/`to.sql`, and the underlying `PandasSqlReader`/`PandasSqlWriter`)
now capture enough datasource context on their own for table-level lineage,
without any custom loader or hand-maintained mapping.

Motivating use case: a graph reads a query joining `orders` and `customers`
from a sales database, aggregates daily revenue in Python, and writes
`daily_revenue` to a separate warehouse database. Previously the emitted SQL
metadata had no table name for a query read, and the OpenLineage adapter
filed writes under the job namespace with a bare table name, so a report
written by one job and read by another could not be connected, and same-named
tables in different databases were indistinguishable. Users had to write a
custom loader and duplicate connection details to get real lineage.

Core (`hamilton/io/utils.py`):
- `get_sql_metadata()` gains optional keyword `db_connection`/`schema`
  params and a version-1.1.0 `sql_metadata` shape: `schema`, `operation`
  (read/write), `source` (dialect, host, port, database, default schema —
  scalars only, never credentials or live objects), and `notes` explaining
  an unresolved source. The legacy two-argument call keeps its keys and
  row-count semantics.
- New `get_sql_source()` inspects SQLAlchemy URL strings, Engines and
  Connections, and raw `sqlite3` connections, read-only: no new connection,
  no transaction, no session-setting query. In-memory SQLite is deliberately
  left unidentified rather than given a shared/fabricated identity.

Pandas SQL I/O (`hamilton/plugins/pandas_extensions.py`):
- `PandasSqlReader`/`PandasSqlWriter` pass their connection and (for the
  writer) explicit schema into the new metadata; the reader's connection
  type is widened to `Any` to match what it already accepted in practice
  (SQLAlchemy Engines/Connections, not just `str | sqlite3.Connection`).

OpenLineage integration (`hamilton/plugins/h_openlineage.py`):
- New reusable `sql_datasets(sql_metadata, operation=None) -> SqlDatasets`
  converts Hamilton's SQL metadata into OpenLineage datasets using the
  `openlineage-sql` parser: every physical table a query reads/writes is
  named per the OpenLineage naming convention (`postgres://host:port` +
  `database.schema.table` for PostgreSQL, `sqlite://path` + `table` for
  SQLite), aliases and CTEs excluded, quoting/case-folding respected.
  Emits nothing, opens nothing, so a future Airflow provider (or anything
  else) can call it directly.
- `OpenLineageAdapter` routes SQL nodes through it, so dataset identity
  comes from the datasource rather than the job namespace — a read of a
  report now resolves to the same identity it was written under. A
  conversion failure is logged (exception type only, never a message that
  might quote a connection string) and the run event is still emitted
  without datasets; the underlying SQL I/O is never affected.

Packaging/CI:
- `openlineage` extra gains `openlineage-sql` (no Windows wheel; marked
  `sys_platform != 'win32'` so the base extra still installs there).
- `.github/workflows/hamilton-main.yml` adds a disposable `postgres:16`
  service and a "Test openlineage" step across the Python matrix.

Tests: `tests/io/test_utils.py` and `tests/plugins/test_pandas_extensions.py`
extended; new `tests/plugins/test_h_openlineage.py` covers dataset naming
(SQLite and PostgreSQL, env-gated via `HAMILTON_TEST_POSTGRES_URL`), schema
precedence, credential-leak negatives, and that metadata/parser failures
never turn a successful SQL read/write into a failed node. A disposable
per-test PostgreSQL schema fixture is added to `tests/conftest.py`.

Docs: `docs/concepts/materialization.rst` gains a "SQL metadata and
lineage" section (fields, supported connections, precedence, failure
behavior); `docs/reference/lifecycle-hooks/OpenLineageAdapter.rst` documents
dataset naming, the identity change from job-scoped SQL datasets, and the
reusable `sql_datasets()` entry point. The `examples/openlineage` example
is rewritten around the revenue-reporting story on two local SQLite files,
runnable with no external server.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant