Conversation
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Related: #1564
Problem
A team uses Hamilton's built-in SQL loaders to read
ordersandcustomers, 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:@load_from.sql(query_or_table=...)) records the SQL text but no table name at all.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.ordersandsales.public.customerson a source PostgreSQL server, computes daily revenue totals in Python, and writesanalytics.reporting.daily_revenueon a separate warehouse server:No custom loader, no manually-supplied lineage mapping — just the normal
@load_from.sql/@save_to.sqldecorators and the connections the author already has. With this change, Hamilton's OpenLineage adapter reports:examples/openlineageis 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-onlydb_connectionandschemaparams. Thesql_metadatadict it returns is now version1.1.0: on top of the existingrows/query/table_name/timestamp, it addsschema(the writer's explicit schema, if any),operation("read"/"write"/Nonefor legacy calls),source(a plain dict ofdialect/host/port/database/default_schema— scalars only, never a connection object or credentials), andnotesexplaining whysourceisNonewhen it is.get_sql_source()inspects SQLAlchemy URL strings,Engines,Connections, and rawsqlite3.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.Pandas SQL I/O (
hamilton/plugins/pandas_extensions.py)PandasSqlReader/PandasSqlWriternow pass theirdb_connection(and, for the writer, its explicitschema) into the new metadata call. The reader'sdb_connectiontype annotation is widened fromstr | sqlite3.ConnectiontoAny, since it already accepted SQLAlchemyEngine/Connectionobjects in practice.OpenLineage integration (
hamilton/plugins/h_openlineage.py)sql_datasets(sql_metadata, operation=None) -> SqlDatasetsconverts Hamilton's SQL metadata into OpenLineage datasets, using theopenlineage-sqlparser 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), andsqlite://{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.OpenLineageAdapternow routes SQL loader/saver nodes through this function, so SQL dataset identity comes from the datasource rather than the job namespace — a later read ofdaily_revenueresolves 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, missingopenlineage-sqlinstall) 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
openlineageextra gainsopenlineage-sql, markedsys_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.ymladds a disposablepostgres:16service and a "Test openlineage" step to the existing matrix.Tests
tests/io/test_utils.pyandtests/plugins/test_pandas_extensions.pyextended for the new metadata fields, connection forms (URL string,Engine,Connection, rawsqlite3), and credential-leak negatives.tests/plugins/test_h_openlineage.py: dataset naming for SQLite and PostgreSQL (the latter env-gated viaHAMILTON_TEST_POSTGRES_URL, skipped otherwise), schema precedence (SQL-qualified → writerschema=→ connection default), an end-to-end revenue-reporting run with realFileTransport-captured OpenLineage events, and failure-injection tests proving metadata/parser errors never fail a node that already succeeded.tests/conftest.pygains apostgres_schemafixture that creates and drops its own throwaway schema on whatever serverHAMILTON_TEST_POSTGRES_URLnames.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 reusablesql_datasets()entry point.examples/openlineage: rewritten end-to-end around the revenue-reporting use case above; runs locally with no external service viaOpenLineageClient'sFileTransport.Compatibility
get_sql_metadata(query_or_table, results)call still works and returns the same keys with the same meanings.SELECT" to "contains whitespace", so a lower-caseselect ...statement is now correctly recognized as a query (documented as a behavior change).OpenLineageAdapterwere 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 inOpenLineageAdapter.rstrather than presented as a silent compatibility change.Checklist
AI Disclosure
Codex (GPT-6-Astra) was used to author this PR.