From 5b087e9f8225e5763236070a56de67e9b75ef2cd Mon Sep 17 00:00:00 2001 From: Josh Taillon Date: Fri, 4 Sep 2026 10:08:41 -0600 Subject: [PATCH 1/2] feat(py): import a warehouse catalog when a source is built A Snowflake or Databricks engine now resolves its selection against the warehouse rather than leaving the readers unused: the session identity is read first, explicitly named relations are access-checked before anything is described, the listing is folded into the authored dictionary, and the session is read again at the end. A selection defaults to the namespace the connection already points at, and exclude drops relations from it. The merge re-keys the dictionary to the warehouse's labels, so a table's definitions are now looked up under the name the author gave it rather than the label it ended up with. Definitions over a column the warehouse spells differently are refused for now, because the compiler cannot yet bind an authored name to the discovered one. --- pkg-py/src/commons/_catalog/_import.py | 202 ++++++++++++++++++++ pkg-py/src/commons/_data_source.py | 128 +++++++++++-- pkg-py/src/commons/_definitions/_compile.py | 8 +- pkg-py/tests/_warehouse.py | 80 ++++++++ pkg-py/tests/test_catalog_import.py | 119 ++++++++++++ pkg-py/tests/test_data_source_warehouse.py | 142 ++++++++++++++ 6 files changed, 658 insertions(+), 21 deletions(-) create mode 100644 pkg-py/src/commons/_catalog/_import.py create mode 100644 pkg-py/tests/_warehouse.py create mode 100644 pkg-py/tests/test_catalog_import.py create mode 100644 pkg-py/tests/test_data_source_warehouse.py diff --git a/pkg-py/src/commons/_catalog/_import.py b/pkg-py/src/commons/_catalog/_import.py new file mode 100644 index 00000000..9c904cb8 --- /dev/null +++ b/pkg-py/src/commons/_catalog/_import.py @@ -0,0 +1,202 @@ +"""Turning a warehouse connection into the tables a data source exposes. + +The order is what matters here. The session identity is read first, because +every access answer that follows was decided for it; explicitly named +relations are checked before anything is described, so a name the caller got +wrong fails at construction rather than mid-conversation; and the session is +read again at the end, because a role that moved during discovery invalidates +what was just learned. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .._data_source import TableId +from . import _databricks, _snowflake +from ._core import ( + Manifest, + MergedDictionary, + Relation, + Selector, + check_exclude, + id_type, + merge_dictionary, + table_registry, +) +from ._security import ( + SessionSnapshot, + check_session, + require_queryable, + require_queryable_relations, + session_snapshot, +) + +__all__ = ["ImportedCatalog", "import_catalog", "is_warehouse"] + +# Each warehouse's reader, and the case its identifiers fold to. +_READERS = { + "snowflake": (_snowflake, "upper"), + "databricks": (_databricks, "lower"), +} + + +@dataclass +class ImportedCatalog: + """What a warehouse listing contributes to a data source.""" + + tables: list[str] + table_ids: dict[str, TableId] + relations: dict[str, Relation] + manifest: Manifest + dictionary: Any | None + definition_bindings: dict[str, Any] | None + session: SessionSnapshot | None + + +def is_warehouse(backend: Any) -> bool: + return backend.dialect() in _READERS + + +def import_catalog( + backend: Any, + tables: Any = None, + exclude: list[str] | None = None, + dictionary: Any = None, +) -> ImportedCatalog: + """Resolve a selection against a warehouse and fold it into a dictionary.""" + reader, identifier_case = _READERS[backend.dialect()] + check_exclude(exclude) + session = session_snapshot(backend) + + registry = table_registry( + _selectors(backend, reader, tables), + exact_relation=lambda selector: reader.exact_relation(backend, selector), + list_relations=lambda selector: reader.list_relations(backend, selector), + exclude=exclude, + ) + if registry.validate: + require_queryable_relations(backend, registry.validate, registry.relations) + if not registry.relations: + raise ValueError("The resolved catalog selection contains no objects.") + + # A relation named in the selection has just been probed, so the merge + # only has to check the ones the listing brought in. + queryable = set(registry.validate) + + def access_check(table_id: TableId, label: str) -> None: + if label in queryable: + return + require_queryable(backend, table_id, label) + queryable.add(label) + + merged = merge_dictionary( + dictionary, + registry.relations, + describe_relation=lambda table_id: reader.describe_relation(backend, table_id), + identifier_case=identifier_case, + access_check=access_check, + ) + check_session(backend, session) + _check_definitions_bound(merged) + + manifest = Manifest.build( + merged.relations, namespace_selected=registry.namespace_selected + ) + # What was probed on the way in is known to be readable, and a first + # touch of it should not pay for the same round trip again. + for label in queryable: + if label in manifest.access: + manifest.access[label] = "queryable" + + return ImportedCatalog( + tables=list(merged.relations), + table_ids={label: item.id for label, item in merged.relations.items()}, + relations=merged.relations, + manifest=manifest, + dictionary=merged.dictionary, + definition_bindings=merged.definition_bindings, + session=session, + ) + + +def _check_definitions_bound(merged: MergedDictionary) -> None: + """Refuse definitions the merge renamed out from under. + + The merge re-keys an authored dictionary to the warehouse's own labels + and column spellings, but a definition's expression still names what the + author wrote. Lowering it as written would emit SQL against identifiers + the warehouse does not have, so it is refused until the compiler can bind + the two together. + """ + bindings = merged.definition_bindings + exports = getattr(merged.dictionary, "definition_exports", None) or {} + if not bindings or not exports: + return + for authored_table, definitions in exports.items(): + if not definitions: + continue + # An authored table that matched nothing is dropped by the merge, so + # its definitions would go with it and the agent would never be told. + if bindings["tables"].get(authored_table) is None: + raise ValueError( + f"Authored table {authored_table!r} declares definitions, and " + f"does not match an exposed relation. Name it as the data " + f"source selects it, or drop it from the data dictionary." + ) + # Every authored column is checked, not only the ones a definition + # reads: which columns an expression touches is in the compiler's + # parse tree, and the refusal is temporary either way. A column the + # warehouse never reported is caught here too, since a definition + # over it would lower to SQL naming nothing at all. + columns = bindings["columns"].get(authored_table) or {} + unbound = [name for name, discovered in columns.items() if discovered != name] + if unbound: + raise NotImplementedError( + f"Table {authored_table!r} declares definitions, and the " + f"warehouse does not have column {unbound[0]!r} under that " + f"name. Binding a definition to the discovered spelling is " + f"not available yet." + ) + + +def _selectors(backend: Any, reader: Any, tables: Any) -> list[Selector]: + """Read a `tables` selection, defaulting to the connection's namespace. + + With nothing named, the schema the connection already points at is the + selection: it is the one namespace the caller has already chosen. + """ + if tables is None: + return [reader.current_namespace(backend)] + entries = list(tables) if isinstance(tables, (list, tuple)) else [tables] + if not entries: + raise ValueError("tables must name at least one relation or namespace.") + return [_selector(entry) for entry in entries] + + +def _selector(entry: Any) -> Selector: + """One selection entry as a `Selector`, whatever it was spelled as. + + A string is read the way it is everywhere else, as a relation whose dots + qualify it. A namespace has to be a `Selector`, because `ANALYTICS.PUBLIC` + on its own does not say whether PUBLIC is a schema or a table. + """ + if isinstance(entry, Selector): + id_type(entry) + return entry + if isinstance(entry, TableId): + return Selector(catalog=entry.catalog, schema=entry.schema, table=entry.table) + if isinstance(entry, str) and entry: + parts = entry.split(".") + if len(parts) > 3 or any(part == "" for part in parts): + raise ValueError( + "A relation is named catalog.schema.table, with no empty or " + f"skipped components, got {entry!r}." + ) + padded = [None] * (3 - len(parts)) + parts + return Selector(catalog=padded[0], schema=padded[1], table=padded[2]) + raise TypeError( + "Each entry in tables must be a relation name, a TableId, or a " + f"Selector naming a namespace, got {entry!r}." + ) diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py index 5d0583c2..ffc8fefd 100644 --- a/pkg-py/src/commons/_data_source.py +++ b/pkg-py/src/commons/_data_source.py @@ -95,6 +95,13 @@ class DataSource: table_ids: dict[str, TableId] = field(default_factory=dict) pending: _PendingPins | None = None dictionary: DataDictionary | None = None + # Set only for a warehouse source: what its catalog reported, what is + # known about access to each relation, and the connection identity all + # of that was decided for. + relations: dict[str, Any] | None = None + manifest: Any | None = None + session: Any | None = None + definition_bindings: dict[str, Any] | None = None @classmethod def from_frames(cls, **frames: Any) -> DataSource: @@ -122,29 +129,87 @@ def from_frames(cls, **frames: Any) -> DataSource: ) @classmethod - def from_engine(cls, engine: sqlalchemy.Engine, tables: Any = None) -> DataSource: + def from_engine( + cls, + engine: sqlalchemy.Engine, + tables: Any = None, + exclude: list[str] | None = None, + dictionary: DataDictionary | None = None, + ) -> DataSource: """Query a caller's database directly. Nothing is copied. - With `tables` unset the backend's own listing is taken as given: it - reports what exists, so there is nothing to check and no round trip - worth paying for. + A Snowflake or Databricks engine imports its catalog: the selection is + resolved against the warehouse, access to it is verified for the + current principal, and what it reports is folded into `dictionary`. + + With `tables` unset on any other backend, its own listing is taken as + given: it reports what exists, so there is nothing to check and no + round trip worth paying for. + + `dictionary` is taken here because the warehouse listing is folded + into it during construction. Its definitions are lowered by + `data_source()`, once, so attach one through there rather than here. """ + from ._catalog._import import is_warehouse + backend = EngineBackend(engine) + if is_warehouse(backend): + return cls._from_warehouse(backend, tables, exclude, dictionary) + if exclude is not None: + raise ValueError( + "exclude selects out of a warehouse catalog listing, and is " + f"supported only for Snowflake and Databricks. This engine is " + f"{backend.dialect()}." + ) if tables is None: discovered = backend.list_tables() return cls( backend=backend, tables=discovered, table_ids={name: TableId(table=name) for name in discovered}, + dictionary=dictionary, ) registry = normalize_table_registry(tables) _check_tables_exist(backend, registry) - return cls(backend=backend, tables=list(registry), table_ids=registry) + return cls( + backend=backend, + tables=list(registry), + table_ids=registry, + dictionary=dictionary, + ) + + @classmethod + def _from_warehouse( + cls, + backend: Backend, + tables: Any, + exclude: list[str] | None, + dictionary: DataDictionary | None, + ) -> DataSource: + from ._catalog._import import import_catalog + + imported = import_catalog(backend, tables, exclude, dictionary) + return cls( + backend=backend, + tables=imported.tables, + table_ids=imported.table_ids, + dictionary=imported.dictionary, + relations=imported.relations, + manifest=imported.manifest, + session=imported.session, + definition_bindings=imported.definition_bindings, + ) @classmethod - def from_board(cls, board: Any, tables: Any) -> DataSource: - """Expose a pins board's pins as tables, each read on first use.""" + def from_board( + cls, board: Any, tables: Any, dictionary: DataDictionary | None = None + ) -> DataSource: + """Expose a pins board's pins as tables, each read on first use. + + As with `from_engine()`, a dictionary's definitions are lowered by + `data_source()` rather than here. + """ if not isinstance(tables, dict): raise TypeError( "For a pins board, tables must be a mapping of table name to " @@ -176,11 +241,15 @@ def from_board(cls, board: Any, tables: Any) -> DataSource: tables=labels, table_ids={label: TableId(table=label) for label in labels}, pending=_PendingPins(board=board, pins=dict(tables)), + dictionary=dictionary, ) def query(self, sql: str) -> list[dict[str, Any]]: """Run one read-only statement, rejecting anything else first.""" + from ._catalog import check_session + check_query(sql, dialect=self.backend.dialect()) + check_session(self.backend, self.session) if self.pending is None: return self.backend.query(sql) return self._query_loading_pins(sql) @@ -255,6 +324,7 @@ def dialect(self) -> str: def data_source( *args: Any, tables: Any = None, + exclude: Any = None, dictionary: Any = None, **frames: Any, ) -> DataSource: @@ -262,11 +332,12 @@ def data_source( A thin dispatcher over the constructors, which are the documented way in. - `tables` and `dictionary` are keyword-only options, so both names are - reserved in every form: a frame passed under either name is rejected - with a TypeError naming it, never silently consumed. `tables` selects - tables of the engine and board forms; `dictionary` attaches a data - dictionary to any form. To use either as a frame name, call + `tables`, `exclude`, and `dictionary` are keyword-only options, so all + three names are reserved in every form: a frame passed under one of them + is rejected with a TypeError naming it, never silently consumed. + `tables` selects tables of the engine and board forms, `exclude` drops + relations from a warehouse catalog listing by glob, and `dictionary` + attaches a data dictionary to any form. To use one as a frame name, call `DataSource.from_frames()` directly. A dictionary's governed definitions are compiled for the source's @@ -288,29 +359,48 @@ def data_source( "Pass either a connection or named data frames, not both. " f"Got a positional argument and the frames {sorted(frames)}." ) - source = _from_positional(args[0], tables) + source = _from_positional(args[0], tables, exclude, resolved) else: if tables is not None: raise TypeError( "`tables` selects tables of an engine or pins board; with " "named frames there is nothing for it to select." ) + if exclude is not None: + raise TypeError( + "`exclude` drops relations from a warehouse catalog listing; " + "with named frames there is no listing to drop them from." + ) source = DataSource.from_frames(**frames) + source.dictionary = resolved - source.dictionary = resolved - if resolved is not None: + # A warehouse source merges the authored dictionary with what its catalog + # reported, so the one to compile against is the source's, not the one + # that was passed in. + if source.dictionary is not None: # The dialect is only known now, which is why lowering waits for it. from ._definitions import attach_compiled_definitions - attach_compiled_definitions(resolved, source.dialect(), set(source.tables)) + attach_compiled_definitions( + source.dictionary, source.dialect(), set(source.tables) + ) return source -def _from_positional(value: Any, tables: Any) -> DataSource: +def _from_positional( + value: Any, tables: Any, exclude: Any = None, dictionary: Any = None +) -> DataSource: if isinstance(value, sqlalchemy.Engine): - return DataSource.from_engine(value, tables=tables) + return DataSource.from_engine( + value, tables=tables, exclude=exclude, dictionary=dictionary + ) if hasattr(value, "pin_list") and hasattr(value, "pin_read"): - return DataSource.from_board(value, tables) + if exclude is not None: + raise TypeError( + "`exclude` drops relations from a warehouse catalog listing; " + "a pins board has no listing to drop them from." + ) + return DataSource.from_board(value, tables, dictionary=dictionary) raise TypeError( "data_source() takes a SQLAlchemy Engine, a pins board, or named data " f"frames. Got {type(value).__name__}." diff --git a/pkg-py/src/commons/_definitions/_compile.py b/pkg-py/src/commons/_definitions/_compile.py index cb2551c3..655683ce 100644 --- a/pkg-py/src/commons/_definitions/_compile.py +++ b/pkg-py/src/commons/_definitions/_compile.py @@ -95,8 +95,12 @@ def attach_compiled_definitions( getattr(dictionary, "definition_exports", None) or {} ) compiled: dict[str, list[ExportRecord]] = {} - for table_name in dictionary.tables: - definitions = exports.get(table_name) or {} + for table_name, entry in dictionary.tables.items(): + # A catalog import re-keys the dictionary to the warehouse's labels + # and records what the author called each table, which is how its + # exports are still found afterwards. + authored = getattr(entry, "authored_name", None) or table_name + definitions = exports.get(authored) or {} if not definitions: compiled[table_name] = [] continue diff --git a/pkg-py/tests/_warehouse.py b/pkg-py/tests/_warehouse.py new file mode 100644 index 00000000..4a1d7542 --- /dev/null +++ b/pkg-py/tests/_warehouse.py @@ -0,0 +1,80 @@ +"""A warehouse backend that answers from canned rows. + +It stands in for the network, not for a warehouse's semantics: the row shapes +are the ones Snowflake really returns, and refusing a relation is how a real +warehouse refuses one, so what is exercised is commons' handling of both. +""" + +from commons._data_source import TableId + +__all__ = ["FakeWarehouse"] + + +class FakeWarehouse: + """A Snowflake standing in for the network, answering from canned rows.""" + + def __init__( + self, dialect="snowflake", relations=None, role="REPORTER", columns=("ID",) + ): + self._dialect = dialect + self._relations = relations if relations is not None else ["SALES", "ORDERS"] + self._columns = list(columns) + self.role = role + self.queries: list[str] = [] + self.refuse: set[str] = set() + + def query(self, sql: str): + self.queries.append(sql) + if sql.startswith("SELECT CURRENT_USER()"): + return [ + { + "PRINCIPAL": "ANALYST", + "ROLE": self.role, + "SECONDARY_ROLES": "ALL", + "CATALOG": "ANALYTICS", + "SCHEMA": "PUBLIC", + } + ] + if sql.startswith("SELECT CURRENT_DATABASE()"): + return [{"CATALOG": "ANALYTICS", "SCHEMA": "PUBLIC"}] + if sql.startswith("SHOW OBJECTS"): + return [ + { + "name": name, + "database_name": "ANALYTICS", + "schema_name": "PUBLIC", + "kind": "TABLE", + "comment": "", + } + for name in self._relations + if f"'{name}'" in sql or "LIKE" not in sql + ] + if sql.startswith("DESC TABLE"): + return [ + { + "name": name, + "type": "NUMBER(38,0)", + "kind": "COLUMN", + "null?": "N", + "comment": "", + } + for name in self._columns + ] + if sql.startswith("SELECT * FROM"): + for name in self.refuse: + if name in sql: + raise PermissionError(f"Insufficient privileges on {name}") + return [] + raise AssertionError(f"unexpected query: {sql}") + + def list_tables(self) -> list[str]: + return list(self._relations) + + def quote(self, table_id: TableId) -> str: + return ".".join(f'"{part}"' for part in table_id.parts) + + def dialect(self) -> str: + return self._dialect + + def inspector(self): + return None diff --git a/pkg-py/tests/test_catalog_import.py b/pkg-py/tests/test_catalog_import.py new file mode 100644 index 00000000..4b1a5dcb --- /dev/null +++ b/pkg-py/tests/test_catalog_import.py @@ -0,0 +1,119 @@ +"""Importing a warehouse catalog for a data source. + +The reader queries are already covered per backend; what is checked here is +the order the import puts them in, since that order is what makes the source +safe: nothing is described before its access is verified, and the session is +the same one at the end as at the start. +""" + +import pytest + +from commons._catalog import CatalogSessionChangedError, Selector +from commons._catalog._import import import_catalog, is_warehouse +from commons._data_dictionary import DataDictionary +from tests._warehouse import FakeWarehouse + + +def test_only_warehouse_backends_import_a_catalog(): + assert is_warehouse(FakeWarehouse()) + assert is_warehouse(FakeWarehouse(dialect="databricks")) + assert not is_warehouse(FakeWarehouse(dialect="duckdb")) + + +def test_an_unset_selection_takes_the_current_namespace(): + backend = FakeWarehouse() + + imported = import_catalog(backend) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES", "ANALYTICS.PUBLIC.ORDERS"] + assert imported.session is not None + assert imported.session.principal == "ANALYST" + + +def test_a_namespace_is_named_with_a_selector(): + backend = FakeWarehouse() + + imported = import_catalog(backend, Selector(catalog="ANALYTICS", schema="PUBLIC")) + + assert list(imported.relations) == [ + "ANALYTICS.PUBLIC.SALES", + "ANALYTICS.PUBLIC.ORDERS", + ] + assert imported.manifest.access == dict.fromkeys(imported.tables, "unknown") + + +def test_a_dotted_string_names_one_relation(): + backend = FakeWarehouse() + + imported = import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_named_relation_is_access_checked_before_the_source_exists(): + backend = FakeWarehouse() + backend.refuse = {"SALES"} + + with pytest.raises(Exception, match="not authorized"): + import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + +def test_what_the_import_probed_is_remembered_as_readable(): + # A relation checked on the way in should not be probed again the first + # time an agent touches it. + imported = import_catalog(FakeWarehouse(), "ANALYTICS.PUBLIC.SALES") + + assert imported.manifest.access == {"ANALYTICS.PUBLIC.SALES": "queryable"} + + +def test_a_selection_resolving_to_nothing_is_an_error(): + backend = FakeWarehouse(relations=[]) + + with pytest.raises(ValueError, match="no objects"): + import_catalog(backend) + + +def test_an_excluded_relation_is_left_out(): + backend = FakeWarehouse() + + imported = import_catalog(backend, exclude=["ORD*"]) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_session_that_moves_during_discovery_is_refused(): + class Moving(FakeWarehouse): + def query(self, sql): + rows = super().query(sql) + if sql.startswith("SELECT CURRENT_USER()"): + self.role = "ADMIN" + return rows + + with pytest.raises(CatalogSessionChangedError): + import_catalog(Moving()) + + +def test_an_authored_dictionary_is_rekeyed_to_the_warehouse_labels(): + backend = FakeWarehouse() + dictionary = DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Authored prose"}]} + ) + + imported = import_catalog(backend, dictionary=dictionary) + + assert imported.dictionary is not None + assert list(imported.dictionary.tables) == ["ANALYTICS.PUBLIC.SALES"] + table = imported.dictionary.tables["ANALYTICS.PUBLIC.SALES"] + assert table.description == "Authored prose" + assert list(table.columns) == ["ID"] + + +def test_an_unauthorized_table_stops_the_dictionary_merge(): + backend = FakeWarehouse() + backend.refuse = {"SALES"} + dictionary = DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Prose"}]} + ) + + with pytest.raises(Exception, match="not authorized"): + import_catalog(backend, dictionary=dictionary) diff --git a/pkg-py/tests/test_data_source_warehouse.py b/pkg-py/tests/test_data_source_warehouse.py new file mode 100644 index 00000000..c7e1d341 --- /dev/null +++ b/pkg-py/tests/test_data_source_warehouse.py @@ -0,0 +1,142 @@ +"""A data source over a warehouse connection. + +What the catalog import produces is covered against the import itself; this +is about the source it becomes: which tables it exposes, what it remembers +about access, and what it refuses once the connection is no longer the one +the catalog was read for. +""" + +import pytest +import sqlalchemy + +from commons import data_source +from commons._catalog import CatalogSessionChangedError +from commons._data_dictionary import DataDictionary +from commons._data_source import DataSource +from tests._warehouse import FakeWarehouse + + +def warehouse_source(backend=None, **kwargs) -> DataSource: + return DataSource._from_warehouse( + backend or FakeWarehouse(), + kwargs.pop("tables", None), + kwargs.pop("exclude", None), + kwargs.pop("dictionary", None), + ) + + +def test_a_warehouse_source_exposes_its_catalog_labels(): + source = warehouse_source() + + assert source.tables == ["ANALYTICS.PUBLIC.SALES", "ANALYTICS.PUBLIC.ORDERS"] + assert source.table_ids["ANALYTICS.PUBLIC.SALES"].catalog == "ANALYTICS" + assert source.manifest is not None + assert source.relations is not None + + +def test_a_warehouse_source_carries_the_session_it_was_built_for(): + backend = FakeWarehouse() + source = warehouse_source(backend) + + assert source.session is not None + + backend.role = "ADMIN" + with pytest.raises(CatalogSessionChangedError): + source.query("SELECT * FROM sales") + + +def test_a_warehouse_engine_reaches_the_catalog_import(monkeypatch): + # The backend is the seam: everything above it is the real dispatch, from + # data_source() through from_engine()'s warehouse branch. + backend = FakeWarehouse() + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + + source = data_source(sqlalchemy.create_engine("sqlite://"), exclude=["ORD*"]) + + assert source.tables == ["ANALYTICS.PUBLIC.SALES"] + assert source.session is not None + assert source.manifest is not None + + +def test_a_source_over_anything_else_has_no_session_to_check(tmp_path): + engine = sqlalchemy.create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") + with engine.begin() as connection: + connection.execute(sqlalchemy.text("CREATE TABLE sales (id INTEGER)")) + source = DataSource.from_engine(engine) + + assert source.session is None + assert source.query("SELECT * FROM sales") == [] + + +def test_exclude_is_refused_by_a_backend_with_no_catalog_listing(tmp_path): + engine = sqlalchemy.create_engine(f"sqlite:///{tmp_path / 'db.sqlite'}") + + with pytest.raises(ValueError, match="Snowflake and Databricks"): + DataSource.from_engine(engine, exclude=["staging_*"]) + + +def test_definitions_the_warehouse_spells_differently_are_refused(): + # Until the compiler can bind an authored name to the discovered one, a + # definition over a renamed column would lower to SQL naming a column + # that is not there, so construction fails instead. + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": "sales", + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(NotImplementedError, match="does not have column"): + warehouse_source(dictionary=dictionary) + + +def sales_dictionary(**table): + return DataDictionary.model_validate( + { + "tables": [ + { + "name": "sales", + "columns": [{"name": "id", "type": "number(quantity)"}], + **table, + } + ] + } + ) + + +def test_a_definition_survives_the_rekeying_a_merge_does(monkeypatch): + # The merge re-keys the dictionary to the warehouse's label, and the + # definition's export records are still keyed by the authored name. + backend = FakeWarehouse(columns=["id"]) + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + dictionary = sales_dictionary(definitions=[{"name": "total", "expr": "sum(id)"}]) + + source = data_source(sqlalchemy.create_engine("sqlite://"), dictionary=dictionary) + + assert source.dictionary is not None + entry = source.dictionary.tables["ANALYTICS.PUBLIC.SALES"] + assert [record.name for record in entry.compiled_definitions] == ["total"] + + +def test_a_definition_on_a_table_that_matched_nothing_is_refused(monkeypatch): + backend = FakeWarehouse(columns=["id"]) + monkeypatch.setattr("commons._data_source.EngineBackend", lambda engine: backend) + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": "unmatched", + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(ValueError, match="does not match an exposed relation"): + data_source(sqlalchemy.create_engine("sqlite://"), dictionary=dictionary) From 1c55edfe6e5357c6cbba3dfbd37cd37c24f67968 Mon Sep 17 00:00:00 2001 From: Josh Taillon Date: Sun, 6 Sep 2026 14:06:16 -0600 Subject: [PATCH 2/2] fix(py): test the order construction depends on, and compile definitions wherever a source is built Review findings on this branch. The order the import runs in is what makes the source safe, and nothing asserted it. Three separate inversions each passed the whole suite: reading the session identity after the registry rather than before it, probing named relations after the merge rather than before, and moving the closing session re-read ahead of the merge, which is the one thing it exists to cover. Four tests now read the query log as a sequence of steps, and each inversion fails exactly one of them. Only Snowflake was ever exercised. The fake could not answer a Databricks query at all, so wiring `databricks` to the Snowflake reader passed everything. There is now a `FakeDatabricks`, and three tests drive a real Databricks import through it. Both fakes take their row shapes from `catalog-rows.json` rather than restating them, so a change to what a warehouse returns reaches them instead of leaving them answering with a shape the readers no longer expect. `from_engine()` and `from_board()` took a `dictionary` and never compiled its definitions: only `data_source()` did, so a source built through the documented constructors silently carried none, and the grain and unexposed-table checks never ran. Lowering now happens at the end of every constructor. The manifest no longer starts with the relations construction probed marked readable. Carrying that answer forward saved a round trip on first touch at the cost of serving a grant revoked in between, and the sibling implementation re-probes for the same reason. Also: `_selector()` delegates its spelling rules to `_table_entry_id()` rather than restating them; a selection emptied by `exclude`, and a table `exclude` dropped out from under its own definitions, each say so instead of reporting that nothing matched; and the four warehouse-only attributes on `DataSource` carry their real types. Docs: `from_engine()` documents `exclude`; `data_source()` no longer claims DuckDB is the only dialect with an emitter, and names the two refusals this stack added; `query()` says it can now fail because the connection identity moved; and `from_board()` says why it takes a dictionary it does not merge. Tests also cover the four `_selector` spellings and their refusals, a named relation the warehouse lacks, and the two `exclude` refusals that had none. --- pkg-py/src/commons/_catalog/_core.py | 13 +- pkg-py/src/commons/_catalog/_import.py | 76 +++++--- pkg-py/src/commons/_data_source.py | 124 ++++++++----- pkg-py/tests/_warehouse.py | 162 +++++++++++++---- pkg-py/tests/test_catalog_import.py | 196 ++++++++++++++++++++- pkg-py/tests/test_data_source_warehouse.py | 19 ++ 6 files changed, 483 insertions(+), 107 deletions(-) diff --git a/pkg-py/src/commons/_catalog/_core.py b/pkg-py/src/commons/_catalog/_core.py index 0456919e..f0ebe675 100644 --- a/pkg-py/src/commons/_catalog/_core.py +++ b/pkg-py/src/commons/_catalog/_core.py @@ -76,6 +76,11 @@ class TableRegistry: relations: dict[str, Relation] validate: dict[str, TableId] namespace_selected: bool + # The relations the listing reported and exclude then removed, so a + # caller left with nothing, or with a dictionary entry that now matches + # nothing, can say which it was. A name the warehouse never had is not in + # here: exclude is not what made it absent. + dropped: list[Relation] = field(default_factory=list) @dataclass @@ -221,6 +226,9 @@ def table_registry( relations.extend(list_relations(selector)) keep = excluded([item.name for item in relations], exclude) + dropped = [ + item for item, hidden in zip(relations, keep) if hidden and item.discovered + ] relations = [item for item, hidden in zip(relations, keep) if not hidden] validate = [ table_id @@ -252,6 +260,7 @@ def table_registry( relations=labelled, validate={table_id.label: table_id for table_id in validate}, namespace_selected=namespace_selected, + dropped=dropped, ) @@ -475,14 +484,14 @@ def _match_one( relative = [ label for label in labels - if _has_suffix(relations[label], suffix, identifier_case) + if has_suffix(relations[label], suffix, identifier_case) ] if len(relative) > 1: _abort_ambiguous(authored_name, relative) return relative[0] if relative else None -def _has_suffix( +def has_suffix( relation: Relation, suffix: list[str], identifier_case: str | None ) -> bool: path = relation.id.parts diff --git a/pkg-py/src/commons/_catalog/_import.py b/pkg-py/src/commons/_catalog/_import.py index 9c904cb8..696f6413 100644 --- a/pkg-py/src/commons/_catalog/_import.py +++ b/pkg-py/src/commons/_catalog/_import.py @@ -21,6 +21,7 @@ Relation, Selector, check_exclude, + has_suffix, id_type, merge_dictionary, table_registry, @@ -79,6 +80,11 @@ def import_catalog( if registry.validate: require_queryable_relations(backend, registry.validate, registry.relations) if not registry.relations: + if registry.dropped: + raise ValueError( + "The resolved catalog selection contains no objects: exclude " + f"dropped every relation it resolved to. Narrow {exclude!r}." + ) raise ValueError("The resolved catalog selection contains no objects.") # A relation named in the selection has just been probed, so the merge @@ -99,17 +105,15 @@ def access_check(table_id: TableId, label: str) -> None: access_check=access_check, ) check_session(backend, session) - _check_definitions_bound(merged) + _check_definitions_bound(merged, registry.dropped, exclude, identifier_case) + # The manifest starts with every relation unknown, including the ones + # just probed. Carrying the construction-time answer forward would save a + # round trip on first touch at the cost of serving a grant revoked in + # between, and the sibling implementation re-probes for the same reason. manifest = Manifest.build( merged.relations, namespace_selected=registry.namespace_selected ) - # What was probed on the way in is known to be readable, and a first - # touch of it should not pay for the same round trip again. - for label in queryable: - if label in manifest.access: - manifest.access[label] = "queryable" - return ImportedCatalog( tables=list(merged.relations), table_ids={label: item.id for label, item in merged.relations.items()}, @@ -121,7 +125,12 @@ def access_check(table_id: TableId, label: str) -> None: ) -def _check_definitions_bound(merged: MergedDictionary) -> None: +def _check_definitions_bound( + merged: MergedDictionary, + dropped: list[Relation] | None = None, + exclude: list[str] | None = None, + identifier_case: str | None = None, +) -> None: """Refuse definitions the merge renamed out from under. The merge re-keys an authored dictionary to the warehouse's own labels @@ -140,6 +149,12 @@ def _check_definitions_bound(merged: MergedDictionary) -> None: # An authored table that matched nothing is dropped by the merge, so # its definitions would go with it and the agent would never be told. if bindings["tables"].get(authored_table) is None: + if _was_excluded(authored_table, dropped, identifier_case): + raise ValueError( + f"Authored table {authored_table!r} declares definitions, " + f"and exclude dropped it from the catalog listing. Narrow " + f"{exclude!r}, or drop the table from the data dictionary." + ) raise ValueError( f"Authored table {authored_table!r} declares definitions, and " f"does not match an exposed relation. Name it as the data " @@ -161,6 +176,23 @@ def _check_definitions_bound(merged: MergedDictionary) -> None: ) +def _was_excluded( + authored_table: str, dropped: list[Relation] | None, identifier_case: str | None +) -> bool: + """Whether exclude is what removed the relation an authored name meant. + + Compared against the relations exclude actually dropped rather than + against the patterns: a pattern is written in the warehouse's spelling + and an authored name need not be, so only the folded names line up. The + authored name may be qualified, so it is matched as a suffix, by the + same rule the merge uses to find the relation in the first place. + """ + suffix = authored_table.split(".") + return any( + has_suffix(item, suffix, identifier_case) for item in dropped or [] + ) + + def _selectors(backend: Any, reader: Any, tables: Any) -> list[Selector]: """Read a `tables` selection, defaulting to the connection's namespace. @@ -178,25 +210,19 @@ def _selectors(backend: Any, reader: Any, tables: Any) -> list[Selector]: def _selector(entry: Any) -> Selector: """One selection entry as a `Selector`, whatever it was spelled as. - A string is read the way it is everywhere else, as a relation whose dots - qualify it. A namespace has to be a `Selector`, because `ANALYTICS.PUBLIC` - on its own does not say whether PUBLIC is a schema or a table. + A string or a `TableId` is read the way it is everywhere else, as a + relation whose dots qualify it, so the spelling rules and their wording + come from the one place that owns them. A namespace has to be a + `Selector`, because `ANALYTICS.PUBLIC` on its own does not say whether + PUBLIC is a schema or a table. """ if isinstance(entry, Selector): + # Raises unless the selector names a relation or a namespace. id_type(entry) return entry - if isinstance(entry, TableId): - return Selector(catalog=entry.catalog, schema=entry.schema, table=entry.table) - if isinstance(entry, str) and entry: - parts = entry.split(".") - if len(parts) > 3 or any(part == "" for part in parts): - raise ValueError( - "A relation is named catalog.schema.table, with no empty or " - f"skipped components, got {entry!r}." - ) - padded = [None] * (3 - len(parts)) + parts - return Selector(catalog=padded[0], schema=padded[1], table=padded[2]) - raise TypeError( - "Each entry in tables must be a relation name, a TableId, or a " - f"Selector naming a namespace, got {entry!r}." + from .._data_source import _table_entry_id + + table_id = _table_entry_id(entry) + return Selector( + catalog=table_id.catalog, schema=table_id.schema, table=table_id.table ) diff --git a/pkg-py/src/commons/_data_source.py b/pkg-py/src/commons/_data_source.py index ffc8fefd..4fe6cc40 100644 --- a/pkg-py/src/commons/_data_source.py +++ b/pkg-py/src/commons/_data_source.py @@ -18,6 +18,7 @@ from ._sql_guard import check_query if TYPE_CHECKING: + from ._catalog import Manifest, Relation, SessionSnapshot from ._data_dictionary import DataDictionary __all__ = ["DataSource", "TableId", "data_source", "list_tables"] @@ -39,6 +40,25 @@ def _fold(name: str) -> str: return name.translate(_ASCII_FOLD) +def _with_compiled_definitions(source: DataSource) -> DataSource: + """Lower a dictionary's governed definitions, once construction is done. + + The dialect and the final table set are only known here, which is why + lowering waits for them. It runs at the end of every constructor rather + than in `data_source()`, so that a source built through a constructor + directly carries its definitions instead of silently dropping them. A + warehouse source folds its catalog listing into the authored dictionary, + so the one to compile against is the source's, not the one passed in. + """ + if source.dictionary is not None: + from ._definitions import attach_compiled_definitions + + attach_compiled_definitions( + source.dictionary, source.dialect(), set(source.tables) + ) + return source + + @dataclass(frozen=True) class TableId: """A table's identity, qualified as far as the backend has levels. @@ -96,11 +116,12 @@ class DataSource: pending: _PendingPins | None = None dictionary: DataDictionary | None = None # Set only for a warehouse source: what its catalog reported, what is - # known about access to each relation, and the connection identity all - # of that was decided for. - relations: dict[str, Any] | None = None - manifest: Any | None = None - session: Any | None = None + # known about access to each relation, the connection identity all of + # that was decided for, and how the merge re-keyed authored names onto + # the warehouse's own. + relations: dict[str, Relation] | None = None + manifest: Manifest | None = None + session: SessionSnapshot | None = None definition_bindings: dict[str, Any] | None = None @classmethod @@ -146,15 +167,23 @@ def from_engine( given: it reports what exists, so there is nothing to check and no round trip worth paying for. - `dictionary` is taken here because the warehouse listing is folded - into it during construction. Its definitions are lowered by - `data_source()`, once, so attach one through there rather than here. + `exclude` takes unqualified object-name globs to drop from a + warehouse catalog listing, such as `"TMP_*"`. Only a warehouse has a + listing to drop from, so any other engine refuses it. + + On a warehouse `dictionary` is taken here because the catalog listing + is folded into it during construction; on any other engine it is + simply attached. Its governed definitions are lowered once the + dialect and the final table set are known, at the end of + construction. """ from ._catalog._import import is_warehouse backend = EngineBackend(engine) if is_warehouse(backend): - return cls._from_warehouse(backend, tables, exclude, dictionary) + return _with_compiled_definitions( + cls._from_warehouse(backend, tables, exclude, dictionary) + ) if exclude is not None: raise ValueError( "exclude selects out of a warehouse catalog listing, and is " @@ -163,20 +192,24 @@ def from_engine( ) if tables is None: discovered = backend.list_tables() - return cls( - backend=backend, - tables=discovered, - table_ids={name: TableId(table=name) for name in discovered}, - dictionary=dictionary, + return _with_compiled_definitions( + cls( + backend=backend, + tables=discovered, + table_ids={name: TableId(table=name) for name in discovered}, + dictionary=dictionary, + ) ) registry = normalize_table_registry(tables) _check_tables_exist(backend, registry) - return cls( - backend=backend, - tables=list(registry), - table_ids=registry, - dictionary=dictionary, + return _with_compiled_definitions( + cls( + backend=backend, + tables=list(registry), + table_ids=registry, + dictionary=dictionary, + ) ) @classmethod @@ -207,8 +240,10 @@ def from_board( ) -> DataSource: """Expose a pins board's pins as tables, each read on first use. - As with `from_engine()`, a dictionary's definitions are lowered by - `data_source()` rather than here. + `dictionary` is taken here so that the argument survives the + dispatcher; a board has no catalog listing to fold into it. Its + governed definitions are lowered at the end of construction, once + the dialect and the final table set are known. """ if not isinstance(tables, dict): raise TypeError( @@ -236,20 +271,28 @@ def from_board( labels = list(tables) _check_labels_distinct(labels) _check_labels_free(con, labels) - return cls( - backend=DuckDBBackend(con), - tables=labels, - table_ids={label: TableId(table=label) for label in labels}, - pending=_PendingPins(board=board, pins=dict(tables)), - dictionary=dictionary, + return _with_compiled_definitions( + cls( + backend=DuckDBBackend(con), + tables=labels, + table_ids={label: TableId(table=label) for label in labels}, + pending=_PendingPins(board=board, pins=dict(tables)), + dictionary=dictionary, + ) ) def query(self, sql: str) -> list[dict[str, Any]]: - """Run one read-only statement, rejecting anything else first.""" + """Run one read-only statement, rejecting anything else first. + + On a warehouse source the connection identity is checked before the + statement is read: access to these tables was decided for one + principal, role, and namespace, so a query raises rather than runs + once any of those has moved. + """ from ._catalog import check_session - check_query(sql, dialect=self.backend.dialect()) check_session(self.backend, self.session) + check_query(sql, dialect=self.backend.dialect()) if self.pending is None: return self.backend.query(sql) return self._query_loading_pins(sql) @@ -341,9 +384,13 @@ def data_source( `DataSource.from_frames()` directly. A dictionary's governed definitions are compiled for the source's - dialect here, so construction raises if the dialect has no emitter - (only DuckDB does today), if a definition sits on a table the source - does not expose, or if a metric mixes row and aggregate grain. + dialect during construction, so construction raises if the dialect has + no emitter (DuckDB, Snowflake, and Databricks have one), if a definition + sits on a table the source does not expose, or if a metric mixes row and + aggregate grain. On a warehouse it also raises if a table declaring + definitions matched no exposed relation, and, until the compiler can + bind an authored name to the discovered one, if the warehouse spells one + of that table's columns differently. """ from ._data_dictionary import as_data_dictionary @@ -374,16 +421,9 @@ def data_source( source = DataSource.from_frames(**frames) source.dictionary = resolved - # A warehouse source merges the authored dictionary with what its catalog - # reported, so the one to compile against is the source's, not the one - # that was passed in. - if source.dictionary is not None: - # The dialect is only known now, which is why lowering waits for it. - from ._definitions import attach_compiled_definitions - - attach_compiled_definitions( - source.dictionary, source.dialect(), set(source.tables) - ) + # The frames form attaches its dictionary here rather than in the + # constructor, so this is where its definitions can be lowered. + source = _with_compiled_definitions(source) return source diff --git a/pkg-py/tests/_warehouse.py b/pkg-py/tests/_warehouse.py index 4a1d7542..92e689eb 100644 --- a/pkg-py/tests/_warehouse.py +++ b/pkg-py/tests/_warehouse.py @@ -1,27 +1,104 @@ """A warehouse backend that answers from canned rows. It stands in for the network, not for a warehouse's semantics: the row shapes -are the ones Snowflake really returns, and refusing a relation is how a real -warehouse refuses one, so what is exercised is commons' handling of both. +come from `tests/shared/catalog-rows.json`, which is the file that governs how +both packages read a listing, and refusing a relation is how a real warehouse +refuses one, so what is exercised is commons' handling of both. + +`FakeWarehouse` speaks Snowflake and `FakeDatabricks` speaks Databricks. They +answer different queries because the two readers ask different ones, and a +fake that answered both would hide a reader wired to the wrong warehouse. """ from commons._data_source import TableId +from tests._shared import load_shared_fixture + +__all__ = [ + "FakeDatabricks", + "FakeWarehouse", + "column_row", + "databricks_column_row", + "databricks_relation_row", + "relation_row", +] + + +def _shape(backend: str, group: str, accept) -> dict: + """A row of the shared fixture the readers accept, as a template to vary. + + Taking the shape from the fixture rather than restating it means a change + to what a warehouse returns reaches these fakes instead of leaving them + answering with a shape the readers no longer expect. + """ + rows = load_shared_fixture("catalog-rows")[backend][group]["rows"] + for row in rows: + if accept(row): + return dict(row) + raise AssertionError(f"no usable {backend} {group} row") + + +def relation_row(name: str, catalog: str = "ANALYTICS", schema: str = "PUBLIC") -> dict: + row = _shape("snowflake", "relations", lambda row: row["kind"] == "TABLE") + row.update(name=name, database_name=catalog, schema_name=schema, comment="") + return row + -__all__ = ["FakeWarehouse"] +def column_row(name: str) -> dict: + row = _shape("snowflake", "columns", lambda row: row["kind"] == "COLUMN") + row.update(name=name, type="NUMBER(38,0)", comment="") + row["null?"] = "N" + return row -class FakeWarehouse: +def databricks_relation_row(name: str, catalog: str = "main", schema: str = "sales"): + row = _shape("databricks", "relations", lambda row: row["table_type"] != "VIEW") + row.update( + table_catalog=catalog, table_schema=schema, table_name=name, comment="" + ) + return row + + +def databricks_column_row(name: str) -> dict: + # A DESCRIBE reply runs on past the columns, so the template is a row + # from before the metadata marker. + row = _shape( + "databricks", + "columns", + lambda row: row["col_name"] not in ("", "# Partition Information"), + ) + row.update(col_name=name, data_type="bigint", comment="") + return row + + +class _Recorder: + def __init__(self): + self.queries: list[str] = [] + self.refuse: set[str] = set() + + def _probe(self, sql: str): + for name in self.refuse: + if name in sql: + raise PermissionError(f"Insufficient privileges on {name}") + return [] + + def quote(self, table_id: TableId) -> str: + return ".".join(f'"{part}"' for part in table_id.parts) + + def inspector(self): + return None + + +class FakeWarehouse(_Recorder): """A Snowflake standing in for the network, answering from canned rows.""" def __init__( self, dialect="snowflake", relations=None, role="REPORTER", columns=("ID",) ): + super().__init__() self._dialect = dialect self._relations = relations if relations is not None else ["SALES", "ORDERS"] self._columns = list(columns) self.role = role - self.queries: list[str] = [] - self.refuse: set[str] = set() def query(self, sql: str): self.queries.append(sql) @@ -39,42 +116,63 @@ def query(self, sql: str): return [{"CATALOG": "ANALYTICS", "SCHEMA": "PUBLIC"}] if sql.startswith("SHOW OBJECTS"): return [ - { - "name": name, - "database_name": "ANALYTICS", - "schema_name": "PUBLIC", - "kind": "TABLE", - "comment": "", - } + relation_row(name) for name in self._relations if f"'{name}'" in sql or "LIKE" not in sql ] if sql.startswith("DESC TABLE"): - return [ - { - "name": name, - "type": "NUMBER(38,0)", - "kind": "COLUMN", - "null?": "N", - "comment": "", - } - for name in self._columns - ] + return [column_row(name) for name in self._columns] if sql.startswith("SELECT * FROM"): - for name in self.refuse: - if name in sql: - raise PermissionError(f"Insufficient privileges on {name}") - return [] + return self._probe(sql) raise AssertionError(f"unexpected query: {sql}") def list_tables(self) -> list[str]: return list(self._relations) - def quote(self, table_id: TableId) -> str: - return ".".join(f'"{part}"' for part in table_id.parts) - def dialect(self) -> str: return self._dialect - def inspector(self): - return None + +class FakeDatabricks(_Recorder): + """A Databricks standing in for the network. + + Unity Catalog only: `hive_metastore` takes a different listing path and + nothing here selects it. + """ + + def __init__(self, relations=None, columns=("id",)): + super().__init__() + self._relations = relations if relations is not None else ["sales", "orders"] + self._columns = list(columns) + + def query(self, sql: str): + self.queries.append(sql) + if sql.startswith("SELECT CURRENT_USER()"): + return [ + {"principal": "analyst@example.com", "catalog": "main", "schema": "sales"} + ] + if sql.startswith("SELECT CURRENT_CATALOG()"): + return [{"catalog": "main", "schema": "sales"}] + if "system.information_schema.columns" in sql: + # DESCRIBE TABLE does not report nullability, so the reader asks + # the information schema for it separately. + return [ + {"column_name": name, "is_nullable": "YES"} for name in self._columns + ] + if "system.information_schema.tables" in sql: + return [ + databricks_relation_row(name) + for name in self._relations + if f"'{name}'" in sql or "table_name =" not in sql + ] + if sql.startswith("DESCRIBE TABLE"): + return [databricks_column_row(name) for name in self._columns] + if sql.startswith("SELECT * FROM"): + return self._probe(sql) + raise AssertionError(f"unexpected query: {sql}") + + def list_tables(self) -> list[str]: + return list(self._relations) + + def dialect(self) -> str: + return "databricks" diff --git a/pkg-py/tests/test_catalog_import.py b/pkg-py/tests/test_catalog_import.py index 4b1a5dcb..f44c1e95 100644 --- a/pkg-py/tests/test_catalog_import.py +++ b/pkg-py/tests/test_catalog_import.py @@ -11,12 +11,13 @@ from commons._catalog import CatalogSessionChangedError, Selector from commons._catalog._import import import_catalog, is_warehouse from commons._data_dictionary import DataDictionary -from tests._warehouse import FakeWarehouse +from commons._data_source import TableId +from tests._warehouse import FakeDatabricks, FakeWarehouse def test_only_warehouse_backends_import_a_catalog(): assert is_warehouse(FakeWarehouse()) - assert is_warehouse(FakeWarehouse(dialect="databricks")) + assert is_warehouse(FakeDatabricks()) assert not is_warehouse(FakeWarehouse(dialect="duckdb")) @@ -58,12 +59,13 @@ def test_a_named_relation_is_access_checked_before_the_source_exists(): import_catalog(backend, "ANALYTICS.PUBLIC.SALES") -def test_what_the_import_probed_is_remembered_as_readable(): - # A relation checked on the way in should not be probed again the first - # time an agent touches it. +def test_construction_probes_are_not_carried_into_the_manifest(): + # Every relation starts unknown, including the ones construction just + # probed. Carrying that answer forward would serve a grant revoked + # between construction and first touch, so first touch re-probes. imported = import_catalog(FakeWarehouse(), "ANALYTICS.PUBLIC.SALES") - assert imported.manifest.access == {"ANALYTICS.PUBLIC.SALES": "queryable"} + assert imported.manifest.access == {"ANALYTICS.PUBLIC.SALES": "unknown"} def test_a_selection_resolving_to_nothing_is_an_error(): @@ -117,3 +119,185 @@ def test_an_unauthorized_table_stops_the_dictionary_merge(): with pytest.raises(Exception, match="not authorized"): import_catalog(backend, dictionary=dictionary) + + +def _prose() -> DataDictionary: + """A dictionary that matches SALES, so the merge describes it.""" + return DataDictionary.model_validate( + {"tables": [{"name": "sales", "description": "Authored prose"}]} + ) + + +def _kinds(queries: list[str]) -> list[str]: + """Each query as the step it belongs to, so order can be asserted.""" + steps = [] + for sql in queries: + if sql.startswith(("SELECT CURRENT_USER()",)): + steps.append("session") + elif sql.startswith(("SELECT CURRENT_DATABASE()", "SELECT CURRENT_CATALOG()")): + steps.append("namespace") + elif sql.startswith("SHOW OBJECTS") or "information_schema.tables" in sql: + steps.append("list") + elif ( + sql.startswith(("DESC TABLE", "DESCRIBE TABLE")) + or "information_schema.columns" in sql + ): + steps.append("describe") + elif sql.startswith("SELECT * FROM"): + steps.append("probe") + else: + raise AssertionError(f"unclassified query: {sql}") + return steps + + +def test_the_session_is_read_before_anything_it_decides(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES") + + # Every access answer that follows was decided for this identity, so + # reading it second would be reading it about a different connection. + assert _kinds(backend.queries)[0] == "session" + + +def test_a_named_relation_is_probed_before_it_is_described(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES", dictionary=_prose()) + + steps = _kinds(backend.queries) + # A name the caller got wrong should fail at construction, which only + # holds if the probe precedes the describe that would otherwise reveal it. + assert steps.index("probe") < steps.index("describe") + + +def test_the_session_is_read_again_after_discovery(): + backend = FakeWarehouse() + + import_catalog(backend, "ANALYTICS.PUBLIC.SALES", dictionary=_prose()) + + steps = _kinds(backend.queries) + # A role that moved during discovery invalidates what was just learned, + # so the second read has to come after the last thing it invalidates. + assert steps[-1] == "session" + assert steps.count("session") == 2 + assert steps.index("describe") < len(steps) - 1 + + +def test_a_databricks_selection_is_imported_the_same_way(): + backend = FakeDatabricks() + + imported = import_catalog(backend, dictionary=_databricks_prose()) + + assert imported.tables == ["main.sales.sales", "main.sales.orders"] + assert imported.session is not None + assert imported.session.principal == "analyst@example.com" + # Databricks reports no role, so there is none to snapshot. + assert imported.session.role is None + assert _kinds(backend.queries)[0] == "session" + assert _kinds(backend.queries)[-1] == "session" + + +def test_a_databricks_authored_name_matches_whatever_its_case(): + # Databricks reports lower-cased identifiers, so an authored name in any + # other case still has to find the relation it describes. + backend = FakeDatabricks() + + imported = import_catalog(backend, dictionary=_databricks_prose("SALES")) + + assert imported.dictionary is not None + table = imported.dictionary.tables["main.sales.sales"] + assert table.description == "Authored prose" + + +def test_a_databricks_relation_is_probed_before_it_is_described(): + backend = FakeDatabricks() + + import_catalog(backend, "main.sales.orders", dictionary=_databricks_prose("orders")) + + steps = _kinds(backend.queries) + assert steps.index("probe") < steps.index("describe") + + +def _databricks_prose(name: str = "sales") -> DataDictionary: + return DataDictionary.model_validate( + {"tables": [{"name": name, "description": "Authored prose"}]} + ) + + +def test_an_empty_selection_is_refused(): + with pytest.raises(ValueError, match="at least one relation or namespace"): + import_catalog(FakeWarehouse(), []) + + +def test_a_selection_entry_with_an_empty_component_is_refused(): + with pytest.raises(ValueError, match="empty name components"): + import_catalog(FakeWarehouse(), "ANALYTICS..SALES") + + +def test_a_selection_entry_of_the_wrong_type_is_refused(): + with pytest.raises(TypeError, match="table name or a TableId"): + import_catalog(FakeWarehouse(), 5) + + +def test_a_table_id_names_one_relation(): + imported = import_catalog( + FakeWarehouse(), TableId(catalog="ANALYTICS", schema="PUBLIC", table="SALES") + ) + + assert imported.tables == ["ANALYTICS.PUBLIC.SALES"] + + +def test_a_named_relation_the_warehouse_lacks_fails_construction(): + with pytest.raises(ValueError, match="does not have"): + import_catalog(FakeWarehouse(), "ANALYTICS.PUBLIC.MISSING") + + +def test_exclude_that_empties_the_selection_says_so(): + with pytest.raises(ValueError, match="dropped every relation"): + import_catalog(FakeWarehouse(), exclude=["*"]) + + +def test_an_exclude_that_matched_nothing_does_not_blame_exclude(): + # The namespace was already empty, so exclude is not what emptied it. + backend = FakeWarehouse(relations=[]) + + with pytest.raises(ValueError, match="contains no objects") as refusal: + import_catalog(backend, exclude=["NOTHING_*"]) + + assert "exclude dropped" not in str(refusal.value) + + +@pytest.mark.parametrize( + "authored", ["sales", "PUBLIC.sales", "ANALYTICS.PUBLIC.sales"] +) +def test_definitions_on_an_excluded_table_name_exclude_however_qualified(authored): + # An authored name may be qualified, and a glob may not be, so the two + # are matched by the same suffix rule the merge uses. + dictionary = DataDictionary.model_validate( + { + "tables": [ + { + "name": authored, + "columns": [{"name": "id", "type": "number(quantity)"}], + "definitions": [{"name": "total", "expr": "sum(id)"}], + } + ] + } + ) + + with pytest.raises(ValueError, match="exclude dropped it"): + import_catalog(FakeWarehouse(), exclude=["SALES"], dictionary=dictionary) + + + + +def test_excluding_a_name_the_warehouse_never_had_does_not_blame_exclude(): + # The relation was absent, not dropped, so exclude is not the reason + # there is nothing left to expose. + with pytest.raises(ValueError, match="contains no objects") as refusal: + import_catalog( + FakeWarehouse(), "ANALYTICS.PUBLIC.MISSING", exclude=["MISSING"] + ) + + assert "exclude dropped" not in str(refusal.value) diff --git a/pkg-py/tests/test_data_source_warehouse.py b/pkg-py/tests/test_data_source_warehouse.py index c7e1d341..708bbf9e 100644 --- a/pkg-py/tests/test_data_source_warehouse.py +++ b/pkg-py/tests/test_data_source_warehouse.py @@ -75,6 +75,25 @@ def test_exclude_is_refused_by_a_backend_with_no_catalog_listing(tmp_path): DataSource.from_engine(engine, exclude=["staging_*"]) +def test_exclude_is_refused_for_a_pins_board(): + class Board: + def pin_list(self): + return ["sales"] + + def pin_read(self, name): + raise AssertionError("not reached") + + with pytest.raises(TypeError, match="warehouse catalog listing"): + data_source(Board(), tables={"sales": "sales"}, exclude=["tmp_*"]) + + +def test_exclude_is_refused_for_named_frames(): + import pandas as pd + + with pytest.raises(TypeError, match="no listing to drop them from"): + data_source(sales=pd.DataFrame({"id": [1]}), exclude=["tmp_*"]) + + def test_definitions_the_warehouse_spells_differently_are_refused(): # Until the compiler can bind an authored name to the discovered one, a # definition over a renamed column would lower to SQL naming a column