OpenLineage: Automatic SQL datasource metadata and table lineage

[Dev-iL]

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.
• New optional operation="read"|"write" lets callers say what the string is. The pandas reader and writer pass it, so a written table name such as daily revenue is always the table written.
• query/table_name keep the 1.0.0 filing ("SELECT" in text) for writes and for the original two-argument call. The only change is on reads: a statement starting (after comments) with select/with in any case is now filed as a query.
• For a raw sqlite3 connection, source["attached"] maps each attached database to its file. It is None where that can't be known without opening a connection.

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

• PandasSqlReader/PandasSqlWriter now pass their db_connection, their operation, 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(..., sql_dataset_identity="legacy" | "datasource"). The default, "legacy", keeps today's SQL dataset names exactly (job namespace + bare table name) and needs no openlineage-sql. Leaving the option unset emits one FutureWarning per adapter pointing to the migration docs; the default flips only in a future major release. create_input_dataset/create_output_dataset keep their existing SQL handling and return shapes.
• With "datasource", the adapter routes SQL loader/saver nodes through sql_datasets(), 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: the sql_dataset_identity option, the datasource naming table, a "Migrating to the datasource identity" section (what changes for a lineage backend, which datasets stop appearing, and step-by-step migration), and the reusable sql_datasets() entry point.
• examples/openlineage: rewritten end-to-end around the revenue-reporting use case above, opting into sql_dataset_identity="datasource"; runs locally with no external service via OpenLineageClient's FileTransport.

Compatibility

• OpenLineage dataset identity is unchanged by default. OpenLineageAdapter built as before emits the same SQL datasets as the previous release: same namespace, name, and facets. That is pinned in tests against main's adapter for query reads (including lower-case select/with), table reads, table writes (including names containing SELECT or spaces), and hand-built metadata. The datasource identity is opt-in via sql_dataset_identity="datasource". The one-time FutureWarning for the unset default never fails a node, even under -W error. The default flips only in a future major release; the adapter docs describe the migration.
• openlineage-sql is needed only for the datasource identity. The default path doesn't import it, so locked environments and Windows keep working unchanged.
• get_sql_metadata(query_or_table, results) still works and returns the same keys with the same values. New keys (schema, operation, source, notes) are additive, and __version__ is 1.1.0.
• PandasSqlReader/PandasSqlWriter metadata keeps the 1.0.0 query/table_name filing, with one exception: a read starting with select/with in any case is now filed as a query (previously a lower-case one was filed as a table name). Legacy-mode OpenLineage names re-derive the 1.0.0 filing, so they are unaffected.
• The reader's db_connection annotation is widened to Any. It previously rejected SQLAlchemy Engine inputs at graph validation, so this is purely additive.

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. The review-driven revisions (staged OpenLineage identity, explicit SQL operation, SQLite attached databases) were made with Claude Code (Claude Opus 5.5).

[skrawcz]

AI-generated review: the two inline comments below were produced with AI assistance and should be evaluated accordingly. They are based on local reproduction against commit a0e4f3f7.

[skrawcz]

AI-generated review comment (validated with a local reproduction): Treating every string containing whitespace as a statement misclassifies valid table names. For example, PandasSqlWriter(table_name="daily revenue", ...) successfully writes that table, but this helper records query="daily revenue", table_name=None, and operation="read"; the OpenLineage parser then reports an error and no output dataset is emitted. Quoted or otherwise unusual identifiers have the same ambiguity. Could the reader/writer pass the operation explicitly (and ideally whether this argument is a table target), rather than inferring both concepts from the string/result shape? Please add a regression test for a table name containing whitespace.

[Dev-iL]

Good catch. Readers and writers now pass the operation explicitly: get_sql_metadata(..., operation="read"|"write"). PandasSqlWriter(table_name="daily revenue") records operation="write", and in datasource mode (with openlineage-sql installed) emits one output dataset for daily revenue. The whitespace heuristic is gone.

For compatibility, query/table_name keep the 1.0.0 filing, so a written name like USER_SELECTIONS is still filed under query. operation is what marks it as the target table, and sql_datasets() names it as a table. The only new filing is on reads: a statement starting with select/with in any case, after comments, counts as a query. A read string that isn't a plain name, like (select ...), is parsed rather than used as a dataset name. For writes, a plain identifier is the table. Anything else is parsed: a statement contributes the tables it writes, and a string that names no table is the table name. Without the parser (e.g. Windows), such a non-plain name is left out with a note rather than guessed. There are regression tests for daily revenue and SELECTED_ROWS end to end, plus the classification matrix.

Changes and this reply were prepared with AI assistance (Claude).

[skrawcz]

AI-generated review comment (validated with a local reproduction): For SQLite, schema can name an attached database, but the namespace here always uses source["database"], which is the main file. A query against reporting.orders stored in attached.db is therefore emitted as sqlite://.../main.db + reporting.orders, attributing the dataset to the wrong physical datasource. Since the docs explicitly advertise attached-database naming, could metadata inspection retain the PRAGMA database_list mapping and resolve the schema to the attached file's namespace? Please cover this with a raw SQLite connection that attaches a second file.

[Dev-iL]

