diff --git a/docs/reference/lifecycle-hooks/OpenLineageAdapter.rst b/docs/reference/lifecycle-hooks/OpenLineageAdapter.rst
index ff8f00ebc..35eb77774 100644
--- a/docs/reference/lifecycle-hooks/OpenLineageAdapter.rst
+++ b/docs/reference/lifecycle-hooks/OpenLineageAdapter.rst
@@ -18,16 +18,18 @@ beyond the client.
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``:
+SQL loaders and savers (``@load_from.sql``, ``@save_to.sql``, the pandas SQL materializers and the
+Polars database 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.
+ without the warning. The Polars database materializers keep their earlier names too: the read is a
+ dataset named after the loader node, with the query as its ``dataSource`` URI, and the write is
+ named after the table. Unlike the pandas query read above, the read has no ``sql`` job facet.
- ``"datasource"``: datasets are named after the datasource, following the `OpenLineage naming
conventions `_, 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
@@ -37,6 +39,9 @@ is set by ``sql_dataset_identity``:
adapter = OpenLineageAdapter(client, "my_namespace", "my_job", sql_dataset_identity="datasource")
+Datasource mode for the Polars database materializers requires ``polars>=1.0``. With an older
+Polars they log a warning asking you to upgrade and emit no dataset.
+
The rest of this section describes the ``"datasource"`` identity:
.. list-table::
@@ -68,6 +73,11 @@ query. A read by another name that is not a plain identifier (``daily revenue``)
that table when no SQL statement starts at it; one starting with ``table`` is left out, since
Postgres' ``TABLE t`` is a statement ``openlineage-sql`` does not know.
+A write is handled alike, for the pandas and Polars writers: a table name that is not a plain identifier is
+parsed, and is the table written unless it parses as a statement that names tables, such as a table literally
+called ``INSERT INTO orders SELECT * FROM customers``. The tables that statement writes are emitted instead, or
+none when it only reads. Give such tables plain names.
+
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:
diff --git a/hamilton/plugins/polars_post_1_0_0_extensions.py b/hamilton/plugins/polars_post_1_0_0_extensions.py
index dab1a3973..e2916d7dc 100644
--- a/hamilton/plugins/polars_post_1_0_0_extensions.py
+++ b/hamilton/plugins/polars_post_1_0_0_extensions.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import csv
import dataclasses
from collections.abc import Collection, Mapping, Sequence
from io import BytesIO, IOBase, TextIOWrapper
@@ -807,7 +808,14 @@ def load_data(self, type_: type) -> tuple[DATAFRAME_TYPE, dict[str, Any]]:
connection=self.connection,
**self._get_loading_kwargs(),
)
- metadata = utils.get_file_and_dataframe_metadata(self.query, df)
+ # with iter_batches=True, pl.read_database returns a generator, so there is no row count
+ rows = len(df) if isinstance(df, pl.DataFrame) else None
+ metadata = {
+ **utils.get_file_and_dataframe_metadata(self.query, df),
+ **utils.get_sql_metadata(
+ self.query, rows, db_connection=self.connection, operation="read"
+ ),
+ }
return df, metadata
@classmethod
@@ -842,12 +850,26 @@ def save_data(self, data: DATAFRAME_TYPE | pl.LazyFrame) -> dict[str, Any]:
if isinstance(data, pl.LazyFrame):
data = data.collect()
- data.write_database(
+ rows = data.write_database(
table_name=self.table_name,
connection=self.connection,
**self._get_saving_kwargs(),
)
- return utils.get_file_and_dataframe_metadata(self.table_name, data)
+ # polars unpacks table_name as [[catalog.]schema.]table, honouring quotes; lineage wants
+ # the schema recorded apart, and the connection already identifies the catalog
+ components = next(csv.reader([self.table_name], delimiter="."))
+ table = components[-1]
+ schema = components[-2] if len(components) > 1 else None
+ return {
+ **utils.get_file_and_dataframe_metadata(self.table_name, data),
+ **utils.get_sql_metadata(
+ table,
+ rows,
+ db_connection=self.connection,
+ schema=schema or None,
+ operation="write",
+ ),
+ }
@classmethod
def name(cls) -> str:
diff --git a/hamilton/plugins/polars_pre_1_0_0_extension.py b/hamilton/plugins/polars_pre_1_0_0_extension.py
index 4dd92dc43..f40ae83b0 100644
--- a/hamilton/plugins/polars_pre_1_0_0_extension.py
+++ b/hamilton/plugins/polars_pre_1_0_0_extension.py
@@ -47,6 +47,7 @@
from polars.type_aliases import CsvEncoding, SchemaDefinition
else:
CsvEncoding = type
+ SchemaDefinition = type
if has_alias and hasattr(pl.type_aliases, "CsvQuoteStyle"):
from polars.type_aliases import CsvQuoteStyle
else:
@@ -729,6 +730,17 @@ def name(cls) -> str:
return "spreadsheet"
+def _sql_metadata_without_datasource(
+ query_or_table: str, rows: int | None, operation: Literal["read", "write"]
+) -> dict[str, Any]:
+ """``sql_metadata`` for database I/O, whose datasource polars<1.0 cannot be asked for."""
+ metadata = utils.get_sql_metadata(query_or_table, rows, operation=operation)
+ metadata["sql_metadata"]["notes"] = (
+ "Datasource lineage for Polars database I/O requires polars>=1.0; upgrade polars"
+ )
+ return metadata
+
+
@dataclasses.dataclass
class PolarsDatabaseReader(DataLoader):
"""
@@ -768,7 +780,12 @@ def load_data(self, type_: type) -> tuple[DATAFRAME_TYPE, dict[str, Any]]:
connection=self.connection,
**self._get_loading_kwargs(),
)
- metadata = utils.get_file_and_dataframe_metadata(self.query, df)
+ # with iter_batches=True, pl.read_database returns a generator, so there is no row count
+ rows = len(df) if isinstance(df, pl.DataFrame) else None
+ metadata = {
+ **utils.get_file_and_dataframe_metadata(self.query, df),
+ **_sql_metadata_without_datasource(self.query, rows, "read"),
+ }
return df, metadata
@classmethod
@@ -803,12 +820,15 @@ def save_data(self, data: DATAFRAME_TYPE | pl.LazyFrame) -> dict[str, Any]:
if isinstance(data, pl.LazyFrame):
data = data.collect()
- data.write_database(
+ rows = data.write_database(
table_name=self.table_name,
connection=self.connection,
**self._get_saving_kwargs(),
)
- return utils.get_file_and_dataframe_metadata(self.table_name, data)
+ return {
+ **utils.get_file_and_dataframe_metadata(self.table_name, data),
+ **_sql_metadata_without_datasource(self.table_name, rows, "write"),
+ }
@classmethod
def name(cls) -> str:
diff --git a/tests/plugins/test_h_openlineage.py b/tests/plugins/test_h_openlineage.py
index 198622369..e863821d7 100644
--- a/tests/plugins/test_h_openlineage.py
+++ b/tests/plugins/test_h_openlineage.py
@@ -18,6 +18,7 @@
import json
import logging
import sqlite3
+import typing
import warnings
import pandas as pd
@@ -36,6 +37,13 @@
from hamilton.plugins import h_openlineage # noqa: E402
from hamilton.plugins.pandas_extensions import PandasSqlReader # noqa: E402
+try:
+ import polars as pl
+
+ from hamilton.plugins import polars_post_1_0_0_extensions as polars_post
+except ImportError: # pragma: no cover - polars is optional
+ pl = None
+
REVENUE_QUERY = """
-- daily revenue per customer country
WITH paid AS (SELECT * FROM orders WHERE status = 'paid')
@@ -845,3 +853,187 @@ def orders(df: pd.DataFrame) -> pd.DataFrame:
assert dataset_ids(events, "inputs") == [
(f"sqlite://{other_path.resolve().as_posix()}", "orders")
]
+
+
+requires_polars = pytest.mark.skipif(pl is None, reason="polars is not installed")
+
+
+class SubscriptableStandIn:
+ """Stands in for a name a type alias refers to when sqlalchemy has no class of that name."""
+
+ def __class_getitem__(cls, item):
+ return cls
+
+
+def sqlalchemy_class(name):
+ """The sqlalchemy class polars' alias means by ``name`` (or ``Alchemy``), if any."""
+ import sqlalchemy
+ import sqlalchemy.orm
+
+ modules = [sqlalchemy, sqlalchemy.orm]
+ try:
+ import sqlalchemy.ext.asyncio
+
+ modules.append(sqlalchemy.ext.asyncio)
+ except ImportError: # the asyncio extension needs greenlet
+ pass
+ for candidate in (name, name.removeprefix("Alchemy")):
+ for module in modules:
+ if hasattr(module, candidate):
+ return getattr(module, candidate)
+ return SubscriptableStandIn
+
+
+@pytest.fixture
+def polars_connection_type_hints(monkeypatch):
+ """Resolves the names polars' ``ConnectionOrCursor`` alias only imports for type checking.
+
+ Hamilton reads the database classes' type hints to build ``@load_from.database`` and
+ ``@save_to.database``, which fails on those unresolved names. They differ between polars
+ versions, so each one that is missing is bound to the sqlalchemy class it means.
+ """
+ for _ in range(50):
+ try:
+ typing.get_type_hints(polars_post.PolarsDatabaseReader)
+ typing.get_type_hints(polars_post.PolarsDatabaseWriter)
+ return
+ except NameError as e:
+ monkeypatch.setattr(polars_post, e.name, sqlalchemy_class(e.name), raising=False)
+ raise AssertionError("polars database type hints still do not resolve")
+
+
+def polars_database_modules():
+ @save_to.database(
+ table_name=value("orders"),
+ connection=source("engine"),
+ if_table_exists=value("replace"),
+ output_name_="saved_orders",
+ )
+ def orders_to_save(raw_orders: pl.DataFrame) -> pl.DataFrame:
+ return raw_orders
+
+ @load_from.database(query=value("SELECT * FROM orders"), connection=source("engine"))
+ def orders_read(orders_df: pl.DataFrame) -> pl.DataFrame:
+ return orders_df
+
+ return (
+ ad_hoc_utils.create_temporary_module(orders_to_save),
+ ad_hoc_utils.create_temporary_module(orders_read),
+ )
+
+
+def run_polars_database_flow(tmp_path, namespace, **adapter_kwargs):
+ """Writes ``orders`` then reads it back through Polars' database materializers."""
+ db_path = tmp_path / "shop.db"
+ inputs = {
+ "engine": create_engine(f"sqlite:///{db_path}"),
+ "raw_orders": pl.DataFrame({"id": [1, 2, 3], "amount": [10.0, 5.0, 7.0]}),
+ }
+ writer, reader = polars_database_modules()
+ run_with_lineage(
+ tmp_path, namespace, inputs, module=writer, final_vars=("saved_orders",), **adapter_kwargs
+ )
+ result, events = run_with_lineage(
+ tmp_path, namespace, inputs, module=reader, final_vars=("orders_read",), **adapter_kwargs
+ )
+ return db_path, result, [strip_private(e) for e in events if e["eventType"] == "RUNNING"]
+
+
+@requires_polars
+def test_polars_database_datasource_identity_names_the_sqlite_table(
+ tmp_path, polars_connection_type_hints
+):
+ db_path, _, (write, read) = run_polars_database_flow(
+ tmp_path, "polars_ns", sql_dataset_identity="datasource"
+ )
+ namespace = f"sqlite://{db_path.resolve().as_posix()}"
+ assert [(d["namespace"], d["name"]) for d in write["outputs"]] == [(namespace, "orders")]
+ assert [(d["namespace"], d["name"]) for d in read["inputs"]] == [(namespace, "orders")]
+ assert read["job"]["facets"]["sql"]["query"] == "SELECT * FROM orders"
+ for dataset in write["outputs"] + read["inputs"]:
+ facets = dataset.get("facets", {})
+ assert "storage" not in facets
+
+
+@requires_polars
+def test_polars_database_default_identity_keeps_legacy_datasets(
+ tmp_path, polars_connection_type_hints
+):
+ """Pinned to ``main``'s datasets: file metadata still wins over the SQL metadata.
+
+ Only the ``FutureWarning`` is new, since the metadata now carries ``sql_metadata``.
+ """
+ with pytest.warns(FutureWarning, match="sql_dataset_identity='datasource'"):
+ _, _, (write, read) = run_polars_database_flow(tmp_path, "polars_legacy_ns")
+ assert (read["inputs"][0]["namespace"], read["inputs"][0]["name"]) == (
+ "polars_legacy_ns",
+ "orders_read.load_data.orders_df",
+ )
+ assert set(read["inputs"][0]["facets"]) == {"storage", "dataSource", "schema"}
+ assert read["inputs"][0]["facets"]["storage"]["storageLayer"] == "FileSystem"
+ assert read["inputs"][0]["facets"]["dataSource"]["uri"] == "SELECT * FROM orders"
+ assert "sql" not in read["job"]["facets"]
+ assert (write["outputs"][0]["namespace"], write["outputs"][0]["name"]) == (
+ "polars_legacy_ns",
+ "orders",
+ )
+ assert set(write["outputs"][0]["facets"]) == {"storage", "dataSource", "schema"}
+ assert write["outputs"][0]["facets"]["storage"]["storageLayer"] == "FileSystem"
+ assert write["outputs"][0]["facets"]["dataSource"]["uri"] == "orders"
+
+
+@requires_polars
+def test_pre_1_0_polars_database_asks_datasource_users_to_upgrade(tmp_path, caplog):
+ """The pre-1.0 classes keep file metadata and add a connection-less ``sql_metadata``."""
+ from hamilton.plugins import polars_pre_1_0_0_extension as pre_1_0
+
+ connection = sqlite3.connect(tmp_path / "old.db")
+ frame = pl.DataFrame({"id": [1, 2]})
+ write_metadata = pre_1_0.PolarsDatabaseWriter(
+ table_name="orders",
+ connection=f"sqlite:///{tmp_path / 'old.db'}",
+ if_table_exists="replace",
+ ).save_data(frame)
+ _, read_metadata = pre_1_0.PolarsDatabaseReader(
+ query="SELECT * FROM orders", connection=connection
+ ).load_data(pl.DataFrame)
+
+ for metadata, operation in ((write_metadata, "write"), (read_metadata, "read")):
+ assert metadata["file_metadata"]["path"] == (
+ "orders" if operation == "write" else "SELECT * FROM orders"
+ )
+ assert metadata["sql_metadata"]["source"] is None
+ assert metadata["sql_metadata"]["operation"] == operation
+ node_ = type("FakeNode", (), {"name": f"{operation}_node"})()
+ caplog.clear()
+ with caplog.at_level(logging.WARNING, logger="hamilton.plugins.h_openlineage"):
+ datasets, _ = h_openlineage._sql_lineage(metadata, operation, node_)
+ assert datasets == []
+ assert "requires polars>=1.0" in caplog.text
+
+
+@requires_polars
+@pytest.mark.parametrize(
+ ("table_name", "query", "name"),
+ [
+ ("main.orders", "SELECT * FROM main.orders", "orders"),
+ ('"my-tbl"', 'SELECT * FROM "my-tbl"', "my-tbl"),
+ ('main."my-tbl"', 'SELECT * FROM main."my-tbl"', "my-tbl"),
+ ],
+)
+def test_polars_database_write_resolves_to_the_table_a_query_reads(
+ tmp_path, table_name, query, name
+):
+ path = tmp_path / "qualified.db"
+ engine = create_engine(f"sqlite:///{path}")
+ written = polars_post.PolarsDatabaseWriter(
+ table_name=table_name, connection=engine, if_table_exists="replace"
+ ).save_data(pl.DataFrame({"id": [1, 2]}))
+ _, read = polars_post.PolarsDatabaseReader(query=query, connection=engine).load_data(
+ pl.DataFrame
+ )
+
+ output = h_openlineage.sql_datasets(written).outputs
+ input_ = h_openlineage.sql_datasets(read).inputs
+ assert identities(output) == identities(input_)
+ assert identities(output) == [(f"sqlite://{path.resolve().as_posix()}", name)]
diff --git a/tests/plugins/test_polars_extensions.py b/tests/plugins/test_polars_extensions.py
index 2709dd85c..1bfa8fdc0 100644
--- a/tests/plugins/test_polars_extensions.py
+++ b/tests/plugins/test_polars_extensions.py
@@ -17,6 +17,7 @@
import io
import pathlib
+import sqlite3
import sys
import typing
@@ -221,3 +222,104 @@ def test_getting_type_hints_spreadsheetwriter():
"""Tests that types can be resolved at run time."""
type_hints = typing.get_type_hints(PolarsSpreadsheetWriter)
assert type_hints["workbook"] == typing.Union[Workbook, io.BytesIO, pathlib.Path, str]
+
+
+def test_polars_database_records_row_counts(df: pl.DataFrame, tmp_path: pathlib.Path) -> None:
+ connector = create_engine(f"sqlite:///{tmp_path}/rows.db")
+ writer = PolarsDatabaseWriter(
+ table_name="rows_table", connection=connector, if_table_exists="replace"
+ )
+ write_metadata = writer.save_data(df)
+ assert write_metadata["sql_metadata"]["rows"] == 2
+ assert write_metadata["sql_metadata"]["table_name"] == "rows_table"
+ assert write_metadata["sql_metadata"]["operation"] == "write"
+
+ reader = PolarsDatabaseReader(query="SELECT * FROM rows_table", connection=connector)
+ _, read_metadata = reader.load_data(pl.DataFrame)
+ assert read_metadata["sql_metadata"]["rows"] == 2
+ assert read_metadata["sql_metadata"]["operation"] == "read"
+ assert read_metadata["sql_metadata"]["query"] == "SELECT * FROM rows_table"
+
+
+def test_polars_database_batched_read_records_no_row_count(
+ df: pl.DataFrame, tmp_path: pathlib.Path
+) -> None:
+ connector = create_engine(f"sqlite:///{tmp_path}/batched.db")
+ PolarsDatabaseWriter(
+ table_name="batched", connection=connector, if_table_exists="replace"
+ ).save_data(df)
+
+ reader = PolarsDatabaseReader(
+ query="SELECT * FROM batched", connection=connector, iter_batches=True, batch_size=1
+ )
+ batches, metadata = reader.load_data(pl.DataFrame)
+
+ assert metadata["sql_metadata"]["rows"] is None
+ assert metadata["sql_metadata"]["source"]["dialect"] == "sqlite"
+ assert sum(len(batch) for batch in batches) == 2
+
+
+def test_polars_database_keeps_file_metadata(df: pl.DataFrame, tmp_path: pathlib.Path) -> None:
+ connector = create_engine(f"sqlite:///{tmp_path}/legacy.db")
+ write_metadata = PolarsDatabaseWriter(
+ table_name="legacy", connection=connector, if_table_exists="replace"
+ ).save_data(df)
+ _, read_metadata = PolarsDatabaseReader(
+ query="SELECT * FROM legacy", connection=connector
+ ).load_data(pl.DataFrame)
+
+ assert write_metadata["file_metadata"]["path"] == "legacy"
+ assert read_metadata["file_metadata"]["path"] == "SELECT * FROM legacy"
+ assert "dataframe_metadata" in write_metadata and "dataframe_metadata" in read_metadata
+
+
+@pytest.mark.parametrize(
+ ("table_name", "expected_table", "expected_schema"),
+ [
+ ("orders", "orders", None),
+ ("main.orders", "orders", "main"),
+ ('"my-tbl"', "my-tbl", None),
+ ('main."my-tbl"', "my-tbl", "main"),
+ ('"main".orders', "orders", "main"),
+ ('"a.b"', "a.b", None),
+ ],
+)
+def test_polars_database_writer_records_the_table_and_schema_as_polars_unpacks_them(
+ df: pl.DataFrame,
+ tmp_path: pathlib.Path,
+ table_name: str,
+ expected_table: str,
+ expected_schema: str | None,
+) -> None:
+ connector = create_engine(f"sqlite:///{tmp_path}/qualified.db")
+ metadata = PolarsDatabaseWriter(
+ table_name=table_name, connection=connector, if_table_exists="replace"
+ ).save_data(df)
+
+ assert metadata["sql_metadata"]["table_name"] == expected_table
+ assert metadata["sql_metadata"]["schema"] == expected_schema
+ assert metadata["file_metadata"]["path"] == table_name
+
+
+def test_pre_1_0_polars_database_records_row_counts(
+ df: pl.DataFrame, tmp_path: pathlib.Path
+) -> None:
+ from hamilton.plugins import polars_pre_1_0_0_extension as pre_1_0
+
+ url = f"sqlite:///{tmp_path}/pre.db"
+ write_metadata = pre_1_0.PolarsDatabaseWriter(
+ table_name="pre", connection=url, if_table_exists="replace"
+ ).save_data(df)
+ assert write_metadata["sql_metadata"]["rows"] == 2
+
+ connection = sqlite3.connect(tmp_path / "pre.db")
+ _, read_metadata = pre_1_0.PolarsDatabaseReader(
+ query="SELECT * FROM pre", connection=connection
+ ).load_data(pl.DataFrame)
+ assert read_metadata["sql_metadata"]["rows"] == 2
+
+ batches, batched_metadata = pre_1_0.PolarsDatabaseReader(
+ query="SELECT * FROM pre", connection=connection, iter_batches=True, batch_size=1
+ ).load_data(pl.DataFrame)
+ assert batched_metadata["sql_metadata"]["rows"] is None
+ assert sum(len(batch) for batch in batches) == 2
diff --git a/writeups/specs/2609-01-polars-sql-metadata.md b/writeups/specs/2609-01-polars-sql-metadata.md
new file mode 100644
index 000000000..c50f44fda
--- /dev/null
+++ b/writeups/specs/2609-01-polars-sql-metadata.md
@@ -0,0 +1,236 @@
+
+
+# Definition: Polars database I/O records SQL metadata (apache/hamilton#1740)
+
+## 1. Intent
+- **Problem:** A team reads `orders` with `PolarsDatabaseReader` and writes a report table with `PolarsDatabaseWriter` against a SQLAlchemy Engine, and runs Hamilton with `OpenLineageAdapter(sql_dataset_identity="datasource")` as #1720 introduced. Lineage for those nodes is wrong: both classes describe database I/O with `get_file_and_dataframe_metadata(self.query | self.table_name, df)` (`polars_post_1_0_0_extensions.py:810`, `:850`; `polars_pre_1_0_0_extension.py:771`, `:811`), so there is no `sql_metadata`, datasource mode cannot identify the database, and the adapter emits fake FileSystem datasets (a read named after the node with the SQL text as its URI) and no `sql` job facet. Reproduced: `sorted(md) == ['dataframe_metadata', 'file_metadata']`, `md["file_metadata"]["path"] == "SELECT * FROM orders"`. The issue text predates #1720's final form; the fix follows what #1720 merged — legacy mode stays compatible with existing conventions (ADR `writeups/adr/2609-01-staged-sql-dataset-identity.md`).
+- **Appetite:** Small. Four call sites in two Polars plugin files, their tests, and one docs page. No change to shared helpers or the adapter.
+- **Out of bounds:**
+ - Recognising more connection types in `get_sql_source` (issue item 2: ADBC connections, connectorx URIs, cursors, Sessions) — deferred; those already drop out of lineage with a note, and SQLAlchemy-parsing connectorx SQLite URIs would name the wrong file.
+ - Removing `file_metadata` from the Polars database classes — reserved for the major release that flips the `sql_dataset_identity` default.
+ - Opening, pushing, or driving a pull request; commenting on #1740. The run ends at a local branch.
+
+## 2. Initial Approach
+- **Architecture:**
+ - Post-1.0 reader/writer return `{**utils.get_file_and_dataframe_metadata(, df), **utils.get_sql_metadata(, rows, db_connection=self.connection, operation="read"|"write")}`. Legacy builders in `h_openlineage.py` check `file_metadata` before `sql_metadata` (`:363`, `:416`), so legacy output is unchanged; datasource mode reads `sql_metadata`.
+ - Read rows: `len(df)` only when `df` is a `pl.DataFrame` (with `iter_batches=True` `pl.read_database` returns a generator), else `None`. Write rows: the `int` that `write_database` returns. Pass the count as `results`; do not broaden `get_sql_metadata`'s row counting (`tests/io/test_utils.py` pins `rows is None` for non-frame input).
+ - Pre-1.0 reader/writer: keep `file_metadata`; add `get_sql_metadata(, rows, operation=...)` with no `db_connection`, then set its `notes` to an upgrade message (e.g. "Datasource lineage for Polars database I/O requires polars>=1.0; upgrade polars"). In datasource mode `sql_datasets` returns no dataset and `_sql_lineage` logs the note; verified in a scratch run.
+ - The pre-1.0 module imports and its database classes run under the installed Polars 1.41 when given a `sqlite3.Connection` (verified), so they can be exercised directly in tests even though the plugin registry only loads that module for Polars <1.0.
+ - Reuse `run_with_lineage` / `FileTransport` harness in `tests/plugins/test_h_openlineage.py:318-333` for end-to-end tests.
+ - Work on a local branch `2609/polars-sql-metadata` off `main`. Commit message body carries the release-note line: Polars database I/O now also records `sql_metadata`; default-mode OpenLineage users may see the identity `FutureWarning`, silenced by `sql_dataset_identity="legacy"`.
+
+## 3. Global Invariants
+
+### INV-G1 — The change stops at what this Manifest authorized
+
+Read this Manifest. This is a conformance check: take its intent as given, and do not judge
+whether the work was necessary, motivated, or worthwhile.
+
+Done when the work this run added carries nothing the Deliverables, Acceptance Criteria, and
+Global Invariants — this one excluded — required, and nothing that nominally serves one of
+them while far exceeding the surface the Appetite allows.
+
+Read Problem, Appetite, and Out of bounds as the intent. Read the Deliverables and the other
+gates as what the work owes. Read the Initial Approach and Process Guidance as mechanisms that
+were *authorized rather than owed* — the ones expected, never the only ones permitted, so work
+reaching a Deliverable by a route this Manifest does not name is required by that Deliverable.
+
+Required although no criterion names it: work inherited rather than added — the artifact this
+Manifest was synthesized over, and anything arriving from outside the run such as a base branch
+merged into the head — and work discharging what a criterion required, including sweeping a
+changed rule into every copy and surface that holds it.
+
+FAIL only on work none of the above accounts for. Treat an unclear case as required work, and
+leave small, incidental, or imperfect changes inside an artifact already in scope alone.
+
+Why: every other gate states a floor, so a contract bounded on one side only gives an executor
+disposed to thoroughness nothing to read as a limit.
+
+Judgment gate.
+
+### INV-G2 — Legacy-mode OpenLineage output for Polars database nodes is unchanged
+
+Done when, for every Polars database reader and writer class in `hamilton/plugins/polars_post_1_0_0_extensions.py` and `hamilton/plugins/polars_pre_1_0_0_extension.py`, the legacy-mode dataset the adapter emits — namespace, name, and the full facet set with values, excluding only timestamp-derived values — is identical on this branch and on `main`. Procedure: in a disposable `git worktree` of `main` and in the branch working tree, run the same script with the repo's `.venv/bin/python` that, against a fresh SQLite file, writes a table with each writer and reads it back with each reader (pre-1.0 classes given a `sqlite3.Connection`, post-1.0 classes given a SQLAlchemy Engine), passes each returned metadata dict through `h_openlineage.create_input_dataset` / `create_output_dataset`, and prints the serialized datasets; the two outputs must match. Also FAIL if the returned metadata on this branch lacks a `file_metadata` key or its `file_metadata` values differ from `main` (timestamps excepted). Remove the worktree afterwards with `git worktree remove --force`.
+
+Why: ADR 2609-01 promises that the default legacy mode emits exactly what earlier releases emitted; the user ruled that this fix must keep that promise.
+
+Deterministic gate.
+
+### INV-G3 — Recording metadata never fails a successful database read or write
+
+Done when no code path added to the Polars database classes' `load_data` / `save_data` can raise after the Polars read or write has succeeded, for any connection form those classes accept (SQLAlchemy Engine, Connection, Session, URL string; `sqlite3` Connection or Cursor; ADBC connections; other DBAPI connections) and for `iter_batches=True` reads that return a generator. Inspect the diff of both plugin files and trace each added call (`len`, `get_sql_metadata`, dict merging, notes assignment) against those inputs; `get_sql_source` already catches inspection errors, so the hiding place is anything computed outside it — row counting especially. The branch's tests must include an `iter_batches=True` read through `PolarsDatabaseReader` that returns without raising.
+
+Judgment gate.
+
+### INV-G4 — Shared SQL helpers and the adapter are untouched
+
+Done when `git diff main --stat -- hamilton/io/utils.py hamilton/plugins/h_openlineage.py` prints nothing.
+
+Why: #1720's merged behaviour is the contract this fix conforms to.
+
+Deterministic gate.
+
+### INV-G5 — Project lint, format and the affected test suites pass
+
+Done when both succeed on the branch working tree: `pre-commit run --files $(git diff --name-only main)` exits 0, and `.venv/bin/python -m pytest tests/plugins/test_polars_extensions.py tests/plugins/test_polars_lazyframe_extensions.py tests/plugins/test_h_openlineage.py tests/io/test_utils.py -q` exits 0 with no test newly skipped relative to `main`. No type-checker is configured for this project (mypy is commented out in `pyproject.toml`), so none is run.
+
+Deterministic gate.
+
+### INV-G6 — The change says what it does
+
+Done when the manifest-dev:review-code skill, activated with dimension=change-intent, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G7 — No mechanical bugs
+
+Done when the manifest-dev:review-code skill, activated with dimension=code-bugs, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G8 — The defect class is closed
+
+Done when the manifest-dev:review-code skill, activated with dimension=defect-class, reports nothing at or above that dimension's threshold. The class: a Hamilton data loader or saver that performs SQL database I/O but describes it with file metadata instead of `sql_metadata`, anywhere under `hamilton/plugins/`.
+
+Judgment gate.
+
+### INV-G9 — Metadata contract stays consistent with #1720
+
+Done when the manifest-dev:review-code skill, activated with dimension=contracts, reports nothing at or above that dimension's threshold. The contract surfaces are the metadata dicts the Polars database classes return, as consumed by `hamilton/plugins/h_openlineage.py` (`sql_datasets`, `create_input_dataset`, `create_output_dataset`) and `hamilton/plugins/h_mlflow.py`.
+
+Judgment gate.
+
+### INV-G10 — Type hints stay sound
+
+Done when the manifest-dev:review-code skill, activated with dimension=type-safety, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G11 — Operational readiness
+
+Done when the manifest-dev:review-code skill, activated with dimension=operational-readiness, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G12 — Maintainability
+
+Done when the manifest-dev:review-code skill, activated with dimension=code-maintainability, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G13 — Simplicity
+
+Done when the manifest-dev:review-code skill, activated with dimension=code-simplicity, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G14 — Design fitness
+
+Done when the manifest-dev:review-code skill, activated with dimension=code-design, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G15 — Testability
+
+Done when the manifest-dev:review-code skill, activated with dimension=code-testability, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G16 — Test quality
+
+Done when the manifest-dev:review-code skill, activated with dimension=test-quality, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G17 — Documentation
+
+Done when the manifest-dev:review-code skill, activated with dimension=docs, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G18 — Prose value
+
+Done when the manifest-dev:review-code skill, activated with dimension=prose-value, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+### INV-G19 — Context-file adherence
+
+Done when the manifest-dev:review-code skill, activated with dimension=context-file-adherence, reports nothing at or above that dimension's threshold.
+
+Judgment gate.
+
+## 4. Process Guidance
+- [PG-1] Run the four test files in INV-G5 on `main` before touching tests, so pre-existing failures are known.
+- [PG-2] Write each new test first and watch it fail against the current code before changing the plugin (red loop: the current classes return no `sql_metadata`).
+- [PG-3] Assert behaviour through the public seams — the classes' `load_data`/`save_data` return values and the events a Hamilton driver emits through `OpenLineageAdapter` — not private helpers.
+- [PG-4] Commit only on the local branch `2609/polars-sql-metadata`; never push. Name the mechanism (file metadata used for database I/O) in the commit message.
+
+## 5. Known Assumptions
+- [ASM-1] The release-note line lives in the commit message body, since there is no CHANGELOG file and no PR in scope | Default: commit body | Impact if wrong: note moved to wherever the user ships it; no code impact.
+- [ASM-2] (auto) The pre-1.0 upgrade note wording | Default: "Datasource lineage for Polars database I/O requires polars>=1.0; upgrade polars" | Impact if wrong: wording edit.
+- [ASM-3] Exercising the pre-1.0 classes under Polars 1.41 with a `sqlite3.Connection` stands in for real Polars <1.0 | Default: accepted; real Polars <1.0 is not installable in this environment | Impact if wrong: a pre-1.0-specific behaviour difference in `read_database`/`write_database` return values goes unseen; the metadata code does not depend on it beyond the row-count guard.
+
+## 6. Deliverables
+- **Order rationale:** Deliverable 1 carries the approach's main bet (additive metadata keeps legacy output while datasource mode works end-to-end through the real driver); Deliverable 2 reuses its pattern; Deliverable 3 documents both.
+
+### Deliverable 1: Post-1.0 Polars database nodes get datasource lineage
+
+*What it is, and how it is exercised end-to-end:* the post-1.0 `PolarsDatabaseReader`/`PolarsDatabaseWriter` record `sql_metadata`; exercised by a Hamilton driver with `OpenLineageAdapter` running a DAG that saves through `PolarsDatabaseWriter` and loads through `PolarsDatabaseReader` against a SQLite file via a SQLAlchemy Engine, with the emitted events read back from a `FileTransport` log.
+
+#### AC-1.1 — Datasource mode names the SQLite tables read and written
+
+Done when a test on the branch runs a Hamilton driver with `OpenLineageAdapter(..., sql_dataset_identity="datasource")` over a DAG that writes table `orders` with `PolarsDatabaseWriter` and reads `SELECT * FROM orders` with `PolarsDatabaseReader` against a SQLite file through a SQLAlchemy Engine, and asserts from the emitted events: the write's output dataset is namespace `sqlite://{absolute file path}` + name `orders`; the read's input dataset is the same namespace + `orders`; the read's job carries a `sql` facet with the query; no emitted dataset for these nodes has a `storage` facet with `storageLayer="FileSystem"`. The test passes under INV-G5's command.
+
+Deterministic gate.
+
+#### AC-1.2 — Row counts are recorded where they exist
+
+Done when tests on the branch assert that `sql_metadata["rows"]` equals the number of rows read for a plain `PolarsDatabaseReader` read, equals the number of rows written for a `PolarsDatabaseWriter` write, and is `None` for an `iter_batches=True` read, and those tests pass under INV-G5's command.
+
+Deterministic gate.
+
+#### AC-1.3 — Legacy output is pinned by a test
+
+Done when a test on the branch runs the post-1.0 reader and writer through a Hamilton driver with `OpenLineageAdapter` in default (legacy) mode and asserts the exact namespace, name and facet keys of each emitted dataset as they are on `main` (read named after the loader node with `storage`, `dataSource`, `schema` facets; write named after the table with the same facet keys), so reordering the `file_metadata`/`sql_metadata` branches in the adapter would fail it. It passes under INV-G5's command.
+
+Why: the additive design relies on the adapter checking `file_metadata` first; nothing else guards that.
+
+Judgment gate.
+
+### Deliverable 2: Pre-1.0 Polars database nodes stay legacy-only and tell datasource users to upgrade
+
+*What it is, and how it is exercised end-to-end:* the pre-1.0 reader/writer keep file metadata and add a connection-less `sql_metadata` with an upgrade note; exercised by calling the pre-1.0 classes' `load_data`/`save_data` against a SQLite file with a `sqlite3.Connection` and feeding the returned metadata through the adapter in both identity modes.
+
+#### AC-2.1 — Datasource mode emits no dataset and logs the upgrade note
+
+Done when a test on the branch passes metadata returned by the pre-1.0 `PolarsDatabaseReader.load_data` and `PolarsDatabaseWriter.save_data` through `OpenLineageAdapter` (or `h_openlineage._sql_lineage` / `sql_datasets`) in datasource mode and asserts: no input or output dataset is produced for those nodes, and a warning from the `hamilton.plugins.h_openlineage` logger contains text telling the user to upgrade to polars>=1.0. The same test or a sibling asserts that `sql_metadata["source"]` is `None` for both classes. Passes under INV-G5's command.
+
+Deterministic gate.
+
+### Deliverable 3: Docs describe Polars database lineage
+
+*What it is, and how it is exercised end-to-end:* `docs/reference/lifecycle-hooks/OpenLineageAdapter.rst` read by a user choosing an identity mode for a pipeline that uses Polars database materializers.
+
+#### AC-3.1 — A reader learns what each mode does for Polars database I/O
+
+Done when `docs/reference/lifecycle-hooks/OpenLineageAdapter.rst` states, in the sections a reader choosing `sql_dataset_identity` reads (the SQL datasets overview at line ~21 and the legacy-mode description): that the Polars database materializers record the datasource; that in legacy mode they keep their earlier dataset names (the read named after the loader node, the write after the table), unlike the pandas SQL query-read case described there; and that datasource mode for them requires polars>=1.0, with older Polars logging an upgrade note and emitting no dataset. No existing statement on the page is left contradicting these, and the page builds without new Sphinx warnings for that file if a docs build is available (otherwise the evaluator checks the RST syntax by reading it).
+
+Judgment gate.