Automatic SQL datasource metadata and OpenLineage table lineage

.github/workflows/hamilton-main.yml

@@ -51,6 +51,21 @@ jobs:
           - '3.14'
     env:
       HAMILTON_TELEMETRY_ENABLED: false
+    services:
+      # disposable server for the SQL lineage tests; they create and drop their own schemas
+      postgres:
+        image: postgres:16-alpine
+        env:
+          POSTGRES_USER: hamilton
+          POSTGRES_PASSWORD: hamilton
+          POSTGRES_DB: sales
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
 
     steps:
         - name: Install Graphviz on Linux
@@ -96,6 +111,14 @@ jobs:
             uv sync --group test --extra mcp
             uv run pytest tests/plugins/test_h_mcp.py
 
+        - name: Test openlineage
+          env:
+            HAMILTON_TEST_POSTGRES_URL: postgresql+psycopg2://hamilton:hamilton@localhost:5432/sales
+          run: |
+            uv sync --group test --extra openlineage
+            uv pip install psycopg2-binary
+            uv run pytest tests/io/test_utils.py tests/plugins/test_pandas_extensions.py tests/plugins/test_h_openlineage.py
+
         - name: Test pandas
           run: |
             uv sync --group test

docs/concepts/materialization.rst

@@ -146,3 +146,178 @@ Here are simplified snippets for saving and loading an XGBoost model to/from JSO
    +----------------------------------------------+-----------------------------------------------+
 
 To define your own DataSaver and DataLoader, the Apache Hamilton `XGBoost extension <https://github.com/apache/hamilton/blob/main/hamilton/plugins/xgboost_extensions.py>`_ provides a good example