Fixed. For a raw sqlite3 connection, metadata now keeps the PRAGMA database_list mapping as source["attached"] (schema name → absolute file). reporting.orders in an attached file is named sqlite://<attached file> + orders, and main/unqualified tables keep the main file's namespace. Where the mapping can't be known without opening a connection (URL strings, SQLAlchemy engines/connections), a schema-qualified table is left out with a note rather than attributed to the main file. temp is also left out, and an in-memory main database still resolves its attached files.

Covered by a raw-connection test that attaches a second file, both at the sql_datasets() level and end to end through the adapter.

Changes and this reply were prepared with AI assistance (Claude).

[skrawcz]

AI-generated compatibility review comment: One additional concern for existing OpenLineageAdapter users: this changes SQL dataset identity from the adapter's job namespace plus a bare table name to a datasource namespace plus a qualified name. Existing lineage history will therefore not connect automatically, and downstream queries, ownership, tags, alerts, or policies keyed to the old identity may stop matching. Locked environments must also add openlineage-sql; on Windows, or when a source cannot be fully identified, SQL datasets may now be omitted rather than retaining the previous identity.

Could we provide a staged compatibility path instead of changing the default immediately? For example, an adapter option such as sql_dataset_identity="legacy" | "datasource" could default to "legacy" for the current release, emit a one-time deprecation warning with the exact migration steps, and switch the default only in a major release. Users could then opt into the new identity deliberately and migrate backend metadata/history first. If retaining both modes is not feasible, the upgrade should at least emit a prominent one-time warning whenever the new identity is used, including the old and new namespace/name and a link to the migration documentation.

I would avoid turning lineage conversion or identity migration into a hard execution error: the SQL operation has already succeeded, and failing the Hamilton node would violate the PR's stated non-interference guarantee. A targeted warning (with strict/error behavior available only as an explicit opt-in, if desired) seems safer. Please also add upgrade tests covering the legacy mode/default and the opt-in datasource mode.

[Dev-iL]

Thanks. The identity change is now staged instead of immediate, as you suggested:

• OpenLineageAdapter(..., sql_dataset_identity="legacy" | "datasource"). The default is "legacy": SQL datasets keep the job namespace and bare table name. I pinned this against main's adapter for query reads (including lower-case select/with), table reads, and table writes (including names containing SELECT and names with spaces). The emitted events are identical. Legacy mode doesn't need openlineage-sql.
• Leaving the option unset emits one FutureWarning per adapter. It names the option, says the default flips in a future major release, and links the new migration section in OpenLineageAdapter.rst. That section covers installing openlineage-sql and doing a dry run with FileTransport. It also says to move ownership, tags, alerts and policies to the new names, and then switch. Passing either value explicitly silences the warning.
• Lineage stays non-fatal in both modes. Conversion errors, and even the warning escalated with -W error, are logged, and the node still succeeds. There's no strict mode.
• create_input_dataset / create_output_dataset keep main's SQL handling and return shapes.
• Upgrade tests cover the default, explicit legacy, explicit datasource, invalid values (ValueError at construction), and warning-once / no-warning behaviour.

Changes and this reply were prepared with AI assistance (Claude).

[skrawcz]

AI-generated review question (validated with a local reproduction): Is this the intended behavior in this case? A PandasSqlReader("daily revenue", engine) successfully reads the SQLite table, and its metadata says operation="read", query=None, and table_name="daily revenue". This branch then reparses that known table name as SQL; openlineage-sql rejects it, so datasource mode emits no input dataset. Since the producer has already classified it as a table read, should this instead emit the named table directly, including names with whitespace or quoting? If so, test_read_by_a_non_plain_table_name_reports_no_statement could become a positive identity regression and the corresponding documentation could be updated.

[Dev-iL]

You're right that this read loses its lineage, and nothing documented that. I've fixed it, though not exactly as suggested. Changes:

• test_read_by_a_non_plain_table_name_reports_no_statement is now a positive identity test through the real PandasSqlReader.
• New regression tests on SQLite and Postgres check that each read query above is never named as a table, and that without the parser the read is still left out.
• The OpenLineageAdapter docs and ADR 2609-03 are updated to match.

Detailed reasoning

On a read, table_name doesn't mean the table was already identified. It only means get_sql_metadata found no SELECT, select or with; pandas decides table versus query at run time via has_table. Read queries such as (select * from t), values (1), explain select * from t and pragma table_info(t) all land in table_name too. Emitting that value directly would name datasets after SQL text.

What the fix does instead: a read string in table_name that isn't a plain identifier is still parsed. It is emitted as the table read only when openlineage-sql reports that no statement starts at its very first character (e.g. daily revenue).

• Postgres' TABLE t is the one row-returning statement the parser also rejects at its first word, so a string starting with table is never taken as a name.
• If the parse fails, openlineage-sql is missing, or the error message is unexpected, the read is still left out with a note.

On quoting: PandasSqlReader('"daily revenue"', engine) and '[daily revenue]' fail inside pd.read_sql with a DatabaseError, because has_table misses the quoted string and running it as SQL fails. No metadata is produced, so there's nothing for lineage to emit. With a raw sqlite3 connection, even daily revenue fails, since pandas always runs the string as a query there.

[Dev-iL]

Also verified that openlineage-sql integration works on Windows. It requires building a wheel using a Rust toolchain. Here's the wheel I built: openlineage_sql-1.53.0-cp310-abi3-win_amd64.zip (change the suffix to .whl then pip install).

[skrawcz]

LGTM just remove the write ups committed.

[skrawcz]

NVM -- sure design decisions.