+
+
+.. _sql-metadata-and-lineage:
+
+SQL metadata and lineage
+------------------------
+
+The built-in SQL materializers (``@load_from.sql``, ``@save_to.sql``, ``from_.sql``, ``to.sql`` and the
+``PandasSqlReader`` / ``PandasSqlWriter`` behind them) return ``sql_metadata`` describing what was read
+or written. Since version ``1.1.0`` of that metadata, they also record *where*: the database the
+connection points at. Lineage consumers such as the :doc:`OpenLineage adapter
+<../reference/lifecycle-hooks/OpenLineageAdapter>` use this to name the physical tables a query
+reads, without any custom loader or hand-maintained mapping.
+
+Take a graph that reads a query joining ``orders`` and ``customers`` from a sales database,
+aggregates daily revenue in Python, and writes ``daily_revenue`` to a reporting database:
+
+.. code-block:: python
+
+    @load_from.sql(query_or_table=value(REVENUE_QUERY), db_connection=source("sales_db"))
+    def order_lines(df: pd.DataFrame) -> pd.DataFrame:
+        return df
+
+    @save_to.sql(table_name=value("daily_revenue"), schema=value("reporting"),
+                 db_connection=source("warehouse_db"), output_name_="saved_revenue")
+    def daily_revenue(order_lines: pd.DataFrame) -> pd.DataFrame:
+        ...
+
+With metadata version ``1.0.0``, the loader's metadata was ``{"rows": 3, "query": "...", "table_name": None}``: no table, no
+server, no database. A join reported no inputs at all, and the saver's ``daily_revenue`` could not
+be told apart from a table of the same name elsewhere. With version ``1.1.0``, the same code, with the same
+connections, yields:
+
+.. code-block:: python
+
+    {"sql_metadata": {
+        "rows": 3,
+        "query": "WITH paid AS (...) SELECT ... FROM paid p JOIN customers c ON ...",
+        "table_name": None,
+        "schema": None,
+        "operation": "read",
+        "source": {"dialect": "postgresql", "host": "source.example", "port": 5432,
+                   "database": "sales", "default_schema": "public"},
+        "notes": "",
+        "timestamp": 1758470400.0,
+        "__version__": "1.1.0",
+    }}
+
+and the OpenLineage adapter, with ``sql_dataset_identity="datasource"`` (see
+:ref:`sql-dataset-identity-change`), reports ``sales.public.orders`` and ``sales.public.customers`` under
+``postgres://source.example:5432`` as inputs, and ``analytics.reporting.daily_revenue`` under the
+warehouse's namespace as output. The same module produces the same identities whether it runs from
+a script, a notebook or an orchestrator.
+
+Fields
+~~~~~~
+
+The ``sql_metadata`` entry holds the following keys:
+
+.. list-table::
+   :header-rows: 1
+   :widths: 18 82
+
+   * - Key
+     - Meaning
+   * - ``rows``
+     - Rows read (``len`` of the DataFrame) or the row count the write returned; ``None`` when unknown.
+   * - ``query``
+     - The statement executed, or ``None`` when a bare table name was read or written. As in 1.0.0,
+       a string containing the upper-case text ``SELECT`` anywhere is filed here, and anything else
+       under ``table_name``. Since 1.1.0 a read that starts (after comments) with ``select`` or
+       ``with`` in any case is also filed here; 1.0.0 recorded a lower-case ``select ...`` as a table
+       name. Writes and the two-argument form of :func:`~hamilton.io.utils.get_sql_metadata` keep the
+       1.0.0 rule. Lineage consumers should use ``operation``: the string a write names is the table
+       written, whichever of the two keys holds it (``"USER_SELECTIONS"`` is filed under ``query``).
+       A written name that is not a plain identifier (``daily revenue``, ``SELECT results``) is
+       parsed with ``openlineage-sql``: a statement that names tables contributes the tables it
+       writes (or is left out, with a note, if it writes none), and a string naming no table is the
+       table name. Without ``openlineage-sql``, such a name is left out of lineage with a note.
+   * - ``table_name``
+     - The bare table name read or written, or ``None`` for a statement.
+   * - ``schema``
+     - The schema explicitly passed to the writer (``PandasSqlWriter(schema=...)``), else ``None``.
+       *New in 1.1.0.*
+   * - ``operation``
+     - ``"read"`` or ``"write"``; ``None`` when the helper was called in its original two-argument
+       form and the direction is unknown. *New in 1.1.0.*
+   * - ``source``
+     - The datasource, or ``None`` when it could not be identified. It holds ``dialect``, the SQLAlchemy
+       backend name (``postgresql`` or ``sqlite``); ``host``; ``port``; ``database``, which is the
+       absolute file path for SQLite; and ``default_schema``, the schema unqualified names resolve
+       against. ``default_schema`` is set only when SQLAlchemy already established it on the connection,
+       and is ``None`` otherwise. SQLite sources also hold ``attached``, a mapping of attached database
+       name to absolute file path, or ``None`` when the attached databases cannot be known (see
+       below). *New in 1.1.0.*
+   * - ``notes``
+     - Why ``source`` is ``None``, for example ``"In-memory SQLite database has no stable identity"``
+       or ``"Unsupported connection type for SQL metadata: MyConn"``; empty otherwise. *New in 1.1.0.*
+   * - ``timestamp``
+     - When the metadata was produced (POSIX seconds).
+   * - ``__version__``
+     - ``"1.1.0"``. Added keys bump the minor version; a change to an existing key's meaning bumps the
+       major version.
+
+Supported connections
+~~~~~~~~~~~~~~~~~~~~~
+
+The connection object determines what ``source`` can hold:
+
+.. list-table::
+   :header-rows: 1
+   :widths: 40 60
+
+   * - Connection passed as ``db_connection``
+     - What ``source`` holds
+   * - SQLAlchemy ``Engine`` or ``Connection`` (PostgreSQL, SQLite)
+     - dialect, host, port, database from the URL; ``default_schema`` as SQLAlchemy determined it on
+       connect (PostgreSQL ``current_schema()``).
+   * - SQLAlchemy URL string (``"postgresql+psycopg2://..."``, ``"sqlite:///path.db"``)
+     - dialect, host, port, database from the parsed URL; ``default_schema`` is ``None`` because pandas
+       discards the temporary engine it built.
+   * - Standard-library ``sqlite3.Connection`` on a file
+     - ``dialect="sqlite"``, ``database`` = absolute file path, and ``attached`` = the file of every
+       database attached to it, read with ``PRAGMA database_list`` on that same connection (no
+       transaction is started). Only this connection form can see attached databases; for the others
+       ``attached`` is ``None``.
+   * - In-memory SQLite (``:memory:``, ``sqlite://``, ``sqlite:///:memory:``)
+     - ``None`` with a note: two unrelated in-memory databases must not share an identity. A raw
+       ``sqlite3`` in-memory connection with files attached keeps ``database=""`` and ``attached``, so
+       tables in the attached files can still be named.
+   * - Anything else (other DBAPI connections, mocks, and similar objects)
+     - ``None`` with a note naming the type. Data loading and saving are unaffected.
+
+Inspection never opens a connection, runs a write, commits, rolls back, or changes session settings,
+and never raises: a failure during inspection becomes a ``notes`` entry naming the exception type
+only. ``source`` holds scalars, never the connection object or the URL string, so it serializes
+with ``json.dumps`` and cannot carry a username, password or URL query parameter. The ``query``
+field is the SQL text you supplied, as before: keep secrets out of literals or strip them
+downstream.
+
+Schema precedence and unknown cases
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+A table referenced by a statement is qualified from, in order: the qualification written in the
+SQL (``sales.public.orders``), the ``schema`` argument given to the writer, then ``source["default_schema"]``.
+For SQLite, a schema names a database file rather than a namespace inside one: ``main`` (or no
+schema) is the connection's file, and an attached database's tables are named in that file's
+namespace, resolved through ``source["attached"]``. When the attached file cannot be known (a URL
+or SQLAlchemy connection), or the schema is ``temp``, the table is left out rather than
+attributed to the main file.
+The default schema is what SQLAlchemy read from the server, not an assumption that PostgreSQL uses
+``public``; when the connection's ``search_path`` spans several schemas, qualify table names in
+the SQL to remove the ambiguity. When no schema can be determined, the table is left out of
+lineage and the reason is reported. Only ``SELECT``-style statements pandas can execute are in
+scope; SQL run inside ordinary Python functions is not observed.
+
+Call the helper yourself
+~~~~~~~~~~~~~~~~~~~~~~~~
+
+Custom ``@dataloader`` / ``@datasaver`` functions can produce the same metadata:
+
+.. code-block:: python
+
+    from hamilton.io import utils
+
+    @dataloader()
+    def orders(sales_db: Engine) -> tuple[pd.DataFrame, dict]:
+        query = "SELECT * FROM sales.public.orders"
+        df = pd.read_sql(query, sales_db)
+        return df, utils.get_sql_metadata(query, df, db_connection=sales_db, operation="read")
+
+Pass ``operation="write"`` from a saver so the table name is never mistaken for a statement.
+The original two-argument call ``get_sql_metadata(query_or_table, results)`` keeps working and
+keeps its keys, their values and row-count semantics; it simply reports ``source=None`` with a note and
+``operation=None``, so consumers diagnose it as incomplete rather than guessing.

docs/reference/lifecycle-hooks/OpenLineageAdapter.rst

@@ -2,7 +2,127 @@
 plugins.h_openlineage.OpenLineageAdapter
 ========================================
 
+Install with ``pip install "apache-hamilton[openlineage]"``. The extra brings ``openlineage-python``
+(the client) and, on Linux and macOS, ``openlineage-sql`` (the parser used to find the tables a
+query reads). ``openlineage-sql`` publishes no Windows wheel, so the extra skips it there and SQL
+queries, and written table names that are not plain identifiers, are reported without table
+datasets on Windows. The parser is used only with
+``sql_dataset_identity="datasource"``; the default identity needs neither it nor anything else
+beyond the client.
+
 .. autoclass:: hamilton.plugins.h_openlineage.OpenLineageAdapter
    :special-members: __init__
    :members:
    :inherited-members:
+
+SQL datasets
+------------
+
+SQL loaders and savers (``@load_from.sql``, ``@save_to.sql`` and the pandas SQL materializers) record
+the datasource they used (see :ref:`sql-metadata-and-lineage`). How the adapter names their datasets
+is set by ``sql_dataset_identity``:
+
+- ``"legacy"``, the default: datasets are named as in earlier Hamilton releases, under the adapter's
+  *job* namespace with the bare ``table_name``. As before, a query read containing ``SELECT``
+  produces a dataset with no name and the query in the job's ``sql`` facet, and other strings are
+  used as the dataset name. Leaving the option unset warns once per adapter (a
+  ``FutureWarning``) because the default will change; pass ``"legacy"`` explicitly to keep these names
+  without the warning.
+- ``"datasource"``: datasets are named after the datasource, following the `OpenLineage naming
+  conventions <https://openlineage.io/docs/spec/naming/>`_, and every physical table a query reads is
+  reported. A report written by one job and read by another then resolves to the same dataset, and
+  two tables with the same name in different databases stay distinct.
+
+.. code-block:: python
+
+    adapter = OpenLineageAdapter(client, "my_namespace", "my_job", sql_dataset_identity="datasource")
+
+The rest of this section describes the ``"datasource"`` identity:
+
+.. list-table::
+   :header-rows: 1
+   :widths: 15 35 50
+
+   * - Dialect
+     - Namespace
+     - Name
+   * - PostgreSQL
+     - ``postgres://{host}:{port}`` (port defaults to 5432)
+     - ``{database}.{schema}.{table}``; unquoted identifiers are folded to lower case, as the server does
+   * - SQLite
+     - ``sqlite://{absolute file path}``
+     - ``{table}``. A table in an attached database (``reporting.orders`` in the SQL, or the writer's
+       ``schema``) is named in the attached file's namespace, which is known only for a
+       standard-library ``sqlite3`` connection; otherwise it is left out.
+
+Aliases and common table expressions are not reported as tables. Each dataset carries a
+``dataSource`` facet with the namespace; the ``schema`` facet from ``dataframe_metadata`` is
+attached only when the node maps to a single table. The job keeps the ``sql`` facet with the
+statement's text; a node that read or wrote a table by name has no statement to report. The
+exception is a table read by a name that the metadata files as a query (one containing
+``SELECT``, or whose first word is ``select`` or ``with``, such as ``SELECT_LOG`` or
+``select-log``): it can't be told apart from a statement, so the name is reported as the job's
+SQL and no input dataset is emitted for it. Give such tables plain names, or read them with a
+query.
+
+Whatever cannot be fully identified is left out and logged as a warning from the
+``hamilton.plugins.h_openlineage`` logger, never guessed. The following cases are left out:
+
+- an unknown datasource (in-memory SQLite, an unsupported connection object, legacy two-argument metadata)
+- an unsupported dialect
+- a table whose schema cannot be determined
+- a statement ``openlineage-sql`` cannot parse
+- a missing ``openlineage-sql`` install
+
+A failure inside the conversion is logged with its exception type and the run event is still
+emitted without datasets. The node itself has already succeeded and is never failed by lineage.
+
+.. _sql-dataset-identity-change:
+
+Migrating to the datasource identity
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+The default stays ``"legacy"`` for now and will become ``"datasource"`` in a future major release.
+Switching changes every SQL dataset's namespace and name (for example from ``my_namespace`` +
+``daily_revenue`` to ``postgres://warehouse.example:5432`` + ``analytics.reporting.daily_revenue``).
+A lineage backend shows the new names as new datasets: history recorded under the old names does
+not connect to them, and nothing is rewritten automatically. Some datasets also stop appearing:
+
+- loaders and savers whose metadata has no ``source`` (custom functions using the two-argument
+  helper, in-memory databases)
+- any of the unidentifiable cases above, including every SQL query and every written table name
+  that is not a plain identifier on Windows, where ``openlineage-sql`` is not installed
+
+To migrate:
+
+1. Install ``openlineage-sql`` where you run Hamilton. ``apache-hamilton[openlineage]`` includes it
+   everywhere except Windows. Pin it in locked environments.
+2. Run the pipeline once with ``sql_dataset_identity="datasource"`` against a test backend, or read
+   the events with ``FileTransport``, and note the new namespace and name of each dataset. Check the
+   ``hamilton.plugins.h_openlineage`` warnings for anything left out.
+3. In your lineage backend, move what is keyed to the old names (ownership, tags, alerts, policies,
+   saved queries) to the new ones, or link old and new datasets where the backend supports it.
+4. Pass ``sql_dataset_identity="datasource"`` in production.
+
+To stay on the current names, pass ``sql_dataset_identity="legacy"``. It silences the warning.
+
+Reuse the conversion
+--------------------
+
+:func:`~hamilton.plugins.h_openlineage.sql_datasets` is the boundary another integration (an
+orchestrator provider, a custom adapter) can call on metadata Hamilton produced, with no
+``OpenLineageClient``, ``Driver`` or database connection involved:
+
+.. code-block:: python
+
+    from hamilton.plugins.h_openlineage import sql_datasets
+
+    lineage = sql_datasets(node_result_metadata)   # the dict a SQL loader/saver returned
+    lineage.inputs    # list[openlineage.client.event_v2.Dataset]
+    lineage.outputs
+    lineage.notes     # why anything was left out
+
+.. autofunction:: hamilton.plugins.h_openlineage.sql_datasets
+
+.. autoclass:: hamilton.plugins.h_openlineage.SqlDatasets
+   :members:

examples/openlineage/.gitignore

@@ -0,0 +1,4 @@
+# generated by run.py
+pipeline.json
+sales.db
+warehouse.db